UnicastProcessor extends FlowableProcessor<T> { @CheckReturnValue @NonNull public static <T> UnicastProcessor<T> create() { return new UnicastProcessor<T>(bufferSize()); } UnicastProcessor(int capacityHint);   UnicastProcessor(int capacityHint, Runnable onTerminate);   UnicastProcessor(int capacityHint, Runnable onTerminate, boolean delayError); @CheckReturnValue @NonNull static UnicastProcessor<T> create(); @CheckReturnValue @NonNull static UnicastProcessor<T> create(int capacityHint); @CheckReturnValue @Experimental @NonNull static UnicastProcessor<T> create(boolean delayError); @CheckReturnValue @NonNull static UnicastProcessor<T> create(int capacityHint, Runnable onCancelled); @CheckReturnValue @Experimental @NonNull static UnicastProcessor<T> create(int capacityHint, Runnable onCancelled, boolean delayError); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override boolean hasSubscribers(); @Override @Nullable Throwable getThrowable(); @Override boolean hasComplete(); @Override boolean hasThrowable();  }
@Test public void onTerminateCalledWhenCanceled() { final AtomicBoolean didRunOnTerminate = new AtomicBoolean(); UnicastProcessor<Integer> us = UnicastProcessor.create(Observable.bufferSize(), new Runnable() { @Override public void run() { didRunOnTerminate.set(true); } }); final Disposable subscribe = us.subscribe(); assertEquals(false, didRunOnTerminate.get()); subscribe.dispose(); assertEquals(true, didRunOnTerminate.get()); }
@Test(expected = NullPointerException.class) public void nullOnTerminate() { UnicastProcessor.create(5, null); }
@Test(expected = IllegalArgumentException.class) public void negativeCapacityHint() { UnicastProcessor.create(-1); }
@Test(expected = IllegalArgumentException.class) public void zeroCapacityHint() { UnicastProcessor.create(0); }
@Test public void rejectSyncFusion() { UnicastProcessor<Object> p = UnicastProcessor.create(); TestSubscriber<Object> ts = SubscriberFusion.newTest(QueueFuseable.SYNC); p.subscribe(ts); SubscriberFusion.assertFusion(ts, QueueFuseable.NONE); }
@Test public void cancelOnArrival() { UnicastProcessor.create() .test(0L, true) .assertEmpty(); }
@Test public void subscribeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final UnicastProcessor<Integer> us = UnicastProcessor.create(); final TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>(); final TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(); Runnable r1 = new Runnable() { @Override public void run() { us.subscribe(ts1); } }; Runnable r2 = new Runnable() { @Override public void run() { us.subscribe(ts2); } }; TestHelper.race(r1, r2); if (ts1.errorCount() == 0) { ts2.assertFailure(IllegalStateException.class); } else if (ts2.errorCount() == 0) { ts1.assertFailure(IllegalStateException.class); } else { fail("Neither TestObserver failed"); } } }
@Test public void unicastSubscriptionBadRequest() { UnicastProcessor<Integer> us = UnicastProcessor.create(false); UnicastProcessor<Integer>.UnicastQueueSubscription usc = us.new UnicastQueueSubscription(); List<Throwable> errors = TestHelper.trackPluginErrors(); try { usc.request(-1); TestHelper.assertError(errors, 0, IllegalArgumentException.class); } finally { RxJavaPlugins.reset(); } }