BEAL-572 different strategies of handling hot-observable

rxjava/src/main/java/com/baelding/rxjava/ColdObservableBackpressure.java

@@ -0,0 +1,44 @@
+package com.baelding.rxjava;
+
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.schedulers.Schedulers;
+
+public class ColdObservableBackPressure {
+    public static void main(String[] args) throws InterruptedException {
+        Observable.range(1, 1_000_000)
+                .observeOn(Schedulers.computation())
+                .subscribe(v -> ComputeFunction.compute(v), Throwable::printStackTrace);
+
+        Thread.sleep(10_000);
+
+
+//        Observable.range(1, 1_000_000) //implementation of reactive pull backpressure on cold observable
+//                .subscribe(new Subscriber<Integer>() {
+//                    @Override
+//                    public void onStart() {
+//                        request(1);
+//                    }
+//
+//                    public void onNext(Integer v) {
+//                        compute(v);
+//
+//                        request(1);
+//                    }
+//
+//                    @Override
+//                    public void onError(Throwable ex) {
+//                        ex.printStackTrace();
+//                    }
+//
+//                    @Override
+//                    public void onCompleted() {
+//                        System.out.println("Done!");
+//                    }
+//                });
+
+    }
+
+
+}

rxjava/src/main/java/com/baelding/rxjava/ComputeFunction.java

@@ -0,0 +1,37 @@
+package com.baelding.rxjava;
+
+
+import rx.Observable;
+
+import java.util.List;
+
+public class ComputeFunction {
+    public static void compute(Integer v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compute(List<Integer> v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compute(Observable<Integer> v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+
+}

rxjava/src/main/java/com/baelding/rxjava/HotObservableBackPressureBatching.java

@@ -0,0 +1,22 @@
+package com.baelding.rxjava;
+
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+public class HotObservableBackPressureBatching {
+    public static void main(String[] args) throws InterruptedException {
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //buffer
+        source.window(500)
+                .observeOn(Schedulers.computation())
+                .subscribe(ComputeFunction::compute, Throwable::printStackTrace);
+
+
+        for (int i = 0; i < 1_000_000; i++) {
+            source.onNext(i);
+        }
+        Thread.sleep(10_000);
+    }
+
+}

rxjava/src/main/java/com/baelding/rxjava/HotObservableBackPressureBuffering.java

@@ -0,0 +1,23 @@
+package com.baelding.rxjava;
+
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+
+public class HotObservableBackPressureBuffering {
+    public static void main(String[] args) throws InterruptedException {
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //buffer
+        source
+                .buffer(1024)
+                .observeOn(Schedulers.computation())
+                .subscribe(ComputeFunction::compute, Throwable::printStackTrace);
+
+
+        for (int i = 0; i < 1_000_000; i++) {
+            source.onNext(i);
+        }
+        Thread.sleep(10_000);
+    }
+}

rxjava/src/main/java/com/baelding/rxjava/HotObservableBackPressureSkipping.java

@@ -0,0 +1,27 @@
+package com.baelding.rxjava;
+
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+import java.util.concurrent.TimeUnit;
+
+public class HotObservableBackPressureSkipping {
+    public static void main(String[] args) throws InterruptedException {
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //buffer
+        source
+//              .debounce(1, TimeUnit.SECONDS)
+//              .sample(1, TimeUnit.SECONDS)
+//              .throttleFirst(100, TimeUnit.MILLISECONDS)
+                .throttleLast(100, TimeUnit.MILLISECONDS)
+                .observeOn(Schedulers.computation())
+                .subscribe(ComputeFunction::compute, Throwable::printStackTrace);
+
+
+        for (int i = 0; i < 1_000_000; i++) {
+            source.onNext(i);
+        }
+        Thread.sleep(10_000);
+    }
+}

rxjava/src/main/java/com/baelding/rxjava/HotObservableBackpressure.java

@@ -1,30 +0,0 @@
-package com.baelding.rxjava;
-
-
-import rx.schedulers.Schedulers;
-import rx.subjects.PublishSubject;
-
-public class HotObservableBackPressure {
-    public static void main(String[] args) throws InterruptedException {
-        PublishSubject<Integer> source = PublishSubject.create();
-
-        source
-                .observeOn(Schedulers.computation())
-                .subscribe(v -> compute(v), Throwable::printStackTrace);
-
-        for (int i = 0; i < 1_000_000; i++) {
-            source.onNext(i);
-        }
-
-        Thread.sleep(10_000);
-    }
-
-    private static void compute(Integer v) {
-        try {
-            System.out.println("compute integer v: "+ v);
-            Thread.sleep(1000);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-    }
-}

rxjava/src/main/java/com/baelding/rxjava/HotObservableWithoutBackPressure.java

@@ -0,0 +1,21 @@
+package com.baelding.rxjava;
+
+
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+public class HotObservableWithoutBackPressure {
+    public static void main(String[] args) throws InterruptedException {
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //buffer
+        source.observeOn(Schedulers.computation())
+              .subscribe(ComputeFunction::compute, Throwable::printStackTrace);
+
+
+        for (int i = 0; i < 1_000_000; i++) {
+            source.onNext(i);
+        }
+        Thread.sleep(10_000);
+    }
+}