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

Author: zsxwing

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

@@ -50,6 +50,7 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
         child.add(worker);
 
         SamplerSubscriber<T> sampler = new SamplerSubscriber<T>(s);
+        child.add(sampler);
         worker.schedulePeriodically(sampler, time, time, unit);
 
         return sampler;

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();
+    }
 }