Lost Subscriptions

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() {

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);
                 }
             }

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() {

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

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() {

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

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() {

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

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;

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() {

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

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;
     }

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

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

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

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