rxjs merge: вы указали «undefined» там, где ожидался поток

Question

Я получаю сообщение об ошибке You provided 'undefined' where a stream was expected., когда пытаюсь объединить наблюдаемые объекты.

Я создаю свое наблюдаемое обновление здесь:

refresh$: Observable<Funnel[]>; 

тогда у меня есть таймер ...

TimerObservable.create(0, (this.storageService.refRate * 1000))
    .takeWhile(() => this.interval)
    .subscribe(() => {
        this.refresh$ = this.funnelService.getStoryFunnels({}).map(response => [...response]);
    });

, а затем, когда я пытаюсь объединить наблюдаемое, выдается ошибка:

this.funnels$ = this.funnelService.getStoryFunnels({}).pipe(
    merge(this.refresh$, funnelCreation$.pipe(
            filter(funnel => !!funnel),
            map(funnel => [funnel])
        )
    )
);

обновление на основе удобного ответа:

this.refresh$ = TimerObservable.create(0, (this.storageService.refRate * 500))
    .takeWhile(() => this.interval)
    .switchMap(() =>  this.funnelService.getStoryFunnels({}).map(response => [...response]));

и затем:

this.funnels$ = this.funnelService.getStoryFunnels({}).pipe(
        merge(this.refresh$, funnelCreation$.pipe(
                filter(funnel => !!funnel),
                map(funnel => [funnel])
            )
        )
    );

Answer 1

Это потому, что this.funnels$ пока не наблюдается, , когда он был объявлен (не инициирован никакими значениями), потому что ему будет присвоено значение только после выполнения таймера.Однако код объявления выполняется почти сразу (синхронно).Вот почему вы получаете undefined.

. То, что вы должны сделать, это то, что вы можете использовать switchMap и присвоить ему this.refresh$:

this.refresh$ = TimerObservable.create(0, (this.storageService.refRate * 1000))
    .takeWhile(() => this.interval)
    .switchMap(() =>  this.funnelService.getStoryFunnels({}).map(response => [...response]));