Merge pull request #5724 from tinomthomas/master

BAEL - 1511

rxjava-2/pom.xml

@@ -1,10 +1,11 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
+
 	<artifactId>rxjava-2</artifactId>
 	<version>1.0-SNAPSHOT</version>
-	<name>rxjava-2</name>
 
 	<parent>
 		<groupId>com.baeldung</groupId>
@@ -34,6 +35,12 @@
 			<artifactId>rxrelay</artifactId>
 			<version>${rxrelay.version}</version>
 		</dependency>
+		<!-- https://mvnrepository.com/artifact/com.github.akarnokd/rxjava2-extensions -->
+		<dependency>
+			<groupId>com.github.akarnokd</groupId>
+			<artifactId>rxjava2-extensions</artifactId>
+			<version>${rxjava2.ext.version}</version>
+		</dependency>
 	</dependencies>
 
 	<properties>
@@ -41,5 +48,6 @@
 		<rx.java2.version>2.2.2</rx.java2.version>
 		<awaitility.version>1.7.0</awaitility.version>
 		<rxrelay.version>2.0.0</rxrelay.version>
+		<rxjava2.ext.version>0.20.4</rxjava2.ext.version>
 	</properties>
 </project>

rxjava-2/src/test/java/com/baeldung/rxjava/AsyncAndSyncToObservableIntegrationTest.java

@@ -0,0 +1,107 @@
+package com.baeldung.rxjava;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import hu.akarnokd.rxjava2.async.AsyncObservable;
+import io.reactivex.Observable;
+
+public class AsyncAndSyncToObservableIntegrationTest {
+
+    AtomicInteger counter = new AtomicInteger();
+    Callable<Integer> callable = () -> counter.incrementAndGet();
+
+    /* Method will execute every time it gets subscribed*/
+    @Test
+    public void givenSyncMethod_whenConvertedWithFromCallable_thenReturnObservable() {
+
+        Observable<Integer> source = Observable.fromCallable(callable);
+
+        for (int i = 1; i < 5; i++) {
+            source.test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertResult(i);
+
+            assertEquals(i, counter.get());
+        }
+    }
+
+    /* Method will execute only once and cache its result.*/
+    @Test
+    public void givenSyncMethod_whenConvertedWithStart_thenReturnObservable() {
+
+        Observable<Integer> source = AsyncObservable.start(callable);
+
+        for (int i = 1; i < 5; i++) {
+            source.test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertResult(1);
+
+            assertEquals(1, counter.get());
+        }
+    }
+
+    /* Method will execute only once and cache its result.*/
+    @Test
+    public void givenAsyncMethod_whenConvertedWithFromFuture_thenRetrunObservble() { 
+
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        Future<Integer> future = executor.submit(callable);
+        Observable<Integer> source = Observable.fromFuture(future);
+
+        for (int i = 1; i < 5; i++) {
+            source.test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertResult(1);
+
+            assertEquals(1, counter.get());
+        }
+
+        executor.shutdown();
+    }
+
+    /* Method will execute every time it gets subscribed*/
+    @Test
+    public void givenAsyncMethod_whenConvertedWithStartFuture_thenRetrunObservble() {
+
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        Observable<Integer> source = AsyncObservable.startFuture(() -> executor.submit(callable));
+
+        for (int i = 1; i < 5; i++) {
+            source.test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertResult(i);
+
+            assertEquals(i, counter.get());
+        }
+
+        executor.shutdown();
+    }
+
+    /*Method will execute only once and cache its result.*/
+    @Test
+    public void givenAsyncMethod_whenConvertedWithDeferFuture_thenRetrunObservble() { 
+        List<Integer> list = Arrays.asList(new Integer[] { counter.incrementAndGet(), counter.incrementAndGet(), counter.incrementAndGet() });
+        ExecutorService exec = Executors.newSingleThreadExecutor();
+        Callable<Observable<Integer>> callable = () -> Observable.fromIterable(list);
+        Observable<Integer> source = AsyncObservable.deferFuture(() -> exec.submit(callable));
+        for (int i = 1; i < 4; i++) {
+            source.test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertResult(1, 2, 3);
+        }
+
+        exec.shutdown();
+    }
+
+}