2.x: introduce op-fusion to Observable + disposable-chaining fixes

Author: akarnokd

src/main/java/io/reactivex/Observable.java

@@ -1659,7 +1659,7 @@ public final Observable<T> doOnLifecycle(final Consumer<? super Disposable> onSu
         return lift(new NbpOperator<T, T>() {
             @Override
             public Observer<? super T> apply(Observer<? super T> s) {
-                return new NbpSubscriptionLambdaSubscriber<T>(s, onSubscribe, onCancel);
+                return new SubscriptionLambdaObserver<T>(s, onSubscribe, onCancel);
             }
         });
     }
@@ -2626,7 +2626,7 @@ public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super T
         Objects.requireNonNull(onComplete, "onComplete is null");
         Objects.requireNonNull(onSubscribe, "onSubscribe is null");
 
-        NbpLambdaSubscriber<T> ls = new NbpLambdaSubscriber<T>(onNext, onError, onComplete, onSubscribe);
+        LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
 
         unsafeSubscribe(ls);
 

src/main/java/io/reactivex/internal/operators/observable/ObservableAll.java

@@ -32,7 +32,7 @@ protected void subscribeActual(Observer<? super Boolean> t) {
         source.subscribe(new AllSubscriber<T>(t, predicate));
     }
 
-    static final class AllSubscriber<T> implements Observer<T> {
+    static final class AllSubscriber<T> implements Observer<T>, Disposable {
         final Observer<? super Boolean> actual;
         final Predicate<? super T> predicate;
 
@@ -48,7 +48,7 @@ public AllSubscriber(Observer<? super Boolean> actual, Predicate<? super T> pred
         public void onSubscribe(Disposable s) {
             if (DisposableHelper.validate(this.s, s)) {
                 this.s = s;
-                actual.onSubscribe(s);
+                actual.onSubscribe(this);
             }
         }
 
@@ -93,5 +93,15 @@ public void onComplete() {
             actual.onNext(true);
             actual.onComplete();
         }
+        
+        @Override
+        public void dispose() {
+            s.dispose();
+        }
+        
+        @Override
+        public boolean isDisposed() {
+            return s.isDisposed();
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/observable/ObservableAny.java

@@ -30,7 +30,7 @@ protected void subscribeActual(Observer<? super Boolean> t) {
         source.subscribe(new AnySubscriber<T>(t, predicate));
     }
 
-    static final class AnySubscriber<T> implements Observer<T> {
+    static final class AnySubscriber<T> implements Observer<T>, Disposable {
 
         final Observer<? super Boolean> actual;
         final Predicate<? super T> predicate;
@@ -47,7 +47,7 @@ public AnySubscriber(Observer<? super Boolean> actual, Predicate<? super T> pred
         public void onSubscribe(Disposable s) {
             if (DisposableHelper.validate(this.s, s)) {
                 this.s = s;
-                actual.onSubscribe(s);
+                actual.onSubscribe(this);
             }
         }
 
@@ -89,5 +89,15 @@ public void onComplete() {
                 actual.onComplete();
             }
         }
+
+        @Override
+        public void dispose() {
+            s.dispose();
+        }
+        
+        @Override
+        public boolean isDisposed() {
+            return s.isDisposed();
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/observable/ObservableBuffer.java

@@ -48,7 +48,7 @@ protected void subscribeActual(Observer<? super U> t) {
         }
     }
 
-    static final class BufferExactSubscriber<T, U extends Collection<? super T>> implements Observer<T> {
+    static final class BufferExactSubscriber<T, U extends Collection<? super T>> implements Observer<T>, Disposable {
         final Observer<? super U> actual;
         final int count;
         final Supplier<U> bufferSupplier;
@@ -98,10 +98,20 @@ boolean createBuffer() {
         public void onSubscribe(Disposable s) {
             if (DisposableHelper.validate(this.s, s)) {
                 this.s = s;
-                actual.onSubscribe(s);
+                actual.onSubscribe(this);
             }
         }
+
+        @Override
+        public void dispose() {
+            s.dispose();
+        }
 
+        @Override
+        public boolean isDisposed() {
+            return s.isDisposed();
+        }
+
         @Override
         public void onNext(T t) {
             U b = buffer;
@@ -137,7 +147,7 @@ public void onComplete() {
     }
 
     static final class BufferSkipSubscriber<T, U extends Collection<? super T>> 
-    extends AtomicBoolean implements Observer<T> {
+    extends AtomicBoolean implements Observer<T>, Disposable {
         /** */
         private static final long serialVersionUID = -8223395059921494546L;
         final Observer<? super U> actual;
@@ -163,10 +173,21 @@ public BufferSkipSubscriber(Observer<? super U> actual, int count, int skip, Sup
         public void onSubscribe(Disposable s) {
             if (DisposableHelper.validate(this.s, s)) {
                 this.s = s;
-                actual.onSubscribe(s);
+                actual.onSubscribe(this);
             }
         }
 
+
+        @Override
+        public void dispose() {
+            s.dispose();
+        }
+        
+        @Override
+        public boolean isDisposed() {
+            return s.isDisposed();
+        }
+
         @Override
         public void onNext(T t) {
             if (index++ % skip == 0) {

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundary.java

@@ -52,7 +52,7 @@ protected void subscribeActual(Observer<? super U> t) {
     }
 
     static final class BufferBoundarySubscriber<T, U extends Collection<? super T>, Open, Close>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Disposable {
+    extends QueueDrainObserver<T, U, U> implements Disposable {
         final ObservableConsumable<? extends Open> bufferOpen;
         final Function<? super Open, ? extends ObservableConsumable<? extends Close>> bufferClose;
         final Supplier<U> bufferSupplier;

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundarySupplier.java

@@ -44,7 +44,7 @@ protected void subscribeActual(Observer<? super U> t) {
     }
 
     static final class BufferBondarySupplierSubscriber<T, U extends Collection<? super T>, B>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Observer<T>, Disposable {
+    extends QueueDrainObserver<T, U, U> implements Observer<T>, Disposable {
         /** */
         final Supplier<U> bufferSupplier;
         final Supplier<? extends ObservableConsumable<B>> boundarySupplier;

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferExactBoundary.java

@@ -42,7 +42,7 @@ protected void subscribeActual(Observer<? super U> t) {
     }
 
     static final class BufferExactBondarySubscriber<T, U extends Collection<? super T>, B>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Observer<T>, Disposable {
+    extends QueueDrainObserver<T, U, U> implements Observer<T>, Disposable {
         /** */
         final Supplier<U> bufferSupplier;
         final ObservableConsumable<B> boundary;

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferTimed.java

@@ -25,7 +25,7 @@
 import io.reactivex.functions.Supplier;
 import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.queue.MpscLinkedQueue;
-import io.reactivex.internal.subscribers.observable.NbpQueueDrainSubscriber;
+import io.reactivex.internal.subscribers.observable.QueueDrainObserver;
 import io.reactivex.internal.util.QueueDrainHelper;
 import io.reactivex.observers.SerializedObserver;
 
@@ -80,7 +80,7 @@ protected void subscribeActual(Observer<? super U> t) {
     }
 
     static final class BufferExactUnboundedSubscriber<T, U extends Collection<? super T>>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Runnable, Disposable {
+    extends QueueDrainObserver<T, U, U> implements Runnable, Disposable {
         final Supplier<U> bufferSupplier;
         final long timespan;
         final TimeUnit unit;
@@ -242,7 +242,7 @@ public void accept(Observer<? super U> a, U v) {
     }
 
     static final class BufferSkipBoundedSubscriber<T, U extends Collection<? super T>>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Runnable, Disposable {
+    extends QueueDrainObserver<T, U, U> implements Runnable, Disposable {
         final Supplier<U> bufferSupplier;
         final long timespan;
         final long timeskip;
@@ -408,7 +408,7 @@ public void accept(Observer<? super U> a, U v) {
     }
 
     static final class BufferExactBoundedSubscriber<T, U extends Collection<? super T>>
-    extends NbpQueueDrainSubscriber<T, U, U> implements Runnable, Disposable {
+    extends QueueDrainObserver<T, U, U> implements Runnable, Disposable {
         final Supplier<U> bufferSupplier;
         final long timespan;
         final TimeUnit unit;

src/main/java/io/reactivex/internal/operators/observable/ObservableCollect.java

@@ -47,7 +47,7 @@ protected void subscribeActual(Observer<? super U> t) {
 
     }
 
-    static final class CollectSubscriber<T, U> implements Observer<T> {
+    static final class CollectSubscriber<T, U> implements Observer<T>, Disposable {
         final Observer<? super U> actual;
         final BiConsumer<? super U, ? super T> collector;
         final U u;
@@ -64,10 +64,22 @@ public CollectSubscriber(Observer<? super U> actual, U u, BiConsumer<? super U,
         public void onSubscribe(Disposable s) {
             if (DisposableHelper.validate(this.s, s)) {
                 this.s = s;
-                actual.onSubscribe(s);
+                actual.onSubscribe(this);
             }
         }
 
+
+        @Override
+        public void dispose() {
+            s.dispose();
+        }
+        
+        @Override
+        public boolean isDisposed() {
+            return s.isDisposed();
+        }
+
+        
         @Override
         public void onNext(T t) {
             try {