Merge branch 'tomekl007-BAEL-572'

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,41 +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, 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/HotObservableBackpressureBatching.java

@@ -1,21 +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,20 +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,22 +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/HotObservableOnBackpressure.java

@@ -1,27 +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/HotObservableWithoutBackpressure.java

@@ -1,19 +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);
+
+    }
+}

spring-core/README.md

@@ -3,4 +3,4 @@
 - [Exploring the Spring BeanFactory API](http://www.baeldung.com/spring-beanfactory)
 - [How to use the Spring FactoryBean?](http://www.baeldung.com/spring-factorybean)
 - [Constructor Dependency Injection in Spring](http://www.baeldung.com/constructor-injection-in-spring)
-- [Constructor Injection in Spring with Lombok](http://inprogress.baeldung.com/constructor-injection-in-spring-with-lombok)
+- [Constructor Injection in Spring with Lombok](http://www.baeldung.com/spring-injection-lombok)

spring-security-cache-control/src/test/java/com/baeldung/cachecontrol/ResourceEndpointTest.java

@@ -11,7 +11,7 @@ import static com.jayway.restassured.RestAssured.given;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = AppRunner.class)
-public class ResourceEndpointLiveTest {
+public class ResourceEndpointTest {
 
     @LocalServerPort
     private int serverPort;
@@ -66,7 +66,7 @@ public class ResourceEndpointLiveTest {
     }
 
     private String getBaseUrl() {
-        return "http://localhost:" + serverPort;
+        return String.format("http://localhost:%d", serverPort);
     }
 
 }