@Test public void testResubscribeShouldAlwaysWork() throws Exception { HttpSource<ServerSentEvent, ServerSentEvent> source = HttpSources .source( HttpClientFactories.sseClientFactory(), HttpRequestFactories.createGetFactory("test/stream")) .withServerProvider(localServerProvider) .withActivityObserver(sourceObserver) .build(); int totalCount = 5; final CountDownLatch latch = new CountDownLatch(totalCount); Observable<ServerSentEvent> stream = Observable.merge(source.call(new Context(), new Index(1, 1))); Subscription sub = stream.subscribe(new Action1<ServerSentEvent>() { @Override public void call(ServerSentEvent event) { latch.countDown(); } }); long waitSeconds = 10; boolean countedDown = latch.await(waitSeconds, TimeUnit.SECONDS); if (!countedDown) { fail(String.format("Waited too long to receive %d events within %d seconds. Total counted: %d", totalCount, waitSeconds, latch.getCount())); } sub.unsubscribe(); final CountDownLatch newLatch = new CountDownLatch(totalCount); sub = stream.subscribe(new Action1<ServerSentEvent>() { @Override public void call(ServerSentEvent event) { newLatch.countDown(); } }); countedDown = newLatch.await(5, TimeUnit.SECONDS); if (!countedDown) { fail("Waited too long to receive enough events. Counted: " + latch.getCount()); } sub.unsubscribe(); }
@Override public Observable<Observable<T>> call(Context context, Index index) { return impl.call(context, index); }
HttpSource implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return impl.call(context, index); } }
HttpSource implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return impl.call(context, index); } private  HttpSource(HttpSourceImpl<ByteBuf, E, T> impl); }
HttpSource implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return impl.call(context, index); } private  HttpSource(HttpSourceImpl<ByteBuf, E, T> impl); static Builder<E, T> builder(HttpSourceImpl.Builder<ByteBuf, E, T> builderImpl); static Builder<E, E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory); static Builder<E, E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory, ClientResumePolicy<ByteBuf, E> policy); @Override Observable<Observable<T>> call(Context context, Index index); }
HttpSource implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return impl.call(context, index); } private  HttpSource(HttpSourceImpl<ByteBuf, E, T> impl); static Builder<E, T> builder(HttpSourceImpl.Builder<ByteBuf, E, T> builderImpl); static Builder<E, E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory); static Builder<E, E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory, ClientResumePolicy<ByteBuf, E> policy); @Override Observable<Observable<T>> call(Context context, Index index);  }