@Test public void testRemovedServerWillBeUnsubscribed() throws Exception { HttpSourceImpl<ByteBuf, ServerSentEvent, ServerContext<ServerSentEvent>> source = createStreamingSource("test/infStream"); final AtomicInteger counter = new AtomicInteger(); final CountDownLatch done = new CountDownLatch(1); final ConcurrentHashMap<String, AtomicInteger> result = new ConcurrentHashMap<>(); final ConcurrentMap<ServerInfo, CountDownLatch> serverRemovalLatch = new ConcurrentHashMap<>(); for (Server server : localServerProvider.getServers()) { serverRemovalLatch.put(toServerInfo(server), new CountDownLatch(1)); } Observable.merge(source.call(new Context(), new Index(1, 1))) .doOnNext(new Action1<ServerContext<ServerSentEvent>>() { @Override public void call(ServerContext<ServerSentEvent> pair) { try { assertTrue(pair.getValue().contentAsString().contains("line")); counter.incrementAndGet(); String msg = pair.getValue().contentAsString(); result.putIfAbsent(msg, new AtomicInteger()); result.get(msg).incrementAndGet(); } finally { serverRemovalLatch.get(pair.getServer()).countDown(); } } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail("Unexpected failure: " + throwable); } }) .doAfterTerminate(new Action0() { @Override public void call() { done.countDown(); } }) .subscribe(); for (Server server : localServerProvider.getServers()) { serverRemovalLatch.get(toServerInfo(server)).await(); localServerProvider.removeServer(toServerInfo(server)); } long waitSeconds = 5; boolean timedout = !done.await(waitSeconds, TimeUnit.SECONDS); if (timedout) { fail(String.format("Waited at least %d seconds for the test to finish. Connection to at least one server is not unsubscribed.", waitSeconds)); } assertTrue(String.format("Each server should emit at least one event before being canceled. Expected counter >= %d, actual counter: %d", localServerProvider.serverSize(), counter.get()), counter.get() >= localServerProvider.serverSize()); for (Server server : localServerProvider.getServers()) { ServerInfo serverInfo = toServerInfo(server); assertEquals("There should be one completion per server", 1, sourceObserver.getCount(serverInfo, EventType.SOURCE_COMPLETED)); assertEquals("There should be one un-subscription per server", 1, sourceObserver.getCount(serverInfo, EventType.CONNECTION_UNSUBSCRIBED)); assertEquals("There should be no error", 0, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_FAILED)); assertEquals("There should be one connection per server", 1, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_ESTABLISHED)); assertEquals(String.format("There should be exactly one cancellation event per server for %d servers. ", localServerProvider.serverSize()), 1, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_CANCELED)); } assertEquals("The source should emit exactly one completion event", 1, sourceObserver.getCompletionCount()); assertEquals("The server should not have any error event", 0, sourceObserver.getErrorCount()); Set<EventType> events = sourceObserver.getEvents(); assertEquals("Each server should have one occurrence per event type except server failure event", (EventType.values().length - 1), events.size()); for (EventType event : events) { assertEquals("Each event should be recorded exactly once", localServerProvider.serverSize(), sourceObserver.getEventCount(event)); } }
@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);  }