ReplayProcessor extends FlowableProcessor<T> { public Object[] getValues() { @SuppressWarnings("unchecked") T[] a = (T[])EMPTY_ARRAY; T[] b = getValues(a); if (b == EMPTY_ARRAY) { return new Object[0]; } return b; } @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 testGetValues() { ReplayProcessor<Object> rs = ReplayProcessor.create(); Object[] expected = new Object[10]; for (int i = 0; i < expected.length; i++) { expected[i] = i; rs.onNext(i); assertArrayEquals(Arrays.copyOf(expected, i + 1), rs.getValues()); } rs.onComplete(); assertArrayEquals(expected, rs.getValues()); }