2.x: add assembly tracking, minor fixes and cleanup (#4417)

* 2.x: add assembly tracking, minor fixes and cleanup

* Add missing header, add more time to test

Author: akarnokd

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

@@ -20,6 +20,7 @@
 import io.reactivex.functions.Consumer;
 import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.operators.flowable.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 /**
  * A {@code ConnectableObservable} resembles an ordinary {@link Flowable}, except that it does not begin
@@ -75,7 +76,7 @@ public void accept(Disposable d) {
      * @see <a href="http://reactivex.io/documentation/operators/refcount.html">ReactiveX documentation: RefCount</a>
      */
     public Flowable<T> refCount() {
-        return new FlowableRefCount<T>(this);
+        return RxJavaPlugins.onAssembly(new FlowableRefCount<T>(this));
     }
 
     /**
@@ -119,8 +120,8 @@ public Flowable<T> autoConnect(int numberOfSubscribers) {
     public Flowable<T> autoConnect(int numberOfSubscribers, Consumer<? super Disposable> connection) {
         if (numberOfSubscribers <= 0) {
             this.connect(connection);
-            return this;
+            return RxJavaPlugins.onAssembly(this);
         }
-        return new FlowableAutoConnect<T>(this, numberOfSubscribers, connection);
+        return RxJavaPlugins.onAssembly(new FlowableAutoConnect<T>(this, numberOfSubscribers, connection));
     }
 }

src/main/java/io/reactivex/Flowable.java

@@ -22,6 +22,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 /**
  * An observable which auto-connects to another observable, caches the elements
@@ -35,28 +36,28 @@ public final class FlowableCache<T> extends AbstractFlowableWithUpstream<T, T> {
 
     final AtomicBoolean once;
     /**
-     * Creates a cached Observable with a default capacity hint of 16.
+     * Creates a cached Flowable with a default capacity hint of 16.
      * @param <T> the value type
      * @param source the source Observable to cache
      * @return the CachedObservable instance
      */
-    public static <T> FlowableCache<T> from(Flowable<T> source) {
+    public static <T> Flowable<T> from(Flowable<T> source) {
         return from(source, 16);
     }
 
     /**
-     * Creates a cached Observable with the given capacity hint.
+     * Creates a cached Flowable with the given capacity hint.
      * @param <T> the value type
      * @param source the source Observable to cache
      * @param capacityHint the hint for the internal buffer size
      * @return the CachedObservable instance
      */
-    public static <T> FlowableCache<T> from(Flowable<T> source, int capacityHint) {
+    public static <T> Flowable<T> from(Flowable<T> source, int capacityHint) {
         if (capacityHint < 1) {
             throw new IllegalArgumentException("capacityHint > 0 required");
         }
         CacheState<T> state = new CacheState<T>(source, capacityHint);
-        return new FlowableCache<T>(source, state);
+        return RxJavaPlugins.onAssembly(new FlowableCache<T>(source, state));
     }
 
     /**

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

@@ -18,10 +18,12 @@
 
 import org.reactivestreams.*;
 
+import io.reactivex.Flowable;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
 import io.reactivex.internal.subscriptions.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class FlowableDistinct<T, K> extends AbstractFlowableWithUpstream<T, T> {
     final Function<? super T, K> keySelector;
@@ -33,7 +35,7 @@ public FlowableDistinct(Publisher<T> source, Function<? super T, K> keySelector,
         this.keySelector = keySelector;
     }
 
-    public static <T, K> FlowableDistinct<T, K> withCollection(Publisher<T> source, Function<? super T, K> keySelector, final Callable<? extends Collection<? super K>> collectionSupplier) {
+    public static <T, K> Flowable<T> withCollection(Publisher<T> source, Function<? super T, K> keySelector, final Callable<? extends Collection<? super K>> collectionSupplier) {
         Callable<? extends Predicate<? super K>> p = new Callable<Predicate<K>>() {
             @Override
             public Predicate<K> call() throws Exception {
@@ -52,10 +54,10 @@ public boolean test(K t) {
             }
         };
 
-        return new FlowableDistinct<T, K>(source, keySelector, p);
+        return RxJavaPlugins.onAssembly(new FlowableDistinct<T, K>(source, keySelector, p));
     }
 
-    public static <T> FlowableDistinct<T, T> untilChanged(Publisher<T> source) {
+    public static <T> Flowable<T> untilChanged(Publisher<T> source) {
         Callable<? extends Predicate<? super T>> p = new Callable<Predicate<T>>() {
             Object last;
             @Override
@@ -75,10 +77,10 @@ public boolean test(T t) {
                 };
             }
         };
-        return new FlowableDistinct<T, T>(source, Functions.<T>identity(), p);
+        return RxJavaPlugins.onAssembly(new FlowableDistinct<T, T>(source, Functions.<T>identity(), p));
     }
 
-    public static <T, K> FlowableDistinct<T, K> untilChanged(Publisher<T> source, Function<? super T, K> keySelector) {
+    public static <T, K> Flowable<T> untilChanged(Publisher<T> source, Function<? super T, K> keySelector) {
         Callable<? extends Predicate<? super K>> p = new Callable<Predicate<K>>() {
             Object last;
             @Override
@@ -98,7 +100,7 @@ public boolean test(K t) {
                 };
             }
         };
-        return new FlowableDistinct<T, K>(source, keySelector, p);
+        return RxJavaPlugins.onAssembly(new FlowableDistinct<T, K>(source, keySelector, p));
     }
 
     @Override

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

@@ -117,7 +117,7 @@ public void subscribe(Subscriber<? super T> child) {
                 }
             }
         };
-        return new FlowablePublish<T>(onSubscribe, source, curr, bufferSize);
+        return RxJavaPlugins.onAssembly(new FlowablePublish<T>(onSubscribe, source, curr, bufferSize));
     }
 
     private FlowablePublish(Publisher<T> onSubscribe, Publisher<T> source,

src/main/java/io/reactivex/flowables/ConnectableFlowable.java

@@ -39,7 +39,7 @@ public void subscribeActual(Subscriber<? super T> s) {
 
         SerializedSubscriber<T> z = new SerializedSubscriber<T>(s);
 
-        FlowProcessor<Object> processor = new UnicastProcessor<Object>(8).toSerialized();
+        FlowableProcessor<Object> processor = new UnicastProcessor<Object>(8).toSerialized();
 
         Publisher<?> when;
 
@@ -137,13 +137,13 @@ static abstract class WhenSourceSubscriber<T, U> extends SubscriptionArbiter imp
 
         protected final Subscriber<? super T> actual;
 
-        protected final FlowProcessor<U> processor;
+        protected final FlowableProcessor<U> processor;
 
         protected final Subscription receiver;
 
         private long produced;
 
-        public WhenSourceSubscriber(Subscriber<? super T> actual, FlowProcessor<U> processor,
+        public WhenSourceSubscriber(Subscriber<? super T> actual, FlowableProcessor<U> processor,
                 Subscription receiver) {
             this.actual = actual;
             this.processor = processor;
@@ -183,7 +183,7 @@ static final class RepeatWhenSubscriber<T> extends WhenSourceSubscriber<T, Objec
         /** */
         private static final long serialVersionUID = -2680129890138081029L;
 
-        public RepeatWhenSubscriber(Subscriber<? super T> actual, FlowProcessor<Object> processor,
+        public RepeatWhenSubscriber(Subscriber<? super T> actual, FlowableProcessor<Object> processor,
                 Subscription receiver) {
             super(actual, processor, receiver);
         }

src/main/java/io/reactivex/internal/operators/flowable/FlowableCache.java

@@ -113,7 +113,7 @@ public void accept(Disposable r) {
      */
     public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> co, final Scheduler scheduler) {
         final Flowable<T> observable = co.observeOn(scheduler);
-        return new ConnectableFlowable<T>() {
+        return RxJavaPlugins.onAssembly(new ConnectableFlowable<T>() {
             @Override
             public void connect(Consumer<? super Disposable> connection) {
                 co.connect(connection);
@@ -123,7 +123,7 @@ public void connect(Consumer<? super Disposable> connection) {
             protected void subscribeActual(Subscriber<? super T> s) {
                 observable.subscribe(s);
             }
-        };
+        });
     }
 
     /**
@@ -248,7 +248,7 @@ public void subscribe(Subscriber<? super T> child) {
                 }
             }
         };
-        return new FlowableReplay<T>(onSubscribe, source, curr, bufferFactory);
+        return RxJavaPlugins.onAssembly(new FlowableReplay<T>(onSubscribe, source, curr, bufferFactory));
     }
 
     private FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,