doOnEach - reduce allocations, report to RxJavaHooks, use OnSubscribe (#4262)

Author: davidmoten

src/main/java/rx/Observable.java

@@ -5507,9 +5507,9 @@ public final Observable<T> distinctUntilChanged(Func2<? super T, ? super T, Bool
     public final Observable<T> doOnCompleted(final Action0 onCompleted) {
         Action1<T> onNext = Actions.empty();
         Action1<Throwable> onError = Actions.empty();
-        Observer<T> observer = new ActionSubscriber<T>(onNext, onError, onCompleted);
+        Observer<T> observer = new ActionObserver<T>(onNext, onError, onCompleted);
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**
@@ -5531,8 +5531,7 @@ public final Observable<T> doOnCompleted(final Action0 onCompleted) {
      */
     public final Observable<T> doOnEach(final Action1<Notification<? super T>> onNotification) {
         Observer<T> observer = new ActionNotificationObserver<T>(onNotification);
-
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**
@@ -5559,7 +5558,7 @@ public final Observable<T> doOnEach(final Action1<Notification<? super T>> onNot
      * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
      */
     public final Observable<T> doOnEach(Observer<? super T> observer) {
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**
@@ -5585,9 +5584,9 @@ public final Observable<T> doOnEach(Observer<? super T> observer) {
     public final Observable<T> doOnError(final Action1<Throwable> onError) {
         Action1<T> onNext = Actions.empty();
         Action0 onCompleted = Actions.empty();
-        Observer<T> observer = new ActionSubscriber<T>(onNext, onError, onCompleted);
+        Observer<T> observer = new ActionObserver<T>(onNext, onError, onCompleted);
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**
@@ -5610,9 +5609,9 @@ public final Observable<T> doOnError(final Action1<Throwable> onError) {
     public final Observable<T> doOnNext(final Action1<? super T> onNext) {
         Action1<Throwable> onError = Actions.empty();
         Action0 onCompleted = Actions.empty();
-        Observer<T> observer = new ActionSubscriber<T>(onNext, onError, onCompleted);
+        Observer<T> observer = new ActionObserver<T>(onNext, onError, onCompleted);
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**
@@ -5693,9 +5692,9 @@ public final Observable<T> doOnTerminate(final Action0 onTerminate) {
         Action1<T> onNext = Actions.empty();
         Action1<Throwable> onError = Actions.toAction1(onTerminate);
 
-        Observer<T> observer = new ActionSubscriber<T>(onNext, onError, onTerminate);
+        Observer<T> observer = new ActionObserver<T>(onNext, onError, onTerminate);
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return create(new OnSubscribeDoOnEach<T>(this, observer));
     }
 
     /**

src/main/java/rx/Single.java

@@ -2413,7 +2413,7 @@ public void onNext(T t) {
             }
         };
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return Observable.create(new OnSubscribeDoOnEach<T>(this.toObservable(), observer)).toSingle();
     }
 
     /**
@@ -2449,7 +2449,7 @@ public void onNext(T t) {
             }
         };
 
-        return lift(new OperatorDoOnEach<T>(observer));
+        return Observable.create(new OnSubscribeDoOnEach<T>(this.toObservable(), observer)).toSingle();
     }
 
     /**

src/main/java/rx/internal/operators/OnSubscribeDoOnEach.java

@@ -0,0 +1,104 @@
+/**
+ * Copyright 2014 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.internal.operators;
+
+import java.util.Arrays;
+
+import rx.*;
+import rx.Observable.OnSubscribe;
+import rx.exceptions.*;
+import rx.plugins.RxJavaHooks;
+
+/**
+ * Calls specified actions for each notification.
+ * 
+ * @param <T> the value type
+ */
+public class OnSubscribeDoOnEach<T> implements OnSubscribe<T> {
+    private final Observer<? super T> doOnEachObserver;
+    private final Observable<T> source;
+
+    public OnSubscribeDoOnEach(Observable<T> source, Observer<? super T> doOnEachObserver) {
+        this.source = source;
+        this.doOnEachObserver = doOnEachObserver;
+    }
+
+    @Override
+    public void call(final Subscriber<? super T> subscriber) {
+        source.unsafeSubscribe(new DoOnEachSubscriber<T>(subscriber, doOnEachObserver));
+    }
+    
+    private static final class DoOnEachSubscriber<T> extends Subscriber<T> {
+
+        private final Subscriber<? super T> subscriber;
+        private final Observer<? super T> doOnEachObserver;
+        
+        private boolean done;
+
+        DoOnEachSubscriber(Subscriber<? super T> subscriber, Observer<? super T> doOnEachObserver) {
+            super(subscriber);
+            this.subscriber = subscriber;
+            this.doOnEachObserver = doOnEachObserver;
+        }
+
+        @Override
+        public void onCompleted() {
+            if (done) {
+                return;
+            }
+            try {
+                doOnEachObserver.onCompleted();
+            } catch (Throwable e) {
+                Exceptions.throwOrReport(e, this);
+                return;
+            }
+            // Set `done` here so that the error in `doOnEachObserver.onCompleted()` can be noticed by observer
+            done = true;
+            subscriber.onCompleted();
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            if (done) {
+                RxJavaHooks.onError(e);
+                return;
+            }
+            done = true;
+            try {
+                doOnEachObserver.onError(e);
+            } catch (Throwable e2) {
+                Exceptions.throwIfFatal(e2);
+                subscriber.onError(new CompositeException(Arrays.asList(e, e2)));
+                return;
+            }
+            subscriber.onError(e);
+        }
+
+        @Override
+        public void onNext(T value) {
+            if (done) {
+                return;
+            }
+            try {
+                doOnEachObserver.onNext(value);
+            } catch (Throwable e) {
+                Exceptions.throwOrReport(e, this, value);
+                return;
+            }
+            subscriber.onNext(value);
+        }        
+    }
+}

src/main/java/rx/internal/operators/OperatorDoOnEach.java

@@ -0,0 +1,51 @@
+/**
+ * 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 rx.internal.util;
+
+import rx.Observer;
+import rx.functions.*;
+
+/**
+ * An Observer that forwards the onXXX method calls to callbacks.
+ * @param <T> the value type
+ */
+public final class ActionObserver<T> implements Observer<T> {
+
+    final Action1<? super T> onNext;
+    final Action1<Throwable> onError;
+    final Action0 onCompleted;
+
+    public ActionObserver(Action1<? super T> onNext, Action1<Throwable> onError, Action0 onCompleted) {
+        this.onNext = onNext;
+        this.onError = onError;
+        this.onCompleted = onCompleted;
+    }
+
+    @Override
+    public void onNext(T t) {
+        onNext.call(t);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        onError.call(e);
+    }
+
+    @Override
+    public void onCompleted() {
+        onCompleted.call();
+    }
+}