PublishProcessor extends FlowableProcessor<T> { @CheckReturnValue @NonNull public static <T> PublishProcessor<T> create() { return new PublishProcessor<T>(); } @SuppressWarnings("unchecked")  PublishProcessor(); @CheckReturnValue @NonNull static PublishProcessor<T> create(); @Override void subscribeActual(Subscriber<? super T> t); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @SuppressWarnings("unchecked") @Override void onError(Throwable t); @SuppressWarnings("unchecked") @Override void onComplete(); @Experimental boolean offer(T t); @Override boolean hasSubscribers(); @Override @Nullable Throwable getThrowable(); @Override boolean hasThrowable(); @Override boolean hasComplete();  }
@Test public void requestValidation() { TestHelper.assertBadRequestReported(PublishProcessor.create()); }
@Test public void addRemoveRance() throws Exception { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishProcessor<Integer> pp = PublishProcessor.create(); final TestSubscriber<Integer> ts = pp.test(); Runnable r1 = new Runnable() { @Override public void run() { pp.subscribe(); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; TestHelper.race(r1, r2); } }