Как изменить наблюдаемый поток другим наблюдаемым в RXJS? Угловой

Я хочу изменить вложенный флаг в моем ships$, когда изменится флаг areShipsExpanded$. Как я могу это сделать. Вот что у меня есть на данный момент:

areShipsExpanded$: Observable<boolean>;
ships$: Observable<Ship>;

constructor() {
   this.ships$ = this.shipsDataSource.getData().pipe(shareReplay());
}

ngOnInit(): void {
    this.areShipsExpanded$.pipe(
        map(flag => {
            if (flag) {
                this.ships$.pipe(map(s => s.items.forEach(r => r.isExpanded = flag)));
            }
        })
    ).subscribe();
}

К сожалению, похоже, что sizes $ не изменен, или, может быть, мне нужно как-то обновить эту ships$ наблюдаемую, поскольку я использую ее в своем представлении таким образом: <ng-container *ngFor="let ship of (ships$ | async)?.items">

@РЕДАКТИРОВАТЬ

Я так поступил, это правильно или это плохая практика?

this.areShipsExpanded$.pipe(
    map(flag => {
        if (flag) {
            this.ships$ = this.ships$.pipe(map(s => {
                s.items.forEach(r => r.isExpanded = flag);
                return s;
            }));
        }
    })
).subscribe();

@ EDIT2

Я использовал combineLatest, теперь все должно быть в порядке?

shipsData$: Observable<Ship>;
ships$: Observable<Ship>;

constructor() {
   this.shipsData$ = this.shipsDataSource.getData().pipe(shareReplay());
}

this.ships$ = combineLatest(this.areShipsExpanded$, this.shipsData$).pipe
    map(([shipsExpanded, ships]) => {
        ships.items.forEach(r => r.isExpanded = shipsExpanded);
        return ships;
    }),
    share()
);

Но проблема combineLatest в том, что вначале функции будут запускаться 2 раза, поскольку оба наблюдаемых объекта испускают значения. Есть ли какой-либо альтернативный оператор, который ведет себя как combineLatest - поэтому, когда какой-либо наблюдаемый выдает значение, испускать последнее значение из каждого, но ТОЛЬКО КОГДА все наблюдаемые будут готовы. Чтобы не повторяться в начале.