2.x: Add doOnEvent to Single & Completable (#4479)

Author: vanniktech

src/main/java/io/reactivex/Completable.java

@@ -984,6 +984,33 @@ public final Completable doOnError(Consumer<? super Throwable> onError) {
                 Functions.EMPTY_ACTION, Functions.EMPTY_ACTION);
     }
 
+    /**
+     * Returns a Completable which calls the given onEvent callback with the (throwable) for an onError
+     * or (null) for an onComplete signal from this Completable before delivering said signal to the downstream.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doOnEvent} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param onEvent the event callback
+     * @return the new Completable instance
+     * @throws NullPointerException if onEvent is null
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Completable doOnEvent(final Consumer<? super Throwable> onEvent) {
+        ObjectHelper.requireNonNull(onEvent, "onEvent is null");
+        return doOnLifecycle(Functions.emptyConsumer(), new Consumer<Throwable>() {
+            @Override
+            public void accept(final Throwable throwable) throws Exception {
+                onEvent.accept(throwable);
+            }
+        }, new Action() {
+            @Override
+            public void run() throws Exception {
+                onEvent.accept(null);
+            }
+        }, Functions.EMPTY_ACTION, Functions.EMPTY_ACTION, Functions.EMPTY_ACTION);
+    }
+
     /**
      * Returns a Completable instance that calls the various callbacks on the specific
      * lifecycle events.

src/main/java/io/reactivex/Maybe.java

@@ -859,7 +859,7 @@ public final Maybe<T> doOnError(Consumer<? super Throwable> onError) {
     }
 
     /**
-     * Calls the given onEvent callback with the (success value, null) for an onSuccess, (null, throwabe) for
+     * Calls the given onEvent callback with the (success value, null) for an onSuccess, (null, throwable) for
      * an onError or (null, null) for an onComplete signal from this Maybe before delivering said
      * signal to the downstream.
      * <p>

src/main/java/io/reactivex/Single.java

@@ -1591,6 +1591,22 @@ public final Single<T> doOnSuccess(final Consumer<? super T> onSuccess) {
         return RxJavaPlugins.onAssembly(new SingleDoOnSuccess<T>(this, onSuccess));
     }
 
+    /**
+     * Calls the shared consumer with the error sent via onError or the value
+     * via onSuccess for each SingleObserver that subscribes to the current Single.
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code doOnSubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param onEvent the consumer called with the success value of onEvent
+     * @return the new Single instance
+     * @since 2.0
+     */
+    public final Single<T> doOnEvent(final BiConsumer<? super T, ? super Throwable> onEvent) {
+        ObjectHelper.requireNonNull(onEvent, "onEvent is null");
+        return RxJavaPlugins.onAssembly(new SingleDoOnEvent<T>(this, onEvent));
+    }
+
     /**
      * Calls the shared consumer with the error sent via onError for each
      * SingleObserver that subscribes to the current Single.

src/main/java/io/reactivex/internal/operators/single/SingleDoOnEvent.java

@@ -0,0 +1,68 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.single;
+
+import io.reactivex.Single;
+import io.reactivex.SingleObserver;
+import io.reactivex.SingleSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.CompositeException;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.BiConsumer;
+
+public final class SingleDoOnEvent<T> extends Single<T> {
+    final SingleSource<T> source;
+
+    final BiConsumer<? super T, ? super Throwable> onEvent;
+
+    public SingleDoOnEvent(SingleSource<T> source, BiConsumer<? super T, ? super Throwable> onEvent) {
+        this.source = source;
+        this.onEvent = onEvent;
+    }
+
+    @Override
+    protected void subscribeActual(final SingleObserver<? super T> s) {
+
+        source.subscribe(new SingleObserver<T>() {
+            @Override
+            public void onSubscribe(Disposable d) {
+                s.onSubscribe(d);
+            }
+
+            @Override
+            public void onSuccess(T value) {
+                try {
+                    onEvent.accept(value, null);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    s.onError(ex);
+                    return;
+                }
+
+                s.onSuccess(value);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                try {
+                    onEvent.accept(null, e);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    e = new CompositeException(ex, e);
+                }
+                s.onError(e);
+            }
+        });
+    }
+}

src/test/java/io/reactivex/completable/CompletableTest.java

@@ -4463,4 +4463,41 @@ public void run() {
             exec.shutdown();
         }
     }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnEventNullEvent() {
+        Completable.complete().doOnEvent(null);
+    }
+
+    @Test
+    public void doOnEventComplete() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+
+        Completable.complete().doOnEvent(new Consumer<Throwable>() {
+            @Override
+            public void accept(final Throwable throwable) throws Exception {
+                if (throwable == null) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void doOnEventError() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+
+        Completable.error(new RuntimeException()).doOnEvent(new Consumer<Throwable>() {
+            @Override
+            public void accept(final Throwable throwable) throws Exception {
+                if (throwable != null) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
 }

src/test/java/io/reactivex/single/SingleTest.java

@@ -452,5 +452,42 @@ public void testToObservable() {
     	ts.assertNoErrors();
     	ts.assertComplete();
     }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnEventNullEvent() {
+        Single.just(1).doOnEvent(null);
+    }
+
+    @Test
+    public void doOnEventComplete() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+
+        Single.just(1).doOnEvent(new BiConsumer<Integer, Throwable>() {
+            @Override
+            public void accept(final Integer integer, final Throwable throwable) throws Exception {
+                if (integer != null) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void doOnEventError() {
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+
+        Single.error(new RuntimeException()).doOnEvent(new BiConsumer<Object, Throwable>() {
+            @Override
+            public void accept(final Object o, final Throwable throwable) throws Exception {
+                if (throwable != null) {
+                    atomicInteger.incrementAndGet();
+                }
+            }
+        }).subscribe();
+
+        assertEquals(1, atomicInteger.get());
+    }
 }