rx-java - 如何使用 RxScala/RxJava 从单个计算步骤构建 Observable？

Question

我目前有以下代码：

def method(): Future[State] = Future {
  // some processing
  State.Completed
}

但现在我注意到我实际上想“发布”一组中间状态：

def method(): Observable[State] = ? {
  // some processing
  publish State.State1
  // some processing
  publish State.State2
  // some processing
  publish State.Completed
}

有没有一种简单的方法可以实现这一目标？尽管我将其描述为 3 种状态转换，但实际上我可能会经历更多或更少的转换。我希望从 Future 到 Observable 的更改意味着对我当前的“命令式”代码的更改最少。

另外，我希望这些“事件”能够实时发布，而不仅仅是从方法返回时。

Answer 1

每当Observable.create：

Observable<State> stateSource = Observable.create(emitter -> {
     // some processing
     emitter.onNext(State.State1);

     // some processing
     emitter.onNext(State.State2);

     // some processing
     emitter.onNext(State.Completed);

     // no further state changes
     emitter.onComplete();
});