Merge pull request #1527 from mattrjacobs/reduce-fails-on-backpressure

Failing unit test for reduce, showing it does not implement backpressure correctly

rxjava-core/src/test/java/rx/internal/operators/OperatorReduceTest.java

@@ -16,6 +16,7 @@
 
 package rx.internal.operators;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -114,4 +115,24 @@ public Integer call(Integer t1) {
         verify(observer, times(1)).onError(any(TestException.class));
     }
 
+    @Test
+    public void testBackpressureWithNoInitialValue() throws InterruptedException {
+        Observable<Integer> source = Observable.from(1, 2, 3, 4, 5, 6);
+        Observable<Integer> reduced = source.reduce(sum);
+
+        Integer r = reduced.toBlocking().first();
+        assertEquals(21, r.intValue());
+    }
+
+    @Test
+    public void testBackpressureWithInitialValue() throws InterruptedException {
+        Observable<Integer> source = Observable.from(1, 2, 3, 4, 5, 6);
+        Observable<Integer> reduced = source.reduce(0, sum);
+
+        Integer r = reduced.toBlocking().first();
+        assertEquals(21, r.intValue());
+    }
+
+
+
 }