在我们的客户端应用程序中，我们通过 SignalR.Client 发送和接收实时事件。从服务器订阅事件是通过 SignalR 完成的。在大多数情况下，订阅事件通过 SignalR 客户端发送到 SignalR 集线器，然后使用响应式扩展，将响应此类事件的应用程序部分订阅正确的 IObservable。
然而，我们现在有一些情况，同一个 observable 共享多个观察者。处理对可观察对象的每个内部订阅都很好，但我想在处理最后一个观察者后取消订阅服务器上的给定事件。我曾考虑为所有这些订阅使用单独的连接，但在这个特定的网络环境中，在连接变为活动状态之前存在相当长的延迟。
例如，客户端应用程序具有组件 A、B 和 C，它们都希望附加到从服务器接收的 SomeEvent。一旦其中任何一个变为活动状态，如果它是第一个活动的，则应用程序应订阅远程 SignalR 集线器上的事件。当其中任何一个变为活动状态时，它应该使用 Rx 在内部订阅我们已经实现的可观察集合中的相关可观察对象（如果不存在则添加一个）。当 A、B 或 C 变为非活动状态时，它显然应该处置其 IDisposable 订阅。如果它是剩下的最后一个，它也应该从 SignalR 集线器服务器退订。
Tl;dr：我想做一个托管取消订阅，看看离开的观察者是否是最后一个，并且此时仅在服务器集线器上取消订阅。有没有一种有效的方法来计算 IObservable 的“活”观察者的数量？或者其他更好的方法来实现所有这些？非常感谢！