1.x: Completable add doOnEach (#4460)

* 1.x: Completable add doOnEach

* Fix Generics

Author: vanniktech

src/main/java/rx/Completable.java

@@ -1442,7 +1442,31 @@ public void onSubscribe(Subscription d) {
     public final Completable doOnCompleted(Action0 onCompleted) {
         return doOnLifecycle(Actions.empty(), Actions.empty(), onCompleted, Actions.empty(), Actions.empty());
     }
-    
+
+    /**
+     * Returns a Completable which calls the given onNotification callback when this Completable emits an error or completes.
+     * @param onNotification the notification callback
+     * @return the new Completable instance
+     * @throws NullPointerException if onNotification is null
+     */
+    public final Completable doOnEach(final Action1<Notification<Object>> onNotification) {
+        if (onNotification == null) {
+            throw new IllegalArgumentException("onNotification is null");
+        }
+
+        return doOnLifecycle(Actions.empty(), new Action1<Throwable>() {
+            @Override
+            public void call(final Throwable throwable) {
+                onNotification.call(Notification.createOnError(throwable));
+            }
+        }, new Action0() {
+            @Override
+            public void call() {
+                onNotification.call(Notification.createOnCompleted());
+            }
+        }, Actions.empty(), Actions.empty());
+    }
+
     /**
      * Returns a Completable which calls the given onUnsubscribe callback if the child subscriber cancels
      * the subscription.

src/test/java/rx/CompletableTest.java

@@ -4174,4 +4174,39 @@ public Object call(Completable completable) {
         assertSame(expectedResult, actualResult);
         assertSame(c, completableRef.get());
     }
-}
+
+    @Test(expected = IllegalArgumentException.class)
+    public void doOnEachNullAction() {
+        Completable.complete().doOnEach(null);
+    }
+
+    @Test
+    public void doOnEachCompleted() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+        Completable.complete().doOnEach(new Action1<Notification<Object>>() {
+            @Override
+            public void call(final Notification<Object> notification) {
+                if (notification.isOnCompleted()) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void doOnEachError() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+        Completable.error(new RuntimeException("What?")).doOnEach(new Action1<Notification<Object>>() {
+            @Override
+            public void call(final Notification<Object> notification) {
+                if (notification.isOnError()) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
+}