RxJS 管道过滤到 2+ 个分支

Question

假设我们有一个 Subject，它可以返回一组给定的值，所有这些都可能需要不同的方法来处理它们。我们可以辩论是否应该如此，但不是我调查的重点。

所以...为了处理这个问题，我们可能会让我们的代码看起来像这样：

var sub = new Subject(); 

sub.subscribe( 
    data => {
        if (caseA) {
            // do something
        }
        if (caseB) {
            // do something else
        }
     });

这一切都很好，很棒……但我想知道是否有一些我们可以链接的操作员使它更 rx-ish？考虑过滤器，但链接 caseA 和 caseB 只会使它不起作用（显然），因为它最终会过滤掉两者。

所以，我的问题归结为：是否有可能有任何类似于下面的伪代码的东西？您认识的任何操作员都会像这样工作吗？

var sub = new Subject(); 

sub.pipe(
    magicOperator(//handles caseA),
    magicOperator(//handles caseB),
)
subscribe( 
    data => {
        // handles common thread
     });

Answer 1

我认为解决此问题的更多 rxjs 方法类似于@ritaj 建议的方法，但通过对分区流使用静态合并运算符（如果您想按顺序处理它们，则使用 concat ）

let sub = new Subject();
// i'm using typescript syntax, but you can refer to them by index
let [leftPartition, rightPartition] = sub.pipe( partition( [predicate] ) )
// apply custom logic to the left and right partition
rightPartition = rightPartition.pipe( map( x => x + 1 ) )
leftPartition = leftPartition.pipe( map( x => x - 1 ) )

merge(
     leftPartition,
     rightPartition
).pipe(
    tap( 
        ( value ) => {
            // this stream consumes both right and left partition outputs
        }
    )
).subscribe();

Partition Operator
Merge ( static )

静态合并版本是从 rxjs 包中导入的，并接受任意数量的参数（您甚至可以使用 spread 传递数组参数）

在应用 map 运算符后，我将 observable 重新分配给同一个变量。我不知道这是否是一个好习惯，但如果您需要完整的源，可以改进或分离到不同的变量）

附言。我通常尽量让订阅回调不受任何实现的影响，因为您在流结束时所做的大部分事情都是副作用，我将它们放入一个点击运算符（与 subscribe 相同的参数）

Answer 2

只需创建一个具有大小写切换功能的对象就可以使代码更漂亮

const handler={
  case1:()=>..handle case1,
  case2:()=>...handle case2
}

sub.map(result=>handler[result])

// experimental switchCase operator 
const switchCase=handlers=>(source)=>source.map(val=>handlers[val]())

// usage
sub.pipe(switchCase({
   case1:....,
   case2:....
}))