Collect Operator

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -15,6 +15,7 @@
  */
 package rx;
 
+import static org.junit.Assert.*;
 import static rx.util.functions.Functions.*;
 
 import java.util.ArrayList;
@@ -123,6 +124,7 @@
 import rx.util.Timestamped;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
+import rx.util.functions.Action2;
 import rx.util.functions.Async;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
@@ -5350,6 +5352,29 @@ public <R> Observable<R> reduce(R initialValue, Func2<R, ? super T, R> accumulat
         return create(OperationScan.scan(this, initialValue, accumulator)).takeLast(1);
     }
 
+    /**
+     * Collect values into a single mutable data structure.
+     * <p>
+     * A simplified version of `reduce` that does not need to return the state on each pass.
+     * <p>
+     * 
+     * @param state
+     * @param collector
+     * @return
+     */
+    public <R> Observable<R> collect(R state, final Action2<R, ? super T> collector) {
+        Func2<R, T, R> accumulator = new Func2<R, T, R>() {
+
+            @Override
+            public R call(R state, T value) {
+                collector.call(state, value);
+                return state;
+            }
+
+        };
+        return reduce(state, accumulator);
+    }
+    
     /**
      * Synonymous with <code>reduce()</code>.
      * <p>

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -43,6 +43,7 @@
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action1;
+import rx.util.functions.Action2;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
@@ -1091,5 +1092,37 @@ public String answer(InvocationOnMock invocation) throws Throwable {
 
         verify(func, times(1)).call();
     }
+    
+    @Test
+    public void testCollectToList() {
+        List<Integer> list = Observable.from(1, 2, 3).collect(new ArrayList<Integer>(), new Action2<List<Integer>, Integer>() {
+
+            @Override
+            public void call(List<Integer> list, Integer v) {
+                list.add(v);
+            }
+        }).toBlockingObservable().last();
+
+        assertEquals(3, list.size());
+        assertEquals(1, list.get(0).intValue());
+        assertEquals(2, list.get(1).intValue());
+        assertEquals(3, list.get(2).intValue());
+    }
+
+    @Test
+    public void testCollectToString() {
+        String value = Observable.from(1, 2, 3).collect(new StringBuilder(), new Action2<StringBuilder, Integer>() {
+
+            @Override
+            public void call(StringBuilder sb, Integer v) {
+                if (sb.length() > 0) {
+                    sb.append("-");
+                }
+                sb.append(v);
+            }
+        }).toBlockingObservable().last().toString();
+
+        assertEquals("1-2-3", value);
+    }
 
 }