c# - 使用来自 SignalR 的共享 IObservable

Question

假设我有一个 IObservable<Something>服务器端，存储在静态字段或其他内容中。

我们还假设我有一个 SignalR 集线器，它有一个 Subscribe方法和具有 notify 的 signalR 客户端功能。

public static IObservable<string> Events;
protected void Application_Start()
{
   //dummy observable just to generate events for me..
   Events = Observable
       .Timer(TimeSpan.Zero, TimeSpan.FromMilliseconds(50))
       .Select(l => l.ToString());
...snip..
}

和:

public class MyHub1 : Hub
{        
    public void Subscribe()
    {
        Clients.All.notify("start");
        WebApiApplication.Events
            .Subscribe(s => Clients.Caller.notify(s));
    }
}

和:

myHub.client.notify = function (event) {
     console.info(event);
};

我需要做什么才能让所有客户端共享可观察对象？也就是说，我希望每个连接的客户端在订阅后都能收到最近 200 个事件。然后每个客户端都应该实时看到相同的事件。

我想 Replay IObservable 的方法应该以某种方式使用。我希望它的行为非常像聊天，用户可以实时查看最后的 x 条消息和每个新事件。

除了如何实际编写可观察查询之外，在 Asp.NET 中存储和设置共享事件流的最佳方法是什么？

Answer 1

这是一个非常常见的架构问题 - 将实时流与“世界状态”相结合。您想要做的是利用 SignalR 向当前订阅者广播实时消息(它擅长的)，并有一个单独的 API 调用来加入客户端以获取历史消息。

在客户端中，您提供的逻辑首先订阅实时 SignalR 消息流，然后请求已发生消息的历史记录(“世界状态”)——通常最好将其作为简单的有序列表拉回。

存在一种固有的竞争条件，可能会导致在实时流和历史记录中收到一条消息 - 因此您必须注意删除重复或“删除重复”您的消息列表。

如何持久化消息可以作为一个完全独立的问题来处理。

这种历史记录和实时流的分离为您处理两者提供了灵 active ，并提供了提高效率的机会，例如分页到历史记录而不是抓取整个内容。

有几个问题和答案讨论了如何利用 Rx 来结合历史数据和实时数据——您需要在客户端 javascript 中执行此操作，而我对 rx js 不是很好。

看看Merging historical and live stock price data with Rx关于这个问题的一些讨论，还有Are these two Observable Operations Equivalent? - 我在后一种情况下有一些示例代码，这是一个纯 .NET 场景。