Angular 9 with RxJs - combining 2 observables results in data but async pipe always sees a null result

Question

I have 2 observables that return data from Firebase, and no matter what type of combination operator I use (forkJoin, CombineLatest etc) the async pipe does not receive the data.

I have an AuthService, with a method that initialises the user from a Firebase realtime database:

  private initUser(firebaseUser: firebase.User) {
    const myUser: User = {account: {}, profile: {}};
    const userId = `US_${firebaseUser.uid}`;

    const accountSnapshot: Observable<UserAccount> = this.db
      .object(`/users/accounts/${userId}`)
      .valueChanges()
      .pipe(first()) as Observable<UserAccount>;

    const profileSnapshot: Observable<UserProfile> = this.db
      .object(`/users/profiles/${userId}`)
      .valueChanges()
      .pipe(first()) as Observable<UserProfile>;

    forkJoin([accountSnapshot, profileSnapshot])
      .subscribe((result: [UserAccount, UserProfile]) => {
        if (result) {
          myUser.account = result[0];
          myUser.profile = result[1];
          this.setCurrentUser(myUser);
        }
      });
  }

CurrentUser is set up as a subject with getter/setter:

  private currentUser = new Subject<User>();

  public get getCurrentUser(): Observable<User> {
    return this.currentUser.asObservable();
  }

  public setCurrentUser(user: User): void {
    this.currentUser.next(user);
  }

In the consuming component:

  currentUser$: Observable<User>;
  user: User;


  ngOnInit(): void {
    this.currentUser$ = this.authService.getCurrentUser;
    this.currentUser$.subscribe(user => {
      this.user = user;
    });
  }

and in the template:

<p>Email: {{ ((currentUser$ | async)?.account).email }}</p>. <- ** DOES NOT WORK **
<p>Email: {{ (user?.account).email }}</p> <- ** DOES WORK **

Now, here is the strange part. If I change the initUser method to NOT combine the observables from firebase, but to get them separately (and set the users profile and account as separate operations), then async works. Like this:

accountSnapshot.subscribe((account: UserAccount) => {
  if (account) {
    myUser.account = account;
    this.setCurrentUser(myUser);
  }
});
profileSnapshot.subscribe((profile: UserProfile) => {
  if (profile) {
    myUser.profile = profile;
    this.setCurrentUser(myUser);
  }
});

then the async pipe works.

I have tried changing the forkJoin to CombineLatest and zip. I've also tried removing the pipe(first()) but the async pipe never works.

Why is this?

Is there something wrong with async or with my code, or maybe my understanding of how async works...?

Poul Kruijt Poul Kruijt · Accepted Answer · 2020-06-01T11:41:18

It's because the Subject fires before your view is ready. You should change it to a ReplaySubject so the latest value(s) get replayed on subscription.

private currentUser = new ReplaySubject<User>(1);

(out of scope of question)

Besides that, it seems a bit odd to use a nested subscribe like you have in your initUser. To me you can simplify your code to the following:

private readonly initUser$ = new Subject<firebase.User>();

readonly currentUser$: Observable<User> = this.initUser$.pipe(
  map((user) => `US_${user.uid}`),
  switchMap((userId) => combineLatest([
    this.db.object<UserAccount>(`/users/accounts/${userId}`).valueChanges(),
    this.db.object<UserProfile>(`/users/profiles/${userId}`).valueChanges()
  ])),
  map(([ account, profile ]) => ({ account, profile } as User)),
  shareReplay(1)
);

private initUser(firebaseUser: firebase.User) {
  this.initUser$.next(firebaseUser);
}

In your consuming component you can then just use the service currentUser$ observable to get the user details

Angular 9 with RxJs - combining 2 observables results in data but async pipe always sees a null result

2 Answers