android - 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 并订阅那个 observable，但没有帮助。

Answer 1

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

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

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