@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 testIfCloseObserverIfSubscriptionRequestFailed() throws Exception {
        CountDownLatch errorReceived = new CountDownLatch(1);
        AtomicReference<Throwable> actualThrowable = new AtomicReference<>();

        runAsync(
                () ->
                        subscribeToEvents()
                                .subscribe(
                                        new Subscriber<NewHeadsNotification>() {
                                            @Override
                                            public void onComplete() {}

                                            @Override
                                            public void onError(Throwable e) {
                                                actualThrowable.set(e);
                                                errorReceived.countDown();
                                            }

                                            @Override
                                            public void onSubscribe(Subscription s) {}

                                            @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());
    }