1.x: fix unsubscription and producer issues in sample(other)

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

@@ -16,9 +16,9 @@
 package rx.internal.operators;
 
 import java.util.concurrent.atomic.AtomicReference;
-import rx.Observable;
+
+import rx.*;
 import rx.Observable.Operator;
-import rx.Subscriber;
 import rx.observers.SerializedSubscriber;
 
 /**
@@ -44,7 +44,9 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
 
         final AtomicReference<Object> value = new AtomicReference<Object>(EMPTY_TOKEN);
 
-        Subscriber<U> samplerSub = new Subscriber<U>(child) {
+        final AtomicReference<Subscription> main = new AtomicReference<Subscription>();
+
+        final Subscriber<U> samplerSub = new Subscriber<U>() {
             @Override
             public void onNext(U t) {
                 Object localValue = value.getAndSet(EMPTY_TOKEN);
@@ -58,15 +60,17 @@ public void onNext(U t) {
             @Override
             public void onError(Throwable e) {
                 s.onError(e);
-                unsubscribe();
+                // no need to null check, main is assigned before any of the two gets subscribed
+                main.get().unsubscribe();
             }
 
             @Override
             public void onCompleted() {
+                // onNext(null); // emit the very last value?
                 s.onCompleted();
-                unsubscribe();
+                // no need to null check, main is assigned before any of the two gets subscribed
+                main.get().unsubscribe();
             }
-
         };
 
         Subscriber<T> result = new Subscriber<T>() {
@@ -78,17 +82,23 @@ public void onNext(T t) {
             @Override
             public void onError(Throwable e) {
                 s.onError(e);
-                unsubscribe();
+
+                samplerSub.unsubscribe();
             }
 
             @Override
             public void onCompleted() {
+                // samplerSub.onNext(null); // emit the very last value?
                 s.onCompleted();
-                unsubscribe();
+
+                samplerSub.unsubscribe();
             }
         };
 
+        main.lazySet(result);
+
         child.add(result);
+        child.add(samplerSub);
 
         sampler.unsafeSubscribe(samplerSub);
 

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

@@ -19,15 +19,18 @@
 import static org.mockito.Mockito.*;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.*;
 
 import org.junit.*;
 import org.mockito.InOrder;
 
 import rx.*;
 import rx.Observable.OnSubscribe;
 import rx.functions.*;
+import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
+import rx.subscriptions.Subscriptions;
 
 public class OperatorSampleTest {
     private TestScheduler scheduler;
@@ -312,4 +315,155 @@ public void call(Long t) {
 
         Assert.assertEquals(Long.MAX_VALUE, requested[0]);
     }
+
+    @Test
+    public void dontUnsubscribeChild1() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        PublishSubject<Integer> sampler = PublishSubject.create();
+
+        source.sample(sampler).unsafeSubscribe(ts);
+
+        source.onCompleted();
+
+        Assert.assertFalse("Source has subscribers?", source.hasObservers());
+        Assert.assertFalse("Sampler has subscribers?", sampler.hasObservers());
+
+        Assert.assertFalse("TS unsubscribed?", ts.isUnsubscribed());
+    }
+
+    @Test
+    public void dontUnsubscribeChild2() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        PublishSubject<Integer> sampler = PublishSubject.create();
+
+        source.sample(sampler).unsafeSubscribe(ts);
+
+        sampler.onCompleted();
+
+        Assert.assertFalse("Source has subscribers?", source.hasObservers());
+        Assert.assertFalse("Sampler has subscribers?", sampler.hasObservers());
+
+        Assert.assertFalse("TS unsubscribed?", ts.isUnsubscribed());
+    }
+
+    @Test
+    public void neverSetProducer() {
+        Observable<Integer> neverBackpressure = Observable.create(new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> t) {
+                t.setProducer(new Producer() {
+                    @Override
+                    public void request(long n) {
+                        // irrelevant in this test
+                    }
+                });
+            }
+        });
+
+        final AtomicInteger count = new AtomicInteger();
+
+        neverBackpressure.sample(neverBackpressure).unsafeSubscribe(new Subscriber<Integer>() {
+            @Override
+            public void onNext(Integer t) {
+                // irrelevant
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                // irrelevant
+            }
+
+            @Override
+            public void onCompleted() {
+                // irrelevant
+            }
+
+            @Override
+            public void setProducer(Producer p) {
+                count.incrementAndGet();
+            }
+        });
+
+        Assert.assertEquals(0, count.get());
+    }
+
+    @Test
+    public void unsubscribeMainAfterCompleted() {
+        final AtomicBoolean unsubscribed = new AtomicBoolean();
+
+        Observable<Integer> source = Observable.create(new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> t) {
+                t.add(Subscriptions.create(new Action0() {
+                    @Override
+                    public void call() {
+                        unsubscribed.set(true);
+                    }
+                }));
+            }
+        });
+
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onCompleted() {
+                if (unsubscribed.get()) {
+                    onError(new IllegalStateException("Resource unsubscribed!"));
+                } else {
+                    super.onCompleted();
+                }
+            }
+        };
+
+        PublishSubject<Integer> sampler = PublishSubject.create();
+
+        source.sample(sampler).unsafeSubscribe(ts);
+
+        sampler.onCompleted();
+
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    @Test
+    public void unsubscribeSamplerAfterCompleted() {
+        final AtomicBoolean unsubscribed = new AtomicBoolean();
+
+        Observable<Integer> source = Observable.create(new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> t) {
+                t.add(Subscriptions.create(new Action0() {
+                    @Override
+                    public void call() {
+                        unsubscribed.set(true);
+                    }
+                }));
+            }
+        });
+
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onCompleted() {
+                if (unsubscribed.get()) {
+                    onError(new IllegalStateException("Resource unsubscribed!"));
+                } else {
+                    super.onCompleted();
+                }
+            }
+        };
+
+        PublishSubject<Integer> sampled = PublishSubject.create();
+
+        sampled.sample(source).unsafeSubscribe(ts);
+
+        sampled.onCompleted();
+
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
 }