@Test public void testStreamingErrorFromServerWillNotCompleteSource() throws Exception { int eventCount = 20; HttpSourceImpl<ByteBuf, ServerSentEvent, ServerContext<ServerSentEvent>> source = createStreamingSource("test/finiteStream?count=" + eventCount); final CountDownLatch latch = new CountDownLatch(1); final AtomicReference<Throwable> ex = new AtomicReference<>(); final ConcurrentHashMap<ServerInfo, Set<String>> items = new ConcurrentHashMap<>(); Observable.merge(source.call(new Context(), new Index(1, 1))) .subscribe(new Subscriber<ServerContext<ServerSentEvent>>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { ex.set(e); latch.countDown(); } @Override public void onNext(ServerContext<ServerSentEvent> pair) { items.putIfAbsent(pair.getServer(), new HashSet<String>()); items.get(pair.getServer()).add(pair.getValue().contentAsString()); } }); if (latch.await(5, TimeUnit.SECONDS)) { fail("The test case should not finish at all"); } Assert.assertNull("The timeout error should be captured by the client so it does not surface to the source", ex.get()); for (Server server : localServerProvider.getServers()) { ServerInfo serverInfo = toServerInfo(server); assertEquals("There should be no source level error", 0, sourceObserver.getErrorCount()); assertEquals("There should be one connection attempt per server", 1, sourceObserver.getCount(serverInfo, EventType.CONNECTION_ATTEMPTED)); assertEquals("There should be one established connection per server ", 1, sourceObserver.getCount(serverInfo, EventType.CONNECTION_ESTABLISHED)); assertEquals("There should no subscribed server because of read timeout", 1, sourceObserver.getCount(serverInfo, EventType.SUBSCRIPTION_ESTABLISHED)); assertEquals(String.format("There should be %d item before simulated error per server", eventCount), eventCount, items.get(serverInfo).size()); } }
@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);  }