javascript - ¿Cómo emitir selectivamente observables calientes (compartidos) en RxJS?

En mi código, tengo un observable caliente (observable frío canalizado a través del operador share()) que emite valores con identificadores (es decir, el tipo de carga útil contiene una propiedad de cadena "id"). Un grupo de observadores en esto luego filtran los valores emitidos basándose en dicho id.

Ahora creo que mi código está sufriendo una gran pérdida de rendimiento, ya que cada observador tiene que filtrar cada valor que se emite. O más bien, se crea un nuevo observable filtrado cada vez que necesito que un nuevo observador se suscriba al observable compartido. El código más o menos se ve así:

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.

Mi solución inicial es crear mi propio tipo de observable caliente/compartido que maneje el filtrado basado en un mapa interno de identificadores a listas de observadores que deberían recibir los datos. Eso reduciría el costo de tener que ejecutar alguna función de filtro de predicado para cada observador que se suscribe. Debajo, el observable se vería algo así (o vagamente así):

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

Antes de ir y ejecutar esto, me preguntaba si ya existe una manera adecuada de manejar este escenario. En otras palabras, ¿RxJS ya tiene algún mecanismo (por ejemplo, un Subject especial, un operador sofisticado) para facilitar este tipo de observable escalable "filtrado y compartido"?

AVISO: Los nombres y la sintaxis de los fragmentos de código anteriores no son precisos con lo que realmente estoy haciendo. Son solo ejemplos para el propósito de esta publicación.

Editar: Aquí están los números reales respecto a la latencia de manejo de emisiones al usar el operador filter. Para "PostFilterLatency", las cargas emitidas se filtran por IDs que corresponden a observadores individuales en el observable compartido. Para "PreFilterLatency", las emisiones se filtran manualmente a través del mapa id->Observer, luego se llama a observer.next() con la carga correspondiente. Los números representan el tiempo que tomó desde que las cargas fueron emitidas hasta que fueron manejadas por sus correspondientes observadores (el tiempo está en milisegundos).

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

Editar: aquí hay un enlace al código que usé para ejecutar estas pruebas.