### Question:
SseWorkerConnectionFunction implements WorkerConnectionFunc<MantisServerSentEvent> { @Override public WorkerConnection<MantisServerSentEvent> call(String hostname, Integer port) { return call(hostname, port, null, null, 5); } SseWorkerConnectionFunction(boolean reconnectUponConnectionRest, Action1<Throwable> connectionResetHandler);  SseWorkerConnectionFunction(boolean reconnectUponConnectionRest, Action1<Throwable> connectionResetHandler, SinkParameters sinkParameters); @Override WorkerConnection<MantisServerSentEvent> call(String hostname, Integer port); @Override WorkerConnection<MantisServerSentEvent> call(final String hostname, final Integer port,
                                                        final Action1<Boolean> updateConxStatus,
                                                        final Action1<Boolean> updateDataRecvngStatus, final long dataRecvTimeoutSecs);  }### Answer:
@Test public void testSseConnection() throws InterruptedException { final int serverPort = TestSseServerFactory.getServerPort(); final CountDownLatch errorLatch = new CountDownLatch(1); final CountDownLatch latch = new CountDownLatch(1); final boolean reconnectOnConnReset = true; SseWorkerConnectionFunction connectionFunction = new SseWorkerConnectionFunction(reconnectOnConnReset, new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("connection was reset, should be retried", throwable); errorLatch.countDown(); } }); final WorkerConnection<MantisServerSentEvent> conn = connectionFunction.call("localhost", serverPort); final Observable<MantisServerSentEvent> events = conn.call(); events .doOnNext(new Action1<MantisServerSentEvent>() { @Override public void call(MantisServerSentEvent e) { logger.info("got event {}", e.getEventAsString()); assertEquals(TEST_EVENT_DATA, e.getEventAsString()); latch.countDown(); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.error("caught error ", throwable); fail("unexpected error"); } }) .doOnCompleted(new Action0() { @Override public void call() { logger.warn("onCompleted"); } }) .subscribe(); errorLatch.await(30, TimeUnit.SECONDS); TestSseServerFactory.newServerWithInitialData(serverPort, TEST_EVENT_DATA); latch.await(30, TimeUnit.SECONDS); TestSseServerFactory.stopAllRunning(); }