Merge pull request #1417 from benjchristensen/onStart

Proposal: Subscriber.onStart

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -157,7 +157,10 @@ public final <R> Observable<R> lift(final Operator<? extends R, ? super T> lift)
             @Override
             public void call(Subscriber<? super R> o) {
                 try {
-                    onSubscribe.call(hook.onLift(lift).call(o));
+                    Subscriber<? super T> st = hook.onLift(lift).call(o);
+                    // new Subscriber created and being subscribed with so 'onStart' it
+                    st.onStart();
+                    onSubscribe.call(st);
                 } catch (Throwable e) {
                     // localized capture of errors rather than it skipping all operators 
                     // and ending up in the try/catch of the subscribe method which then
@@ -6837,6 +6840,8 @@ public void onNext(T t) {
      */
     public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
         try {
+            // new Subscriber so onStart it
+            subscriber.onStart();
             // allow the hook to intercept and/or decorate
             hook.onSubscribeStart(this, onSubscribe).call(subscriber);
             return hook.onSubscribeReturn(subscriber);
@@ -6908,6 +6913,10 @@ public final Subscription subscribe(Subscriber<? super T> subscriber) {
              * so I won't mention that in the exception
              */
         }
+        
+        // new Subscriber so onStart it
+        subscriber.onStart();
+        
         /*
          * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
          * to user code from within an Observer"

rxjava-core/src/main/java/rx/Subscriber.java

@@ -52,23 +52,11 @@ protected Subscriber() {
         this.cs = new SubscriptionList();
     }
 
-    protected Subscriber(int bufferRequest) {
-        this.op = null;
-        this.cs = new SubscriptionList();
-        request(bufferRequest);
-    }
-
     protected Subscriber(Subscriber<?> op) {
         this.op = op;
         this.cs = op.cs;
     }
 
-    protected Subscriber(Subscriber<?> op, int bufferRequest) {
-        this.op = op;
-        this.cs = op.cs;
-        request(bufferRequest);
-    }
-
     /**
      * Adds a {@link Subscription} to this Subscriber's list of subscriptions if this list is not marked as
      * unsubscribed. If the list <em>is</em> marked as unsubscribed, {@code add} will indicate this by
@@ -95,6 +83,10 @@ public final boolean isUnsubscribed() {
         return cs.isUnsubscribed();
     }
 
+    public void onStart() {
+        // do nothing by default
+    }
+    
     public final void request(int n) {
         Producer shouldRequest = null;
         synchronized (this) {

rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java

@@ -77,16 +77,21 @@ private static final class MergeSubscriber<T> extends Subscriber<Observable<? ex
          */
 
         public MergeSubscriber(Subscriber<? super T> actual) {
-            // we request backpressure so we can handle long-running Observables that are enqueueing, such as flatMap use cases
-            // passing the request value in decouples the Producer chain while keeping the Subscription chain together (perf benefit)
-            super(actual, RxRingBuffer.SIZE);
+            super(actual);
             this.actual = actual;
             this.mergeProducer = new MergeProducer<T>(this);
             // decoupled the subscription chain because we need to decouple and control backpressure
             actual.add(this);
             actual.setProducer(mergeProducer);
         }
 
+        @Override
+        public void onStart() {
+            // we request backpressure so we can handle long-running Observables that are enqueueing, such as flatMap use cases
+            // we decouple the Producer chain while keeping the Subscription chain together (perf benefit) via super(actual)
+            request(RxRingBuffer.SIZE);
+        }
+        
         /*
          * This is expected to be executed sequentially as per the Rx contract or it will not work.
          */

rxjava-core/src/main/java/rx/internal/operators/OperatorObserveOn.java

@@ -83,8 +83,6 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
         // do NOT pass the Subscriber through to couple the subscription chain ... unsubscribing on the parent should
         // not prevent anything downstream from consuming, which will happen if the Subscription is chained
         public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child) {
-         // signal that this is an async operator capable of receiving this many
-            super(RxRingBuffer.SIZE);
             this.child = child;
             this.recursiveScheduler = scheduler.createWorker();
             this.scheduledUnsubscribe = new ScheduledUnsubscribe(recursiveScheduler, queue);
@@ -103,6 +101,12 @@ public void request(int n) {
             child.add(this);
 
         }
+        
+        @Override
+        public void onStart() {
+            // signal that this is an async operator capable of receiving this many
+            request(RxRingBuffer.SIZE);
+        }
 
         @Override
         public void onNext(final T t) {

rxjava-core/src/main/java/rx/internal/operators/OperatorTakeLast.java

@@ -47,8 +47,14 @@ public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
         final QueueProducer<T> producer = new QueueProducer<T>(notification, deque, subscriber);
         subscriber.setProducer(producer);
 
-        // no backpressure up as it wants to receive and discard all but the last
-        return new Subscriber<T>(subscriber, -1) {
+        return new Subscriber<T>(subscriber) {
+
+            // no backpressure up as it wants to receive and discard all but the last
+            @Override
+            public void onStart() {
+                // we do this to break the chain of the child subscriber being passed through
+                request(-1);
+            }
 
             @Override
             public void onCompleted() {

rxjava-core/src/perf/java/rx/operators/OperatorRangePerf.java

@@ -40,8 +40,13 @@ public void setup(final Blackhole bh) {
         }
 
         public Subscriber<Integer> newSubscriber() {
-            return new Subscriber<Integer>(size) {
+            return new Subscriber<Integer>() {
 
+                @Override
+                public void onStart() {
+                    request(size);
+                }
+                
                 @Override
                 public void onCompleted() {
 

rxjava-core/src/test/java/rx/BackpressureTests.java

@@ -173,10 +173,12 @@ public Observable<Integer> call(Integer i) {
         }).take(NUM).subscribe(ts);
         ts.awaitTerminalEvent();
         ts.assertNoErrors();
-        System.out.println("testFlatMapAsync => Received: " + ts.getOnNextEvents().size() + "  Emitted: " + c.get());
+        System.out.println("testFlatMapAsync => Received: " + ts.getOnNextEvents().size() + "  Emitted: " + c.get() + " Size: " + RxRingBuffer.SIZE);
         assertEquals(NUM, ts.getOnNextEvents().size());
-        // expect less than 1 buffer since the flatMap is emitting 10 each time, so it is NUM/10 that will be taken.
-        assertTrue(c.get() <= RxRingBuffer.SIZE);
+        // even though we only need 10, it will request at least RxRingBuffer.SIZE, and then as it drains keep requesting more
+        // and then it will be non-deterministic when the take() causes the unsubscribe as it is scheduled on 10 different schedulers (threads)
+        // normally this number is ~250 but can get up to ~1200 when RxRingBuffer.SIZE == 1024
+        assertTrue(c.get() <= RxRingBuffer.SIZE * 2);
     }
 
     @Ignore
@@ -303,8 +305,13 @@ public void testUserSubscriberUsingRequestSync() {
         final AtomicInteger totalReceived = new AtomicInteger();
         final AtomicInteger batches = new AtomicInteger();
         final AtomicInteger received = new AtomicInteger();
-        incrementingIntegers(c).subscribe(new Subscriber<Integer>(100) {
+        incrementingIntegers(c).subscribe(new Subscriber<Integer>() {
 
+            @Override
+            public void onStart() {
+                request(100);
+            }
+            
             @Override
             public void onCompleted() {
 
@@ -344,8 +351,14 @@ public void testUserSubscriberUsingRequestAsync() throws InterruptedException {
         final AtomicInteger received = new AtomicInteger();
         final AtomicInteger batches = new AtomicInteger();
         final CountDownLatch latch = new CountDownLatch(1);
-        incrementingIntegers(c).subscribeOn(Schedulers.newThread()).subscribe(new Subscriber<Integer>(100) {
+        incrementingIntegers(c).subscribeOn(Schedulers.newThread()).subscribe(new Subscriber<Integer>() {
 
+            @Override
+            public void onStart() {
+                request(100);
+            }
+            
+            
             @Override
             public void onCompleted() {
                 latch.countDown();