RxJS 合并 2 个订阅答案

【问题标题】：RxJS Combine 2 SubscriptionsRxJS 合并 2 个订阅
【发布时间】：2018-06-26 15:35:44
【问题描述】：

我有一个输出 2 个 observables 的服务。 1 只用于获取当前经过身份验证的用户，另一个用于获取所有活动用户。在主屏幕上，我想调用这两个 observables 来使用，这将（当前）通过订阅这两个 observables 来完成，但是第一个 observables 并不总是运行。我正在寻找一种解决方案，让这两个可观察对象在我们或页面上加载任何错误之前发出一个值。我在下面的示例中尝试了 forkjoin，但它不起作用，我怀疑这是因为 forkjoin 期望发出一些东西但没有发出任何东西，但是订阅两个 observables 确实输出了我需要的数据。这是我尝试过的：

    const tt =  Observable.forkJoin(
        this.auth.getActiveUsers().map((res) => res = res  ),
        this.auth.user.map((res) => res = res ),        
    );
    console.log("running" ); 
    tt.subscribe( (data) => {
      console.log("data ", data );      
        console.log("data[0] ", data[0] );
        console.log("data[1] ", data[1] );        
      },
      err => console.error(err),
      () => console.error("complete"),
    );

这是订阅在我的 service.ts 文件中的布局方式：

    Filename: Service.ts

    ...

    // first Observable  
    constructor( private afAuth: AngularFireAuth,
                private db: AngularFirestore ) {

        this.user = this.afAuth.authState.switchMap(user => {
            if (user) {
                return this.db.doc<iUser>(`users/${user.uid}`).valueChanges()
            } else {
                return Observable.of(null)
            }
            })     

    }

    ...

    // second Observable
    getActiveUsers(): Observable<iUser[]> {
    return this.db.collection<iUser>('users').valueChanges().catch(this.catchError);                 
    }

    ...

【问题讨论】：

标签： javascript angular ionic-framework rxjs ionic3

【解决方案1】：

使用 mergeMap + forkJoin：

import { mergeMap } from 'rxjs/operators';
import { forkJoin, of } from 'rxjs';

const myPromise = val =>
  new Promise(resolve =>
    setTimeout(() => resolve(`Promise Resolved: ${val}`), 5000)
  );

const source = of([1, 2, 3, 4, 5]);
const example = source.pipe(mergeMap(q => forkJoin(...q.map(myPromise))));
/*
  output:
  [
   "Promise Resolved: 1",
   "Promise Resolved: 2",
   "Promise Resolved: 3",
   "Promise Resolved: 4",
   "Promise Resolved: 5"
  ]
*/
const subscribe = example.subscribe(val => console.log(val));

Full example

【讨论】：

【解决方案2】：

我猜你的auth.user observable 是一个热门的（在发出值时不完整），而 forkJoin 仅在所有传递的 observable都完成时才会发出。不发出值但完成。

如果我的猜测是正确的，在这种情况下你可以做的只是传递auth.user.first()，这将从原始的 observable 创建一个完整的 observable，并且你的 forkJoin 将按预期工作

【讨论】：

【解决方案3】：

forkJoin 需要一个数组，所以我认为这就是它在您的代码中不起作用的原因：

 const tt = Observable.forkJoin([
                this.auth.getActiveUsers(),
                this.auth.user
            ]);
 // ...

【讨论】：

来自Rxjs6+可以直接使用import { forkJoin } from 'rxjs';,See docs