javascript - Kuinka valikoivasti lähettää kuumia (jaettuja) observables RxJS:ssä?

Koodissani minulla on kuuma observable (kylmä observable, joka on putkistettu share()-operaattorin läpi), joka lähettää arvoja tunnisteiden kanssa (eli datapaketin tyyppi sisältää "id"-merkkijonomuuttujan). Useat havainnoijat suodattavat sitten lähetetyt arvot kyseisen id:n perusteella.

Nyt luulen, että koodini ottaa suuren suorituskykyiskun, koska jokaisen havainnoijan täytyy suodattaa jokainen arvo, joka lähetetään. Tai oikeastaan joka kerta, kun tarvitsen uuden havaitsijan tilaamaan jaetun havaittavan kohteen, luodaan uusi suodatettu havaittava kohde. Koodi näyttää jotenkin tältä:

const source = new Observable();
const sharedObservable = source.pipe(share());
sharedObservable
    .pipe(filter(payload => payload.id === "id1"))
    .subscribe(observerOne);
sharedObservable
    .pipe(filter(payload => payload.id === "id2"))
    .subscribe(observerTwo);
// etc.

Alkuperäinen ratkaisuni on luoda oma tyyppinen hot/shared-observable, joka käsittelee suodatuksen sisäisen id:istä observereihin -listaan perustuvan kartan avulla, jotka pitäisi vastaanottaa payloadit. Tämä vähentäisi kustannuksia, jotka liittyvät sen predikaattifilterifunktion ajamiseen jokaiselle yksittäiselle tilaajalle. Observable alla näyttäisi joltakin (tai epämääräisesti siltä) tältä:

class FilteredHotObservable {
    constructor(sourceObservable) {
        this.observerMap = new Map();
        this.source = sourceObservable;
        this.source.subscribe(payload => {
            const observers = this.observerMap.get(payload.id);
            observers.forEach(observer => observer.next(payload));
        });
    }

    subscribe(observer, desiredIdToFilterBy) {
        const observers = this.observerMap.get(desiredIdToFilterBy);
        observers.push(observer);
    }
}

Ennen kuin menen ja toteutan tämän, mietin, onko tähän tilanteeseen jo olemassa kunnollista tapaa. Toisin sanoen, onko RxJS:llä jo mekanismia (esim. erityinen Subject, hieno operaattori) tämänkaltaisen skaalautuvan "suodatetun ja jaetun" havaittavan helpottamiseksi?

VASTUUVAPAUSLAUSEKE: Yllä olevien koodipätkien nimet ja syntaksi eivät vastaa sitä, mitä oikeasti teen. Ne ovat vain näytteitä tämän postauksen vuoksi.

Muokattu: Tässä ovat todelliset luvut filter-operaattorin käyttämisen päästönkäsittelyviiveestä. Arvolla "PostFilterLatency" lähetettyjä tietopaketteja suodatetaan ID:iden perusteella, jotka vastaavat yksittäisiä tarkkailijoita jaetussa observable:ssa. Arvolla "PreFilterLatency" päästöt suodatetaan manuaalisesti id->Observer-kartan avulla, ja sen jälkeen kutsutaan observer.next() sopivalla tietopaketilla. Luvut kuvaavat aikaa, joka kului siitä, kun tietopaketit lähetettiin, siihen kun ne käsiteltiin vastaavien tarkkailijoiden toimesta (aika millisekunneissa).

ObserverCount  PostFilterLatency  PreFilterLatency
-------------  -----------------  ----------------
1              0                  0
10             0                  0
100            0.03               0
1000           0.039              0.001
10000          1.1849             0.0005
100000         19.90902           0.002

Muokkaus: tässä on linkki koodiin, jota käytin näiden testien suorittamiseen.