javascript - 组合/合并可观察对象

Question

假设我正在开发一个聊天应用程序。我有观察 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$发出，我想获取 Array<Threads> 的最新值( threads$ )并通过改变一个线程的状态来映射它并立即发出映射集合。

我迷失了 Rx 的 combineLatest , mergeMap , zip和类似的，所以如果有人能帮助我在这种情况下实现组合(更多的是 Rx 方式)，我将不胜感激

Answer 1

我认为你可以使用 multicast() 来实现它:

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

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