Observable direct rest (ReactiveX#4145)

* Switch the remaining Observable operators to direct impl.

* Last set of operators, test renames

Author: akarnokd

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

@@ -12,39 +12,39 @@
  */
 package io.reactivex.internal.operators.observable;
 
-import io.reactivex.Observable.NbpOperator;
-import io.reactivex.Observer;
+import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.*;
 import io.reactivex.internal.disposables.*;
-import io.reactivex.internal.subscribers.observable.NbpEmptySubscriber;
 
-public final class NbpOperatorCollect<T, U> implements NbpOperator<U, T> {
+public final class ObservableCollect<T, U> extends ObservableSource<T, U> {
     final Supplier<? extends U> initialSupplier;
     final BiConsumer<? super U, ? super T> collector;
 
-    public NbpOperatorCollect(Supplier<? extends U> initialSupplier, BiConsumer<? super U, ? super T> collector) {
+    public ObservableCollect(ObservableConsumable<T> source, 
+            Supplier<? extends U> initialSupplier, BiConsumer<? super U, ? super T> collector) {
+        super(source);
         this.initialSupplier = initialSupplier;
         this.collector = collector;
     }
 
     @Override
-    public Observer<? super T> apply(Observer<? super U> t) {
-        
+    protected void subscribeActual(Observer<? super U> t) {
         U u;
         try {
             u = initialSupplier.get();
         } catch (Throwable e) {
             EmptyDisposable.error(e, t);
-            return NbpEmptySubscriber.DISPOSED;
+            return;
         }
 
         if (u == null) {
             EmptyDisposable.error(new NullPointerException("The inital supplier returned a null value"), t);
-            return NbpEmptySubscriber.DISPOSED;
+            return;
         }
 
-        return new CollectSubscriber<T, U>(t, u, collector);
+        source.subscribe(new CollectSubscriber<T, U>(t, u, collector));
+        
     }
 
     static final class CollectSubscriber<T, U> implements Observer<T> {

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

@@ -16,27 +16,27 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import io.reactivex.*;
-import io.reactivex.Observable.NbpOperator;
 import io.reactivex.disposables.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.queue.SpscLinkedArrayQueue;
 import io.reactivex.observers.SerializedObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class NbpOperatorConcatMap<T, U> implements NbpOperator<U, T> {
+public final class ObservableConcatMap<T, U> extends ObservableSource<T, U> {
     final Function<? super T, ? extends ObservableConsumable<? extends U>> mapper;
     final int bufferSize;
-    public NbpOperatorConcatMap(Function<? super T, ? extends ObservableConsumable<? extends U>> mapper, int bufferSize) {
+    public ObservableConcatMap(ObservableConsumable<T> source, Function<? super T, ? extends ObservableConsumable<? extends U>> mapper, int bufferSize) {
+        super(source);
         this.mapper = mapper;
         this.bufferSize = Math.max(8, bufferSize);
     }
     @Override
-    public Observer<? super T> apply(Observer<? super U> s) {
+    public void subscribeActual(Observer<? super U> s) {
         SerializedObserver<U> ssub = new SerializedObserver<U>(s);
         SerialDisposable sa = new SerialDisposable();
         ssub.onSubscribe(sa);
-        return new SourceSubscriber<T, U>(ssub, sa, mapper, bufferSize);
+        source.subscribe(new SourceSubscriber<T, U>(ssub, sa, mapper, bufferSize));
     }
 
     static final class SourceSubscriber<T, U> extends AtomicInteger implements Observer<T> {

…ators/observable/NbpOperatorCollect.java …rators/observable/ObservableCollect.javasrc/main/java/io/reactivex/internal/operators/observable/NbpOperatorCollect.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableCollect.java src/main/java/io/reactivex/internal/operators/observable/NbpOperatorCollect.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableCollect.java

@@ -13,22 +13,18 @@
 
 package io.reactivex.internal.operators.observable;
 
-import io.reactivex.Observable.NbpOperator;
-import io.reactivex.Observer;
+import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 
-public enum NbpOperatorCount implements NbpOperator<Long, Object> {
-    INSTANCE;
-    
-    @SuppressWarnings("unchecked")
-    public static <T> NbpOperator<Long, T> instance() {
-        return (NbpOperator<Long, T>)INSTANCE;
+public final class ObservableCount<T> extends ObservableSource<T, Long> {
+    public ObservableCount(ObservableConsumable<T> source) {
+        super(source);
     }
-    
+
     @Override
-    public Observer<? super Object> apply(Observer<? super Long> t) {
-        return new CountSubscriber(t);
+    public void subscribeActual(Observer<? super Long> t) {
+        source.subscribe(new CountSubscriber(t));
     }
 
     static final class CountSubscriber implements Observer<Object> {

…ors/observable/NbpOperatorConcatMap.java …tors/observable/ObservableConcatMap.javasrc/main/java/io/reactivex/internal/operators/observable/NbpOperatorConcatMap.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java src/main/java/io/reactivex/internal/operators/observable/NbpOperatorConcatMap.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java

@@ -16,24 +16,24 @@
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.*;
-import io.reactivex.Observable.NbpOperator;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.subscribers.observable.DisposableObserver;
 import io.reactivex.observers.SerializedObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class NbpOperatorDebounce<T, U> implements NbpOperator<T, T> {
+public final class ObservableDebounce<T, U> extends ObservableSource<T, T> {
     final Function<? super T, ? extends ObservableConsumable<U>> debounceSelector;
 
-    public NbpOperatorDebounce(Function<? super T, ? extends ObservableConsumable<U>> debounceSelector) {
+    public ObservableDebounce(ObservableConsumable<T> source, Function<? super T, ? extends ObservableConsumable<U>> debounceSelector) {
+        super(source);
         this.debounceSelector = debounceSelector;
     }
 
     @Override
-    public Observer<? super T> apply(Observer<? super T> t) {
-        return new DebounceSubscriber<T, U>(new SerializedObserver<T>(t), debounceSelector);
+    public void subscribeActual(Observer<? super T> t) {
+        source.subscribe(new DebounceSubscriber<T, U>(new SerializedObserver<T>(t), debounceSelector));
     }
 
     static final class DebounceSubscriber<T, U> 

…erators/observable/NbpOperatorCount.java …perators/observable/ObservableCount.javasrc/main/java/io/reactivex/internal/operators/observable/NbpOperatorCount.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableCount.java src/main/java/io/reactivex/internal/operators/observable/NbpOperatorCount.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableCount.java

@@ -17,29 +17,29 @@
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.*;
-import io.reactivex.Observable.NbpOperator;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.observers.SerializedObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public final class NbpOperatorDebounceTimed<T> implements NbpOperator<T, T> {
+public final class ObservableDebounceTimed<T> extends ObservableSource<T, T> {
     final long timeout;
     final TimeUnit unit;
     final Scheduler scheduler;
 
-    public NbpOperatorDebounceTimed(long timeout, TimeUnit unit, Scheduler scheduler) {
+    public ObservableDebounceTimed(ObservableConsumable<T> source, long timeout, TimeUnit unit, Scheduler scheduler) {
+        super(source);
         this.timeout = timeout;
         this.unit = unit;
         this.scheduler = scheduler;
     }
 
     @Override
-    public Observer<? super T> apply(Observer<? super T> t) {
-        return new DebounceTimedSubscriber<T>(
+    public void subscribeActual(Observer<? super T> t) {
+        source.subscribe(new DebounceTimedSubscriber<T>(
                 new SerializedObserver<T>(t), 
-                timeout, unit, scheduler.createWorker());
+                timeout, unit, scheduler.createWorker()));
     }
 
     static final class DebounceTimedSubscriber<T>

…tors/observable/NbpOperatorDebounce.java …ators/observable/ObservableDebounce.javasrc/main/java/io/reactivex/internal/operators/observable/NbpOperatorDebounce.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableDebounce.java src/main/java/io/reactivex/internal/operators/observable/NbpOperatorDebounce.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableDebounce.java

@@ -16,30 +16,28 @@
 import java.util.concurrent.TimeUnit;
 
 import io.reactivex.*;
-import io.reactivex.Observable.NbpOperator;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.observers.SerializedObserver;
 
-public final class NbpOperatorDelay<T> implements NbpOperator<T, T> {
+public final class ObservableDelay<T> extends ObservableSource<T, T> {
     final long delay;
     final TimeUnit unit;
     final Scheduler scheduler;
     final boolean delayError;
 
-    public NbpOperatorDelay(long delay, TimeUnit unit, Scheduler scheduler, boolean delayError) {
+    public ObservableDelay(ObservableConsumable<T> source, long delay, TimeUnit unit, Scheduler scheduler, boolean delayError) {
+        super(source);
         this.delay = delay;
         this.unit = unit;
         this.scheduler = scheduler;
         this.delayError = delayError;
     }
 
-
-
     @Override
     @SuppressWarnings("unchecked")
-    public Observer<? super T> apply(Observer<? super T> t) {
+    public void subscribeActual(Observer<? super T> t) {
         Observer<T> s;
         if (delayError) {
             s = (Observer<T>)t;
@@ -49,7 +47,7 @@ public Observer<? super T> apply(Observer<? super T> t) {
 
         Scheduler.Worker w = scheduler.createWorker();
 
-        return new DelaySubscriber<T>(s, delay, unit, w, delayError);
+        source.subscribe(new DelaySubscriber<T>(s, delay, unit, w, delayError));
     }
 
     static final class DelaySubscriber<T> implements Observer<T>, Disposable {

…observable/NbpOperatorDebounceTimed.java …/observable/ObservableDebounceTimed.javasrc/main/java/io/reactivex/internal/operators/observable/NbpOperatorDebounceTimed.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableDebounceTimed.java src/main/java/io/reactivex/internal/operators/observable/NbpOperatorDebounceTimed.java renamed to src/main/java/io/reactivex/internal/operators/observable/ObservableDebounceTimed.java

@@ -14,22 +14,19 @@
 package io.reactivex.internal.operators.observable;
 
 import io.reactivex.*;
-import io.reactivex.Observable.NbpOperator;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
-public enum NbpOperatorDematerialize implements NbpOperator<Object, Try<Optional<Object>>> {
-    INSTANCE;
-
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    public static <T> NbpOperator<T, Try<Optional<T>>> instance() {
-        return (NbpOperator)INSTANCE;
-    }
+public final class ObservableDematerialize<T> extends ObservableSource<Try<Optional<T>>, T> {
 
+    public ObservableDematerialize(ObservableConsumable<Try<Optional<T>>> source) {
+        super(source);
+    }
+
     @Override
-    public Observer<? super Try<Optional<Object>>> apply(Observer<? super Object> t) {
-        return new DematerializeSubscriber<Object>(t);
+    public void subscribeActual(Observer<? super T> t) {
+        source.subscribe(new DematerializeSubscriber<T>(t));
     }
 
     static final class DematerializeSubscriber<T> implements Observer<Try<Optional<T>>> {