1.x: fix subscribe(Action1 [, Action1]) to report isUnsubscribed (#4716)

Author: akarnokd

src/main/java/rx/Single.java

@@ -1535,19 +1535,7 @@ public final Single<T> onErrorResumeNext(final Func1<Throwable, ? extends Single
      * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
      */
     public final Subscription subscribe() {
-        return subscribe(new SingleSubscriber<T>() {
-
-            @Override
-            public final void onError(Throwable e) {
-                throw new OnErrorNotImplementedException(e);
-            }
-
-            @Override
-            public final void onSuccess(T args) {
-                // do nothing
-            }
-
-        });
+        return subscribe(Actions.empty(), Actions.errorNotImplemented());
     }
 
     /**
@@ -1567,23 +1555,7 @@ public final void onSuccess(T args) {
      * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
      */
     public final Subscription subscribe(final Action1<? super T> onSuccess) {
-        if (onSuccess == null) {
-            throw new IllegalArgumentException("onSuccess can not be null");
-        }
-
-        return subscribe(new SingleSubscriber<T>() {
-
-            @Override
-            public final void onError(Throwable e) {
-                throw new OnErrorNotImplementedException(e);
-            }
-
-            @Override
-            public final void onSuccess(T args) {
-                onSuccess.call(args);
-            }
-
-        });
+        return subscribe(onSuccess, Actions.errorNotImplemented());
     }
 
     /**
@@ -1617,12 +1589,20 @@ public final Subscription subscribe(final Action1<? super T> onSuccess, final Ac
 
             @Override
             public final void onError(Throwable e) {
-                onError.call(e);
+                try {
+                    onError.call(e);
+                } finally {
+                    unsubscribe();
+                }
             }
 
             @Override
             public final void onSuccess(T args) {
-                onSuccess.call(args);
+                try {
+                    onSuccess.call(args);
+                } finally {
+                    unsubscribe();
+                }
             }
 
         });

src/main/java/rx/functions/Actions.java

@@ -15,6 +15,8 @@
  */
 package rx.functions;
 
+import rx.exceptions.OnErrorNotImplementedException;
+
 /**
  * Utility class for the Action interfaces.
  */
@@ -566,4 +568,20 @@ public void call(T t) {
             action.call();
         }
     }
+
+    enum NotImplemented implements Action1<Throwable> {
+        INSTANCE;
+        @Override
+        public void call(Throwable t) {
+            throw new OnErrorNotImplementedException(t);
+        }
+    }
+
+    /**
+     * Returns an action which throws OnErrorNotImplementedException.
+     * @return the the shared action
+     */
+    public static Action1<Throwable> errorNotImplemented() {
+        return NotImplemented.INSTANCE;
+    }
 }

src/test/java/rx/SingleTest.java

@@ -2136,4 +2136,51 @@ public void call(final Notification<? extends Integer> notification) {
 
         assertEquals(1, atomicInteger.get());
     }
+
+    @Test
+    public void isUnsubscribedAfterSuccess() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        final int[] calls = { 0 };
+
+        Subscription s = ps.toSingle().subscribe(new Action1<Integer>() {
+            @Override
+            public void call(Integer t) {
+                calls[0]++;
+            }
+        });
+
+        assertFalse(s.isUnsubscribed());
+
+        ps.onNext(1);
+        ps.onCompleted();
+
+        assertTrue(s.isUnsubscribed());
+
+        assertEquals(1, calls[0]);
+    }
+
+    @Test
+    public void isUnsubscribedAfterError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        final int[] calls = { 0 };
+
+        Action1<Integer> a = Actions.empty();
+
+        Subscription s = ps.toSingle().subscribe(a, new Action1<Throwable>() {
+            @Override
+            public void call(Throwable t) {
+                calls[0]++;
+            }
+        });
+
+        assertFalse(s.isUnsubscribed());
+
+        ps.onError(new TestException());
+
+        assertTrue(s.isUnsubscribed());
+
+        assertEquals(1, calls[0]);
+    }
 }