2.x: add Flowable.doAfterNext operator + 3 doX TCKs

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

@@ -7353,6 +7353,32 @@ public final Flowable<T> doFinally(Action onFinally) {
         return RxJavaPlugins.onAssembly(new FlowableDoFinally<T>(this, onFinally));
     }
 
+    /**
+     * Calls the specified consumer with the current item after this item has been emitted to the downstream.
+     * <p>Note that the {@code onAfterNext} action is shared between subscriptions and as such
+     * should be thread-safe.
+     * <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 doAfterNext} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <td><b>Operator-fusion:</b></dt>
+     *  <dd>This operator supports normal and conditional Subscribers as well as boundary-limited
+     *  synchronous or asynchronous queue-fusion.</dd>
+     * </dl>
+     * @param onAfterNext the Consumer that will be called after emitting an item from upstream to the downstream
+     * @return the new Flowable instance
+     * @since 2.0.1 - experimental
+     */
+    @BackpressureSupport(BackpressureKind.PASS_THROUGH)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Flowable<T> doAfterNext(Consumer<? super T> onAfterNext) {
+        ObjectHelper.requireNonNull(onAfterNext, "onAfterNext is null");
+        return RxJavaPlugins.onAssembly(new FlowableDoAfterNext<T>(this, onAfterNext));
+    }
+
     /**
      * Registers an {@link Action} to be called when this Publisher invokes either
      * {@link Subscriber#onComplete onComplete} or {@link Subscriber#onError onError}.

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

@@ -0,0 +1,131 @@
+/**
+ * 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 org.reactivestreams.*;
+
+import io.reactivex.annotations.Experimental;
+import io.reactivex.functions.Consumer;
+import io.reactivex.internal.fuseable.ConditionalSubscriber;
+import io.reactivex.internal.subscribers.*;
+
+/**
+ * Calls a consumer after pushing the current item to the downstream.
+ * @param <T> the value type
+ * @since 2.0.1 - experimental
+ */
+@Experimental
+public final class FlowableDoAfterNext<T> extends AbstractFlowableWithUpstream<T, T> {
+
+    final Consumer<? super T> onAfterNext;
+
+    public FlowableDoAfterNext(Publisher<T> source, Consumer<? super T> onAfterNext) {
+        super(source);
+        this.onAfterNext = onAfterNext;
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super T> s) {
+        if (s instanceof ConditionalSubscriber) {
+            source.subscribe(new DoAfterConditionalSubscriber<T>((ConditionalSubscriber<? super T>)s, onAfterNext));
+        } else {
+            source.subscribe(new DoAfterSubscriber<T>(s, onAfterNext));
+        }
+    }
+
+    static final class DoAfterSubscriber<T> extends BasicFuseableSubscriber<T, T> {
+
+        final Consumer<? super T> onAfterNext;
+
+        DoAfterSubscriber(Subscriber<? super T> actual, Consumer<? super T> onAfterNext) {
+            super(actual);
+            this.onAfterNext = onAfterNext;
+        }
+
+        @Override
+        public void onNext(T t) {
+            actual.onNext(t);
+
+            if (sourceMode == NONE) {
+                try {
+                    onAfterNext.accept(t);
+                } catch (Throwable ex) {
+                    fail(ex);
+                }
+            }
+        }
+
+        @Override
+        public int requestFusion(int mode) {
+            return transitiveBoundaryFusion(mode);
+        }
+
+        @Override
+        public T poll() throws Exception {
+            T v = qs.poll();
+            if (v != null) {
+                onAfterNext.accept(v);
+            }
+            return v;
+        }
+    }
+
+    static final class DoAfterConditionalSubscriber<T> extends BasicFuseableConditionalSubscriber<T, T> {
+
+        final Consumer<? super T> onAfterNext;
+
+        DoAfterConditionalSubscriber(ConditionalSubscriber<? super T> actual, Consumer<? super T> onAfterNext) {
+            super(actual);
+            this.onAfterNext = onAfterNext;
+        }
+
+        @Override
+        public void onNext(T t) {
+            actual.onNext(t);
+
+            if (sourceMode == NONE) {
+                try {
+                    onAfterNext.accept(t);
+                } catch (Throwable ex) {
+                    fail(ex);
+                }
+            }
+        }
+
+        @Override
+        public boolean tryOnNext(T t) {
+            boolean b = actual.tryOnNext(t);
+            try {
+                onAfterNext.accept(t);
+            } catch (Throwable ex) {
+                fail(ex);
+            }
+            return b;
+        }
+
+        @Override
+        public int requestFusion(int mode) {
+            return transitiveBoundaryFusion(mode);
+        }
+
+        @Override
+        public T poll() throws Exception {
+            T v = qs.poll();
+            if (v != null) {
+                onAfterNext.accept(v);
+            }
+            return v;
+        }
+    }
+}