如何在 RxJS 中直接链接发布子系统中的生产者和消费者，而无需主题

>Edit

我认为这个问题的实施相当复杂，现在已经 3 个月无法弄清楚了。我在这里的另一个问题中重新措辞了它：RxJS5 - 如何在不使用主题并排除已完成的主题的情况下缓存聚合流的最后一个值？，希望能更简洁地总结我所追求的内容。谢谢大家的帮助！

源语言

现在，我已经学会了如何处理 RxJS 中的状态，方法是通过扫描将一系列部分应用的状态操作函数映射到我的状态上。但是，我现在想更进一步，通过数据平面(pubsub 层(以声明方式连接"生产者"和"消费者"之间的流，而无需通过主题管道连接流。

网上有很多教程涵盖了这一点，但它们基本上都只是在每个流名称或频道的中间层中强制使用 publish 作为 Subject.next。这是我当前的实现，但它需要为每个永无止境的流创建一个 ReplaySubject(1((用于缓存迟到的使用者的值(，因为任何获得流的使用者都会收到一个引用，如果当前没有生产者流式传输到该名称的通道时删除主题，则该引用将无效。

我想直接连接流，并让使用者接收流，这些流是特定名称的所有活动已发布流的聚合。这仍然需要主体在初始注册的生产者流(格式为 {name, stream} 的创建者流的传入流(中进行管道处理。

然后我想将所有同名流合并到一个每个注册消费者作为参考接收的流中(想想一个过滤器服务接收对filter.add的引用，这是所有创建过滤器的活动生产者的合并( - 但让这个流重新合并，以反应方式，并且消费者到该流的链接仍然有效， 如果具有相同名称的新生产者也注册。任何迟到的使用者还应收到该聚合流的最后一个缓存值。

通过这种方式，每次在 pubsub 层上公开新流时，都需要动态重新评估每个聚合流，因此将流包装在"getActive"函数中(如此处(不起作用，因为这是必不可少的，并且仅在首次获取流时发生一次，而不是在每次发布新流时懒惰地为所有消费者重新评估。

结果应为以：

• 从不完成;
• 聚合特定名称的所有活动流的结果，并丢弃不再活跃的流;
延迟更新收到对该流的引用的所有使用者，
• 以便在发布同名的新流时，使用者的引用仍然有效，并重新评估以合并到该新流中;
• 缓存上次合并的结果，以便新使用者将在订阅时收到上次发出的值。

基本上，我需要"trimToActiveOnly"功能。

function getStream(all$, name) {
  return all$
  .filter(x => x.name === name)
    .map(x => x.stream)
    .map(trimToActiveOnly) // in this way, should be dynamically re-evaluated for all
                 // consumers every time new stream is published or stream ends,
                 // not just run once when the stream is first 'got'
    .flatMapLatest(x => Rx.Observable.merge(x)) // where x is all currently active streams for a particular name, with finished/errored ones discarded
    .publish().refCount(); //this is re-evaluated when a new stream is published or when one of the aggregated streams concludes. So the aggregate stream itself never concludes but may go cold if nothing is subscribed.
}
// desired behavior as followed
const publishStream$ = new Rx.Subject();
const foo$ = getStream(publishStream$, 'foo');
const bar$ = getStream(publishStream$, 'bar');
const fooSourceA$ = new Rx.Subject();
const fooSourceB$ = new Rx.Subject();
const barSourceA$ = new Rx.Subject();
const barSourceB$ = new Rx.Subject();
publishStream$.onNext({ name: 'foo', stream: fooSourceA$ });
publishStream$.onNext({ name: 'foo', stream: fooSourceB$ });
publishStream$.onNext({ name: 'bar', stream: barSourceA$ });
fooSourceA$.onNext('hello');
fooSourceA$.onNext('world');
barSourceA$.onNext('testing');
const fooSub = foo$.subscribe(x => console.log('foo: ' + x)); // should receive cached 'world'
const barSub = bar$.subscribe(x => console.log('bar: ' + x)); // should receive cached 'testing'
publishStream$.onNext({ name: 'bar', stream: barSourceB$ });
barSourceB$.onNext('123'); // barSub should now receive '123' as the aggregated active streams are dynamically re-evaluated on each publish of a new stream!

我在这里也有一个JSBin。

function getStream(all$, name) {
  return active(all$.filter(x => x.name === name).map(x => x.stream))
    .flatMapLatest(x => Rx.Observable.merge(x));
}

const publishStream$ = new Rx.Subject();
const foo$ = getStream(publishStream$, 'foo');
const bar$ = getStream(publishStream$, 'bar');

const fooSub = foo$.subscribe(x => console.log('foo: ' + x));
const barSub = bar$.subscribe(x => console.log('bar: ' + x));

const fooSourceA$ = new Rx.Subject();
const fooSourceB$ = new Rx.Subject();
const barSourceA$ = new Rx.Subject();
const barSourceB$ = new Rx.Subject();
publishStream$.onNext({ name: 'foo', stream: fooSourceA$ });
publishStream$.onNext({ name: 'foo', stream: fooSourceB$ });
publishStream$.onNext({ name: 'bar', stream: barSourceA$ });
publishStream$.onNext({ name: 'bar', stream: barSourceB$ });
fooSourceA$.onNext('hello');
fooSourceA$.onNext('world');
barSourceA$.onNext('testing');
barSourceB$.onNext('123');