Rx js для реализации живой рассылки

Question

мой код довольно прост:
При загрузке страницы я загружаю сообщения пользователя, затем хочу объединить все новые поступающие сообщения.

   this.me$ = this.store
        .pipe(map((state: AppState) => state.auth));

    const allMessages$ = this.me$
        .pipe(mergeMap((me: User) => this.messageService.getMessage$(me.username)));

    const allNewMessages$: Observable<Message[]> = this.me$
        .pipe(mergeMap((me: User) => this.messageService.newMessage$(me.username)))
        .pipe(scan((messages: Message[], newMessage: Message) => messages.concat(newMessage), []))
        .pipe(startWith([]))

    this.messages$ = combineLatest([allMessages$, allNewMessages$])
        .pipe(map(([allMessages, allNewMessages]) => allMessages.concat(allNewMessages)))
        .pipe(map((messages: Message[]) => [...messages].sort((d1, d2) => new Date(d1.date) < new Date(d2.date) ? -1 : 1)));

Проблема в том, что новое сообщение отправляет сообщение. allNewMessages $ наблюдаемый emit вроде:

events[m1, m2, m3] -> emitted[[m1], [m1, m1, m2], [m1, m1, m2, m1, m1, m2, m3]]

и я хочу что-то вроде:

events[m1, m2, m3] -> emitted[[m1], [m1, m2], [m1, m2, m3]]

спасибо

Answer 1

Я наконец нашел решение. Я должен заменить свои mergeMap на switchMap, таким образом, наблюдаемые Http хорошо неписаны и избегают множественных хаотических c вызовов.