Fix for #1791 - don't retry (subscribe) to source if child has unsubscribed.

Author: Alex Wenckus

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

@@ -206,6 +206,10 @@ public void call(final Subscriber<? super T> child) {
         final Action0 subscribeToSource = new Action0() {
             @Override
             public void call() {
+                if (child.isUnsubscribed()) {
+                    return;
+                }
+
                 Subscriber<T> terminalDelegatingSubscriber = new Subscriber<T>() {
                     @Override
                     public void onCompleted() {

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

@@ -244,6 +244,37 @@ public Observable<?> call(Observable<? extends Throwable> t1) {
         inOrder.verifyNoMoreInteractions();
     }
 
+    @Test
+    public void testSingleSubscriptionOnFirst() throws Exception {
+        final AtomicInteger inc = new AtomicInteger(0);
+        Observable.OnSubscribe<Integer> onSubscribe = new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> subscriber) {
+                final int emit = inc.incrementAndGet();
+                subscriber.onNext(emit);
+                subscriber.onCompleted();
+            }
+        };
+
+        int first = Observable.create(onSubscribe)
+                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
+                    @Override
+                    public Observable<?> call(Observable<? extends Throwable> attempt) {
+                        return attempt.zipWith(Observable.just(1), new Func2<Throwable, Integer, Void>() {
+                            @Override
+                            public Void call(Throwable o, Integer integer) {
+                                return null;
+                            }
+                        });
+                    }
+                })
+                .toBlocking()
+                .first();
+
+        assertEquals("Observer did not receive the expected output", 1, first);
+        assertEquals("Subscribe was not called once", 1, inc.get());
+    }
+
     @Test
     public void testOriginFails() {
         @SuppressWarnings("unchecked")