if (subscription != null) {
            subscription.getListeners().forEach(this::unsubscribe);
            subscription.close();
        }
    }

    private void unsubscribeAll() {
        try {
            Collection<AbstractListener<?>> subscribed = subscriptions.values();
            subscribed.stream()
                .flatMap(it -> it.getListeners().stream())