@Test public void canStreamFromMultipleServersWithCorrectContext() throws Exception { ContextualHttpSource<ServerSentEvent> source = HttpSources .contextualSource( HttpClientFactories.sseClientFactory(), HttpRequestFactories.createGetFactory("test/stream")) .withServerProvider(localServerProvider) .withActivityObserver(sourceObserver) .build(); final AtomicInteger counter = new AtomicInteger(); final CountDownLatch done = new CountDownLatch(1); final ConcurrentHashMap<String, AtomicInteger> result = new ConcurrentHashMap<>(); final CopyOnWriteArraySet<ServerInfo> connectedServers = new CopyOnWriteArraySet<>(); Observable.merge(source.call(new Context(), new Index(1, 1))) .doOnNext(new Action1<ServerContext<ServerSentEvent>>() { @Override public void call(ServerContext<ServerSentEvent> pair) { assertTrue(pair.getValue().contentAsString().contains("line")); counter.incrementAndGet(); String msg = pair.getValue().contentAsString(); result.putIfAbsent(msg, new AtomicInteger()); result.get(msg).incrementAndGet(); connectedServers.add(pair.getServer()); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail("Unexpected failure: " + throwable); } }) .doOnCompleted(new Action0() { @Override public void call() { System.out.println("completed"); } }) .doAfterTerminate(new Action0() { @Override public void call() { done.countDown(); } }) .subscribe(); long waitSeconds = 3; boolean timedout = !done.await(waitSeconds, TimeUnit.SECONDS); if (timedout) { fail(String.format("Waited at least %d seconds for the test to finish. Something is wrong", waitSeconds)); } assertEquals("There should be as many as provided servers", localServerProvider.serverSize(), connectedServers.size()); Assert.assertEquals(String.format("%d servers => the result has %d times of a single stream", localServerProvider.serverSize(), localServerProvider.serverSize()), counter.get(), RequestProcessor.smallStreamContent.size() * localServerProvider.serverSize()); for (String data : RequestProcessor.smallStreamContent) { assertEquals(String.format("%d servers => %d identical copies per message", localServerProvider.serverSize(), localServerProvider.serverSize()), localServerProvider.serverSize(), result.get(data).get()); } for (ServerInfo server : localServerProvider.getServerInfos()) { assertEquals("There should be one completion per server", 1, sourceObserver.getCount(server, EventType.SOURCE_COMPLETED)); assertEquals("There should be one un-subscription per server", 1, sourceObserver.getCount(server, EventType.CONNECTION_UNSUBSCRIBED)); assertEquals("There should be no error", 0, sourceObserver.getCount(server, EventType.SUBSCRIPTION_FAILED)); assertEquals("There should be one connection per server", 1, sourceObserver.getCount(server, EventType.CONNECTION_ESTABLISHED)); } assertEquals("There should be one completions", 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<ServerContext<E>>> call(Context context, Index t2) { return impl.call(context, t2); }
ContextualHttpSource implements Source<ServerContext<E>> { @Override public Observable<Observable<ServerContext<E>>> call(Context context, Index t2) { return impl.call(context, t2); } }
ContextualHttpSource implements Source<ServerContext<E>> { @Override public Observable<Observable<ServerContext<E>>> call(Context context, Index t2) { return impl.call(context, t2); } private  ContextualHttpSource(HttpSourceImpl<ByteBuf, E, ServerContext<E>> impl); }
ContextualHttpSource implements Source<ServerContext<E>> { @Override public Observable<Observable<ServerContext<E>>> call(Context context, Index t2) { return impl.call(context, t2); } private  ContextualHttpSource(HttpSourceImpl<ByteBuf, E, ServerContext<E>> impl); static Builder<E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory); static Builder<E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory, ClientResumePolicy<ByteBuf, E> policy); @Override Observable<Observable<ServerContext<E>>> call(Context context, Index t2); }
ContextualHttpSource implements Source<ServerContext<E>> { @Override public Observable<Observable<ServerContext<E>>> call(Context context, Index t2) { return impl.call(context, t2); } private  ContextualHttpSource(HttpSourceImpl<ByteBuf, E, ServerContext<E>> impl); static Builder<E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory); static Builder<E> builder(HttpClientFactory<ByteBuf, E> clientFactory, HttpRequestFactory<ByteBuf> requestFactory, ClientResumePolicy<ByteBuf, E> policy); @Override Observable<Observable<ServerContext<E>>> call(Context context, Index t2);  }