Merge pull request #3351 from vqvu/blocking-iterator-backpressure

Make BlockingOperatorToIterator exert backpressure.

Author: abersnaze

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

@@ -23,8 +23,8 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
-import rx.Subscription;
 import rx.exceptions.Exceptions;
+import rx.internal.util.RxRingBuffer;
 
 /**
  * Returns an Iterator that iterates over all items emitted by a specified Observable.
@@ -47,68 +47,88 @@ private BlockingOperatorToIterator() {
      * @return the iterator that could be used to iterate over the elements of the observable.
      */
     public static <T> Iterator<T> toIterator(Observable<? extends T> source) {
-        final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>();
+        SubscriberIterator<T> subscriber = new SubscriberIterator<T>();
 
         // using subscribe instead of unsafeSubscribe since this is a BlockingObservable "final subscribe"
-        final Subscription subscription = source.materialize().subscribe(new Subscriber<Notification<? extends T>>() {
-            @Override
-            public void onCompleted() {
-                // ignore
-            }
+        source.materialize().subscribe(subscriber);
+        return subscriber;
+    }
 
-            @Override
-            public void onError(Throwable e) {
-                notifications.offer(Notification.<T>createOnError(e));
-            }
+    public static final class SubscriberIterator<T>
+        extends Subscriber<Notification<? extends T>> implements Iterator<T> {
 
-            @Override
-            public void onNext(Notification<? extends T> args) {
-                notifications.offer(args);
-            }
-        });
+        static final int LIMIT = 3 * RxRingBuffer.SIZE / 4;
 
-        return new Iterator<T>() {
-            private Notification<? extends T> buf;
+        private final BlockingQueue<Notification<? extends T>> notifications;
+        private Notification<? extends T> buf;
+        private int received;
 
-            @Override
-            public boolean hasNext() {
-                if (buf == null) {
-                    buf = take();
-                }
-                if (buf.isOnError()) {
-                    throw Exceptions.propagate(buf.getThrowable());
+        public SubscriberIterator() {
+            this.notifications = new LinkedBlockingQueue<Notification<? extends T>>();
+        }
+
+        @Override
+        public void onStart() {
+            request(RxRingBuffer.SIZE);
+        }
+
+        @Override
+        public void onCompleted() {
+            // ignore
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            notifications.offer(Notification.<T>createOnError(e));
+        }
+
+        @Override
+        public void onNext(Notification<? extends T> args) {
+            notifications.offer(args);
+        }
+
+        @Override
+        public boolean hasNext() {
+            if (buf == null) {
+                buf = take();
+                received++;
+                if (received >= LIMIT) {
+                    request(received);
+                    received = 0;
                 }
-                return !buf.isOnCompleted();
             }
+            if (buf.isOnError()) {
+                throw Exceptions.propagate(buf.getThrowable());
+            }
+            return !buf.isOnCompleted();
+        }
 
-            @Override
-            public T next() {
-                if (hasNext()) {
-                    T result = buf.getValue();
-                    buf = null;
-                    return result;
-                }
-                throw new NoSuchElementException();
+        @Override
+        public T next() {
+            if (hasNext()) {
+                T result = buf.getValue();
+                buf = null;
+                return result;
             }
+            throw new NoSuchElementException();
+        }
 
-            private Notification<? extends T> take() {
-                try {
-                    Notification<? extends T> poll = notifications.poll();
-                    if (poll != null) {
-                        return poll;
-                    }
-                    return notifications.take();
-                } catch (InterruptedException e) {
-                    subscription.unsubscribe();
-                    throw Exceptions.propagate(e);
+        private Notification<? extends T> take() {
+            try {
+                Notification<? extends T> poll = notifications.poll();
+                if (poll != null) {
+                    return poll;
                 }
+                return notifications.take();
+            } catch (InterruptedException e) {
+                unsubscribe();
+                throw Exceptions.propagate(e);
             }
+        }
 
-            @Override
-            public void remove() {
-                throw new UnsupportedOperationException("Read-only iterator");
-            }
-        };
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException("Read-only iterator");
+        }
     }
-
 }

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

@@ -26,6 +26,8 @@
 import rx.Observable.OnSubscribe;
 import rx.Subscriber;
 import rx.exceptions.TestException;
+import rx.internal.operators.BlockingOperatorToIterator.SubscriberIterator;
+import rx.internal.util.RxRingBuffer;
 
 public class BlockingOperatorToIteratorTest {
 
@@ -81,4 +83,48 @@ public void call(Subscriber<? super String> subscriber) {
             System.out.println(string);
         }
     }
+
+    @Test
+    public void testIteratorExertBackpressure() {
+        final Counter src = new Counter();
+
+        Observable<Integer> obs = Observable.from(new Iterable<Integer>() {
+            @Override
+            public Iterator<Integer> iterator() {
+                return src;
+            }
+        });
+
+        Iterator<Integer> it = toIterator(obs);
+        while (it.hasNext()) {
+            // Correct backpressure should cause this interleaved behavior.
+            // We first request RxRingBuffer.SIZE. Then in increments of
+            // SubscriberIterator.LIMIT.
+            int i = it.next();
+            int expected = i - (i % SubscriberIterator.LIMIT) + RxRingBuffer.SIZE;
+            expected = Math.min(expected, Counter.MAX);
+
+            assertEquals(expected, src.count);
+        }
+    }
+
+    public static final class Counter implements Iterator<Integer> {
+        static final int MAX = 5 * RxRingBuffer.SIZE;
+        public int count;
+
+        @Override
+        public boolean hasNext() {
+            return count < MAX;
+        }
+
+        @Override
+        public Integer next() {
+            return ++count;
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException();
+        }
+    }
 }