@Test public void testResumeOnCompletion() throws Exception { final int maxRepeat = 10; final CountDownLatch countGate = new CountDownLatch(maxRepeat * localServerProvider.serverSize()); HttpSourceImpl<ByteBuf, ByteBuf, ServerContext<ByteBuf>> source = createSingleEntitySource(new ClientResumePolicy<ByteBuf, ByteBuf>() { @Override public Observable<HttpClientResponse<ByteBuf>> onError(ServerClientContext<ByteBuf, ByteBuf> clientContext, int attempts, Throwable error) { return null; } @Override public Observable<HttpClientResponse<ByteBuf>> onCompleted(ServerClientContext<ByteBuf, ByteBuf> clientContext, int attempts) { countGate.countDown(); if (attempts < maxRepeat) { return clientContext.newResponse(); } else { return null; } } }); final AtomicInteger counter = new AtomicInteger(); final CountDownLatch done = new CountDownLatch(1); Observable.merge(source.call(new Context(), new Index(1, 1))) .map(new Func1<ServerContext<ByteBuf>, ServerContext<String>>() { @Override public ServerContext<String> call(ServerContext<ByteBuf> pair) { try { return new ServerContext<>(pair.getServer(), pair.getValue().retain().toString(Charset.defaultCharset())); } catch (Throwable e) { e.printStackTrace(); throw new RuntimeException(e); } } }) .doOnNext(new Action1<ServerContext<String>>() { @Override public void call(ServerContext<String> pair) { counter.incrementAndGet(); assertEquals(RequestProcessor.SINGLE_ENTITY_RESPONSE, pair.getValue()); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail("Unexpected failure: " + throwable); } }) .doAfterTerminate(new Action0() { @Override public void call() { try { countGate.await(); } catch (InterruptedException e) { e.printStackTrace(); } done.countDown(); } }) .subscribe(); long wait = 5; if (!done.await(5, TimeUnit.SECONDS)) { fail(String.format("All streaming should be done within %d seconds. ", wait)); } assertEquals( String.format("There should be exactly %d response from each of the %d servers", maxRepeat, localServerProvider.serverSize()), maxRepeat * localServerProvider.serverSize(), counter.get()); for (Server server : localServerProvider.getServers()) { ServerInfo serverInfo = toServerInfo(server); assertEquals( String.format("There should be %d completion per server as resumption function should called %d times", maxRepeat, maxRepeat - 1), maxRepeat, sourceObserver.getCount(serverInfo, EventType.SOURCE_COMPLETED)); assertEquals("There should be no error", 0, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_FAILED)); assertEquals( String.format("Connection per server should have been established %d times", maxRepeat), 10, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_ESTABLISHED)); } assertEquals( String.format("There are %d repeats, but there should be only one final completion", maxRepeat - 1), 1, sourceObserver.getCompletionCount()); assertEquals(0, sourceObserver.getErrorCount()); Set<EventType> events = sourceObserver.getEvents(); assertEquals(EXPECTED_EVENTS_SETS, events); assertEquals("Each connection should be unsubscribed once by the subscriber", localServerProvider.serverSize(), sourceObserver.getEventCount(CONNECTION_UNSUBSCRIBED)); for (EventType eventType : new EventType[] {EventType.SOURCE_COMPLETED, EventType.SUBSCRIPTION_ESTABLISHED, EventType.CONNECTION_ATTEMPTED, EventType.CONNECTION_ESTABLISHED}) { assertEquals( String.format("Event %s should be recorded exactly %d times per server", eventType, maxRepeat), maxRepeat * localServerProvider.serverSize(), sourceObserver.getEventCount(eventType)); } assertEquals( "Event SERVER_FOUND should be recorded exactly once per server", localServerProvider.serverSize(), sourceObserver.getEventCount(EventType.SERVER_FOUND)); }
@Override public Observable<Observable<T>> call(Context context, Index index) { return serverProvider .getServersToAdd() .filter((ServerInfo serverInfo) -> !connectionManager.alreadyConnected(serverInfo) && !connectionManager.connectionAlreadyAttempted(serverInfo)) .flatMap((ServerInfo serverInfo) -> { return streamServers(Observable.just(serverInfo)); }) .doOnError((Throwable error) -> { logger.error(String.format("The source encountered an error " + error.getMessage(), error)); observer.onError(error); }) .doAfterTerminate(() -> { observer.onCompleted(); connectionManager.reset(); }) .lift(new Operator<Observable<T>, Observable<T>>() { @Override public Subscriber<? super Observable<T>> call(Subscriber<? super Observable<T>> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { connectionManager.reset(); } })); return subscriber; } }); }
HttpSourceImpl implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return serverProvider .getServersToAdd() .filter((ServerInfo serverInfo) -> !connectionManager.alreadyConnected(serverInfo) && !connectionManager.connectionAlreadyAttempted(serverInfo)) .flatMap((ServerInfo serverInfo) -> { return streamServers(Observable.just(serverInfo)); }) .doOnError((Throwable error) -> { logger.error(String.format("The source encountered an error " + error.getMessage(), error)); observer.onError(error); }) .doAfterTerminate(() -> { observer.onCompleted(); connectionManager.reset(); }) .lift(new Operator<Observable<T>, Observable<T>>() { @Override public Subscriber<? super Observable<T>> call(Subscriber<? super Observable<T>> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { connectionManager.reset(); } })); return subscriber; } }); } }
HttpSourceImpl implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return serverProvider .getServersToAdd() .filter((ServerInfo serverInfo) -> !connectionManager.alreadyConnected(serverInfo) && !connectionManager.connectionAlreadyAttempted(serverInfo)) .flatMap((ServerInfo serverInfo) -> { return streamServers(Observable.just(serverInfo)); }) .doOnError((Throwable error) -> { logger.error(String.format("The source encountered an error " + error.getMessage(), error)); observer.onError(error); }) .doAfterTerminate(() -> { observer.onCompleted(); connectionManager.reset(); }) .lift(new Operator<Observable<T>, Observable<T>>() { @Override public Subscriber<? super Observable<T>> call(Subscriber<? super Observable<T>> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { connectionManager.reset(); } })); return subscriber; } }); } HttpSourceImpl(
            HttpRequestFactory<R> requestFactory,
            HttpServerProvider serverProvider,
            HttpClientFactory<R, E> clientFactory,
            Observer<HttpSourceEvent> observer,
            Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor,
            ClientResumePolicy<R, E> resumePolicy); }
