BehaviorProcessor extends FlowableProcessor<T> { @CheckReturnValue @NonNull public static <T> BehaviorProcessor<T> create() { return new BehaviorProcessor<T>(); } @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 addRemoveRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final BehaviorProcessor<Object> p = BehaviorProcessor.create(); final TestSubscriber<Object> ts = p.test(); Runnable r1 = new Runnable() { @Override public void run() { p.test(); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; TestHelper.race(r1, r2); } }