使用 RxJava 构建 MVI 循环：如何用 scan() 替换 BehaviorSubject

Question

我正在尝试找出一种方法来消除可变状态，从而消除可能的竞争条件。但我似乎无法弄清楚如何以某种方式“交织”两个 Observable，同时还使用“扫描”。

希望通过显示更多代码，我可以给你一个想法：

private val stateRelay: BehaviorRelay<State> = BehaviorRelay.createDefault(initialState ?: DEFAULT_STATE) // maybe this should be `Observable.startWith()` somehow?

fun bindIntents(intents: Observable<Actions>, stateRenderer: StateRenderer) {
    compositeDisposable += intents.concatMap { action ->
        when (action) {
            is Actions.Increment -> {
                Observable.create<Change> { emitter -> 
                   // emit things
                }
            }
            is Actions.Decrement -> {
                Observable.create<Change> { emitter -> 
                   // emit things
                }
            }
        }
    }.map { change ->
        reducer(stateRelay.value, change) // TODO: figure out how to use scan() here, instead of stateRelay.value! :(
    }.subscribeBy { newState ->
        stateRelay.accept(newState) // there is a chance that the relay shouldn't be here if scan is used
    }

    compositeDisposable += 
        stateRelay // TODO: figure out how to use scan() instead of a relay!
            .distinctUntilChanged()
            .subscribeBy { state ->
                stateRenderer(state)
            }
}

fun unbindIntents() {
    compositeDisposable.clear()
}

所以我在这种方法中收到了一个Observable<Actions>，从技术上讲，它在另一边是一个PublishRelay（这应该没问题）。

但是，不知何故，我应该用Observable.scan()（可能用startWith）替换BehaviorRelay，以消除可变状态，但我似乎无法理解我应该做的事情做到这一点。

至于涉及的类型，以备不时之需：

private typealias Reducer = (state: State, change: Change) -> State

private typealias StateRenderer = (state: State) -> Unit

@Parcelize
data class State(val count: Int): Parcelable

---

我如何将intents.concatMap.map 包装为Observable.scan() 的一部分（可能还有startWith() 和replay(1)），以消除我对BehaviorSubject 的使用？

Answer 1

我将在上面详细说明我的评论。 这是对代码的简单重写，以执行您的要求。

fun bindIntents(intents: Observable<Actions>, stateRenderer: StateRenderer) {
    val stateObservable = intents.concatMap { action ->
        when (action) {
            is Actions.Increment -> {
                Observable.create<Change> { emitter ->
                // emit things
                }
            }
            is Actions.Decrement -> {
                Observable.create<Change> { emitter ->
                    // emit things
                }
            }
        }
    }.scan(initialState, { currentState, change -> reducer(currentState, change)})

    compositeDisposable +=
        stateObservable
                .distinctUntilChanged()
                .subscribeBy { state ->
                    stateRenderer(state)
                }
}

请注意，这可以通过在下面的表达式中内联我分配给 stateObservable 的 observable 并使用方法引用作为第二个参数进行扫描来进一步简化

fun bindIntents(intents: Observable<Actions>, stateRenderer: StateRenderer) {
    compositeDisposable +=
            intents.concatMap { action ->
                when (action) {
                    is Actions.Increment -> {
                        Observable.create<Change> { emitter ->
                            // emit things
                        }
                    }
                    is Actions.Decrement -> {
                        Observable.create<Change> { emitter ->
                            // emit things
                        }
                    }
                }
            }.scan(initialState, this::reducer)
                    .distinctUntilChanged()
                    .subscribeBy { state ->
                        stateRenderer(state)
                    }
}