BehaviorProcessor 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; } BehaviorSubscription<T>[] array = subscribers.get(); for (BehaviorSubscription<T> s : array) { if (s.isFull()) { return false; } } Object o = NotificationLite.next(t); setCurrent(o); for (BehaviorSubscription<T> bs : array) { bs.emitNext(o, index); } return true; } @SuppressWarnings("unchecked")  BehaviorProcessor();   BehaviorProcessor(T defaultValue); @CheckReturnValue @NonNull static BehaviorProcessor<T> create(); @CheckReturnValue @NonNull static BehaviorProcessor<T> createDefault(T defaultValue); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Experimental boolean offer(T t); @Override boolean hasSubscribers(); @Override @Nullable Throwable getThrowable(); @Nullable T getValue(); Object[] getValues(); @SuppressWarnings("unchecked") T[] getValues(T[] array); @Override boolean hasComplete(); @Override boolean hasThrowable(); boolean hasValue();  }
@Test public void offer() { BehaviorProcessor<Integer> pp = BehaviorProcessor.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(1); assertTrue(pp.offer(null)); ts.assertFailure(NullPointerException.class, 2); assertTrue(pp.hasThrowable()); assertTrue(pp.getThrowable().toString(), pp.getThrowable() instanceof NullPointerException); }