使用 RxJava 进行基于内容的路由的惯用方式

Question

有人知道在 RxJava 或 Reactive Stream 中 content based routing 的惯用方法吗？

我可以看到如何使用标准运算符（如映射过滤器和转换）创建反应式消息处理管道。

但我不明白为什么要使用运算符在多个可能的管道之间切换。
我希望能够基于一个初始输入对象或映射触发多个处理管道中的一个，以便根据输入映射或对象的内容选择不同的管道。

我的想法是我需要向外部 api 发出几个可能的请求，决定 api 和要触发的管道的因素取决于我的控制器的输入对象。

我知道这可以通过 Spring 集成或 Camel 来完成，但我的用例不需要集成引擎的全部灵活性，因为 munch deen 能够处理网络延迟。

谢谢你，亲切的

Answer 1

基于内容的路由是 flatMap 的工作：

observable.flatMap(x -> {
    if (isA(x))
        return observableA(x);
    else 
        return observableB(x);
});

Answer 2

您还可以使用运算符 GroupBy，它允许您发送不同的项目组 http://reactivex.io/documentation/operators/groupby.html

Idiomatic way to content based routing with RxJava