2.x: Flowable as a Publisher to be fully RS compliant (#5112)

Author: akarnokd

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

@@ -23,8 +23,9 @@
 import io.reactivex.flowables.*;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
-import io.reactivex.internal.fuseable.ScalarCallable;
+import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.operators.flowable.*;
+import io.reactivex.internal.operators.flowable.FlowableStrict.StrictSubscriber;
 import io.reactivex.internal.operators.observable.ObservableFromPublisher;
 import io.reactivex.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.internal.subscribers.*;
@@ -1557,7 +1558,7 @@ public static <T> Flowable<T> concatEager(Publisher<? extends Publisher<? extend
     @SchedulerSupport(SchedulerSupport.NONE)
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public static <T> Flowable<T> concatEager(Publisher<? extends Publisher<? extends T>> sources, int maxConcurrency, int prefetch) {
-        return RxJavaPlugins.onAssembly(new FlowableConcatMapEager(sources, Functions.identity(), maxConcurrency, prefetch, ErrorMode.IMMEDIATE));
+        return RxJavaPlugins.onAssembly(new FlowableConcatMapEagerPublisher(sources, Functions.identity(), maxConcurrency, prefetch, ErrorMode.IMMEDIATE));
     }
 
     /**
@@ -11679,7 +11680,7 @@ public final Flowable<T> retryWhen(
     public final void safeSubscribe(Subscriber<? super T> s) {
         ObjectHelper.requireNonNull(s, "s is null");
         if (s instanceof SafeSubscriber) {
-            subscribe(s);
+            subscribe((SafeSubscriber<? super T>)s);
         } else {
             subscribe(new SafeSubscriber<T>(s));
         }
@@ -12713,13 +12714,15 @@ public final Flowable<T> startWithArray(T... items) {
      * </dl>
      * @return the new Flowable instance
      * @since 2.0.5 - experimental
+     * @deprecated 2.0.7, will be removed in 2.1.0; by default, the Publisher interface is always strict
      */
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     @SchedulerSupport(SchedulerSupport.NONE)
     @Experimental
     @CheckReturnValue
+    @Deprecated
     public final Flowable<T> strict() {
-        return RxJavaPlugins.onAssembly(new FlowableStrict<T>(this));
+        return this;
     }
 
     /**
@@ -12892,13 +12895,61 @@ public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super T
     @SchedulerSupport(SchedulerSupport.NONE)
     @Override
     public final void subscribe(Subscriber<? super T> s) {
+        if (s instanceof FlowableSubscriber) {
+            subscribe((FlowableSubscriber<? super T>)s);
+        } else {
+            ObjectHelper.requireNonNull(s, "s is null");
+            subscribe(new StrictSubscriber<T>(s));
+        }
+    }
+
+    /**
+     * Establish a connection between this Flowable and the given FlowableSubscriber and
+     * start streaming events based on the demand of the FlowableSubscriber.
+     * <p>
+     * This is a "factory method" and can be called multiple times, each time starting a new {@link Subscription}.
+     * <p>
+     * Each {@link Subscription} will work for only a single {@link FlowableSubscriber}.
+     * <p>
+     * If the same {@link FlowableSubscriber} instance is subscribed to multiple {@link Flowable}s and/or the
+     * same {@link Flowable} multiple times, it must ensure the serialization over its {@code onXXX}
+     * methods manually.
+     * <p>
+     * If the {@link Flowable} rejects the subscription attempt or otherwise fails it will signal
+     * the error via {@link FlowableSubscriber#onError(Throwable)}.
+     * <p>
+     * This subscribe method relaxes the following Reactive-Streams rules:
+     * <ul>
+     * <li>§1.3: onNext should not be called concurrently until onSubscribe returns.
+     *     <b>FlowableSubscriber.onSubscribe should make sure a sync or async call triggered by request() is safe.</b></li>
+     * <li>§2.3: onError or onComplete must not call cancel.
+     *     <b>Calling request() or cancel() is NOP at this point.</b></li>
+     * <li>§2.12: onSubscribe must be called at most once on the same instance.
+     *     <b>FlowableSubscriber reuse is not checked and if happens, it is the responsibility of
+     *     the FlowableSubscriber to ensure proper serialization of its onXXX methods.</b></li>
+     * <li>§3.9: negative requests should emit an onError(IllegalArgumentException).
+     *     <b>Non-positive requests signal via RxJavaPlugins.onError and the stream is not affected.</b></li>
+     * </ul>
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The backpressure behavior/expectation is determined by the supplied {@code FlowableSubscriber}.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param s the FlowableSubscriber that will consume signals from this Flowable
+     * @since 2.0.7 - experimental
+     */
+    @BackpressureSupport(BackpressureKind.SPECIAL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final void subscribe(FlowableSubscriber<? super T> s) {
         ObjectHelper.requireNonNull(s, "s is null");
         try {
-            s = RxJavaPlugins.onSubscribe(this, s);
+            Subscriber<? super T> z = RxJavaPlugins.onSubscribe(this, s);
 
-            ObjectHelper.requireNonNull(s, "Plugin returned null Subscriber");
+            ObjectHelper.requireNonNull(z, "Plugin returned null Subscriber");
 
-            subscribeActual(s);
+            subscribeActual(z);
         } catch (NullPointerException e) { // NOPMD
             throw e;
         } catch (Throwable e) {

src/main/java/io/reactivex/FlowableSubscriber.java

@@ -0,0 +1,41 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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;
+
+import org.reactivestreams.*;
+
+import io.reactivex.annotations.Experimental;
+
+/**
+ * Represents a Reactive-Streams inspired Subscriber that is RxJava 2 only
+ * and weakens rules §1.3 and §3.9 of the specification for gaining performance.
+ *
+ * @param <T> the value type
+ * @since 2.0.7 - experimental
+ */
+@Experimental
+public interface FlowableSubscriber<T> extends Subscriber<T> {
+
+    /**
+     * Implementors of this method should make sure everything that needs
+     * to be visible in {@link #onNext(Object)} is established before
+     * calling {@link Subscription#request(long)}. In practice this means
+     * no initialization should happen after the {@code request()} call and
+     * additional behavior is thread safe in respect to {@code onNext}.
+     *
+     * {@inheritDoc}
+     */
+    @Override
+    void onSubscribe(Subscription s);
+}

src/main/java/io/reactivex/Maybe.java

@@ -254,7 +254,7 @@ public static <T> Flowable<T> concat(Publisher<? extends MaybeSource<? extends T
     public static <T> Flowable<T> concat(Publisher<? extends MaybeSource<? extends T>> sources, int prefetch) {
         ObjectHelper.requireNonNull(sources, "sources is null");
         ObjectHelper.verifyPositive(prefetch, "prefetch");
-        return RxJavaPlugins.onAssembly(new FlowableConcatMap(sources, MaybeToPublisher.instance(), prefetch, ErrorMode.IMMEDIATE));
+        return RxJavaPlugins.onAssembly(new FlowableConcatMapPublisher(sources, MaybeToPublisher.instance(), prefetch, ErrorMode.IMMEDIATE));
     }
 
     /**
@@ -827,7 +827,7 @@ public static <T> Flowable<T> merge(Publisher<? extends MaybeSource<? extends T>
     @SchedulerSupport(SchedulerSupport.NONE)
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public static <T> Flowable<T> merge(Publisher<? extends MaybeSource<? extends T>> sources, int maxConcurrency) {
-        return RxJavaPlugins.onAssembly(new FlowableFlatMap(sources, MaybeToPublisher.instance(), false, maxConcurrency, Flowable.bufferSize()));
+        return RxJavaPlugins.onAssembly(new FlowableFlatMapPublisher(sources, MaybeToPublisher.instance(), false, maxConcurrency, Flowable.bufferSize()));
     }
 
     /**

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

@@ -188,8 +188,9 @@ public static <T> Flowable<T> concat(Publisher<? extends SingleSource<? extends
     @SchedulerSupport(SchedulerSupport.NONE)
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public static <T> Flowable<T> concat(Publisher<? extends SingleSource<? extends T>> sources, int prefetch) {
+        ObjectHelper.requireNonNull(sources, "sources is null");
         ObjectHelper.verifyPositive(prefetch, "prefetch");
-        return RxJavaPlugins.onAssembly(new FlowableConcatMap(sources, SingleInternalHelper.toFlowable(), prefetch, ErrorMode.IMMEDIATE));
+        return RxJavaPlugins.onAssembly(new FlowableConcatMapPublisher(sources, SingleInternalHelper.toFlowable(), prefetch, ErrorMode.IMMEDIATE));
     }
 
     /**
@@ -684,7 +685,7 @@ public static <T> Flowable<T> merge(Iterable<? extends SingleSource<? extends T>
     @SchedulerSupport(SchedulerSupport.NONE)
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public static <T> Flowable<T> merge(Publisher<? extends SingleSource<? extends T>> sources) {
-        return RxJavaPlugins.onAssembly(new FlowableFlatMap(sources, SingleInternalHelper.toFlowable(), false, Integer.MAX_VALUE, Flowable.bufferSize()));
+        return RxJavaPlugins.onAssembly(new FlowableFlatMapPublisher(sources, SingleInternalHelper.toFlowable(), false, Integer.MAX_VALUE, Flowable.bufferSize()));
     }
 
     /**

src/main/java/io/reactivex/internal/fuseable/ConditionalSubscriber.java

@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.fuseable;
 
-import org.reactivestreams.Subscriber;
+import io.reactivex.FlowableSubscriber;
 
 /**
  * A Subscriber with an additional onNextIf(T) method that
@@ -25,7 +25,7 @@
  *
  * @param <T> the value type
  */
-public interface ConditionalSubscriber<T> extends Subscriber<T> {
+public interface ConditionalSubscriber<T> extends FlowableSubscriber<T> {
     /**
      * Conditionally takes the value.
      * @param t the value to deliver

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

@@ -42,7 +42,7 @@ public void subscribeActual(CompletableObserver s) {
 
     static final class CompletableConcatSubscriber
     extends AtomicInteger
-    implements Subscriber<CompletableSource>, Disposable {
+    implements FlowableSubscriber<CompletableSource>, Disposable {
         private static final long serialVersionUID = 9032184911934499404L;
 
         final CompletableObserver actual;

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

@@ -32,7 +32,7 @@ protected void subscribeActual(final CompletableObserver cs) {
         flowable.subscribe(new FromPublisherSubscriber<T>(cs));
     }
 
-    static final class FromPublisherSubscriber<T> implements Subscriber<T>, Disposable {
+    static final class FromPublisherSubscriber<T> implements FlowableSubscriber<T>, Disposable {
 
         final CompletableObserver cs;
 

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

@@ -43,7 +43,7 @@ public void subscribeActual(CompletableObserver s) {
 
     static final class CompletableMergeSubscriber
     extends AtomicInteger
-    implements Subscriber<CompletableSource>, Disposable {
+    implements FlowableSubscriber<CompletableSource>, Disposable {
 
         private static final long serialVersionUID = -2108443387387077490L;
 

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

@@ -31,14 +31,14 @@ abstract class AbstractFlowableWithUpstream<T, R> extends Flowable<R> implements
     /**
      * The upstream source Publisher.
      */
-    protected final Publisher<T> source;
+    protected final Flowable<T> source;
 
     /**
      * Constructs a FlowableSource wrapping the given non-null (verified)
      * source Publisher.
      * @param source the source (upstream) Publisher instance, not null (verified)
      */
-    AbstractFlowableWithUpstream(Publisher<T> source) {
+    AbstractFlowableWithUpstream(Flowable<T> source) {
         this.source = ObjectHelper.requireNonNull(source, "source is null");
     }
 

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

@@ -17,20 +17,21 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.*;
 
-import org.reactivestreams.*;
+import org.reactivestreams.Subscription;
 
+import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.MissingBackpressureException;
 import io.reactivex.internal.queue.SpscArrayQueue;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.*;
 
 public final class BlockingFlowableIterable<T> implements Iterable<T> {
-    final Publisher<? extends T> source;
+    final Flowable<? extends T> source;
 
     final int bufferSize;
 
-    public BlockingFlowableIterable(Publisher<? extends T> source, int bufferSize) {
+    public BlockingFlowableIterable(Flowable<? extends T> source, int bufferSize) {
         this.source = source;
         this.bufferSize = bufferSize;
     }
@@ -44,7 +45,7 @@ public Iterator<T> iterator() {
 
     static final class BlockingFlowableIterator<T>
     extends AtomicReference<Subscription>
-    implements Subscriber<T>, Iterator<T>, Runnable, Disposable {
+    implements FlowableSubscriber<T>, Iterator<T>, Runnable, Disposable {
 
         private static final long serialVersionUID = 6695226475494099826L;