scala - 物化可观察对象上的模式匹配图

Question

假设我想将 Observable[T] 转换为 Observable[Try[T]]。我想对物化的原始可观察值进行模式匹配，但我不知道 OnCompleted() 返回什么:

obs.materialized map {
  case OnNext(v) => Success(v)
  case OnError(t) => Failure(t)
  case OnCompleted() => // What do I return here?
}

• 总的来说，当 OnCompleted 的情况实际上并不对应于可观察的“元素”时，我不明白如何映射具体化的可观察。

Answer 1

In Rx Java materialized Observable is a list of notifications - 而 OnCompleted/OnError 是最后也是唯一的一个通知。我认为 RxScala 也是如此。将每条消息威胁为成功似乎是不正确的 - 因为最后一个成功/失败是整个序列的成功/失败，因此最好仅对 OnError/OnCompleted 消息使用 Try (但它将是 Observable[T] => Observable[消息[T]]):

trait Msg[T]
case class Data[T](d: T) extends Msg[T]
case class End[T](t: Try[Observable[T]]) extends Msg[T]
obs.materialize map {
  case OnNext(v) => Data(v)
  case OnError(t) => End(Failure(t))
  case OnCompleted() => End(Success(obs))
}

无论如何，在收到最后一个通知之前，您无法(从逻辑角度)将 Observable[T] 转换为 Observable[Try[T]]，所以实际上您只能将其转换为 Future[Observable[Try[T]] 或只是 Future[Observable[T]] (因为 Observable 完成或失败对于其所有元素):

 val p = promise[Observable[Notification[T]]]

 obs.materialize flatMap {
  case OnNext(v) => Some(v)
  case OnError(t) =>  p.fail(t); None
  case OnCompleted() => p.complete(obs); None
 }

 p.future

请记住， future 的每个类似 map 的操作都将在收到所有通知后应用(例如 doOnTerminate)。