Merge pull request #3653 from akarnokd/SampleUnboundedFix1x

1.x: fix sample(Observable) not requesting Long.MAX_VALUE
ReactiveX · Jan 31, 2016 · c5a4902 · c5a4902
2 parents be493f1 + 995d3f1
commit c5a4902
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 9 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorSampleWithObservable.java b/src/main/java/rx/internal/operators/OperatorSampleWithObservable.java
@@ -69,7 +69,7 @@ public void onCompleted() {
 
         };
 
-        Subscriber<T> result = new Subscriber<T>(child) {
+        Subscriber<T> result = new Subscriber<T>() {
             @Override
             public void onNext(T t) {
                 value.set(t);
@@ -88,6 +88,8 @@ public void onCompleted() {
             }
         };
 
+        child.add(result);
+
         sampler.unsafeSubscribe(samplerSub);
 
         return result;

diff --git a/src/test/java/rx/internal/operators/OperatorSampleTest.java b/src/test/java/rx/internal/operators/OperatorSampleTest.java
@@ -16,21 +16,16 @@
 package rx.internal.operators;
 
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.*;
 import org.mockito.InOrder;
 
 import rx.*;
 import rx.Observable.OnSubscribe;
-import rx.functions.Action0;
+import rx.functions.*;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 
@@ -283,4 +278,38 @@ public void call(Subscriber<? super Integer> subscriber) {
         o.throttleLast(1, TimeUnit.MILLISECONDS).subscribe().unsubscribe();
         verify(s).unsubscribe();
     }
+
+    @Test
+    public void testSampleOtherUnboundedIn() {
+
+        final long[] requested = { -1 };
+
+        PublishSubject.create()
+        .doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long t) {
+                requested[0] = t;
+            }
+        })
+        .sample(PublishSubject.create()).subscribe();
+
+        Assert.assertEquals(Long.MAX_VALUE, requested[0]);
+    }
+
+    @Test
+    public void testSampleTimedUnboundedIn() {
+
+        final long[] requested = { -1 };
+
+        PublishSubject.create()
+        .doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long t) {
+                requested[0] = t;
+            }
+        })
+        .sample(1, TimeUnit.SECONDS).subscribe().unsubscribe();
+
+        Assert.assertEquals(Long.MAX_VALUE, requested[0]);
+    }
 }