add TODOs and comments

Author: davidmoten

src/main/java/rx/internal/operators/OnSubscribeRedo.java

@@ -33,6 +33,7 @@
 
 import static rx.Observable.create;
 
+import java.util.Random;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -339,9 +340,19 @@ public void onNext(Object t) {
                             // if there are outstanding requests 
                             if (consumerCapacity.get() > 0) {
                                 // schedule resubscription
+                                // note that at this point consumerCapacity may have dropped to zero
+                                // but this is a best endeavours check to try not to resubscribe more than
+                                // we need to
                                 worker.schedule(subscribeToSource);
                             } else {
                                 // otherwise we indicate that on the next request we should resubscribe
+                                // because it is a potential waste of resources to schedule resubscription 
+                                // now when there is the possibility that the child may unsubscribe and never use
+                                // that resubscription
+                                
+                                //TODO what if consumerCapacity increases now then we are in a situation where the
+                                // next call of request(n) will do nothing because of the c ==0 check in request(n) 
+                                // this would cause the stream to stall?
                                 resumeBoundary.compareAndSet(false, true);
                             }
                         }
@@ -359,9 +370,16 @@ public void setProducer(Producer producer) {
 
             @Override
             public void request(final long n) {
+                // TODO at this point we usually (?) add some optimizations 
+                // like detecting that we are already on the fast path (non-backpressure) 
+                // and not attempting to process additional requests
                 long c = BackpressureUtils.getAndAddRequest(consumerCapacity, n);
                 Producer producer = currentProducer.get();
                 if (producer != null) {
+                    // TODO what if at this point the subscription finishes and currentProducer 
+                    // is set to null or even the next producer. The request would be added to consumerCapacity but
+                    // if it more requests only come after emission then this call to the old producer could produce 
+                    // nothing and the stream would stall
                     producer.request(n);
                 } else
                 if (c == 0 && resumeBoundary.compareAndSet(true, false)) {

src/test/java/rx/internal/operators/OperatorRetryTest.java

@@ -708,6 +708,7 @@ public void testRetryWithBackpressure() throws InterruptedException {
 
     @Test(timeout = 15000)
     public void testRetryWithBackpressureParallel() throws InterruptedException {
+        final long startTime = System.currentTimeMillis();
         final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
         int ncpu = Runtime.getRuntime().availableProcessors();
         ExecutorService exec = Executors.newFixedThreadPool(Math.max(ncpu / 2, 2));
@@ -718,11 +719,17 @@ public void testRetryWithBackpressureParallel() throws InterruptedException {
 
         int m = 5000;
         final CountDownLatch cdl = new CountDownLatch(m);
+        
         for (int i = 0; i < m; i++) {
             final int j = i;
             exec.execute(new Runnable() {
                 @Override
                 public void run() {
+                    if (System.currentTimeMillis() - startTime >12000) {
+                        //don't start this job if we are close to test timeout
+                        cdl.countDown();
+                        return;
+                    }
                     final AtomicInteger nexts = new AtomicInteger();
                     try {
                         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));