Adds beginnings of doOnEach operator

Author: nullstyle

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

@@ -43,6 +43,7 @@
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
 import rx.operators.OperationDistinctUntilChanged;
+import rx.operators.OperationDoOnEach;
 import rx.operators.OperationElementAt;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
@@ -4777,6 +4778,22 @@ public static <T> Observable<T> amb(Iterable<? extends Observable<? extends T>>
         return create(OperationAmb.amb(sources));
     }
 
+
+    /**
+     * Invokes an action for each element in the observable sequence.
+     *
+     * @param func
+     *            The action to invoke for each element in the source sequence.
+     *
+     * @return
+     *            The source sequence with the side-effecting behavior applied.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229115(v=vs.103).aspx">MSDN: Observable.Amb</a>
+     */
+    public Observable<T> doOnEach(Observer<? super T> observer) {
+        return create(OperationDoOnEach.doOnEach(this, observer));
+    }
+
+
     /**
      * Whether a given {@link Function} is an internal implementation inside rx.* packages or not.
      * <p>

rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java

@@ -0,0 +1,66 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Observable.OnSubscribeFunc;
+import rx.Subscription;
+
+/**
+ * Converts the elements of an observable sequence to the specified type.
+ */
+public class OperationDoOnEach {
+    public static <T> OnSubscribeFunc<T> doOnEach(Observable<? extends T> source, Observer<? super T> observer) {
+        return new DoOnEachObservable<T>(source, observer);
+    }
+
+    private static class DoOnEachObservable<T> implements OnSubscribeFunc<T> {
+
+        private final Observable<? extends T> source;
+        private final Observer<? super T> doOnEachObserver;
+
+        public DoOnEachObservable(Observable<? extends T> source, Observer<? super T> doOnEachObserver) {
+            this.source = source;
+            this.doOnEachObserver = doOnEachObserver;
+        }
+
+        @Override
+        public Subscription onSubscribe(final Observer<? super T> observer) {
+            return source.subscribe(new Observer<T>() {
+                @Override
+                public void onCompleted() {
+                    doOnEachObserver.onCompleted();
+                    observer.onCompleted();
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    doOnEachObserver.onError(e);
+                    observer.onError(e);
+                }
+
+                @Override
+                public void onNext(T value) {
+                    doOnEachObserver.onNext(value);
+                    observer.onNext(value);
+                }
+
+            });
+        }
+
+    }
+}

rxjava-core/src/test/java/rx/operators/OperationDoOnEachTest.java

@@ -0,0 +1,108 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+import static rx.operators.OperationMap.*;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import rx.Observable;
+import rx.Observer;
+import rx.concurrency.Schedulers;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
+
+public class OperationDoOnEachTest {
+
+    @Mock
+    Observer<String> subscribedObserver;
+    @Mock
+    Observer<String> sideEffectObserver;
+
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test
+    public void testDoOnEach() {
+        Observable<String> base     = Observable.from("a", "b", "c");
+        Observable<String> doOnEach = base.doOnEach(sideEffectObserver);
+
+        doOnEach.subscribe(subscribedObserver);
+
+        // ensure the leaf observer is still getting called
+        verify(subscribedObserver, never()).onError(any(Throwable.class));
+        verify(subscribedObserver, times(1)).onNext("a");
+        verify(subscribedObserver, times(1)).onNext("b");
+        verify(subscribedObserver, times(1)).onNext("c");
+        verify(subscribedObserver, times(1)).onCompleted();
+
+        // ensure our injected observer is getting called
+        verify(sideEffectObserver, never()).onError(any(Throwable.class));
+        verify(sideEffectObserver, times(1)).onNext("a");
+        verify(sideEffectObserver, times(1)).onNext("b");
+        verify(sideEffectObserver, times(1)).onNext("c");
+        verify(sideEffectObserver, times(1)).onCompleted();
+    }
+    
+
+
+    @Test
+    public void testDoOnEachWithError() {
+        Observable<String> base = Observable.from("one", "fail", "two", "three", "fail");
+        Observable<String> errs = base.map(new Func1<String, String>() {
+            @Override
+            public String call(String s) {
+                if ("fail".equals(s)) {
+                    throw new RuntimeException("Forced Failure");
+                }
+                return s;
+            }
+        });
+        
+        Observable<String> doOnEach = errs.doOnEach(sideEffectObserver);
+
+
+        doOnEach.subscribe(subscribedObserver);
+        verify(subscribedObserver, times(1)).onNext("one");
+        verify(subscribedObserver, never()).onNext("two");
+        verify(subscribedObserver, never()).onNext("three");
+        verify(subscribedObserver, never()).onCompleted();
+        verify(subscribedObserver, times(1)).onError(any(Throwable.class));
+
+
+        verify(sideEffectObserver, times(1)).onNext("one");
+        verify(sideEffectObserver, never()).onNext("two");
+        verify(sideEffectObserver, never()).onNext("three");
+        verify(sideEffectObserver, never()).onCompleted();
+        verify(sideEffectObserver, times(1)).onError(any(Throwable.class));
+    }
+
+  
+}