java - 如何在订阅/取消订阅时执行 Completable

Question

有我的方法:

public void openWorkshift(WorkshiftSettings workshiftSettings, Subscriber<WorkshiftSettings> subscriber) {
    api.openWorkshift(workshiftSettings)
            .compose(RxOperatorsHelpers.additionalStacktrace())
            .doOnSubscribe(() -> actionsSystem.registerAction(...).await()) // <-
            .doOnUnsubscribe(() -> actionsSystem.unregisterAction(...).await()); // <-
            .subscribeOn(ioScheduler)
            .observeOn(uiScheduler)
            .doOnError(this::handleError)
            .subscribe(subscriber);
}

ActionsSystem.registerAction(...)/ActionsSystem.unregisterActions(...) 看起来像这样:

public Completable registerAction(OperatorAction action) {
    return Completable.fromAction(() -> actions.add(action));
}

public Completable unregisterAction(OperatorAction action) {
    return Completable.fromAction(() -> actions.remove(action));
}

如您所见，我使用 .await() 在源 Observable 的流中执行 Completable。感觉像是错误的解决方案。 我怎样才能做得更优雅？

Answer 1

由于您的 Completable 执行琐碎的操作，您可以简单地将它们的代码内联到 doOnSubscribe 和 doOnUnsubscribe 中:

        .doOnSubscribe(() -> actions.add(action))
        .doOnUnsubscribe(() -> actions.remove(action))

您可以通过从可完成的 andThen 开始 Observable 序列的其余部分来避免 doOnSubscribe:

actionsSyste.registerAction(...)
.andThen(api.openWorkshift(workshiftSettings)
        .compose(RxOperatorsHelpers.additionalStacktrace())
        .doOnUnsubscribe(() -> actionsSyste.unregisterAction(...).await())
        .subscribeOn(ioScheduler)
        .observeOn(uiScheduler)
        .doOnError(this::handleError)
 )
 .subscribe(...)

目前，没有办法在下游取消订阅时执行Completable，也没有简单的方法在序列可能正常终止或出现异常时执行它。