### Question:
MetricsClientImpl implements MetricsClient<T> { @Override public Observable<Observable<T>> getResults() { return Observable .create(new Observable.OnSubscribe<Observable<T>>() { @Override public void call(final Subscriber subscriber) { internalGetResults().subscribe(subscriber); } }) .subscribeOn(Schedulers.io()); } MetricsClientImpl(String jobId, WorkerConnectionFunc<T> workerConnectionFunc, JobWorkerMetricsLocator jobWorkerMetricsLocator,
                      Observable<Integer> numWorkersObservable,
                      Observer<WorkerConnectionsStatus> workerConnectionsStatusObserver, long dataRecvTimeoutSecs); @Override boolean hasError(); @Override String getError(); @Override Observable<Observable<T>> getResults();  }### Answer:
@Test public void testMetricConnections() throws InterruptedException, UnsupportedEncodingException, JsonProcessingException { final String jobId = "test-job-1"; final String testResUsageMetricData = generateMetricJson(MetricStringConstants.RESOURCE_USAGE_METRIC_GROUP); final String testDropDataMetricData = generateMetricJson(MetricStringConstants.DATA_DROP_METRIC_GROUP); final int metricsPort = TestSseServerFactory.newServerWithInitialData(testResUsageMetricData); final AtomicInteger i = new AtomicInteger(0); final Observable<EndpointChange> workerMetricLocationStream = Observable.interval(1, TimeUnit.SECONDS, Schedulers.io()).map(new Func1<Long, EndpointChange>() { @Override public EndpointChange call(Long aLong) { logger.info("emitting endpointChange"); if (i.getAndIncrement() % 10 == 0) { return new EndpointChange(EndpointChange.Type.add, new Endpoint("localhost", 31002)); } else { return new EndpointChange(EndpointChange.Type.add, new WorkerEndpoint("localhost", 31002, 1, metricsPort, 0, 1)); } } }); MetricsClientImpl<MantisServerSentEvent> metricsClient = new MetricsClientImpl<>(jobId, new SseWorkerConnectionFunction(true, new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.error("Metric connection error: " + throwable.getMessage()); try { Thread.sleep(500); } catch (InterruptedException ie) { logger.error("Interrupted waiting for retrying connection"); } } }, new SinkParameters.Builder().withParameter("name", MetricStringConstants.RESOURCE_USAGE_METRIC_GROUP).build()), new JobWorkerMetricsLocator() { @Override public Observable<EndpointChange> locateWorkerMetricsForJob(String jobId) { return workerMetricLocationStream; } }, Observable.just(1), new Observer<WorkerConnectionsStatus>() { @Override public void onCompleted() { logger.info("got onCompleted in WorkerConnStatus obs"); } @Override public void onError(Throwable e) { logger.info("got onError in WorkerConnStatus obs"); } @Override public void onNext(WorkerConnectionsStatus workerConnectionsStatus) { logger.info("got WorkerConnStatus {}", workerConnectionsStatus); } }, 60); final CountDownLatch latch = new CountDownLatch(1); final Observable<Observable<MantisServerSentEvent>> results = metricsClient.getResults(); Observable.merge(results) .doOnNext(new Action1<MantisServerSentEvent>() { @Override public void call(MantisServerSentEvent event) { logger.info("got event {}", event.getEventAsString()); assertEquals(testResUsageMetricData, event.getEventAsString()); latch.countDown(); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.error("got error {}", throwable.getMessage(), throwable); } }) .doOnCompleted(new Action0() { @Override public void call() { logger.info("onComplete"); } }) .subscribe(); latch.await(30, TimeUnit.SECONDS); TestSseServerFactory.stopAllRunning(); }