revert synchronization on consumerCapacity and adjust static helper method

Author: davidmoten

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

@@ -315,12 +315,10 @@ public void onError(Throwable e) {
                     @Override
                     public void onNext(Object t) {
                         if (!isLocked.get() && !child.isUnsubscribed()) {
-                            synchronized(consumerCapacity) {
-                                if (consumerCapacity.get() > 0) {
-                                    worker.schedule(subscribeToSource);
-                                } else {
-                                    resumeBoundary.compareAndSet(false, true);
-                                }
+                            if (consumerCapacity.get() > 0) {
+                                worker.schedule(subscribeToSource);
+                            } else {
+                                resumeBoundary.compareAndSet(false, true);
                             }
                         }
                     }
@@ -337,16 +335,13 @@ public void setProducer(Producer producer) {
 
             @Override
             public void request(final long n) {
-                long c;
-                synchronized(consumerCapacity) {
-                    c = BackpressureUtils.getAndAddRequest(consumerCapacity, n);
-                    Producer producer = currentProducer.get();
-                    if (producer != null) {
-                        producer.request(n);
-                    } else
-                    if (c == 0 && resumeBoundary.compareAndSet(true, false)) {
-                        worker.schedule(subscribeToSource);
-                    }
+                long c = BackpressureUtils.getAndAddRequest(consumerCapacity, n);
+                Producer producer = currentProducer.get();
+                if (producer != null) {
+                    producer.request(n);
+                } else
+                if (c == 0 && resumeBoundary.compareAndSet(true, false)) {
+                    worker.schedule(subscribeToSource);
                 }
             }
         });

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

@@ -706,7 +706,7 @@ public void testRetryWithBackpressure() throws InterruptedException {
         }
     }
 
-    @Test(timeout = 150000)
+    @Test(timeout = 15000)
     public void testRetryWithBackpressureParallel() throws InterruptedException {
         final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
         int ncpu = Runtime.getRuntime().availableProcessors();
@@ -738,7 +738,6 @@ public void run() {
                         }
                         if (ts.getOnNextEvents().size() != NUM_RETRIES + 2) {
                             data.put(j, ts.getOnNextEvents());
-                            System.out.println(j+"->"+ts.getOnNextEvents());
                         }
                     } catch (Throwable t) {
                         timeouts.incrementAndGet();
@@ -784,10 +783,10 @@ static <T> StringBuilder allSequenceFrequency(Map<Integer, List<T>> its) {
     }
     static <T> StringBuilder sequenceFrequency(Iterable<T> it) {
         StringBuilder sb = new StringBuilder();
-        
+
         Object prev = null;
         int cnt = 0;
-        
+
         for (Object curr : it) {
             if (sb.length() > 0) {
                 if (!curr.equals(prev)) {
@@ -806,7 +805,10 @@ static <T> StringBuilder sequenceFrequency(Iterable<T> it) {
             }
             prev = curr;
         }
-        
+        if (cnt > 1) {
+            sb.append(" x ").append(cnt);
+        }
+
         return sb;
     }
     @Test(timeout = 3000)