在 dispose 后订阅 observable

Question

我正在 android repository by Fernando Cejas 上构建我的应用程序，但在调用 dispose 后订阅 observable 时遇到问题。

当我来到仪表板时，我调用方法subscribeOnUserMessages.execute(new Subscriber(), new Params(token))，这是UseCase类中的方法

public void execute(DisposableObserver<T> observer, Params params) {
    Preconditions.checkNotNull(observer);
    final Observable<T> observable = this.buildUseCaseObservable(params)
            .subscribeOn(Schedulers.from(threadExecutor))
            .observeOn(postExecutionThread.getScheduler());
    addDisposable(observable.subscribeWith(observer));
}

在子类SubscribeOnUserMessages 中，我只需像这样调用存储库 return messageRepository.subscribeOnUserMessages(params);

在我的套接字实现中，我这样创建

return Observable.create(emitter -> {

        if (!isThereInternetConnection()) {
            Timber.w("Network connection exception");
            emitter.onError(new NetworkConnectionException());
            return;
        }

        /*
         * Open socket if not opened
         */
        openSocket(params.getToken());



        String channelName = CHANNEL_PRIVATE_USER + params.getAuthenticated().getUuid();

        if (subscribedChannels.contains(channelName)) {
            Timber.d("Channel %s is already subscribed", channelName);
            return;
        }


        JSONObject auth;

        try {
            auth = createAuthJson(CHANNEL, channelName, params.getToken());
        } catch (JSONException e) {
            Timber.e("Couldn't create auth json");
            emitter.onError(e);
            return;
        }

        mSocket.emit(SUBSCRIBE, auth);
        Timber.d("Emitted subscribe with channel: %s ", CHANNEL_PRIVATE_USER + params.getAuthenticated().getUuid());
        subscribedChannels.add(CHANNEL_PRIVATE_USER + params.getAuthenticated().getUuid());
        Timber.d("Subscribing on event: %s\n with user: %s", EVENT_USER_NEW_MESSAGE, params.getAuthenticated().getUuid());

        if (mSocket.hasListeners(EVENT_USER_NEW_MESSAGE)) {
            Timber.v("Socket already has listener on event: %s", EVENT_USER_NEW_MESSAGE);
            return;
        }


        mSocket.on(EVENT_USER_NEW_MESSAGE, args -> {
            if (args[1] == null) {
                emitter.onError(new EmptyResponseException());
            }

            Timber.d("Event - %s %s", EVENT_USER_NEW_MESSAGE, args[1].toString());

            try {
                MessageEntity messageEntity = messageEntityJsonMapper.transform(args[1]);
                emitter.onNext(messageEntity);
            } catch (JSONException e) {
                Timber.e(e, "Could not parse message json");
                emitter.onError(e);
            }
        });

    });

症状是我第一次订阅时，所有内容都进入了表示层。当我在进入第二个屏幕并返回后处理时，我只看到日志进入套接字实现，但没有通过。

我的问题是：有没有办法再次订阅相同的 observable？我已经尝试在我的用例中以单例形式保存该可观察对象并订阅该可观察对象，但没有帮助。

Answer 1

如果没有额外的信息和细节来重新升级套接字实现，很难准确地发现问题，但是，从你发布的代码中，你没有处理逻辑，所以虽然你可以正确地调用 dispose() 到 @ 987654322@ 在正确的生命周期事件中，您的套接字实际上将保持打开状态，并且它可能永远不会正确断开/关闭。
这可能会导致在第二次打开和连接到套接字时出现问题，因为您可能会尝试重新打开已经打开的套接字，并且取决于您的内部套接字实现，这可能是一个问题。
（我可以在评论中看到 openSocket 如果尚未打开，但在其他地方多次调用套接字上的某些方法或设置侦听器时仍然可能存在问题，再次取决于套接字 impl）

作为一般准则，您应该使用emitter.setCancellable()/emitter.setDisposable() 添加处置逻辑，以便在您不再需要套接字资源时正确处置它们，因此 - 再次应用订阅时（无论是否相同）将再次调用您的订阅逻辑，这将重新打开套接字并监听它。

我不清楚您是否希望在移动到其他屏幕时保持套接字打开（我认为这不是一个好习惯，因为您将保持此资源打开并且可能永远不会回到screen 再次使用它），但如果是@Phoenix Wang 提到的情况，您可以使用发布类型运算符多播Observable，因此每个新的Subscriber 都不会尝试重新打开套接字（即调用订阅逻辑) 但只会收到有关在已打开的套接字中运行的消息的通知。