Javascript – How do I use forkJoin with higher-order Observables in Angular?

I’ve followed this guide on how to cancel HTTP requests with the switchMap Operator Guide

I have following Component in my Angular project

import { Component, OnInit } from '@angular/core';
import { BehaviorSubject, Observable, forkJoin, skip, switchMap } from 'rxjs';
import { Todo } from 'src/app/model/todo.model';
import { User } from 'src/app/model/user.model';
import { TodoService } from 'src/app/service/todo.service';
import { UserService } from 'src/app/service/user.service';

@Component({
  selector: 'app-todo',
  templateUrl: './todo.component.html',
  styleUrls: ['./todo.component.scss']
})
export class TodoComponent implements OnInit{

  todos: Todo[] = [];
  users: User[] = [];

  fetchTodos$ = new BehaviorSubject<Observable<Todo[]>>(null);
  fetchUsers$ = new BehaviorSubject<Observable<User[]>>(null);

  public constructor(
    private todoService: TodoService,
    private userService: UserService
  ){}

  ngOnInit(): void {
    this.fetchTodos$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(todos => {
      this.todos = todos;
    });

    this.fetchUsers$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(users => {
      this.users = users;
    });

    this.loadData();
  }

  loadData(){
    const getAllTodosObservable = this.todoService.getAllTodos();
    this.fetchTodos$.next(getAllTodosObservable);

    const getAllUsersObservable = this.userService.getAllUsers();
    this.fetchUsers$.next(getAllUsersObservable);

    forkJoin([this.fetchTodos$, this.fetchUsers$])
    .subscribe(_ => {
      this.processTodos();
    });
  }

  //Only process todos when all todos and users have been loaded
  processTodos(){
    console.log("Processing todos...")
  }
}

The problem is that forkJoin subscribe will never be called because the BehaviourSubjects are not completed.

forkJoin([this.fetchTodos$, this.fetchUsers$])
    .subscribe(_ => {
      this.processTodos();
  });

How would I achieve calling processTodos() when both requests are done without losing the functionality of HTTP cancellation?

I tried calling complete() on the BehaviourSubjects when requests were done like so

ngOnInit(): void {
    this.fetchTodos$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(todos => {
      this.todos = todos;
      this.fetchTodos$.complete();
    });

    this.fetchUsers$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(users => {
      this.users = users;
      this.fetchUsers$.complete();
    });

    this.loadData();
  }

This will actually call the subscribe method of the forkJoin, but only once and not more.

I’ve also tried using combineLatest instead of forkJoin like this

loadData(){
    const getAllTodosObservable = this.todoService.getAllTodos();
    this.fetchTodos$.next(getAllTodosObservable);

    const getAllUsersObservable = this.userService.getAllUsers();
    this.fetchUsers$.next(getAllUsersObservable);

    combineLatest([this.fetchTodos$, this.fetchUsers$])
    .subscribe(_ => {
      this.processTodos();
    });
  }

The problem with that is, processTodos() will be called twice due to two values being emitted.

Last thing I tried was to use forkJoin() with the HTTP observables directly instead of the behaviour subjects like following

loadData(){
    const getAllTodosObservable = this.todoService.getAllTodos();
    this.fetchTodos$.next(getAllTodosObservable);

    const getAllUsersObservable = this.userService.getAllUsers();
    this.fetchUsers$.next(getAllUsersObservable);

    forkJoin([getAllTodosObservable, getAllUsersObservable])
    .subscribe(_ => {
      this.processTodos();
    });
  }

Here I have to problem that the HTTP requests are being sent twice because of two subscriptions.
Subscriptions in ngOnInit:

ngOnInit(): void {
    this.fetchTodos$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(todos => {
      this.todos = todos;
      this.fetchTodos$.complete();
    });

    this.fetchUsers$.pipe(
      skip(1),
      switchMap(innerObservable => innerObservable)
    )
    .subscribe(users => {
      this.users = users;
      this.fetchUsers$.complete();
    });

    this.loadData();
  }