@Test public void testResumeOnCompletionButNotOnRemovedServers() throws Exception { final int maxRepeat = 10; final int cutOff = 5; final CountDownLatch countGate = new CountDownLatch(localServerProvider.serverSize() * (cutOff - 1)); final ConcurrentHashMap<ServerInfo, AtomicInteger> resumptionCounts = new ConcurrentHashMap<>(); final ConcurrentHashMap<ServerInfo, AtomicInteger> counterPerServer = new ConcurrentHashMap<>(); for (Server server : localServerProvider.getServers()) { ServerInfo serverInfo = toServerInfo(server); resumptionCounts.put(serverInfo, new AtomicInteger(0)); counterPerServer.put(serverInfo, new AtomicInteger(0)); } 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) { resumptionCounts.get(clientContext.getServer()).incrementAndGet(); 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) { return new ServerContext<>(pair.getServer(), pair.getValue().toString(Charset.defaultCharset())); } }) .doOnNext(new Action1<ServerContext<String>>() { @Override public void call(ServerContext<String> context) { counter.incrementAndGet(); assertEquals(RequestProcessor.SINGLE_ENTITY_RESPONSE, context.getValue()); ServerInfo server = context.getServer(); counterPerServer.get(server).incrementAndGet(); if (counterPerServer.get(server).get() > cutOff) { localServerProvider.removeServer(server); } } }) .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(wait, TimeUnit.SECONDS)) { fail(String.format("All streaming should be done within %d seconds. ", wait)); } assertEquals("All server should be resumed", localServerProvider.serverSize(), resumptionCounts.size()); for (ServerInfo server : resumptionCounts.keySet()) { assertTrue( String.format("The server %s:%s should be resumed fewer than %d times", server.getHost(), server.getPort(), maxRepeat), maxRepeat > resumptionCounts.get(server).get()); } assertTrue( String.format("There should be at least %d response from each of the %d servers", cutOff + 1, localServerProvider.serverSize()), (cutOff + 1) * localServerProvider.serverSize() <= counter.get()); for (Server server : localServerProvider.getServers()) { ServerInfo serverInfo = toServerInfo(server); assertEquals("There should be no error", 0, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_FAILED)); } assertEquals( String.format("There are %d repeats, but there should be only one final completion", maxRepeat - 1), 1, sourceObserver.getCompletionCount()); assertEquals("There should be no error", 0, sourceObserver.getErrorCount()); Set<EventType> events = sourceObserver.getEvents(); Set<EventType> expectedEvents = new HashSet<>(); expectedEvents.addAll(Arrays.asList( EventType.SUBSCRIPTION_CANCELED, EventType.SERVER_FOUND, EventType.CONNECTION_ATTEMPTED, EventType.CONNECTION_ESTABLISHED, EventType.SUBSCRIPTION_ESTABLISHED, EventType.SOURCE_COMPLETED, EventType.SUBSCRIPTION_ENDED, EventType.CONNECTION_UNSUBSCRIBED)); assertEquals(expectedEvents, 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}) { assertTrue( String.format("Event %s should be recorded at least %d times per server", eventType, cutOff), (cutOff + 1) * localServerProvider.serverSize() <= sourceObserver.getEventCount(eventType)); } for (EventType eventType : new EventType[] {EventType.SERVER_FOUND, EventType.SUBSCRIPTION_CANCELED}) { assertEquals( String.format("Event %s should be recorded exactly once per server", eventType), localServerProvider.serverSize(), sourceObserver.getEventCount(eventType)); } }
@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);  }