sec 08 combining publishers

Author: syntax-tee

src/main/java/chap6/Lec04PublishOn.java

@@ -19,6 +19,7 @@ public static void main(String[] args) {
 
         flux.publishOn(Schedulers.boundedElastic())
                 .doOnNext(i -> printThreadName("next " + i))
+                .publishOn(Schedulers.parallel())
                 .subscribe(v -> printThreadName("sub " + v));
 
         Util.sleepSeconds(5);

src/main/java/chap6/Lec05PubSubOn.java

@@ -0,0 +1,34 @@
+package chap6;
+
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+import utility.Util;
+
+public class Lec05PubSubOn {
+    public static void main(String[] args) {
+
+        Flux<Object> flux = Flux.create(fluxSink -> {
+                    printThreadName("create");
+                    for (int i = 0; i < 4; i++) {
+                        fluxSink.next(i);
+                    }
+                    fluxSink.complete();
+                })
+                .doOnNext(i -> printThreadName("next " + i));
+
+
+        flux
+                .publishOn(Schedulers.parallel())
+                .doOnNext(i -> printThreadName("next " + i))
+                .subscribeOn(Schedulers.boundedElastic())
+                .subscribe(v -> printThreadName("sub " + v));
+
+
+        Util.sleepSeconds(5);
+
+    }
+
+    private static void printThreadName(String msg){
+        System.out.println(msg + "\t\t: Thread : " + Thread.currentThread().getName());
+    }
+}

src/main/java/chap6/Lec06Parallel.java

@@ -0,0 +1,19 @@
+package chap6;
+
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+public class Lec06Parallel {
+
+    public static void main(String[] args) {
+        Flux.range(1, 10)
+                .parallel()
+                .runOn(Schedulers.parallel())
+                .doOnNext(i -> printThreadName("next " + i))
+                .subscribe(v -> printThreadName("next " + v));
+    }
+
+    private static void printThreadName(String msg) {
+        System.out.println(msg + "\t\t: Thread : " + Thread.currentThread().getName());
+    }
+}

src/main/java/chap6/Lec07FluxInterval.java

@@ -0,0 +1,14 @@
+package chap6;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.time.Duration;
+
+public class Lec07FluxInterval {
+
+    public static void main(String[] args) {
+        Flux.interval(Duration.ofSeconds(1)).subscribe(Util.subscriber());
+        Util.sleepSeconds(1);
+    }
+}

src/main/java/chap7/Lec01Demo.java

@@ -0,0 +1,21 @@
+package chap7;
+
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+import utility.Util;
+
+public class Lec01Demo {
+    public static void main(String[] args) {
+
+        Flux.create(fluxSink -> {
+                    for (int i = 0; i < 500; i++) {
+                        fluxSink.next(i);
+                        System.out.println("Pushed : " + i);
+                    }
+                }).publishOn(Schedulers.boundedElastic())
+                .doOnNext(i -> {
+                    Util.sleepMillis(10);
+                }).subscribe(Util.subscriber());
+        Util.sleepSeconds(60);
+    }
+}

src/main/java/chap7/Lec02Drop.java

@@ -0,0 +1,22 @@
+package chap7;
+
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+import utility.Util;
+
+public class Lec02Drop {
+    public static void main(String[] args) {
+
+        Flux.create(fluxSink -> {
+                    for (int i = 0; i < 500; i++) {
+                        fluxSink.next(i);
+                        System.out.println("Pushed : " + i);
+                    }
+                }).onBackpressureBuffer()
+                 .publishOn(Schedulers.boundedElastic())
+                .doOnNext(i -> {
+                    Util.sleepMillis(10);
+                }).subscribe(Util.subscriber());
+        Util.sleepSeconds(60);
+    }
+}

src/main/java/chap7/Lec03latest.java

@@ -0,0 +1,22 @@
+package chap7;
+
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+import utility.Util;
+
+public class Lec03latest {
+    public static void main(String[] args) {
+
+        Flux.create(fluxSink -> {
+                    for (int i = 0; i < 500; i++) {
+                        fluxSink.next(i);
+                        System.out.println("Pushed : " + i);
+                    }
+                }).onBackpressureLatest()
+                 .publishOn(Schedulers.boundedElastic())
+                .doOnNext(i -> {
+                    Util.sleepMillis(10);
+                }).subscribe(Util.subscriber());
+        Util.sleepSeconds(60);
+    }
+}

src/main/java/chap8/Lec01StartWith.java

@@ -0,0 +1,19 @@
+package chap8;
+
+import chap8.helper.NameGenerator;
+import utility.Util;
+
+public class Lec01StartWith {
+
+    public static void main(String[] args) {
+        NameGenerator generator = new NameGenerator();
+        generator.generateNames().take(2).subscribe(Util.subscriber("sam"));
+
+
+        generator.generateNames().take(2).subscribe(Util.subscriber("mike"));
+
+
+        generator.generateNames().filter(n -> n.startsWith("A")).take(1).subscribe(Util.subscriber("marshal"));
+
+    }
+}

src/main/java/chap8/Lec02Concat.java

@@ -0,0 +1,25 @@
+package chap8;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+public class Lec02Concat {
+
+    public static void main(String[] args) {
+        Flux<String> flux1 = Flux.just("a","b","c");
+        Flux<String> flux2 = Flux.just("d","e","f");
+
+        Flux<String> concatFlux = flux1.concatWith(flux2);
+        Flux<String> concatFlux2 = Flux.concat(flux1,flux2);
+        concatFlux.subscribe(System.out::println);
+
+        System.out.println("FLUXXXXXXX2");
+        concatFlux2.subscribe(System.out::println);
+        Flux<String> fluxa = Flux.just("1","2","3");
+        Flux<String> fluxc = Flux.error(new RuntimeException("error"));
+        Flux<String> fluxb = Flux.just("4","5","6");
+
+        Flux<String> fluxalphabet = Flux.concatDelayError(fluxa,fluxb,fluxc);
+        fluxalphabet.subscribe(Util.subscriber());
+    }
+}

src/main/java/chap8/Lec03Merge.java

@@ -0,0 +1,18 @@
+package chap8;
+
+import chap8.helper.American;
+import chap8.helper.Emirate;
+import chap8.helper.Qatar;
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+public class Lec03Merge {
+
+    public static void main(String[] args) {
+
+        Flux<String> merge = Flux.merge(Qatar.getFlights(), Emirate.getFlights(), American.getFlights());
+        merge.subscribe(Util.subscriber());
+        Util.sleepSeconds(10);
+
+    }
+}

src/main/java/chap8/Lec04Zip.java

@@ -0,0 +1,27 @@
+package chap8;
+
+import reactor.core.publisher.Flux;
+import reactor.util.function.Tuple3;
+import utility.Util;
+
+public class Lec04Zip {
+
+    public static void main(String[] args) {
+     Flux<Tuple3<String, String, String>> car = Flux.zip(getBody(),getEngine(),getTires());
+     car.subscribe(Util.subscriber());
+    }
+
+    public static Flux<String> getBody(){
+        return Flux.range(1,2).map(integer -> "Body "+integer);
+    }
+
+    public static Flux<String> getEngine(){
+        return Flux.range(1,5).map(integer -> "Engine "+integer);
+
+    }
+
+    public static Flux<String> getTires(){
+        return Flux.range(1,5).map(integer -> "Tire "+integer);
+
+    }
+}

src/main/java/chap8/Lec05CombineLatest.java

@@ -0,0 +1,26 @@
+package chap8;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.time.Duration;
+
+public class Lec05CombineLatest {
+
+    public static void main(String[] args) {
+      Flux.combineLatest(getNumber(),getString(),(i,s)-> s + i).subscribe(Util.subscriber());
+
+      Util.sleepSeconds(10);
+
+    }
+
+    private static  Flux<String> getString(){
+        return Flux.just("A","B","C").delayElements(Duration.ofSeconds(1));
+    }
+
+
+    private static Flux<Integer> getNumber(){
+      return Flux.just(1,2,3).delayElements(Duration.ofSeconds(3));
+    }
+
+}

src/main/java/chap8/helper/American.java

@@ -0,0 +1,16 @@
+package chap8.helper;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.time.Duration;
+
+public class American {
+
+    public static Flux<String> getFlights() {
+        return Flux.range(1, Util.faker().random().nextInt(10, 20))
+                .delayElements(Duration.ofSeconds(1))
+                .map(i -> "AA " + Util.faker().random().nextInt(100, 999))
+                .filter(i -> Util.faker().random().nextBoolean());
+    }
+}

src/main/java/chap8/helper/Emirate.java

@@ -0,0 +1,16 @@
+package chap8.helper;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.time.Duration;
+
+public class Emirate {
+
+    public static Flux<String> getFlights() {
+        return Flux.range(1, Util.faker().random().nextInt(1, 10))
+                .delayElements(Duration.ofSeconds(1))
+                .map(i -> "Emirate " + Util.faker().random().nextInt(100, 999))
+                .filter(i -> Util.faker().random().nextBoolean());
+    }
+}

src/main/java/chap8/helper/NameGenerator.java

@@ -0,0 +1,30 @@
+package chap8.helper;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class NameGenerator {
+
+
+    private List<String> list = new ArrayList<>();
+
+    public Flux<String> generateNames() {
+        return Flux.generate(stringSynchronousSink -> {
+            System.out.println("generated fresh");
+            Util.sleepSeconds(1);
+            String name = Util.faker().name().firstName();
+            list.add(name);
+            stringSynchronousSink.next(name);
+        }).cast(String.class)
+          .startWith(getFromCache());
+
+    }
+
+    private Flux<String> getFromCache() {
+        System.out.println("from cache");
+        return Flux.fromIterable(list);
+    }
+}

src/main/java/chap8/helper/Qatar.java

@@ -0,0 +1,16 @@
+package chap8.helper;
+
+import reactor.core.publisher.Flux;
+import utility.Util;
+
+import java.time.Duration;
+
+public class Qatar {
+
+    public static Flux<String> getFlights() {
+        return Flux.range(1, Util.faker().random().nextInt(1, 5))
+                .delayElements(Duration.ofSeconds(1))
+                .map(i -> "Qatar " + Util.faker().random().nextInt(100, 999))
+                .filter(i -> Util.faker().random().nextBoolean());
+    }
+}

src/main/java/utility/DefaultSubscriber.java

@@ -22,7 +22,7 @@ public void onSubscribe(Subscription subscription) {
 
     @Override
     public void onNext(Object o) {
-        System.out.println(name +" Received :"+ o);
+        System.out.println(name +"Received :"+ o);
     }
 
     @Override
@@ -32,6 +32,6 @@ public void onError(Throwable throwable) {
 
     @Override
     public void onComplete() {
-        System.out.println(name + " Completed: ");
+        System.out.println(name + "Completed: ");
     }
 }

src/main/java/utility/Util.java

@@ -27,8 +27,12 @@ public static  Faker faker(){
     }
 
     public static void sleepSeconds(int seconds) {
+        sleepMillis(seconds * 1000);
+    }
+
+    public static void sleepMillis(int milli) {
         try {
-            Thread.sleep(seconds * 1000);
+            Thread.sleep(milli );
         } catch (InterruptedException e) {
             e.printStackTrace();
         }