@Override
    public <T extends Notification<?>> Flowable<T> subscribe(
            Request request, String unsubscribeMethod, Class<T> responseType) {
        // We can't use usual Observer since we can call "onError"
        // before first client is subscribed and we need to
        // preserve it
        BehaviorSubject<T> subject = BehaviorSubject.create();

        // We need to subscribe synchronously, since if we return
        // an Flowable to a client before we got a reply
        // a client can unsubscribe before we know a subscription
        // id and this can cause a race condition
        subscribeToEventsStream(request, subject, responseType);

        return subject.doOnDispose(() -> closeSubscription(subject, unsubscribeMethod))
                .toFlowable(BackpressureStrategy.BUFFER);
    }
@Test
    public void testPropagateSubscriptionEvent() throws Exception {
        CountDownLatch eventReceived = new CountDownLatch(1);
        CountDownLatch disposed = new CountDownLatch(1);
        AtomicReference<NewHeadsNotification> actualNotificationRef = new AtomicReference<>();

        runAsync(
                () -> {
                    Disposable disposable =
                            subscribeToEvents()
                                    .subscribe(
                                            newHeadsNotification -> {
                                                actualNotificationRef.set(newHeadsNotification);
                                                eventReceived.countDown();
                                            });
                    try {
                        eventReceived.await(2, TimeUnit.SECONDS);
                        disposable.dispose();
                        disposed.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });

        sendSubscriptionConfirmation();
        sendWebSocketEvent();

        assertTrue(disposed.await(6, TimeUnit.SECONDS));
        assertEquals(
                "0xd9263f42a87",
                actualNotificationRef.get().getParams().getResult().getDifficulty());
        assertEquals(
                "0xcd0c3e8af590364c09d0fa6a1210faf5",
                actualNotificationRef.get().getParams().getSubscription());
    }