ReplaySubject extends Subject<T, T> { @Experimental public int size() { return state.size(); } 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 testNodeListSimpleAddRemove() { ReplaySubject.NodeList<Integer> list = new ReplaySubject.NodeList<Integer>(); assertEquals(0, list.size()); list.addLast(1); assertEquals(1, list.size()); assertEquals((Integer)1, list.removeFirst()); assertEquals(0, list.size()); list.addLast(1); assertEquals(1, list.size()); assertEquals((Integer)1, list.removeFirst()); list.addLast(1); list.addLast(2); assertEquals(2, list.size()); assertEquals((Integer)1, list.removeFirst()); assertEquals((Integer)2, list.removeFirst()); assertEquals(0, list.size()); list.addLast(1); list.addLast(2); assertEquals(2, list.size()); list.clear(); assertEquals(0, list.size()); }