javascript - RxJS 可观察到的.concat : How to know where next result came from?

Question

我的应用程序(使用 TypeScript 编写的带有 RxJS 5 的 Angular 2)要求我以预定义的顺序对 Web 服务进行 2 次调用。 Inspired by this page我决定尝试 RxJS 可观察对象的 concat 运算符。不同之处在于，接收值的 onNext 回调需要根据它们的来源将它们转发给不同的处理程序

我的代码可以简化为:

suscription1 = Observable.return({name:'John', age:7});
subscription2 = Observable.return({name:'Mary', color:'blue'}); 
Observable.concat(subscription1, subscription2).subscribe(
    data=>{ /* need to know which subscripion 
             i'm dealing with to direct to the right handler*/}
);

确定我的数据在每次可观察发射时的来源的最优雅方法是什么？我想到了几种方法，但它们看起来都很笨拙:

1. 在范围内保留一个计数器，也许作为一个类属性。每次收到结果时检查并增加它。由于每个 observable 只发出一次，因此当第一个订阅输出时计数器将具有其初始值，而当第二个订阅输出时它将具有 initial+1 。问题是当我不知道每个 Observable 在完成之前会发射多少次时，这将不起作用。

2. 检查每个结果并根据结果的形状识别其来源。例如，在我的示例中，只有第一个可观察对象的结果具有 age 属性。显然，当结果具有相同的形状时，这变得不切实际。

3. 完全摆脱 concat 运算符；在 subscription1 的 onComplete 处理程序中订阅 subscription2。这可行，但扩展性不佳；如果我有 4 或 5 个订阅要连接，它就会变成嵌套 hell 。

只有 2 个订阅的解决方案 3...

suscription1 = Observable.return({name:'John', age:7});
subscription2 = Observable.return({name:'Mary', age:8}); 
subscription1.subscribe(
    data=> this.handlerOne(data),
    error=>{},
    ()=> {
        subscription2.subscribe(
            data=>this.handlerTwo(data),
            error=>{},
            ()=>console.log('All subscriptions completed!')
        )
    }
);

所以我的问题是:当使用 Observable.concat 依次订阅多个可观察对象时，我的 onNext 处理程序如何确定数据的来源？或者，我应该为此目的使用另一个运算符(operator)吗？我不能使用 forkJoin 运算符，因为完成订阅的顺序很重要。

Answer 1

您可以映射这些描述以了解它的来源。

suscription1 = Observable.return({name:'John', age:7})
         .map(person => { person.source = 1; return person };
subscription2 = Observable.return({name:'Mary', age:8})
         .map(person => { person.source = 2; return person };

然后你可以很容易地确定它来自哪里:

Observable.concat(subscription1, subscription2).subscribe(
    data=> { if (data.source == 1) /* do whatever */ }
);