@Test(timeout = 60000) public void testReplayable() throws Exception { DisposableContentSource<ByteBuf> disposable = new ContentSource<>(sourceRule.getChannel(), sourceRule).replayable(); sourceRule.subscribe(disposable.autoRelease(), 1); sourceRule.subscribe(disposable.autoRelease(), 1); assertThat("Unexpected ref count before dispose.", sourceRule.getData().refCnt(), is(1)); disposable.dispose(); assertThat("Unexpected ref count after dispose.", sourceRule.getData().refCnt(), is(0)); }
public DisposableContentSource<T> replayable() { return DisposableContentSource.createNew(this); }
ContentSource extends Observable<T> { public DisposableContentSource<T> replayable() { return DisposableContentSource.createNew(this); } }
ContentSource extends Observable<T> { public DisposableContentSource<T> replayable() { return DisposableContentSource.createNew(this); } private  ContentSource(final Observable<T> source);  ContentSource(final Channel channel, final Func1<Subscriber<? super T>, Object> subscriptionEventFactory);  ContentSource(final Throwable error); }
ContentSource extends Observable<T> { public DisposableContentSource<T> replayable() { return DisposableContentSource.createNew(this); } private  ContentSource(final Observable<T> source);  ContentSource(final Channel channel, final Func1<Subscriber<? super T>, Object> subscriptionEventFactory);  ContentSource(final Throwable error); Observable<T> autoRelease(); DisposableContentSource<T> replayable(); ContentSource<R> transform(Transformer<T, R> transformer); }
ContentSource extends Observable<T> { public DisposableContentSource<T> replayable() { return DisposableContentSource.createNew(this); } private  ContentSource(final Observable<T> source);  ContentSource(final Channel channel, final Func1<Subscriber<? super T>, Object> subscriptionEventFactory);  ContentSource(final Throwable error); Observable<T> autoRelease(); DisposableContentSource<T> replayable(); ContentSource<R> transform(Transformer<T, R> transformer);  }