组合/合并可观察对象答案

【问题标题】：Combining/merging observables组合/合并可观察对象
【发布时间】：2017-07-18 21:41:45
【问题描述】：

假设我正在开发一个聊天应用程序。我有可观察的threads$ 每 n 秒发出线程数组，可观察的offline$ 通知线程何时离线，可观察的online$ 通知线程何时在线：

enum ConnectionStatus { Offline = 0, Online }

interface Thread {
    id: string;
    status: ConnectionStatus
}

const threads$ = Observable
    .interval(n)
    .switchMap(() => Observable.create((observer: Observer<Array<Thread>>) =>
        getThreads((threads: Array<Thread>) => observer.next(threads))));

const online$ = Observable.create((observer: Observer<Thread>) =>
    onOnline((threadId: string) => observer.next({
        id: threadId,
        status: ConnectionStatus.Online
    })));

const offline$ = Observable.create((observer: Observer<Thread>) =>
    onOffline((threadId: string) => observer.next({
        id: threadId,
        status: ConnectionStatus.Offline
    })));

我想按照这个规则组合这些流：threads$ 应该每 n 秒发出一次数组，但是每当online$ 或offline$ 发出时，我想获取最新值（@987654328 @) 的threads$ 并通过改变一个线程的状态来映射它并立即发出映射集合。

我已经迷失了 Rx 的 combineLatest、mergeMap、zip 和类似的，所以如果有人可以帮助我在这种情况下实现组合，我将不胜感激（在更多的 Rx 方式中）

【问题讨论】：

标签： javascript typescript rxjs observable rxjs5

【解决方案1】：

我认为你可以使用 multicast() 来做到这一点：

const stop$ = Observable.merge(online$, offline$);
threads$
    .multicast(new Subject(), obs => Observable.merge(obs, obs.takeUntil(stop$).takeLast(1).map(...)))
    .subscribe(...);

我显然没有测试它，但也许它会推动你正确的方向。

【讨论】：

【解决方案2】：

这应该在每次threads$ 发出时发出一个Array<Thread>并且在online$ 和offline$ 发出时立即发出。

const threadUpdate$ = Observable.merge(
    threads$,
    Observable.merge(online$, offline$)
        .withLatestFrom(threads$,
            (thread, threads) => threads.map(t => {
                if(t.id === thread.id) {
                    t.status = thread.status
                }
            })));

请注意，threads$ 将继续发射，甚至可能与组合的 online$/offline$ 流同时发射。

【讨论】：