Merge pull request #2244 from davidmoten/take-last-unsub

OperatorTakeLast add check for isUnsubscribed to fast path

Author: zsxwing

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

@@ -70,6 +70,8 @@ void emit(long previousRequested) {
             if (previousRequested == 0) {
                 try {
                     for (Object value : deque) {
+                        if (subscriber.isUnsubscribed())
+                            return;
                         notification.accept(subscriber, value);
                     }
                 } catch (Throwable e) {

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

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.verify;
 
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -264,4 +265,32 @@ public void onNext(Integer integer) {
             }
         });
     }
+    
+    @Test
+    public void testUnsubscribeTakesEffectEarlyOnFastPath() {
+        final AtomicInteger count = new AtomicInteger();
+        Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Integer integer) {
+                count.incrementAndGet();
+                unsubscribe();
+            }
+        });
+        assertEquals(1,count.get());
+    }
 }