使用 RxJava 的状态机? [英] State machine using RxJava?

问题描述

我正在尝试全力投入 RxJava 并解决我遇到的这个问题，但它似乎非常不适合它，因为 RxJava 似乎不想处理任何类型的状态，而只想传递事件并改变它们以处理它们.

I'm trying to go all in on RxJava and solve this problem I have with it, but it seems very unsuited for it, as RxJava seems to not want to deal with having any sort of state, rather only passing along events and mutating them to handle them.

我试图用 RxJava 模拟的基本状态机行为是这样的:

The basic state machine behavior I'm trying to emulate with RxJava is this:

1. 在应用启动事件中，等待下一次应用暂停.
2. 在应用暂停事件中，启动 15 分钟计时器并等待下一个应用恢复.
   • 如果应用在计时器内恢复，请取消并返回第 1 步.

推荐答案

虽然可以在 RxJava 中实现状态机，但它变得非常难看，非常快.您可以使用 switchMap() 运算符来选择一些分支路径，也可以使用其他几个运算符来进行分支和合并.

While it is possible to implement a state machine in RxJava, it gets very ugly, very quickly. You can use the switchMap() operator to select some of the branching paths, and you can use several other operators to branch and merge.

大多数问题是状态机转换看起来更像是go to而不是结构化编程，而且它们根本不像函数式编程.

Most of the issue is that state machine transitions tend to look more like go to than structured programming, and they tend not to resemble functional programming at all.

出于类似的原因，很难对任何非平凡的状态机创建流畅或功能性的描述.有一个通用状态机的 RxJava 实现 RxFsm，看起来相当称职.您将提供计时器作为外部输入，以及其他此类解决方法.

For similar reasons, it is quite difficult to create a fluent or functional description of any non-trivial state machine. There is an RxJava implementation of general state machines RxFsm, and it looks quite competent. You will have provide timers as external inputs, and other such work-arounds.

以下是实现状态机的代码，使用了一些 RxJava 元素，并进行了一些简化假设:

Here is code that implements your state machine, using some RxJava elements, and making some simplifying assumptions:

Observable<AppState> appStateObservable;
AtomicReference<Subscription> timerSubscription = new AtomicReference<>(Subscriptions.empty());

appStateObservable
  .doOnNext( state -> if ( state == START ) doStartup() )
  .filter( state -> state != START )
  .subscribe( state -> if ( state == PAUSE ) {
      timerSubscription.set( Observable.timer(15, MINUTES)
        .subscribe( timerSubscription.get().unsubscribe() ) );
    } else  if ( timerSubscription.get().isUnsubscribed() ) {
      refresh();
    } else {
      timerSubscription.get().unsubscribe();
    } );

它使用 timerSubscription 订阅状态​​来确定计时器是否已触发，从而执行刷新.

It uses the timerSubscription subscription state to determine if the timer has fired, and, thus, perform the refresh.

这篇关于使用 RxJava 的状态机?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！