@Test public void testGettingSingleEntityFromMultipleServers() throws Exception { HttpSourceImpl<ByteBuf, ByteBuf, ServerContext<ByteBuf>> source = createSingleEntitySource(); 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> 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() { done.countDown(); } }) .subscribe(); done.await(3000, TimeUnit.SECONDS); assertEquals( String.format("There should be exactly one response from each of the %d servers", localServerProvider.serverSize()), localServerProvider.serverSize(), counter.get()); 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(1, sourceObserver.getCompletionCount()); assertEquals(0, sourceObserver.getErrorCount()); Set<EventType> events = sourceObserver.getEvents(); assertEquals(EXPECTED_EVENTS_SETS, events); for (EventType event : events) { assertEquals("Each event should be recorded exactly once per server", 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);  }