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

Question

我正在尝试找出一种方法来消除可变状态，从而消除可能的竞争条件。但我似乎无法弄清楚如何以某种方式 "intertwine" 两个 Observables，同时还使用 "scan".

希望通过展示更多代码，我可以给你思路：

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（这应该没问题）。

但是，我应该以某种方式将 BehaviorRelay 替换为 Observable.scan()（可能是 startWith）以消除可变状态，但我似乎无法包装我的想想我应该做些什么来实现它。

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

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 的可观察对象并使用方法引用作为第二个参数来像这样扫描

来进一步简化

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)
                    }
}

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

Building MVI loop with RxJava: how to replace BehaviorSubject with scan()

kotlin

rx-java2

behaviorsubject