Fixed the issue of takeLast(items, 0)

Author: zsxwing

rxjava-core/src/main/java/rx/operators/OperationTakeLast.java

@@ -26,6 +26,7 @@
 
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
+import rx.subscriptions.Subscriptions;
 import rx.Observer;
 import rx.Subscription;
 
@@ -59,6 +60,26 @@ private static class TakeLast<T> implements OnSubscribeFunc<T> {
         }
 
         public Subscription onSubscribe(Observer<? super T> observer) {
+            if(count == 0) {
+                items.subscribe(new Observer<T>() {
+
+                    @Override
+                    public void onCompleted() {
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                    }
+
+                    @Override
+                    public void onNext(T args) {
+                    }
+
+                }).unsubscribe();
+                observer.onCompleted();
+                return Subscriptions.empty();
+            }
+
             return subscription.wrap(items.subscribe(new ItemObserver(observer)));
         }
 
@@ -140,6 +161,19 @@ public void testTakeLast2() {
             verify(aObserver, times(1)).onCompleted();
         }
 
+        @Test
+        public void testTakeLastWithZeroCount() {
+            Observable<String> w = Observable.from("one");
+            Observable<String> take = Observable.create(takeLast(w, 0));
+
+            @SuppressWarnings("unchecked")
+            Observer<String> aObserver = mock(Observer.class);
+            take.subscribe(aObserver);
+            verify(aObserver, never()).onNext("one");
+            verify(aObserver, never()).onError(any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+        }
+
     }
 
 }