Make BlockingOperatorToIterator exert backpressure.

Author: vqvu

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

@@ -23,7 +23,6 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
-import rx.Subscription;
 import rx.exceptions.Exceptions;
 
 /**
@@ -47,68 +46,82 @@ 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);
-            }
-        });
+        private final BlockingQueue<Notification<? extends T>> notifications;
+        private Notification<? extends T> buf;
 
-        return new Iterator<T>() {
-            private Notification<? extends T> buf;
+        public SubscriberIterator() {
+            this.notifications = new LinkedBlockingQueue<Notification<? extends T>>();
+            this.buf = null;
+        }
 
-            @Override
-            public boolean hasNext() {
-                if (buf == null) {
-                    buf = take();
-                }
-                if (buf.isOnError()) {
-                    throw Exceptions.propagate(buf.getThrowable());
-                }
-                return !buf.isOnCompleted();
+        @Override
+        public void onStart() {
+            request(0);
+        }
+
+        @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) {
+                request(1);
+                buf = take();
+            }
+            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

@@ -81,4 +81,46 @@ 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.
+            int i = it.next();
+            assertEquals(i + 1, src.count);
+        }
+    }
+
+    public static final class Counter implements Iterator<Integer> {
+        public int count;
+
+        public Counter() {
+            this.count = 0;
+        }
+
+        @Override
+        public boolean hasNext() {
+            return count < 5;
+        }
+
+        @Override
+        public Integer next() {
+            return count++;
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException();
+        }
+    }
 }