RxJava/RxKotlin 根据子类型拆分流答案

【问题标题】：RxJava/RxKotlin split stream depending on subtypeRxJava/RxKotlin 根据子类型拆分流
【发布时间】：2017-11-14 20:04:16
【问题描述】：

我有一个ResponseMessage 流，它可以是不同的子类型。我想将流拆分为流，以便我可以在自己的流中处理每种类型。

我的第一次尝试导致了这个，我看不出有什么效果。

file.readLines()
        .toObservable()
        .map { mapper.readValue(it, ResponseMessage::class.java) }
        .groupBy { when(it) {
            is MarketChangeMessage -> it::class
            else -> it::class
        }}
        .map { it.????? } //How can possible this work?

我现在的问题是：将流划分为特定子类型的流的惯用方法是什么？

【问题讨论】：

标签： kotlin rx-java rx-kotlin rx-kotlin2 rx-javafx

【解决方案1】：

您可以使用ofType 运算符：

ofType( ) — 仅从源 Observable 中发出属于特定类的那些项目。

例子：

val messages = file.readLines()
    .toObservable()
    .map { mapper.readValue(it, ResponseMessage::class.java) }
    .share() // <-- or other multicasting operator

messages
    .ofType(MarketChangeMessage::class)
    .subscribe()

messages
    .ofType(Other::class)
    .subscribe()

【讨论】：