Observable.from(iterable) should emit onCompleted even if none reques…

…ted when iterable is empty
ReactiveX · Apr 21, 2015 · 1c29e7c · 1c29e7c
1 parent 9d90aed
commit 1c29e7c
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 3 deletions.
diff --git a/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java b/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java
@@ -44,7 +44,10 @@ public OnSubscribeFromIterable(Iterable<? extends T> iterable) {
     @Override
     public void call(final Subscriber<? super T> o) {
         final Iterator<? extends T> it = is.iterator();
-        o.setProducer(new IterableProducer<T>(o, it));
+        if (!it.hasNext() && !o.isUnsubscribed())
+            o.onCompleted();
+        else 
+            o.setProducer(new IterableProducer<T>(o, it));
     }
 
     private static final class IterableProducer<T> implements Producer {

diff --git a/src/test/java/rx/internal/operators/OnSubscribeFromIterableTest.java b/src/test/java/rx/internal/operators/OnSubscribeFromIterableTest.java
@@ -27,6 +27,7 @@
 import java.util.Iterator;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -74,12 +75,12 @@ public Iterator<String> iterator() {
 
                     @Override
                     public boolean hasNext() {
-                        return i++ < 3;
+                        return i < 3;
                     }
 
                     @Override
                     public String next() {
-                        return String.valueOf(i);
+                        return String.valueOf(++i);
                     }
 
                     @Override
@@ -193,5 +194,31 @@ public void onNext(Integer t) {
         assertTrue(latch.await(10, TimeUnit.SECONDS));
     }
 
+    @Test
+    public void testFromEmptyIterableWhenZeroRequestedShouldStillEmitOnCompletedEagerly() {
+        final AtomicBoolean completed = new AtomicBoolean(false);
+        Observable.from(Collections.emptyList()).subscribe(new Subscriber<Object>() {
+
+            @Override
+            public void onStart() {
+                request(0);
+            }
+
+            @Override
+            public void onCompleted() {
+                completed.set(true);
+            }
 
+            @Override
+            public void onError(Throwable e) {
+
+            }
+
+            @Override
+            public void onNext(Object t) {
+
+            }});
+        assertTrue(completed.get());
+    }
+
 }