Merge CancelledSubscriber into EmptySubscriber. (#4083)

Author: JakeWharton

src/main/java/io/reactivex/internal/subscribers/flowable/CancelledSubscriber.java

@@ -18,23 +18,29 @@
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
- * A subscriber that ignores all events (onError is forwarded to RxJavaPlugins though).
+ * A subscriber that ignores all events.
  */
 public enum EmptySubscriber implements Subscriber<Object> {
     /** Empty instance that reports error to the plugins. */
-    INSTANCE(true),
+    INSTANCE(true, false),
     /** Empty instance that doesn't report to the plugins to avoid flooding the test output. */
-    INSTANCE_NOERROR(false);
+    INSTANCE_NOERROR(false, false),
+    /** Empty instance that cancels subscriptions. */
+    CANCELLED(true, true);
 
-    final boolean reportError;
+    private final boolean reportError;
+    private final boolean cancelSubscription;
 
-    EmptySubscriber(boolean reportError) {
+    EmptySubscriber(boolean reportError, boolean cancelSubscription) {
         this.reportError = reportError;
+        this.cancelSubscription = cancelSubscription;
     }
 
     @Override
     public void onSubscribe(Subscription s) {
-        
+        if (cancelSubscription) {
+            s.cancel();
+        }
     }
 
     @Override

src/main/java/io/reactivex/internal/subscribers/flowable/EmptySubscriber.java

@@ -37,7 +37,7 @@ public static <T> Subscriber<T> empty() {
 
     @SuppressWarnings("unchecked")
     public static <T> Subscriber<T> cancelled() {
-        return (Subscriber<T>)CancelledSubscriber.INSTANCE;
+        return (Subscriber<T>)EmptySubscriber.CANCELLED;
     }
 
     public static <T> DisposableSubscriber<T> emptyDisposable() {

src/main/java/io/reactivex/subscribers/Subscribers.java

@@ -29,10 +29,9 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.flowable.TestHelper;
 import io.reactivex.functions.*;
-import io.reactivex.internal.subscribers.flowable.CancelledSubscriber;
 import io.reactivex.processors.ReplayProcessor;
 import io.reactivex.schedulers.*;
-import io.reactivex.subscribers.TestSubscriber;
+import io.reactivex.subscribers.*;
 
 public class FlowableRefCountTest {
 
@@ -441,7 +440,7 @@ public void accept(Long t1) {
 
     @Test
     public void testAlreadyUnsubscribedClient() {
-        Subscriber<Integer> done = CancelledSubscriber.instance();
+        Subscriber<Integer> done = Subscribers.cancelled();
 
         Subscriber<Integer> o = TestHelper.mockSubscriber();
 
@@ -460,7 +459,7 @@ public void testAlreadyUnsubscribedClient() {
     public void testAlreadyUnsubscribedInterleavesWithClient() {
         ReplayProcessor<Integer> source = ReplayProcessor.create();
 
-        Subscriber<Integer> done = CancelledSubscriber.instance();
+        Subscriber<Integer> done = Subscribers.cancelled();
 
         Subscriber<Integer> o = TestHelper.mockSubscriber();
         InOrder inOrder = inOrder(o);