使用来自 SignalR 的共享 IObservable

Question

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

我们还假设我有一个具有 Subscribe 方法的 SignalR 集线器和一个具有 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 个事件。然后每个客户端都应该实时看到相同的事件。

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

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

Answer 1

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

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

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

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

这种历史记录和实时流的分离为您提供了处理两者的灵活性，并提供了提高效率的机会，例如翻阅历史记录而不是抓住整个事件。

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

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