@Test public void pollingSourceWillWork() throws Exception { ServerInfo server = localServerProvider.getServerInfos().get(0); HttpSource<ByteBuf, String> source = HttpSources .pollingSource(server.getHost(), server.getPort(), "test/singleEntity") .withActivityObserver(sourceObserver) .build(); final AtomicInteger counter = new AtomicInteger(); final int maxRepeat = 10; final CountDownLatch done = new CountDownLatch(maxRepeat); final ConcurrentHashMap<String, AtomicInteger> result = new ConcurrentHashMap<>(); Subscription subscription = Observable.merge(source.call(new Context(), new Index(1, 1))) .doOnNext(new Action1<String>() { @Override public void call(String content) { assertEquals(RequestProcessor.SINGLE_ENTITY_RESPONSE, content); counter.incrementAndGet(); result.putIfAbsent(content, new AtomicInteger()); result.get(content).incrementAndGet(); done.countDown(); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail("Unexpected failure: " + throwable); } }) .subscribe(); long waitSeconds = 3; boolean timedout = !done.await(waitSeconds, TimeUnit.SECONDS); if (timedout) { fail(String.format("Waited at least %d seconds for the test to finish. Something is wrong", waitSeconds)); } Assert.assertEquals(String.format("%d servers => the result has %d times of a single stream", localServerProvider.serverSize(), localServerProvider.serverSize()), counter.get(), maxRepeat); assertTrue( String.format("There should be at least %d completions after %d repeats (The last one may not have completion. Actual completion count: %d", maxRepeat - 1, maxRepeat, sourceObserver.getCount(server, EventType.SOURCE_COMPLETED)), maxRepeat - 1 <= sourceObserver.getCount(server, EventType.SOURCE_COMPLETED)); assertEquals("There should be no error", 0, sourceObserver.getCount(server, EventType.SUBSCRIPTION_FAILED)); assertEquals("There should be " + maxRepeat + " connection establishment in total", maxRepeat, sourceObserver.getCount(server, EventType.CONNECTION_ESTABLISHED)); assertEquals("There should no final completion", 0, sourceObserver.getCompletionCount()); assertEquals(0, sourceObserver.getErrorCount()); Set<EventType> events = sourceObserver.getEvents(); assertTrue("Polling Source always has subscriptions, so there won't be subscription_ended event. But other events should all be there", EXPECTED_EVENTS_SETS.containsAll(events)); }
@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);  }