After thinking about ReactiveX#2575 I added a note to all the places …

…where a Subscription is lost from a Scheduler.
abersnaze · Jan 31, 2015 · 4ee0983 · 4ee0983
1 parent fe3cc75
commit 4ee0983
Show file tree

Hide file tree

Showing 16 changed files with 27 additions and 0 deletions.
diff --git a/src/main/java/rx/internal/operators/OnSubscribeDelaySubscription.java b/src/main/java/rx/internal/operators/OnSubscribeDelaySubscription.java
@@ -46,6 +46,7 @@ public void call(final Subscriber<? super T> s) {
         final Worker worker = scheduler.createWorker();
         s.add(worker);
 
+        // FIXME should subscription returned be added to the s composite
         worker.schedule(new Action0() {
             @Override
             public void call() {

diff --git a/src/main/java/rx/internal/operators/OnSubscribeRedo.java b/src/main/java/rx/internal/operators/OnSubscribeRedo.java
@@ -286,6 +286,7 @@ public void setProducer(Producer producer) {
                 }));
 
         // subscribe to the restarts observable to know when to schedule the next redo.
+        // FIXME should subscription returned be added to the child composite
         worker.schedule(new Action0() {
             @Override
             public void call() {
@@ -304,6 +305,7 @@ public void onError(Throwable e) {
                     public void onNext(Object t) {
                         if (!isLocked.get() && !child.isUnsubscribed()) {
                             if (consumerCapacity.get() > 0) {
+                                // FIXME should subscription returned be added to the child composite
                                 worker.schedule(subscribeToSource);
                             } else {
                                 resumeBoundary.compareAndSet(false, true);
@@ -329,6 +331,7 @@ public void request(final long n) {
                     producer.request(n);
                 } else
                 if (c == 0 && resumeBoundary.compareAndSet(true, false)) {
+                    // FIXME should subscription returned be added to the child composite
                     worker.schedule(subscribeToSource);
                 }
             }

diff --git a/src/main/java/rx/internal/operators/OnSubscribeTimerOnce.java b/src/main/java/rx/internal/operators/OnSubscribeTimerOnce.java
@@ -41,6 +41,7 @@ public OnSubscribeTimerOnce(long time, TimeUnit unit, Scheduler scheduler) {
     public void call(final Subscriber<? super Long> child) {
         Worker worker = scheduler.createWorker();
         child.add(worker);
+        // FIXME should subscription returned be added to the child composite
         worker.schedule(new Action0() {
             @Override
             public void call() {

diff --git a/src/main/java/rx/internal/operators/OnSubscribeTimerPeriodically.java b/src/main/java/rx/internal/operators/OnSubscribeTimerPeriodically.java
@@ -43,6 +43,7 @@ public OnSubscribeTimerPeriodically(long initialDelay, long period, TimeUnit uni
     public void call(final Subscriber<? super Long> child) {
         final Worker worker = scheduler.createWorker();
         child.add(worker);
+        // FIXME should subscription returned be added to the child composite
         worker.schedulePeriodically(new Action0() {
             long counter;
             @Override

diff --git a/src/main/java/rx/internal/operators/OperatorBufferWithTime.java b/src/main/java/rx/internal/operators/OperatorBufferWithTime.java
@@ -166,6 +166,7 @@ public void onCompleted() {
             unsubscribe();
         }
         void scheduleChunk() {
+            // FIXME should subscription returned be added to the child composite
             inner.schedulePeriodically(new Action0() {
                 @Override
                 public void call() {
@@ -181,6 +182,7 @@ void startNewChunk() {
                 }
                 chunks.add(chunk);
             }
+            // FIXME should subscription returned be added to the child composite
             inner.schedule(new Action0() {
                 @Override
                 public void call() {
@@ -280,6 +282,7 @@ public void onCompleted() {
             unsubscribe();
         }
         void scheduleExact() {
+            // FIXME should subscription returned be added to the child composite
             inner.schedulePeriodically(new Action0() {
                 @Override
                 public void call() {

diff --git a/src/main/java/rx/internal/operators/OperatorDelay.java b/src/main/java/rx/internal/operators/OperatorDelay.java
@@ -52,6 +52,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
 
             @Override
             public void onCompleted() {
+                // FIXME should subscription returned be added to the child composite
                 worker.schedule(new Action0() {
 
                     @Override
@@ -69,6 +70,7 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(final T t) {
+                // FIXME should subscription returned be added to the child composite
                 worker.schedule(new Action0() {
 
                     @Override

diff --git a/src/main/java/rx/internal/operators/OperatorObserveOn.java b/src/main/java/rx/internal/operators/OperatorObserveOn.java
@@ -148,6 +148,7 @@ public void onError(final Throwable e) {
 
         protected void schedule() {
             if (COUNTER_UPDATER.getAndIncrement(this) == 0) {
+                // FIXME should subscription returned be added to the child composite
                 recursiveScheduler.schedule(new Action0() {
 
                     @Override
@@ -229,6 +230,7 @@ public boolean isUnsubscribed() {
         @Override
         public void unsubscribe() {
             if (ONCE_UPDATER.getAndSet(this, 1) == 0) {
+                // FIXME should subscription returned be added to something
                 worker.schedule(new Action0() {
                     @Override
                     public void call() {

diff --git a/src/main/java/rx/internal/operators/OperatorRetryWithPredicate.java b/src/main/java/rx/internal/operators/OperatorRetryWithPredicate.java
@@ -74,6 +74,7 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(final Observable<T> o) {
+                // FIXME should subscription returned be added to the child composite
                 inner.schedule(new Action0() {
 
                     @Override
@@ -94,6 +95,7 @@ public void onCompleted() {
                             public void onError(Throwable e) {
                                 if (predicate.call(attempts, e) && !inner.isUnsubscribed()) {
                                     // retry again
+                                    // FIXME should subscription returned be added to the child composite
                                     inner.schedule(_self);
                                 } else {
                                     // give up and pass the failure

diff --git a/src/main/java/rx/internal/operators/OperatorSampleWithTime.java b/src/main/java/rx/internal/operators/OperatorSampleWithTime.java
@@ -51,6 +51,7 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
 
         SamplerSubscriber<T> sampler = new SamplerSubscriber<T>(s);
         child.add(sampler);
+        // FIXME should subscription returned be added to the child composite
         worker.schedulePeriodically(sampler, time, time, unit);
 
         return sampler;

diff --git a/src/main/java/rx/internal/operators/OperatorSkipTimed.java b/src/main/java/rx/internal/operators/OperatorSkipTimed.java
@@ -43,6 +43,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
         final Worker worker = scheduler.createWorker();
         child.add(worker);
         final AtomicBoolean gate = new AtomicBoolean();
+        // FIXME should subscription returned be added to the child composite
         worker.schedule(new Action0() {
             @Override
             public void call() {

diff --git a/src/main/java/rx/internal/operators/OperatorSubscribeOn.java b/src/main/java/rx/internal/operators/OperatorSubscribeOn.java
@@ -87,6 +87,7 @@ public void request(final long n) {
                                             // see unit test 'testSetProducerSynchronousRequest' for more context on this
                                             producer.request(n);
                                         } else {
+                                            // FIXME should subscription returned be added to the subscriber composite
                                             inner.schedule(new Action0() {
 
                                                 @Override

diff --git a/src/main/java/rx/internal/operators/OperatorTakeTimed.java b/src/main/java/rx/internal/operators/OperatorTakeTimed.java
@@ -46,6 +46,7 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
         child.add(worker);
 
         TakeSubscriber<T> ts = new TakeSubscriber<T>(new SerializedSubscriber<T>(child));
+        // FIXME should subscription returned be added to the child composite
         worker.schedule(ts, time, unit);
         return ts;
     }

diff --git a/src/main/java/rx/internal/operators/OperatorUnsubscribeOn.java b/src/main/java/rx/internal/operators/OperatorUnsubscribeOn.java
@@ -59,6 +59,7 @@ public void onNext(T t) {
             @Override
             public void call() {
                 final Scheduler.Worker inner = scheduler.createWorker();
+                // FIXME should subscription returned be added to the subscriber composite
                 inner.schedule(new Action0() {
 
                     @Override

diff --git a/src/main/java/rx/internal/operators/OperatorWindowWithTime.java b/src/main/java/rx/internal/operators/OperatorWindowWithTime.java
@@ -275,6 +275,7 @@ public void onCompleted() {
         }
 
         void scheduleExact() {
+            // FIXME should subscription returned be added to the child composite
             worker.schedulePeriodically(new Action0() {
 
                 @Override
@@ -420,6 +421,7 @@ public void onCompleted() {
             child.onCompleted();
         }
         void scheduleChunk() {
+            // FIXME should subscription returned be added to the child composite
             worker.schedulePeriodically(new Action0() {
 
                 @Override
@@ -444,6 +446,7 @@ void startNewChunk() {
                 return;
             }
 
+            // FIXME should subscription returned be added to the child composite
             worker.schedule(new Action0() {
 
                 @Override

diff --git a/src/main/java/rx/internal/util/ObjectPool.java b/src/main/java/rx/internal/util/ObjectPool.java
@@ -55,6 +55,7 @@ private ObjectPool(final int min, final int max, final long validationInterval)
         initialize(min);
 
         schedulerWorker = Schedulers.computation().createWorker();
+        // FIXME should subscription returned be added to some composite
         schedulerWorker.schedulePeriodically(new Action0() {
 
             @Override

diff --git a/src/main/java/rx/subjects/TestSubject.java b/src/main/java/rx/subjects/TestSubject.java
@@ -90,6 +90,7 @@ private void _onCompleted() {
      *         the number of milliseconds in the future relative to "now()" at which to call {@code onCompleted}
      */
     public void onCompleted(long timeInMilliseconds) {
+        // FIXME should subscription returned be added to some composite
         innerScheduler.schedule(new Action0() {
 
             @Override
@@ -125,6 +126,7 @@ private void _onError(final Throwable e) {
      *         the number of milliseconds in the future relative to "now()" at which to call {@code onError}
      */
     public void onError(final Throwable e, long timeInMilliseconds) {
+        // FIXME should subscription returned be added to some composite
         innerScheduler.schedule(new Action0() {
 
             @Override
@@ -158,6 +160,7 @@ private void _onNext(T v) {
      *         the number of milliseconds in the future relative to "now()" at which to call {@code onNext}
      */
     public void onNext(final T v, long timeInMilliseconds) {
+        // FIXME should subscription returned be added to some composite
         innerScheduler.schedule(new Action0() {
 
             @Override