RxJS Observables 只能由一个订阅者处理吗？

Question

我试图了解 Observables 和 RxJS 的工作原理，所以这可能根本不是如何使用它们的重点。

我有一个 Angular2 应用程序，并且还使用 RxJS Observables 来发送事件。现在对于一种特殊类型的错误事件，我想知道该事件是否已经由另一个订阅者处理。 Observable 上可能存在多个订阅者，其中一些可能会对该事件承担全部责任，这样其他订阅者就不会再得到它了。

这个想法来自路由事件在 WPF 中的工作方式。在事件处理程序中，您将获得RoutedEventArgs 参数，该参数具有属性Handled：

如果设置，如果事件被标记为已处理，则设置为true；否则为假。如果读取此值，则 true 表示类处理程序或沿途的某个实例处理程序已标记此事件已处理。 false.表示没有此类处理程序已将事件标记为已处理。

另一个实现示例是中间件在 ASP.NET Core 管道中的工作方式 - https://docs.microsoft.com/en-us/aspnet/core/fundamentals/middleware - 您可以调用下一个中间件或只返回一个结果。

我正在考虑将 Handled 属性添加到我要放入可观察管道的事件中，但我不确定这是否是在 RxJS 中执行此操作的惯用方式。

Answer 1

通常，您使用 observable 的方式是不将 observable 交给每个人并且每个人都订阅它。相反，您给每个感兴趣的方一个“添加到管道”的机会，然后最终订阅一次。有很多方法可以做到这一点。最简单的是实际上不给任何人可观察的。而是让他们为您提供回调：

class Foo {

   observable = ...;
   callbacks = [];

   addCallback(callback) { this.callbacks.push(callback); }

   constructor() {

       // subscribe to the observable and call any registered callbacks
       this.observable.subscribe(e => {
           for (const cb of this.callbacks) {
               // callback returns true if it has "handled" the event
               if (cb(e)) {
                   return; // do not call any other callbacks
               }
           }
       });
   }
}

const foo = new Foo();

// subscriber1 handles "type1" events
foo.addCallback(ev => ev.type === "type1");

// subscriber2
foo.addCallback(ev => ev.type === "type2");

这是最简单的方法。 Foo 还可以通过其他方式为每个客户端公开可观察对象并监控其结果以构建管道。