Test cases for Operators

src/main/java/com/example/reactor/ProjectReactor/entity/Department.java

+package com.example.reactor.ProjectReactor.entity;
+
+import java.util.List;
+
+public class Department {
+
+    private int id;
+
+    private String name;
+
+    private int organizationId;
+
+    private List<Employee> employees;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public int getOrganizationId() {
+        return organizationId;
+    }
+
+    public void setOrganizationId(int organizationId) {
+        this.organizationId = organizationId;
+    }
+
+    public List<Employee> getEmployees() {
+        return employees;
+    }
+
+    public void setEmployees(List<Employee> employees) {
+        this.employees = employees;
+    }
+
+    public Department(int id, String name, int organizationId, List<Employee> employees) {
+        this.id = id;
+        this.name = name;
+        this.organizationId = organizationId;
+        this.employees = employees;
+    }
+
+    public Department(int id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public Department(int id, String name, int organizationId) {
+        this.id = id;
+        this.name = name;
+        this.organizationId = organizationId;
+    }
+
+    public Department() {
+    }
+
+    @Override
+    public String toString() {
+        return "Department{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", organizationId=" + organizationId +
+                ", employees=" + employees +
+                '}';
+    }
+}

src/main/java/com/example/reactor/ProjectReactor/entity/Employee.java

+package com.example.reactor.ProjectReactor.entity;
+
+public class Employee {
+
+    private int id;
+    private String name;
+
+    private int salary;
+
+    private int organizationId;
+
+    private int departmentId;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public int getSalary() {
+        return salary;
+    }
+
+    public void setSalary(int salary) {
+        this.salary = salary;
+    }
+
+    public int getOrganizationId() {
+        return organizationId;
+    }
+
+    public void setOrganizationId(int organizationId) {
+        this.organizationId = organizationId;
+    }
+
+    public int getDepartmentId() {
+        return departmentId;
+    }
+
+    public void setDepartmentId(int departmentId) {
+        this.departmentId = departmentId;
+    }
+
+    public Employee(int id, String name, int salary, int organizationId, int departmentId) {
+        this.id = id;
+        this.name = name;
+        this.salary = salary;
+        this.organizationId = organizationId;
+        this.departmentId = departmentId;
+    }
+
+    public Employee(int id, String name, int salary, int organizationId) {
+        this.id = id;
+        this.name = name;
+        this.salary = salary;
+        this.organizationId = organizationId;
+    }
+
+    public Employee(int id, String name, int salary) {
+        this.id = id;
+        this.name = name;
+        this.salary = salary;
+    }
+
+    public Employee() {
+    }
+
+    @Override
+    public String toString() {
+        return "Employee{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", salary=" + salary +
+                ", organizationId=" + organizationId +
+                ", departmentId=" + departmentId +
+                '}';
+    }
+}

src/main/java/com/example/reactor/ProjectReactor/entity/Organization.java

+package com.example.reactor.ProjectReactor.entity;
+
+import java.util.List;
+
+public class Organization {
+
+    private int id;
+    private String name;
+
+    private List<Employee> employees;
+
+    private List<Department> departments;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public List<Employee> getEmployees() {
+        return employees;
+    }
+
+    public void setEmployees(List<Employee> employees) {
+        this.employees = employees;
+    }
+
+    public List<Department> getDepartments() {
+        return departments;
+    }
+
+    public void setDepartments(List<Department> departments) {
+        this.departments = departments;
+    }
+
+    public Organization(int id, String name, List<Employee> employees, List<Department> departments) {
+        this.id = id;
+        this.name = name;
+        this.employees = employees;
+        this.departments = departments;
+    }
+
+    public Organization(int id, String name, List<Employee> employees) {
+        this.id = id;
+        this.name = name;
+        this.employees = employees;
+    }
+
+    public Organization(int id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public Organization() {
+    }
+
+    @Override
+    public String toString() {
+        return "Organization{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", employees=" + employees +
+                ", departments=" + departments +
+                '}';
+    }
+}

src/test/java/com/example/reactor/ProjectReactor/ContactControllerTest.java

@@ -63,7 +63,7 @@ public class ContactControllerTest {
                 .exchange()
                 .returnResult(Contact.class).getResponseBody().log();
 
-        StepVerifier.create(updateFlux).expectSubscription().expectNextMatches(c -> c.getEmail().equals("wtc@email.com")).verifyComplete();
+        StepVerifierEx.create(updateFlux).expectSubscription().expectNextMatches(c -> c.getEmail().equals("wtc@email.com")).verifyComplete();
     }*/
 
 

src/test/java/com/example/reactor/ProjectReactor/FluxSample.java

@@ -59,6 +59,7 @@ public class FluxSample {
         Flux<Integer> range = Flux.range(1, 5);
         range.subscribe(r -> System.out.println(r));
 
+
         StepVerifier.create(just.log()).expectNext("Ravi").expectNext("Ramu").expectComplete();
 
         Flux.empty().log();

src/test/java/com/example/reactor/ProjectReactor/FluxTest.java

 package com.example.reactor.ProjectReactor;
 
+import com.example.reactor.ProjectReactor.entity.Contact;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.context.SpringBootTest;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.GroupedFlux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
+import java.util.*;
 
 @SpringBootTest
 public class FluxTest {
@@ -143,4 +144,36 @@ public class FluxTest {
                 item -> item.split(":")[1]);
     }
 
+    @Test
+    public void startWith(){
+        Flux<String> just = Flux.just("Ravi", "Reddy");
+        Flux<String> pannala = just.startWith("Pannala");
+        Flux<String> ayaan = pannala.startWith(Arrays.asList("Ayaan"));
+        Flux<String> hello = ayaan.startWith(Mono.just("Hello"));
+        hello.subscribe(s->System.out.println(s));
+
+    }
+
+    @Test
+    public void concatenateWithValues(){
+        Flux<String> stringFlux = Flux.just("avi", "annala").concatWithValues("Reddy", "Pannala");
+        stringFlux.subscribe(s->System.out.println("ConcatenateWith Values-->"+s));
+    }
+
+    @Test
+    public void groupBy(){
+        Flux<Contact> contactFlux = Flux.fromIterable(fetchContact());
+        Flux<GroupedFlux<String, Contact>> groupedFluxFlux = contactFlux.groupBy(Contact::getName);
+        Mono<Map<String, Collection<Contact>>> mapMono = contactFlux.collectMultimap(Contact::getEmail, item -> item);
+        Mono<Map<String, String>> mapMono1 = contactFlux.collectMap(Contact::getName, Contact::getEmail);
+        Mono<List<Contact>> listMono = contactFlux.collectSortedList(Comparator.comparing(Contact::getName));
+        mapMono.subscribe(System.out::println);
+    }
+    public List<Contact> fetchContact(){
+        List<Contact> contacts = new ArrayList<>();
+        contacts.add(new Contact("1","Ravi","ravi@gmail.com","344343"));
+        contacts.add(new Contact("2","Raju","ravi@gmail.com","332"));
+        return contacts;
+    }
+
 }

src/test/java/com/example/reactor/ProjectReactor/Just.java

+package com.example.reactor.ProjectReactor;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Optional;
+
+@SpringBootTest
+public class Just {
+
+    @Test
+    public void just() {
+        Mono<String> ravi = Mono.just("Ravi");
+        Flux<String> ravi1 = Flux.just("Ravi");
+        Flux<String> just = Flux.just("Ravi", "Ravi");
+
+        Mono.just(asyncMethod())
+                .doOnNext(System.out::println)
+                .log();
+    }
+
+    /*
+    An Optional is a container object which may or may not contain a non-null value.
+    The Mono#justOrEmpty flattens the Optional and creates a new Mono that emits the value of Optional
+    if non-null and invokes onNext method.
+    If Optional value is null, then skips the onNext and calls the onComplete directly.
+     */
+    @Test
+    public void justOrEmpty() {
+
+        Mono<Integer> integerMono = Mono.justOrEmpty(asyncMethod())
+                .doOnNext(System.out::println)
+                .log();
+        integerMono.subscribe(s -> System.out.println(s));
+    }
+
+    /*
+    A value can be a null or non-null. The Mono#justOrEmpty(T data) creates a new Mono that emits the value if non-null
+    and invokes onNext method. If the value is null,
+    then skips the onNext and calls onComplete directly.
+     */
+    @Test
+    public void justOrEmptyOptional() {
+
+        Mono<Integer> integerMono = Mono.justOrEmpty(asyncOptional())
+                .doOnNext(System.out::println)
+                .log();
+        integerMono.subscribe(s -> System.out.println(s));
+    }
+
+    public Integer asyncMethod() {
+        return null;
+    }
+
+    public Optional<Integer> asyncOptional() {
+        return Optional.of(10);
+    }
+}

src/test/java/com/example/reactor/ProjectReactor/MonoSample.java

@@ -33,19 +33,33 @@ public class MonoSample {
 
     @Test
     public void createMono() {
+        //Create mono with Just
         Mono<String> mono = Mono.just("Ravi");
         mono.subscribe(a -> System.out.println("Mono with Just--->" + a));
+
+        //Create Mono With Error
         Mono<Object> error = Mono.error(new Throwable());
         error.subscribe(a -> System.out.println("Mono with Error--->" + a));
+
+        //Create mono with empty
         Mono<Object> empty = Mono.empty();
         empty.subscribe(a -> System.out.println("Mono with empty--->" + a));
+
+        //Create Mono from other publisher
         Mono<Integer> from = Mono.from(Flux.range(1, 10));
         from.subscribe(s -> System.out.println("Mono with from --->" + s));
-        Random rd = new Random();
-        Mono<Double> doubleMono = Mono.fromSupplier(rd::nextDouble);
+
+        //Create mono from Supplier
+        Mono<Double> doubleMono = Mono.fromSupplier(Math::random);
         doubleMono.subscribe(s -> System.out.println("From Supplier-->" + s));
+
+        //Create Mono from callable
         Mono<String> hello = Mono.fromCallable(() -> "Hello");
         hello.subscribe(a -> System.out.println("From Callable --->" + a));
+
+        //Create Mono from Runnable
+        Mono<Object> objectMono = Mono.fromRunnable(() -> {});
+        objectMono.subscribe(a->System.out.println(a));
     }
 
     @Test

src/test/java/com/example/reactor/ProjectReactor/StepVerifierEx.java

+package com.example.reactor.ProjectReactor;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Iterator;
+
+@SpringBootTest
+public class StepVerifierEx {
+
+    @Test
+    public void create() {
+
+        StepVerifier.create(Flux.just(1, 2, 3))
+                .expectNext(1).expectNext(2).expectNext(3).expectComplete().verify();
+    }
+
+    @Test
+    public void createWithDuration() {
+
+        //  StepVerifier.create(Flux.just(1,2,3)).expectNextCount(3).expectComplete().verify();
+        StepVerifier.create(Flux.just(1)).expectNextSequence(Arrays.asList(1)).expectComplete().verify();
+    }
+
+    @Test
+    public void expectNextMatches() {
+        Flux<String> map = Flux.just("ravi", "ramu", "Ra", "Zoo").filter(s->s.length()>2).map(String::toUpperCase);
+        StepVerifier.create(map).
+                expectNext("RAVI").
+                expectNextMatches(s -> s.startsWith("RA")).
+                expectComplete().
+                verify();
+    }
+
+    @Test
+    public void expectError(){
+        Flux<Integer> integerFlux = Flux.just(1, 2, 3).concatWith(Flux.error(new ArithmeticException("Error Occured")));
+        //StepVerifier.create(integerFlux).expectNext(1).expectNext(2).expectNext(3).expectError().verify();
+        StepVerifier.create(integerFlux).expectNextCount(3).expectError(ArithmeticException.class).verify();
+        StepVerifier
+                .create(integerFlux).
+                expectNext(1).
+                expectNext(2).
+                expectNext(3).
+                expectErrorMatches(error->error.getMessage().matches("Error Occured")).
+                verify();
+    }
+}

src/test/java/com/example/reactor/ProjectReactor/Transform.java

+package com.example.reactor.ProjectReactor;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.publisher.Flux;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+@SpringBootTest
+public class Transform {
+
+    AtomicInteger ai = new AtomicInteger();
+    @Test
+    public void trnasform(){
+
+        Flux.fromIterable(Arrays.asList("blue","green","orange","purple"))
+                .doOnNext(System.out::println)
+                .transform(function)
+                .subscribe(s->System.out.println("Values are transformed to--->"+s));
+
+    }
+
+    @Test
+    public void trnasformDefered(){
+
+        Flux<String> stringFlux = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
+                .doOnNext(System.out::println)
+                .transformDeferred(filterAndMap);
+        stringFlux.subscribe(s->System.out.println("Values are transformed to--->"+s));
+        stringFlux.subscribe(s->System.out.println("Values are transformed to--->"+s));
+    }
+
+    public Function<Flux<String>, Flux<String>> function = f -> f.filter(color -> !color.equals("orange"))
+            .map(s -> s.toUpperCase());
+
+
+    Function<Flux<String>, Flux<String>> filterAndMap = f -> {
+        if (ai.incrementAndGet() == 1) {
+            return f.filter(color -> !color.equals("orange"))
+                    .map(String::toUpperCase);
+        }
+        return f.filter(color -> !color.equals("purple"))
+                .map(String::toUpperCase);
+    };
+}

src/test/java/com/example/reactor/ProjectReactor/ZipSample.java

+package com.example.reactor.ProjectReactor;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+import reactor.util.function.Tuple3;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Stream;
+
+@SpringBootTest
+public class ZipSample {
+
+    @Test
+    public void zipSingleElementinSource() {
+        Flux<String> f1 = Flux.just("F1");
+        Flux<String> f2 = Flux.just("F2");
+        Flux<String> f3 = Flux.just("F3");
+        Flux<Tuple3<String, String, String>> zip = Flux.zip(f1, f2, f3);
+
+        StepVerifier.create(zip).expectSubscription().expectNextCount(1).verifyComplete();
+    }
+
+    @Test
+    public void zipWithMultipleValueSource() {
+        Flux<String> f1 = Flux.just("F11", "F12");
+        Flux<String> f2 = Flux.fromArray(new String[]{"F21", "F22"});
+        Flux<String> f3 = Flux.fromStream(Stream.of("F31", "F32"));
+        Flux<String> map = Flux.zip(f1, f2, f3).map(s -> {
+            String s1 = s.getT1() + " " + s.getT2() + " " + s.getT3();
+            return s1;
+        });
+        StepVerifier.create(map).expectSubscription().expectNext("F11 F21 F31").expectNext("F12 F22 F32").verifyComplete();
+    }
+
+    @Test
+    public void zipWithEmptySource() {
+        Flux<Object> empty = Flux.empty();
+        Flux<String> fluxString = Flux.just("F11", "F21");
+        Flux<String> from = Flux.from(Flux.just("F22", "F23"));
+        Mono<Void> then = Flux.zip(empty, fluxString, from).map(s -> {
+            return Mono.empty();
+        }).then();
+
+        StepVerifier.create(then).expectSubscription().expectNextCount(0).verifyComplete();
+    }
+
+    @Test
+    public void prematureCompleteEmptySource() {
+        StepVerifier.create(Flux.zip(obj -> 0, Flux.just(1), Mono.empty())).expectSubscription().expectNextCount(0)
+                .verifyComplete();
+    }
+
+    @Test
+    public void createZipWithPrefetchIterable() {
+        List<Flux<Integer>> list = Arrays.asList(Flux.just(1), Flux.just(2));
+        Flux<Integer> f = Flux.zip(list, 123, obj -> 0);
+         f.subscribe(s->System.out.println(s));
+        //StepVerifierEx.create(f).expectSubscription().expectNext()
+    }
+
+    @Test
+    public void failNull() {
+        StepVerifier.create(Flux.zip(obj -> 0, Flux.just(1), null))
+                .verifyError(NullPointerException.class);
+    }
+}

src/test/java/com/example/reactor/ProjectReactor/doExamples.java

+package com.example.reactor.ProjectReactor;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+@SpringBootTest
+public class doExamples {
+
+    @Test
+    public void doNext(){
+        Flux<String> stringFlux = Flux.just("Hello", "Ravinder")
+                .map(String::toUpperCase).doOnNext(s -> System.out.println("Do nExt Elemetns--->" + s))
+                .doOnEach(s -> System.out.println(s));
+
+        StepVerifier.create(stringFlux)
+                .expectNextCount(2)
+                .verifyComplete();
+    }
+
+    @Test
+   public void doOnSubscribe(){
+        Flux.just(1,2,3,4)
+                .doOnSubscribe(subscription -> System.out.println(subscription.toString()))
+                .subscribe(s->System.out.println("Subscribe happens"));
+   }
+
+   @Test
+    public void donOnNext(){
+        Flux.just(1,2,3,4)
+                .doOnNext(s->System.out.println("Do On next element--->"+s))
+                .subscribe(s->System.out.println("Subscription happen"));
+   }
+
+   @Test
+   public void doOnComplete(){
+      Flux.just(1,2,3,4)
+              .doOnNext(s->System.out.println("Do on Next element--->"+s))
+              .doOnComplete(()->System.out.println("Process Complete-->"))
+              .subscribe();
+   }
+
+   @Test
+    public void doOnError(){
+        Flux.error(new RuntimeException("Exception occurred")).doOnError(s->System.out.println("Do On Error"))
+                .subscribe();
+   }
+
+   @Test
+    public void doOnFinnaly(){
+        Flux.just(1)
+                .doFinally(s->System.out.println("Do FInnally"+s))
+                .subscribe();
+
+        Flux.error(new RuntimeException("Exception"))
+                .doFinally(s->System.out.println("Error Result"))
+                .subscribe();
+   }
+
+
+}