EndConsumerHelper { public static String composeMessage(String consumer) { return "It is not allowed to subscribe with a(n) " + consumer + " multiple times. " + "Please create a fresh instance of " + consumer + " and subscribe that to the target source instead."; } private  EndConsumerHelper(); static boolean validate(Disposable upstream, Disposable next, Class<?> observer); static boolean setOnce(AtomicReference<Disposable> upstream, Disposable next, Class<?> observer); static boolean validate(Subscription upstream, Subscription next, Class<?> subscriber); static boolean setOnce(AtomicReference<Subscription> upstream, Subscription next, Class<?> subscriber); static String composeMessage(String consumer); static void reportDoubleSubscription(Class<?> consumer);  }
@Test public void checkDoubleDefaultSubscriber() { Subscriber<Integer> consumer = new DefaultSubscriber<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; BooleanSubscription sub1 = new BooleanSubscription(); consumer.onSubscribe(sub1); assertFalse(sub1.isCancelled()); BooleanSubscription sub2 = new BooleanSubscription(); consumer.onSubscribe(sub2); assertFalse(sub1.isCancelled()); assertTrue(sub2.isCancelled()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDefaultSubscriberNonAnonymous() { Subscriber<Integer> consumer = new EndDefaultSubscriber(); BooleanSubscription sub1 = new BooleanSubscription(); consumer.onSubscribe(sub1); assertFalse(sub1.isCancelled()); BooleanSubscription sub2 = new BooleanSubscription(); consumer.onSubscribe(sub2); assertFalse(sub1.isCancelled()); assertTrue(sub2.isCancelled()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage("io.reactivex.internal.util.EndConsumerHelperTest$EndDefaultSubscriber"), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDisposableSubscriber() { Subscriber<Integer> consumer = new DisposableSubscriber<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; BooleanSubscription sub1 = new BooleanSubscription(); consumer.onSubscribe(sub1); assertFalse(sub1.isCancelled()); BooleanSubscription sub2 = new BooleanSubscription(); consumer.onSubscribe(sub2); assertFalse(sub1.isCancelled()); assertTrue(sub2.isCancelled()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleResourceSubscriber() { Subscriber<Integer> consumer = new ResourceSubscriber<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; BooleanSubscription sub1 = new BooleanSubscription(); consumer.onSubscribe(sub1); assertFalse(sub1.isCancelled()); BooleanSubscription sub2 = new BooleanSubscription(); consumer.onSubscribe(sub2); assertFalse(sub1.isCancelled()); assertTrue(sub2.isCancelled()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDefaultObserver() { Observer<Integer> consumer = new DefaultObserver<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDisposableObserver() { Observer<Integer> consumer = new DisposableObserver<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleResourceObserver() { Observer<Integer> consumer = new ResourceObserver<Integer>() { @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDisposableSingleObserver() { SingleObserver<Integer> consumer = new DisposableSingleObserver<Integer>() { @Override public void onSuccess(Integer t) { } @Override public void onError(Throwable t) { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleResourceSingleObserver() { SingleObserver<Integer> consumer = new ResourceSingleObserver<Integer>() { @Override public void onSuccess(Integer t) { } @Override public void onError(Throwable t) { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDisposableMaybeObserver() { MaybeObserver<Integer> consumer = new DisposableMaybeObserver<Integer>() { @Override public void onSuccess(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleResourceMaybeObserver() { MaybeObserver<Integer> consumer = new ResourceMaybeObserver<Integer>() { @Override public void onSuccess(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleDisposableCompletableObserver() { CompletableObserver consumer = new DisposableCompletableObserver() { @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }
@Test public void checkDoubleResourceCompletableObserver() { CompletableObserver consumer = new ResourceCompletableObserver() { @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Disposable sub1 = Disposables.empty(); consumer.onSubscribe(sub1); assertFalse(sub1.isDisposed()); Disposable sub2 = Disposables.empty(); consumer.onSubscribe(sub2); assertFalse(sub1.isDisposed()); assertTrue(sub2.isDisposed()); TestHelper.assertError(errors, 0, ProtocolViolationException.class); assertEquals(EndConsumerHelper.composeMessage(consumer.getClass().getName()), errors.get(0).getMessage()); assertEquals(errors.toString(), 1, errors.size()); }