UnicastSubject extends Subject<T> { @CheckReturnValue @NonNull public static <T> UnicastSubject<T> create() { return new UnicastSubject<T>(bufferSize(), true); } UnicastSubject(int capacityHint, boolean delayError);   UnicastSubject(int capacityHint, Runnable onTerminate);   UnicastSubject(int capacityHint, Runnable onTerminate, boolean delayError); @CheckReturnValue @NonNull static UnicastSubject<T> create(); @CheckReturnValue @NonNull static UnicastSubject<T> create(int capacityHint); @CheckReturnValue @NonNull static UnicastSubject<T> create(int capacityHint, Runnable onTerminate); @CheckReturnValue @Experimental @NonNull static UnicastSubject<T> create(int capacityHint, Runnable onTerminate, boolean delayError); @CheckReturnValue @Experimental @NonNull static UnicastSubject<T> create(boolean delayError); @Override void onSubscribe(Disposable s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override boolean hasObservers(); @Override @Nullable Throwable getThrowable(); @Override boolean hasThrowable(); @Override boolean hasComplete();  }
@Test public void onTerminateCalledWhenCanceled() { final AtomicBoolean didRunOnTerminate = new AtomicBoolean(); UnicastSubject<Integer> us = UnicastSubject.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() { UnicastSubject.create(5, null); }
@Test(expected = IllegalArgumentException.class) public void negativeCapacityHint() { UnicastSubject.create(-1); }
@Test(expected = IllegalArgumentException.class) public void zeroCapacityHint() { UnicastSubject.create(0); }
@Test public void rejectSyncFusion() { UnicastSubject<Object> p = UnicastSubject.create(); TestObserver<Object> to = ObserverFusion.newTest(QueueFuseable.SYNC); p.subscribe(to); ObserverFusion.assertFusion(to, QueueFuseable.NONE); }
@Test public void cancelOnArrival() { UnicastSubject.create() .test(true) .assertEmpty(); }
@Test public void subscribeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final UnicastSubject<Integer> us = UnicastSubject.create(); final TestObserver<Integer> to1 = new TestObserver<Integer>(); final TestObserver<Integer> to2 = new TestObserver<Integer>(); Runnable r1 = new Runnable() { @Override public void run() { us.subscribe(to1); } }; Runnable r2 = new Runnable() { @Override public void run() { us.subscribe(to2); } }; TestHelper.race(r1, r2); if (to1.errorCount() == 0) { to2.assertFailure(IllegalStateException.class); } else if (to2.errorCount() == 0) { to1.assertFailure(IllegalStateException.class); } else { fail("Neither TestObserver failed"); } } }