Factored out the backpressure management into an experimental class and

reimplemented Buffer and Block strategies with it.

Author: akarnokd

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

@@ -20,8 +20,8 @@
 import java.util.concurrent.BlockingQueue;
 
 import rx.Observable.Operator;
-import rx.Producer;
 import rx.Subscriber;
+import rx.internal.util.BackpressureDrainManager;
 
 /**
  * Operator that blocks the producer thread in case a backpressure is needed.
@@ -38,45 +38,25 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
         return s;
     }
 
-    static final class BlockingSubscriber<T> extends Subscriber<T> {
+    static final class BlockingSubscriber<T> extends Subscriber<T> implements BackpressureDrainManager.BackpressureQueueCallback {
         final NotificationLite<T> nl = NotificationLite.instance();
         final BlockingQueue<Object> queue;
         final Subscriber<? super T> child;
-        /** Guarded by this. */
-        long requestedCount;
-        /** Guarded by this. */
-        boolean emitting;
-        volatile boolean terminated;
-        /** Set before terminated, read after terminated. */
-        Throwable exception;
+        final BackpressureDrainManager manager;
         public BlockingSubscriber(int max, Subscriber<? super T> child) {
             this.queue = new ArrayBlockingQueue<Object>(max);
             this.child = child;
+            this.manager = new BackpressureDrainManager(this);
         }
         void init() {
             child.add(this);
-            child.setProducer(new Producer() {
-                @Override
-                public void request(long n) {
-                    if (n == 0) {
-                        return;
-                    }
-                    synchronized (BlockingSubscriber.this) {
-                        if (n == Long.MAX_VALUE || requestedCount == Long.MAX_VALUE) {
-                            requestedCount = Long.MAX_VALUE;
-                        } else {
-                            requestedCount += n;
-                        }
-                    }
-                    drain();
-                }
-            });
+            child.setProducer(manager);
         }
         @Override
         public void onNext(T t) {
             try {
                 queue.put(nl.next(t));
-                drain();
+                manager.drain();
             } catch (InterruptedException ex) {
                 if (!isUnsubscribed()) {
                     onError(ex);
@@ -85,91 +65,31 @@ public void onNext(T t) {
         }
         @Override
         public void onError(Throwable e) {
-            if (!terminated) {
-                exception = e;
-                terminated = true;
-                drain();
-            }
+            manager.terminateAndDrain(e);
         }
         @Override
         public void onCompleted() {
-            terminated = true;
-            drain();
+            manager.terminateAndDrain();
         }
-        void drain() {
-            long n;
-            boolean term;
-            synchronized (this) {
-                if (emitting) {
-                    return;
-                }
-                emitting = true;
-                n = requestedCount;
-                term = terminated;
-            }
-            boolean skipFinal = false;
-            try {
-                Subscriber<? super T> child = this.child;
-                BlockingQueue<Object> queue = this.queue;
-                while (true) {
-                    int emitted = 0;
-                    while (n > 0 || term) {
-                        Object o;
-                        if (term) {
-                            o = queue.peek();
-                            if (o == null) {
-                                Throwable e = exception;
-                                if (e != null) {
-                                    child.onError(e);
-                                } else {
-                                    child.onCompleted();
-                                }
-                                skipFinal = true;
-                                return;
-                            }
-                            if (n == 0) {
-                                break;
-                            }
-                        }
-                        o = queue.poll();
-                        if (o == null) {
-                            break;
-                        } else {
-                            child.onNext(nl.getValue(o));
-                            n--;
-                            emitted++;
-                        }
-                    }
-                    synchronized (this) {
-                        term = terminated;
-                        boolean more = queue.peek() != null;
-                        // if no backpressure below
-                        if (requestedCount == Long.MAX_VALUE) {
-                            // no new data arrived since the last poll
-                            if (!more && !term) {
-                                skipFinal = true;
-                                emitting = false;
-                                return;
-                            }
-                            n = Long.MAX_VALUE;
-                        } else {
-                            requestedCount -= emitted;
-                            n = requestedCount;
-                            if ((n == 0 || !more) && (!term || more)) {
-                                skipFinal = true;
-                                emitting = false;
-                                return;
-                            }
-                        }
-                    }
-                }
-            } finally {
-                if (!skipFinal) {
-                    synchronized (this) {
-                        emitting = false;
-                    }
-                }
+        @Override
+        public boolean accept(Object value) {
+            return nl.accept(child, value);
+        }
+        @Override
+        public void complete(Throwable exception) {
+            if (exception != null) {
+                child.onError(exception);
+            } else {
+                child.onCompleted();
             }
         }
+        @Override
+        public Object peek() {
+            return queue.peek();
+        }
+        @Override
+        public Object poll() {
+            return queue.poll();
+        }
     }
 }