HttpSourceImpl implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return serverProvider .getServersToAdd() .filter((ServerInfo serverInfo) -> !connectionManager.alreadyConnected(serverInfo) && !connectionManager.connectionAlreadyAttempted(serverInfo)) .flatMap((ServerInfo serverInfo) -> { return streamServers(Observable.just(serverInfo)); }) .doOnError((Throwable error) -> { logger.error(String.format("The source encountered an error " + error.getMessage(), error)); observer.onError(error); }) .doAfterTerminate(() -> { observer.onCompleted(); connectionManager.reset(); }) .lift(new Operator<Observable<T>, Observable<T>>() { @Override public Subscriber<? super Observable<T>> call(Subscriber<? super Observable<T>> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { connectionManager.reset(); } })); return subscriber; } }); } HttpSourceImpl(
            HttpRequestFactory<R> requestFactory,
            HttpServerProvider serverProvider,
            HttpClientFactory<R, E> clientFactory,
            Observer<HttpSourceEvent> observer,
            Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor,
            ClientResumePolicy<R, E> resumePolicy); static Builder<R, E, T> builder(HttpClientFactory<R, E> clientFactory, HttpRequestFactory<R> requestFactory, Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor); static Builder<R, E, T> builder(HttpClientFactory<R, E> clientFactory, HttpRequestFactory<R> requestFactory, Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor, ClientResumePolicy<R, E> resumePolicy); static Func2<ServerContext<HttpClientResponse<E>>, E, ServerContext<E>> contextWrapper(); static Func2<ServerContext<HttpClientResponse<E>>, E, E> identityConverter(); @Override Observable<Observable<T>> call(Context context, Index index); }
HttpSourceImpl implements Source<T> { @Override public Observable<Observable<T>> call(Context context, Index index) { return serverProvider .getServersToAdd() .filter((ServerInfo serverInfo) -> !connectionManager.alreadyConnected(serverInfo) && !connectionManager.connectionAlreadyAttempted(serverInfo)) .flatMap((ServerInfo serverInfo) -> { return streamServers(Observable.just(serverInfo)); }) .doOnError((Throwable error) -> { logger.error(String.format("The source encountered an error " + error.getMessage(), error)); observer.onError(error); }) .doAfterTerminate(() -> { observer.onCompleted(); connectionManager.reset(); }) .lift(new Operator<Observable<T>, Observable<T>>() { @Override public Subscriber<? super Observable<T>> call(Subscriber<? super Observable<T>> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { connectionManager.reset(); } })); return subscriber; } }); } HttpSourceImpl(
            HttpRequestFactory<R> requestFactory,
            HttpServerProvider serverProvider,
            HttpClientFactory<R, E> clientFactory,
            Observer<HttpSourceEvent> observer,
            Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor,
            ClientResumePolicy<R, E> resumePolicy); static Builder<R, E, T> builder(HttpClientFactory<R, E> clientFactory, HttpRequestFactory<R> requestFactory, Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor); static Builder<R, E, T> builder(HttpClientFactory<R, E> clientFactory, HttpRequestFactory<R> requestFactory, Func2<ServerContext<HttpClientResponse<E>>, E, T> postProcessor, ClientResumePolicy<R, E> resumePolicy); static Func2<ServerContext<HttpClientResponse<E>>, E, ServerContext<E>> contextWrapper(); static Func2<ServerContext<HttpClientResponse<E>>, E, E> identityConverter(); @Override Observable<Observable<T>> call(Context context, Index index);  }