BAEL-572 move code to tests

rxjava/pom.xml

@@ -26,9 +26,15 @@
             <artifactId>rxjava</artifactId>
             <version>${rx.java.version}</version>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+        </dependency>
     </dependencies>
 
     <properties>
+        <junit.version>4.12</junit.version>
         <rx.java.version>1.2.5</rx.java.version>
     </properties>
 

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

@@ -1,16 +0,0 @@
-package com.baelding.rxjava;
-
-import rx.Observable;
-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(ComputeFunction::compute);
-
-        Thread.sleep(10_000);
-    }
-
-}

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

@@ -1,18 +0,0 @@
-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();
-
-        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

@@ -1,17 +0,0 @@
-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();
-
-        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

@@ -1,21 +0,0 @@
-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();
-
-        source.sample(100, TimeUnit.MILLISECONDS)
-            //                .throttleFirst(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/HotObservableWithoutBackpressure.java

@@ -1,20 +0,0 @@
-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();
-
-        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);
-    }
-}

rxjava/src/test/java/com/baeldung/rxjava/RxJavaBackpressureTest.java

@@ -0,0 +1,155 @@
+package com.baeldung.rxjava;
+
+import org.junit.Test;
+import rx.BackpressureOverflow;
+import rx.Observable;
+import rx.exceptions.MissingBackpressureException;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertTrue;
+
+public class RxJavaBackpressureTest {
+
+    @Test
+    public void givenColdObservable_shouldNotThrowException() {
+        //given
+        TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
+
+        //when
+        Observable.range(1, 1_000_000)
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        //then
+        testSubscriber.awaitTerminalEvent();
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+
+    @Test
+    public void givenHotObservable_whenBackpressureNotDefined_shouldTrowException() {
+        //given
+        TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        source.observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        //when
+        for (int i = 0; i < 1_000_000; i++) {
+            source.onNext(i);
+
+        }
+
+        //then
+        testSubscriber.awaitTerminalEvent();
+        testSubscriber.assertError(MissingBackpressureException.class);
+    }
+
+    @Test
+    public void givenHotObservable_whenWindowIsDefined_shouldNotThrowException() {
+        //given
+        TestSubscriber<Observable<Integer>> testSubscriber = new TestSubscriber<>();
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //when
+        source.window(500)
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        for (int i = 0; i < 1_000; i++) {
+            source.onNext(i);
+        }
+
+        //then
+        testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+
+    @Test
+    public void givenHotObservable_whenBufferIsDefined_shouldNotThrowException() {
+        //given
+        TestSubscriber<List<Integer>> testSubscriber = new TestSubscriber<>();
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //when
+        source.buffer(1024)
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        for (int i = 0; i < 1_000; i++) {
+            source.onNext(i);
+        }
+
+        //then
+        testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+
+
+    @Test
+    public void givenHotObservable_whenSkippingOperationIsDefined_shouldNotThrowException() {
+        //given
+        TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
+        PublishSubject<Integer> source = PublishSubject.<Integer>create();
+
+        //when
+        source.sample(100, TimeUnit.MILLISECONDS)
+//                .throttleFirst(100, TimeUnit.MILLISECONDS)
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        for (int i = 0; i < 1_000; i++) {
+            source.onNext(i);
+        }
+
+        //then
+        testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+
+    @Test
+    public void givenHotObservable_whenOnBackpressureBufferDefined_shouldNotThrowException() {
+        //given
+        TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
+
+        //when
+        Observable.range(1, 1_000_000)
+                .onBackpressureBuffer(16, () -> {
+                        },
+                        BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST)
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        //then
+        testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+
+
+    @Test
+    public void givenHotObservable_whenOnBackpressureDropDefined_shouldNotThrowException() {
+        //given
+        TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
+
+        //when
+        Observable.range(1, 1_000_000)
+                .onBackpressureDrop()
+                .observeOn(Schedulers.computation())
+                .subscribe(testSubscriber);
+
+        //then
+        testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
+        assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
+
+    }
+}