### Question:
WebSocketService implements Web3jService { @Override public <T extends Notification<?>> Observable<T> subscribe( Request request, String unsubscribeMethod, Class<T> responseType) { BehaviorSubject<T> subject = BehaviorSubject.create(); subscribeToEventsStream(request, subject, responseType); return subject .doOnUnsubscribe(() -> closeSubscription(subject, unsubscribeMethod)); } WebSocketService(String serverUrl, boolean includeRawResponses);  WebSocketService(WebSocketClient webSocketClient,
                     boolean includeRawResponses);   WebSocketService(WebSocketClient webSocketClient,
                     ScheduledExecutorService executor,
                     boolean includeRawResponses); void connect(); @Override T send(Request request, Class<T> responseType); @Override CompletableFuture<T> sendAsync(
            Request request,
            Class<T> responseType); @Override Observable<T> subscribe(
            Request request,
            String unsubscribeMethod,
            Class<T> responseType); @Override void close();  }### Answer:
@Test public void testPropagateSubscriptionEvent() throws Exception { CountDownLatch eventReceived = new CountDownLatch(1); CountDownLatch completedCalled = new CountDownLatch(1); AtomicReference<NewHeadsNotification> actualNotificationRef = new AtomicReference<>(); runAsync(() -> { final Subscription subscription = subscribeToEvents() .subscribe(new Subscriber<NewHeadsNotification>() { @Override public void onCompleted() { completedCalled.countDown(); } @Override public void onError(Throwable e) { } @Override public void onNext(NewHeadsNotification newHeadsNotification) { actualNotificationRef.set(newHeadsNotification); eventReceived.countDown(); } }); try { eventReceived.await(2, TimeUnit.SECONDS); subscription.unsubscribe(); } catch (InterruptedException e) { e.printStackTrace(); } }); sendSubscriptionConfirmation(); sendWebSocketEvent(); assertTrue(completedCalled.await(6, TimeUnit.SECONDS)); assertEquals( "0xd9263f42a87", actualNotificationRef.get().getParams().getResult().getDifficulty()); }
@Test public void testSendUnsubscribeRequest() throws Exception { CountDownLatch unsubscribed = new CountDownLatch(1); runAsync(() -> { Observable<NewHeadsNotification> observable = subscribeToEvents(); observable.subscribe().unsubscribe(); unsubscribed.countDown(); }); sendSubscriptionConfirmation(); sendWebSocketEvent(); assertTrue(unsubscribed.await(2, TimeUnit.SECONDS)); verifyUnsubscribed(); }
@Test public void testIfCloseObserverIfSubscriptionRequestFailed() throws Exception { CountDownLatch errorReceived = new CountDownLatch(1); AtomicReference<Throwable> actualThrowable = new AtomicReference<>(); runAsync(() -> subscribeToEvents().subscribe(new Observer<NewHeadsNotification>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { actualThrowable.set(e); errorReceived.countDown(); } @Override public void onNext(NewHeadsNotification newHeadsNotification) { } })); waitForRequestSent(); sendErrorReply(); assertTrue(errorReceived.await(2, TimeUnit.SECONDS)); Throwable throwable = actualThrowable.get(); assertEquals( IOException.class, throwable.getClass() ); assertEquals( "Subscription request failed with error: Error message", throwable.getMessage()); }