PublishProcessor extends FlowableProcessor<T> { @Experimental public boolean offer(T t) { if (t == null) { onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources.")); return true; } PublishSubscription<T>[] array = subscribers.get(); for (PublishSubscription<T> s : array) { if (s.isFull()) { return false; } } for (PublishSubscription<T> s : array) { s.onNext(t); } return true; } @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 offer() { PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = pp.test(0); assertFalse(pp.offer(1)); ts.request(1); assertTrue(pp.offer(1)); assertFalse(pp.offer(2)); ts.cancel(); assertTrue(pp.offer(2)); ts = pp.test(0); assertTrue(pp.offer(null)); ts.assertFailure(NullPointerException.class); assertTrue(pp.hasThrowable()); assertTrue(pp.getThrowable().toString(), pp.getThrowable() instanceof NullPointerException); }