Examples 3.7 Custom operators

Froussios · Froussios · commit 9a2fc2139903 · 2015-05-07T19:52:13.000+02:00
diff --git a/tests/java/itrx/chapter3/custom/ComposeTest.java b/tests/java/itrx/chapter3/custom/ComposeTest.java
@@ -0,0 +1,104 @@
+package itrx.chapter3.custom;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.observers.TestSubscriber;
+
+public class ComposeTest {
+
+	/**
+	 * A custom operator for calculating a running average
+	 * 
+	 * @author Chris
+	 *
+	 */
+	public static class RunningAverage implements Observable.Transformer<Integer, Double> {
+	    private static class AverageAcc {
+	        public final int sum;
+	        public final int count;
+	        public AverageAcc(int sum, int count) {
+	            this.sum = sum;
+	            this.count = count;
+	        }
+	    }
+
+	    final int threshold;
+
+	    public RunningAverage() {
+	        this.threshold = Integer.MAX_VALUE;
+	    }
+
+	    public RunningAverage(int threshold) {
+	        this.threshold = threshold;
+	    }
+
+	    @Override
+	    public Observable<Double> call(Observable<Integer> source) {
+	        return source
+	            .filter(i -> i< this.threshold)
+	            .scan(
+	                new AverageAcc(0,0),
+	                (acc, v) -> new AverageAcc(acc.sum + v, acc.count + 1))
+	            .filter(acc -> acc.count > 0)
+	            .map(acc -> acc.sum/(double)acc.count);
+	    }
+	}
+	
+	public void exampleComposeFromClass() {
+		Observable.just(2, 3, 10, 12, 4)
+			.compose(new RunningAverage())
+			.subscribe(System.out::println);
+		
+//		2.0
+//		2.5
+//		5.0
+//		6.75
+//		6.2
+	}
+	
+	public void exampleComposeParameterised() {
+		Observable.just(2, 3, 10, 12, 4)
+			.compose(new RunningAverage(5))
+			.subscribe(System.out::println);
+		
+//		2.0
+//		2.5
+//		3.0
+	}
+	
+	
+	//
+	// Test
+	//
+	
+	@Test
+	public void testComposeFromClass() {
+		TestSubscriber<Double> tester = new TestSubscriber<>();
+		
+		Observable.just(2, 3, 10, 12, 4)
+			.compose(new RunningAverage())
+			.subscribe(tester);
+		
+		tester.assertReceivedOnNext(Arrays.asList(2.0, 2.5, 5.0, 6.75, 6.2));
+		tester.assertTerminalEvent();
+		tester.assertNoErrors();
+	}
+	
+	@Test
+	public void testComposeParameterised() {
+		TestSubscriber<Double> tester = new TestSubscriber<>();
+		
+		Observable.just(2, 3, 10, 12, 4)
+			.compose(new RunningAverage(5))
+			.subscribe(tester);
+		
+		tester.assertReceivedOnNext(Arrays.asList(2.0, 2.5, 3.0));
+		tester.assertTerminalEvent();
+		tester.assertNoErrors();
+	}
+	
+
+}
diff --git a/tests/java/itrx/chapter3/custom/LiftTest.java b/tests/java/itrx/chapter3/custom/LiftTest.java
@@ -0,0 +1,80 @@
+package itrx.chapter3.custom;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.observers.TestSubscriber;
+
+public class LiftTest {
+
+	public static class MyMap<T,R> implements Observable.Operator<R, T> {
+	    private Func1<T,R> transformer;
+
+	    public MyMap(Func1<T,R> transformer) {
+	        this.transformer = transformer;
+	    }
+
+	    @Override
+	    public Subscriber<? super T> call(Subscriber<? super R> subscriber) {
+	        return new Subscriber<T>() {
+
+	            @Override
+	            public void onCompleted() {
+	                if (!subscriber.isUnsubscribed())
+	                    subscriber.onCompleted();
+	            }
+
+	            @Override
+	            public void onError(Throwable e) {
+	                if (!subscriber.isUnsubscribed())
+	                    subscriber.onError(e);
+	            }
+
+	            @Override
+	            public void onNext(T t) {
+	                if (!subscriber.isUnsubscribed())
+	                    subscriber.onNext(transformer.call(t));
+	            }
+
+	        };
+	    }
+	    
+	    public static <T,R> MyMap<T,R> create(Func1<T,R> transformer) {
+	    	return new MyMap<T, R>(transformer);
+	    }
+	}
+	
+	public void exampleLift() {
+		Observable.range(0, 5)
+		    .lift(MyMap.create(i -> i + "!"))
+		    .subscribe(System.out::println);
+		
+//		0!
+//		1!
+//		2!
+//		3!
+//		4!
+	}
+	
+	
+	//
+	// Tests
+	//
+	
+	@Test
+	public void testLift() {
+		TestSubscriber<String> tester = new TestSubscriber<>();
+		
+		Observable.range(0, 5)
+		    .lift(MyMap.create(i -> i + "!"))
+		    .subscribe(tester);
+		
+		tester.assertReceivedOnNext(Arrays.asList("0!", "1!", "2!", "3!", "4!"));
+		tester.assertTerminalEvent();
+		tester.assertNoErrors();
+	}
+}
