use ProducerArbiter from @akarnokd, delete OnSubscribeRetry

Author: davidmoten

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

@@ -35,7 +35,6 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Notification;
 import rx.Observable;
@@ -53,7 +52,7 @@
 
 public final class OnSubscribeRedo<T> implements OnSubscribe<T> {
 
-    static final Func1<Observable<? extends Notification<?>>, Observable<?>> REDO_INIFINITE = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+    static final Func1<Observable<? extends Notification<?>>, Observable<?>> REDO_INFINITE = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
         @Override
         public Observable<?> call(Observable<? extends Notification<?>> ts) {
             return ts.map(new Func1<Notification<?>, Notification<?>>() {
@@ -120,17 +119,15 @@ public Notification<Integer> call(Notification<Integer> n, Notification<?> term)
     }
 
     public static <T> Observable<T> retry(Observable<T> source) {
-      // return retry(source, REDO_INIFINITE);
-      return retry(source, Long.MAX_VALUE);
+       return retry(source, REDO_INFINITE);
     }
 
     public static <T> Observable<T> retry(Observable<T> source, final long count) {
         if (count < 0)
             throw new IllegalArgumentException("count >= 0 expected");
         if (count == 0)
             return source;
-//        return retry(source, new RedoFinite(count));
-        return create(new OnSubscribeRetry<T>(source, count));
+        return retry(source, new RedoFinite(count));
     }
 
     public static <T> Observable<T> retry(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler) {
@@ -146,7 +143,7 @@ public static <T> Observable<T> repeat(Observable<T> source) {
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, Scheduler scheduler) {
-        return repeat(source, REDO_INIFINITE, scheduler);
+        return repeat(source, REDO_INFINITE, scheduler);
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, final long count) {
@@ -195,7 +192,6 @@ public void call(final Subscriber<? super T> child) {
         final AtomicBoolean resumeBoundary = new AtomicBoolean(true);
         // incremented when requests are made, decremented when requests are fulfilled
         final AtomicLong consumerCapacity = new AtomicLong(0l);
-        final AtomicReference<Producer> currentProducer = new AtomicReference<Producer>();
 
         final Scheduler.Worker worker = scheduler.createWorker();
         child.add(worker);
@@ -205,6 +201,8 @@ public void call(final Subscriber<? super T> child) {
 
         final PublishSubject<Notification<?>> terminals = PublishSubject.create();
 
+        final ProducerArbiter arbiter = new ProducerArbiter();
+        
         final Action0 subscribeToSource = new Action0() {
             @Override
             public void call() {
@@ -214,43 +212,37 @@ public void call() {
 
                 Subscriber<T> terminalDelegatingSubscriber = new Subscriber<T>() {
                     boolean done;
+
                     @Override
                     public void onCompleted() {
                         if (!done) {
                             done = true;
-                            synchronized (consumerCapacity) {
-                                currentProducer.set(null);
-                                unsubscribe();
-                                terminals.onNext(Notification.createOnCompleted());
-                            }
+                            unsubscribe();
+                            terminals.onNext(Notification.createOnCompleted());
                         }
                     }
 
                     @Override
                     public void onError(Throwable e) {
                         if (!done) {
                             done = true;
-                            synchronized (consumerCapacity) {
-                                currentProducer.set(null);
-                                unsubscribe();
-                                terminals.onNext(Notification.createOnError(e));
-                            }
+                            unsubscribe();
+                            terminals.onNext(Notification.createOnError(e));
                         }
                     }
 
                     @Override
                     public void onNext(T v) {
                         if (!done) {
-                            synchronized (consumerCapacity) {
-                                child.onNext(v);
-                                decrementConsumerCapacity();
-                            }
+                            child.onNext(v);
+                            decrementConsumerCapacity();
+                            arbiter.produced(1);
                         }
                     }
-                    
+
                     private void decrementConsumerCapacity() {
-                        // use a CAS loop because we don't want to decrement the value 
-                        // if it is Long.MAX_VALUE
+                        // use a CAS loop because we don't want to decrement the
+                        // value if it is Long.MAX_VALUE
                         while (true) {
                             long cc = consumerCapacity.get();
                             if (cc != Long.MAX_VALUE) {
@@ -265,13 +257,7 @@ private void decrementConsumerCapacity() {
 
                     @Override
                     public void setProducer(Producer producer) {
-                        synchronized (consumerCapacity) {
-                            currentProducer.set(producer);
-                            long c = consumerCapacity.get();
-                            if (c > 0) {
-                                producer.request(c);
-                            }
-                        }
+                        arbiter.setProducer(producer);
                     }
                 };
                 // new subscription each time so if it unsubscribes itself it does not prevent retries
@@ -337,17 +323,11 @@ public void onError(Throwable e) {
                     @Override
                     public void onNext(Object t) {
                         if (!isLocked.get() && !child.isUnsubscribed()) {
-                            final boolean scheduleNow;
-                            synchronized (consumerCapacity) {
-                                if (consumerCapacity.get() > 0) {
-                                    scheduleNow = true;
-                                } else {
-                                    scheduleNow = false;
-                                    resumeBoundary.compareAndSet(false, true);
-                                }
-                            }
-                            if (scheduleNow)
+                            if (consumerCapacity.get() > 0) {
                                 worker.schedule(subscribeToSource);
+                            } else {
+                                resumeBoundary.compareAndSet(false, true);
+                            }
                         }
                     }
 
@@ -363,26 +343,173 @@ public void setProducer(Producer producer) {
 
             @Override
             public void request(final long n) {
-                final Producer producer;
-                final boolean requestNow;
-                final boolean scheduleNow;
-                synchronized (consumerCapacity) {
+                if (n > 0) {
                     BackpressureUtils.getAndAddRequest(consumerCapacity, n);
-                    producer = currentProducer.get();
-                    if (producer != null) {
-                        requestNow = true;
-                        scheduleNow = false;
-                    } else {
-                        requestNow = false;
-                        scheduleNow = resumeBoundary.compareAndSet(true, false);
-                    }
+                    arbiter.request(n);
+                    if (resumeBoundary.compareAndSet(true, false))
+                        worker.schedule(subscribeToSource);
                 }
-                if (requestNow)
-                    producer.request(n);
-                else if (scheduleNow)
-                    worker.schedule(subscribeToSource);
             }
         });
 
     }
+    
+    static final class ProducerArbiter implements Producer {
+        /** Guarded by this. */
+        boolean emitting;
+        /** The current producer. Accessed while emitting. */
+        Producer currentProducer;
+        /** The current requested count. */
+        long requested;
+        
+        long missedRequested;
+        Producer missedProducer;
+        long missedProd;
+        
+        @Override
+        public void request(long n) {
+            if (n <= 0) {
+                return;
+            }
+            Producer mp;
+            long mprod;
+            synchronized (this) {
+                if (emitting) {
+                    missedRequested += n;
+                    return;
+                }
+                emitting = true;
+                mp = missedProducer;
+                mprod = missedProd;
+                
+                missedProducer = null;
+                missedProd = 0L;
+            }
+            
+            boolean skipFinal = false;
+            try {
+                emit(n, mp, mprod);
+                drainLoop();
+                skipFinal = true;
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+        public void setProducer(Producer p) {
+            if (p == null) {
+                throw new NullPointerException();
+            }
+            
+            long mreq;
+            long mprod;
+            synchronized (this) {
+                if (emitting) {
+                    missedProducer = p;
+                    return;
+                }
+                emitting = true;
+                mreq = missedRequested;
+                mprod = missedProd;
+                
+                missedRequested = 0L;
+                missedProd = 0L;
+            }
+            
+            boolean skipFinal = false;
+            try {
+                emit(mreq, p, mprod);
+                drainLoop();
+                skipFinal = true;
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+        public void produced(long n) {
+            if (n <= 0) {
+                throw new IllegalArgumentException(n + " produced?!");
+            }
+            
+            long mreq;
+            Producer mp;
+            synchronized (this) {
+                if (emitting) {
+                    missedProd += n;
+                    return;
+                }
+                emitting = true;
+                mreq = missedRequested;
+                mp = missedProducer;
+                
+                missedRequested = 0L;
+                missedProducer = null;
+            }
+            
+            boolean skipFinal = false;
+            try {
+                emit(mreq, mp, n);
+                drainLoop();
+                skipFinal = true;
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+        void drainLoop() {
+            for (;;) {
+                long mreq;
+                long mprod;
+                Producer mp;
+                synchronized (this) {
+                    mreq = missedRequested;
+                    mprod = missedProd;
+                    mp = missedProducer;
+                    if (mreq == 0L && mp == null && mprod == 0L) {
+                        emitting = false;
+                        return;
+                    }
+                    missedRequested = 0L;
+                    missedProd = 0L;
+                    missedProducer = null;
+                }
+                emit(mreq, mp, mprod);
+            }
+        }
+        void emit(long mreq, Producer mp, long mprod) {
+            boolean newMp = false;
+            if (mp != null) {
+                newMp = true;
+                currentProducer = mp;
+            } else {
+                mp = currentProducer;
+            }
+            
+            long u = requested + mreq;
+            if (u < 0) {
+                u = Long.MAX_VALUE;
+            }
+            u -= mprod;
+            if (u < 0) {
+                throw new IllegalStateException("More produced than requested");
+            }
+            requested = u;
+            
+            if (mreq > 0 && mp != null) {
+                mp.request(mreq);
+            } else
+            if (newMp && u > 0) {
+                mp.request(u);
+            }
+        }
+    }
 }