take(0) subscribes to its source

ReactiveX · Mar 30, 2013 · 6c1a1ab · 6c1a1ab
1 parent 550005c
commit 6c1a1ab
Showing 1 changed file with 40 additions and 1 deletion.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTake.java b/rxjava-core/src/main/java/rx/operators/OperationTake.java
@@ -23,8 +23,10 @@
 import rx.util.AtomicObservableSubscription;
 import rx.util.functions.Func1;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
@@ -82,6 +84,23 @@ private Take(Observable<T> items, int num) {
         @Override
         public Subscription call(Observer<T> observer) {
             if (num < 1) {
+                items.subscribe(new Observer<T>()
+                {
+                    @Override
+                    public void onCompleted()
+                    {
+                    }
+
+                    @Override
+                    public void onError(Exception e)
+                    {
+                    }
+
+                    @Override
+                    public void onNext(T args)
+                    {
+                    }
+                }).unsubscribe();
                 observer.onCompleted();
                 return Subscriptions.empty();
             }
@@ -178,8 +197,28 @@ public Subscription call(Observer<String> observer)
 
         @Test
         public void testTakeZeroDoesntLeakError() {
-            Observable<String> source = Observable.error(new Exception("test failed"));
+            final AtomicBoolean subscribed = new AtomicBoolean(false);
+            final AtomicBoolean unSubscribed = new AtomicBoolean(false);
+            Observable<String> source = Observable.create(new Func1<Observer<String>, Subscription>()
+            {
+                @Override
+                public Subscription call(Observer<String> observer)
+                {
+                    subscribed.set(true);
+                    observer.onError(new Exception("test failed"));
+                    return new Subscription()
+                    {
+                        @Override
+                        public void unsubscribe()
+                        {
+                            unSubscribed.set(true);
+                        }
+                    };
+                }
+            });
             Observable.create(assertTrustedObservable(take(source, 0))).lastOrDefault("ok");
+            assertTrue("source subscribed", subscribed.get());
+            assertTrue("source unsubscribed", unSubscribed.get());
         }
 
         @Test