ReplayProcessor extends FlowableProcessor<T> { @CheckReturnValue @NonNull public static <T> ReplayProcessor<T> create() { return new ReplayProcessor<T>(new UnboundedReplayBuffer<T>(16)); } @SuppressWarnings("unchecked")  ReplayProcessor(ReplayBuffer<T> buffer); @CheckReturnValue @NonNull static ReplayProcessor<T> create(); @CheckReturnValue @NonNull static ReplayProcessor<T> create(int capacityHint); @CheckReturnValue @NonNull static ReplayProcessor<T> createWithSize(int maxSize); @CheckReturnValue @NonNull static ReplayProcessor<T> createWithTime(long maxAge, TimeUnit unit, Scheduler scheduler); @CheckReturnValue @NonNull static ReplayProcessor<T> createWithTimeAndSize(long maxAge, TimeUnit unit, Scheduler scheduler, int maxSize); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @SuppressWarnings("unchecked") @Override void onError(Throwable t); @SuppressWarnings("unchecked") @Override void onComplete(); @Override boolean hasSubscribers(); @Override @Nullable Throwable getThrowable(); @Experimental void cleanupBuffer(); T getValue(); Object[] getValues(); T[] getValues(T[] array); @Override boolean hasComplete(); @Override boolean hasThrowable(); boolean hasValue();  }
@Test public void createInvalidCapacity() { try { ReplayProcessor.create(-99); fail("Didn't throw IllegalArgumentException"); } catch (IllegalArgumentException ex) { assertEquals("capacityHint > 0 required but it was -99", ex.getMessage()); } }
@Test public void subscribeCancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); final ReplayProcessor<Integer> rp = ReplayProcessor.create(); Runnable r1 = new Runnable() { @Override public void run() { rp.subscribe(ts); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; TestHelper.race(r1, r2); } }
@Test public void subscribeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ReplayProcessor<Integer> rp = ReplayProcessor.create(); Runnable r1 = new Runnable() { @Override public void run() { rp.test(); } }; TestHelper.race(r1, r1); } }
@Test public void invalidRequest() { TestHelper.assertBadRequestReported(ReplayProcessor.create()); }