2.x: Completable enhance doOnEvent to reduce allocations (#4486)

Author: vanniktech

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

@@ -948,7 +948,7 @@ public final Completable doOnComplete(Action onComplete) {
                 onComplete, Functions.EMPTY_ACTION,
                 Functions.EMPTY_ACTION, Functions.EMPTY_ACTION);
     }
-    
+
     /**
      * Returns a Completable which calls the given onDispose callback if the child subscriber cancels
      * the subscription.
@@ -966,7 +966,7 @@ public final Completable doOnDispose(Action onDispose) {
                 Functions.EMPTY_ACTION, Functions.EMPTY_ACTION,
                 Functions.EMPTY_ACTION, onDispose);
     }
-    
+
     /**
      * Returns a Completable which calls the given onError callback if this Completable emits an error.
      * <dl>
@@ -998,17 +998,7 @@ public final Completable doOnError(Consumer<? super Throwable> onError) {
     @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);
+        return RxJavaPlugins.onAssembly(new CompletableDoOnEvent(this, onEvent));
     }
 
     /**
@@ -1027,9 +1017,9 @@ public void run() throws Exception {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     private Completable doOnLifecycle(
-            final Consumer<? super Disposable> onSubscribe, 
-            final Consumer<? super Throwable> onError, 
-            final Action onComplete, 
+            final Consumer<? super Disposable> onSubscribe,
+            final Consumer<? super Throwable> onError,
+            final Action onComplete,
             final Action onTerminate,
             final Action onAfterTerminate,
             final Action onDispose) {

src/main/java/io/reactivex/internal/operators/completable/CompletableDoOnEvent.java

@@ -0,0 +1,67 @@
+/**
+ * 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.completable;
+
+import io.reactivex.Completable;
+import io.reactivex.CompletableObserver;
+import io.reactivex.CompletableSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.CompositeException;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Consumer;
+
+public final class CompletableDoOnEvent extends Completable {
+    final CompletableSource source;
+    final Consumer<? super Throwable> onEvent;
+
+    public CompletableDoOnEvent(final CompletableSource source, final Consumer<? super Throwable> onEvent) {
+        this.source = source;
+        this.onEvent = onEvent;
+    }
+
+    @Override
+    protected void subscribeActual(final CompletableObserver s) {
+        source.subscribe(new CompletableObserver() {
+            @Override
+            public void onComplete() {
+                try {
+                    onEvent.accept(null);
+                } catch (Throwable e) {
+                    Exceptions.throwIfFatal(e);
+                    s.onError(e);
+                    return;
+                }
+                
+                s.onComplete();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                try {
+                    onEvent.accept(e);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    e = new CompositeException(ex, e);
+                }
+                
+                s.onError(e);
+            }
+
+            @Override
+            public void onSubscribe(final Disposable d) {
+                s.onSubscribe(d);
+            }
+        });
+    }
+}

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

@@ -4465,7 +4465,37 @@ public void run() {
     }
 
     @Test(expected = NullPointerException.class)
-    public void doOnEventNullEvent() {
+    public void doOnErrorNullValue() {
+        Completable.complete().doOnError(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnSubscribeNullValue() {
+        Completable.complete().doOnSubscribe(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doAfterTerminateNullValue() {
+        Completable.complete().doAfterTerminate(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnTerminateNullValue() {
+        Completable.complete().doOnTerminate(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnCompleteNullValue() {
+        Completable.complete().doOnComplete(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnDisposeNullValue() {
+        Completable.complete().doOnDispose(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void doOnEventNullValue() {
         Completable.complete().doOnEvent(null);
     }