BAEL-572 BackPressure -> Backpressure; code reformatted; included rxjava in parent pom

pom.xml

@@ -94,6 +94,7 @@
         <module>rest-assured</module>
         <module>rest-testing</module>
         <module>resteasy</module>
+        <module>rxjava</module>
 
         <module>selenium-junit-testng</module>
         <module>solr-fulltext-search</module>

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

@@ -1,19 +1,14 @@
 package com.baelding.rxjava;
 
-
 import rx.Observable;
-import rx.Subscriber;
 import rx.schedulers.Schedulers;
 
-public class ColdObservableBackPressure {
+public class ColdObservableBackpressure {
     public static void main(String[] args) throws InterruptedException {
-        Observable.range(1, 1_000_000)
+        Observable.range(1, 1_000_000).observeOn(Schedulers.computation()).subscribe(v -> ComputeFunction.compute(v), Throwable::printStackTrace);
-                .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
@@ -40,5 +35,4 @@ public class ColdObservableBackPressure {
 
     }
 
-
 }

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

@@ -1,6 +1,5 @@
 package com.baelding.rxjava;
 
-
 import rx.Observable;
 
 import java.util.List;
@@ -33,7 +32,6 @@ public class ComputeFunction {
         }
     }
 
-
     public static void compute(Long v) {
         try {
             System.out.println("compute integer v: " + v);

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

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

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

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

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

@@ -5,15 +5,13 @@ import rx.subjects.PublishSubject;
 
 import java.util.concurrent.TimeUnit;
 
-public class HotObservableBackPressureSkipping {
+public class HotObservableBackpressureSkipping {
     public static void main(String[] args) throws InterruptedException {
         PublishSubject<Integer> source = PublishSubject.<Integer>create();
 
         source.sample(100, TimeUnit.MILLISECONDS)
             //                .throttleFirst(100, TimeUnit.MILLISECONDS)
-                .observeOn(Schedulers.computation())
+            .observeOn(Schedulers.computation()).subscribe(ComputeFunction::compute, Throwable::printStackTrace);
-                .subscribe(ComputeFunction::compute, Throwable::printStackTrace);
-
 
         for (int i = 0; i < 1_000_000; i++) {
             source.onNext(i);

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

@@ -1,28 +0,0 @@
-package com.baelding.rxjava;
-
-
-import rx.BackpressureOverflow;
-import rx.Observable;
-import rx.schedulers.Schedulers;
-
-public class HotObservableOnBackPressure {
-    public static void main(String[] args) throws InterruptedException {
-        Observable.range(1, 1_000_000)
-                .onBackpressureBuffer(16, () -> {
-                        },
-                        BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST)
-                .observeOn(Schedulers.computation())
-                .subscribe(e -> {
-                }, Throwable::printStackTrace);
-
-
-        Observable.range(1, 1_000_000)
-                .onBackpressureDrop()
-                .observeOn(Schedulers.io())
-                .doOnNext(ComputeFunction::compute)
-                .subscribe(v -> {
-                }, Throwable::printStackTrace);
-        Thread.sleep(10_000);
-
-    }
-}

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

@@ -0,0 +1,18 @@
+package com.baelding.rxjava;
+
+import rx.BackpressureOverflow;
+import rx.Observable;
+import rx.schedulers.Schedulers;
+
+public class HotObservableOnBackpressure {
+    public static void main(String[] args) throws InterruptedException {
+        Observable.range(1, 1_000_000).onBackpressureBuffer(16, () -> {
+        }, BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST).observeOn(Schedulers.computation()).subscribe(e -> {
+        }, Throwable::printStackTrace);
+
+        Observable.range(1, 1_000_000).onBackpressureDrop().observeOn(Schedulers.io()).doOnNext(ComputeFunction::compute).subscribe(v -> {
+        }, Throwable::printStackTrace);
+        Thread.sleep(10_000);
+
+    }
+}

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

@@ -4,7 +4,7 @@ package com.baelding.rxjava;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 
-public class HotObservableWithoutBackPressure {
+public class HotObservableWithoutBackpressure {
     public static void main(String[] args) throws InterruptedException {
         PublishSubject<Integer> source = PublishSubject.<Integer>create();