Examples 3.2 Leaving the monad

Froussios · Froussios · commit ca59ab3aeb72 · 2015-05-10T01:00:30.000+02:00
diff --git a/tests/java/itrx/chapter3/leaving/FirstLastSingleTest.java b/tests/java/itrx/chapter3/leaving/FirstLastSingleTest.java
@@ -0,0 +1,76 @@
+package itrx.chapter3.leaving;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import rx.Observable;
+
+public class FirstLastSingleTest {
+
+	public void exampleFirst() {
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS);
+
+		long value = values
+		    .take(5)
+		    .toBlocking()
+		    .first(i -> i>2);
+		System.out.println(value);
+		
+//		3
+	}
+	
+	public void exampleSingleError() {
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS);
+
+		try {
+		    long value = values
+		        .take(5)
+		        .toBlocking()
+		        .single(i -> i>2);
+		    System.out.println(value);
+		}
+		catch (Exception e) {
+		    System.out.println("Caught: " + e);
+		}
+		
+//		Caught: java.lang.IllegalArgumentException: Sequence contains too many elements
+	}
+	
+	
+	//
+	// Tests
+	//
+	
+	@Test
+	public void testFirst() throws InterruptedException {
+		List<Integer> received = new ArrayList<>();
+		
+		Observable<Integer> values = Observable.range(0,5);
+
+		int value = values
+		    .take(5)
+		    .toBlocking()
+		    .first(i -> i>2);
+		received.add(value);
+
+		assertEquals(received, Arrays.asList(3));
+	}
+	
+	@Test(expected = IllegalArgumentException.class)
+	public void testSingleError() {
+		Observable<Integer> values = Observable.range(0, 5);
+
+	    long value = values
+	        .take(5)
+	        .toBlocking()
+	        .single(i -> i>2);
+	    System.out.println(value);
+	}
+
+}
diff --git a/tests/java/itrx/chapter3/leaving/ForEachTest.java b/tests/java/itrx/chapter3/leaving/ForEachTest.java
@@ -0,0 +1,145 @@
+package itrx.chapter3.leaving;
+
+import static org.junit.Assert.*;
+
+import java.lang.Thread.State;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.schedulers.Schedulers;
+import rx.schedulers.TestScheduler;
+
+public class ForEachTest {
+
+	public void exampleObservableForEach() {
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS);
+
+		values
+		    .take(5)
+		    .forEach(
+		        v -> System.out.println(v));
+		System.out.println("Subscribed");
+		
+//		Subscribed
+//		0
+//		1
+//		2
+//		3
+//		4
+	}
+	
+	public void exampleBlockingForEach() {
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS);
+
+		values
+		    .take(5)
+		    .toBlocking()
+		    .forEach(
+		        v -> System.out.println(v));
+		System.out.println("Subscribed");
+		
+//		0
+//		1
+//		2
+//		3
+//		4
+//		Subscribed
+	}
+	
+	public void exampleBlockingForEachError() {
+		Observable<Long> values = Observable.error(new Exception("Oops"));
+
+		try {
+		    values
+		        .take(5)
+		        .toBlocking()
+		        .forEach(
+		            v -> System.out.println(v));
+		}
+		catch (Exception e) {
+		    System.out.println("Caught: " + e.getMessage());
+		}
+		System.out.println("Subscribed");
+		
+//		Caught: java.lang.Exception: Oops
+//		Subscribed
+	}
+	
+	
+	// 
+	// Tests
+	//
+	
+	@Test
+	public void testObservableForEach() {
+		List<Long> received = new ArrayList<>();
+		TestScheduler scheduler = Schedulers.test();
+		
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS, scheduler);
+
+		values
+		    .take(5)
+		    .forEach(
+		        i -> received.add(i));
+		received.add(-1L); // Mark that forEach statement returned
+		
+		assertEquals(received, Arrays.asList(-1L));
+		scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+		assertEquals(received, Arrays.asList(-1L, 0L, 1L, 2L, 3L, 4L));
+	}
+	
+	@Test
+	public void testBlockingForEach() throws InterruptedException {
+		List<Long> received = new ArrayList<>();
+		TestScheduler scheduler = Schedulers.test();
+		
+		Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS, scheduler);
+
+		// Blocking call on new thread
+		Thread thread = new Thread(() -> {
+			values
+			    .take(5)
+			    .toBlocking()
+			    .forEach(
+		    		i -> received.add(i));
+			received.add(-1L); // Mark that forEach statement returned
+			
+		});
+		thread.start();
+		
+		assertEquals(received, Arrays.asList());
+		// Wait for blocking call to block before producing values
+		while (thread.getState() != State.WAITING)
+			Thread.sleep(1);
+		scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+		// Wait for processing to complete
+		thread.join(50);
+		assertEquals(received, Arrays.asList(0L, 1L, 2L, 3L, 4L, -1L));
+	}
+	
+	@Test
+	public void testBlockingForEachError() {
+		boolean caughtException = false;
+		
+		Observable<Long> values = Observable.error(new Exception("Oops"));
+
+		try {
+		    values
+		        .take(5)
+		        .toBlocking()
+		        .forEach(
+		            v -> {});
+		}
+		catch (Exception e) {
+		    caughtException = true;
+		}
+
+		assertTrue(caughtException);
+	}
+
+}
diff --git a/tests/java/itrx/chapter3/leaving/FutureTest.java b/tests/java/itrx/chapter3/leaving/FutureTest.java
@@ -0,0 +1,46 @@
+package itrx.chapter3.leaving;
+
+
+import static org.junit.Assert.*;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import rx.Observable;
+
+public class FutureTest {
+
+	public void exampleFuture() {
+		Observable<Long> values = Observable.timer(500, TimeUnit.MILLISECONDS);
+
+		values.subscribe(v -> System.out.println("Emitted: " + v));
+
+		Future<Long> future = values.toBlocking().toFuture();
+		try {
+			System.out.println(future.get());
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		} catch (ExecutionException e) {
+			e.printStackTrace();
+		}
+		
+//		Emitted: 0
+//		0
+	}
+	
+	
+	//
+	//
+	
+	@Test
+	public void testFuture() throws InterruptedException, ExecutionException {
+		Observable<Integer> sequence = Observable.just(0);
+		Future<Integer> future = sequence.toBlocking().toFuture();
+		int value = future.get();
+		assertEquals(0, value);
+	}
+
+}
diff --git a/tests/java/itrx/chapter3/leaving/IterablesTest.java b/tests/java/itrx/chapter3/leaving/IterablesTest.java
