@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()); }
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."; }
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."; } }
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(); }
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); }
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);  }