ReplaySubject extends Subject<T, T> { @SuppressWarnings("unchecked") @Experimental public Object[] getValues() { return state.toArray((T[])EMPTY_ARRAY); } ReplaySubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> ssm, ReplayState<T, ?> state); static ReplaySubject<T> create(); static ReplaySubject<T> create(int capacity); static ReplaySubject<T> createWithSize(int size); static ReplaySubject<T> createWithTime(long time, TimeUnit unit, final Scheduler scheduler); static ReplaySubject<T> createWithTimeAndSize(long time, TimeUnit unit, int size, final Scheduler scheduler); @Override void onNext(T t); @Override void onError(final Throwable e); @Override void onCompleted(); @Override boolean hasObservers(); @Experimental boolean hasThrowable(); @Experimental boolean hasCompleted(); @Experimental Throwable getThrowable(); @Experimental int size(); @Experimental boolean hasAnyValue(); @SuppressWarnings("unchecked") @Experimental Object[] getValues(); @Experimental T[] getValues(T[] a);  }
@Test public void testGetValues() { ReplaySubject<Object> rs = ReplaySubject.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.onCompleted(); assertArrayEquals(expected, rs.getValues()); }