Fix the issue that Sample doesn't call 'unsubscribe'

ReactiveX · Dec 12, 2014 · 6c958e3 · 6c958e3
1 parent dd73c15
commit 6c958e3
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 4 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorSampleWithTime.java b/src/main/java/rx/internal/operators/OperatorSampleWithTime.java
@@ -68,6 +68,7 @@ static final class SamplerSubscriber<T> extends Subscriber<T> implements Action0
         static final AtomicReferenceFieldUpdater<SamplerSubscriber, Object> VALUE_UPDATER
                 = AtomicReferenceFieldUpdater.newUpdater(SamplerSubscriber.class, Object.class, "value");
         public SamplerSubscriber(Subscriber<? super T> subscriber) {
+            super(subscriber);
             this.subscriber = subscriber;
         }
 

diff --git a/src/test/java/rx/internal/operators/OperatorSampleTest.java b/src/test/java/rx/internal/operators/OperatorSampleTest.java
@@ -28,14 +28,12 @@
 import org.junit.Test;
 import org.mockito.InOrder;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.OnSubscribe;
-import rx.Observer;
-import rx.Scheduler;
-import rx.Subscriber;
 import rx.functions.Action0;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
+import rx.subscriptions.Subscriptions;
 
 public class OperatorSampleTest {
     private TestScheduler scheduler;
@@ -271,4 +269,19 @@ public void sampleWithSamplerThrows() {
         inOrder.verify(observer2, times(1)).onError(any(RuntimeException.class));
         verify(observer, never()).onCompleted();
     }
+
+    @Test
+    public void testSampleUnsubscribe() {
+        final Subscription s = mock(Subscription.class);
+        Observable<Integer> o = Observable.create(
+                new OnSubscribe<Integer>() {
+                    @Override
+                    public void call(Subscriber<? super Integer> subscriber) {
+                        subscriber.add(s);
+                    }
+                }
+        );
+        o.throttleLast(1, TimeUnit.MILLISECONDS).subscribe().unsubscribe();
+        verify(s).unsubscribe();
+    }
 }