javascript - Kombination/Zusammenführung von Observables

Angenommen, ich entwickle eine Chat-App. Ich habe ein beobachtbares threads$, das alle n Sekunden ein Array von Threads ausgibt, ein beobachtbares offline$, das benachrichtigt, wenn ein Thread offline geht, und ein beobachtbares online$, das benachrichtigt, wenn ein Thread online wird:

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
    })));

Ich möchte diese Streams nach dieser Regel kombinieren: threads$ sollte alle n Sekunden ein Array ausgeben, aber immer wenn online$ oder offline$ ausgibt, möchte ich den neuesten Wert (Array<Threads>) von threads$ abrufen und ihn abbilden, indem ich den Status eines Threads ändere und die abgebildete Sammlung sofort ausgeben.

Ich habe bei Rx's combineLatest, mergeMap, zip und ähnlichen den Überblick verloren, daher würde ich es sehr schätzen, wenn mir jemand helfen könnte, die Kombination in diesem Fall zu implementieren (mehr auf eine Rx-Weise).

Antwort #1

Ich denke, du könntest es so mit multicast() machen:

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

Offensichtlich habe ich es nicht getestet, aber vielleicht wird es dich in die richtige Richtung schubsen.

Antwort #2

Dies sollte ein Array<Thread> ausgeben, jedes Mal wenn threads$ ausgibt und sofort, wenn online$ und offline$ ausgeben.

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
                }
            })));

Beachten Sie, dass threads$ weiterhin ausgeben wird und möglicherweise sogar zur gleichen Zeit wie der kombinierte online$/offline$-Stream ausgeben könnte.