2.x: add strict() operator for strong RS conformance

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

@@ -12509,6 +12509,35 @@ public final Flowable<T> startWithArray(T... items) {
         return concatArray(fromArray, this);
     }
 
+    /**
+     * Ensures that the event flow between the upstream and downstream follow
+     * the Reactive-Streams 1.0 specification by honoring the 3 additional rules
+     * (which are omitted in standard operators due to performance reasons).
+     * <ul>
+     * <li>§1.3: onNext should not be called concurrently until onSubscribe returns</li>
+     * <li>§2.3: onError or onComplete must not call cancel</li>
+     * <li>§3.9: negative requests should emit an onError(IllegalArgumentException)</li>
+     * </ul>
+     * In addition, if rule §2.12 (onSubscribe must be called at most once) is violated,
+     * the sequence is cancelled an onError(IllegalStateException) is emitted.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The operator doesn't interfere with backpressure which is determined by the source {@code Publisher}'s backpressure
+     *  behavior.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code strict} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return the new Flowable instance
+     * @since 2.0.5 - experimental
+     */
+    @BackpressureSupport(BackpressureKind.PASS_THROUGH)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    @CheckReturnValue
+    public final Flowable<T> strict() {
+        return RxJavaPlugins.onAssembly(new FlowableStrict<T>(this));
+    }
+
     /**
      * Subscribes to a Publisher and ignores {@code onNext} and {@code onComplete} emissions.
      * <p>

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

@@ -0,0 +1,122 @@
+/**
+ * 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 io.reactivex.internal.operators.flowable;
+
+import java.util.concurrent.atomic.*;
+
+import org.reactivestreams.*;
+
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.*;
+
+/**
+ * Ensures that the event flow between the upstream and downstream follow
+ * the Reactive-Streams 1.0 specification by honoring the 3 additional rules
+ * (which are omitted in standard operators due to performance reasons).
+ * <ul>
+ * <li>§1.3: onNext should not be called concurrently until onSubscribe returns</li>
+ * <li>§2.3: onError or onComplete must not call cancel</li>
+ * <li>§3.9: negative requests should emit an onError(IllegalArgumentException)</li>
+ * </ul>
+ * In addition, if rule §2.12 (onSubscribe must be called at most once) is violated,
+ * the sequence is cancelled an onError(IllegalStateException) is emitted.
+ * @param <T> the value type
+ */
+public final class FlowableStrict<T> extends AbstractFlowableWithUpstream<T, T> {
+
+    public FlowableStrict(Publisher<T> source) {
+        super(source);
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super T> s) {
+        source.subscribe(new StrictSubscriber<T>(s));
+    }
+
+    static final class StrictSubscriber<T>
+    extends AtomicInteger
+    implements Subscriber<T>, Subscription {
+
+        private static final long serialVersionUID = -4945028590049415624L;
+
+        final Subscriber<? super T> actual;
+
+        final AtomicThrowable error;
+
+        final AtomicLong requested;
+
+        final AtomicReference<Subscription> s;
+
+        final AtomicBoolean once;
+
+        volatile boolean done;
+
+        StrictSubscriber(Subscriber<? super T> actual) {
+            this.actual = actual;
+            this.error = new AtomicThrowable();
+            this.requested = new AtomicLong();
+            this.s = new AtomicReference<Subscription>();
+            this.once = new AtomicBoolean();
+        }
+
+        @Override
+        public void request(long n) {
+            if (n <= 0) {
+                cancel();
+                onError(new IllegalArgumentException("§3.9 violated: positive request amount required but it was " + n));
+            } else {
+                SubscriptionHelper.deferredRequest(s, requested, n);
+            }
+        }
+
+        @Override
+        public void cancel() {
+            if (!done) {
+                SubscriptionHelper.cancel(s);
+            }
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (once.compareAndSet(false, true)) {
+
+                actual.onSubscribe(this);
+
+                SubscriptionHelper.deferredSetOnce(this.s, requested, s);
+            } else {
+                s.cancel();
+                cancel();
+                onError(new IllegalStateException("§2.12 violated: onSubscribe must be called at most once"));
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            HalfSerializer.onNext(actual, t, this, error);
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            done = true;
+            HalfSerializer.onError(actual, t, this, error);
+        }
+
+        @Override
+        public void onComplete() {
+            done = true;
+            HalfSerializer.onComplete(actual, this, error);
+        }
+    }
+
+}

src/test/java/io/reactivex/internal/operators/flowable/FlowableStrictTest.java

@@ -0,0 +1,236 @@
+/**
+ * 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 io.reactivex.internal.operators.flowable;
+
+import static org.junit.Assert.*;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+import org.reactivestreams.*;
+
+import io.reactivex.Flowable;
+import io.reactivex.exceptions.TestException;
+import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subscribers.TestSubscriber;
+
+public class FlowableStrictTest {
+
+    @Test
+    public void empty() {
+        Flowable.empty()
+        .strict()
+        .test()
+        .assertResult();
+    }
+
+    @Test
+    public void just() {
+        Flowable.just(1)
+        .strict()
+        .test()
+        .assertResult(1);
+    }
+
+    @Test
+    public void range() {
+        Flowable.range(1, 5)
+        .strict()
+        .test()
+        .assertResult(1, 2, 3, 4, 5);
+    }
+
+    @Test
+    public void take() {
+        Flowable.range(1, 5)
+        .take(2)
+        .strict()
+        .test()
+        .assertResult(1, 2);
+    }
+
+    @Test
+    public void backpressure() {
+        Flowable.range(1, 5)
+        .strict()
+        .test(0)
+        .assertEmpty()
+        .requestMore(1)
+        .assertValue(1)
+        .requestMore(2)
+        .assertValues(1, 2, 3)
+        .requestMore(2)
+        .assertResult(1, 2, 3, 4, 5);
+    }
+
+    @Test
+    public void error() {
+        Flowable.error(new TestException())
+        .strict()
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void observeOn() {
+        Flowable.range(1, 5)
+        .hide()
+        .observeOn(Schedulers.single())
+        .strict()
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(1, 2, 3, 4, 5);
+    }
+
+    @Test
+    public void invalidRequest() {
+        for (int i = 0; i > -100; i--) {
+            final int j = i;
+            final List<Object> items = new ArrayList<Object>();
+
+            Flowable.range(1, 2)
+            .strict()
+            .subscribe(new Subscriber<Integer>() {
+                @Override
+                public void onSubscribe(Subscription s) {
+                    s.request(j);
+                }
+
+                @Override
+                public void onNext(Integer t) {
+                    items.add(t);
+                }
+
+                @Override
+                public void onError(Throwable t) {
+                    items.add(t);
+                }
+
+                @Override
+                public void onComplete() {
+                    items.add("Done");
+                }
+            });
+
+            assertTrue(items.toString(), items.size() == 1);
+            assertTrue(items.toString(), items.get(0) instanceof IllegalArgumentException);
+            assertTrue(items.toString(), items.get(0).toString().contains("§3.9"));
+        }
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        final BooleanSubscription bs1 = new BooleanSubscription();
+        final BooleanSubscription bs2 = new BooleanSubscription();
+
+        TestSubscriber<Object> ts = Flowable.fromPublisher(new Publisher<Object>() {
+            @Override
+            public void subscribe(Subscriber<? super Object> p) {
+                p.onSubscribe(bs1);
+                p.onSubscribe(bs2);
+            }
+        })
+        .strict()
+        .test()
+        .assertFailure(IllegalStateException.class);
+
+        assertTrue(bs1.isCancelled());
+        assertTrue(bs2.isCancelled());
+
+        String es = ts.errors().get(0).toString();
+        assertTrue(es, es.contains("§2.12"));
+    }
+
+    @Test
+    public void noCancelOnComplete() {
+        final BooleanSubscription bs = new BooleanSubscription();
+
+        Flowable.fromPublisher(new Publisher<Object>() {
+            @Override
+            public void subscribe(Subscriber<? super Object> p) {
+                p.onSubscribe(bs);
+                p.onComplete();
+            }
+        })
+        .strict()
+        .subscribe(new Subscriber<Object>() {
+
+            Subscription s;
+
+            @Override
+            public void onSubscribe(Subscription s) {
+                this.s = s;
+            }
+
+            @Override
+            public void onNext(Object t) {
+                // not called
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                // not called
+            }
+
+            @Override
+            public void onComplete() {
+                s.cancel();
+            }
+        });
+
+        assertFalse(bs.isCancelled());
+    }
+
+    @Test
+    public void noCancelOnError() {
+        final BooleanSubscription bs = new BooleanSubscription();
+
+        Flowable.fromPublisher(new Publisher<Object>() {
+            @Override
+            public void subscribe(Subscriber<? super Object> p) {
+                p.onSubscribe(bs);
+                p.onError(new TestException());
+            }
+        })
+        .strict()
+        .subscribe(new Subscriber<Object>() {
+
+            Subscription s;
+
+            @Override
+            public void onSubscribe(Subscription s) {
+                this.s = s;
+            }
+
+            @Override
+            public void onNext(Object t) {
+                // not called
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                s.cancel();
+            }
+
+            @Override
+            public void onComplete() {
+                // not called
+            }
+        });
+
+        assertFalse(bs.isCancelled());
+    }
+}