从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消息流的逻辑,然后请求已经发生的消息的历史记录("世界状态") - 通常最好作为简单的有序列表撤回.

有一种固有的竞争条件可能会导致在直播和历史记录中收到消息 - 因此您必须注意重复删除或"删除"您的消息列表.

然后,如何将消息保留为完全独立的问题.

这种历史和现场直播的分离为您提供了处理这两者的灵活性,并提供了提高效率的机会,例如分页到历史而不是抓住整个事物.

有一些问题和答案讨论了如何利用Rx来结合历史数据和实时数据 - 你需要在客户端javascript中执行此操作,而且我在rx js上并不是很好.

看一下将历史和实时股票价格数据与Rx合并以进行关于此问题的一些讨论,这两个可观察操作是否等效？ - 我在后一种情况下有一些示例代码,这是纯粹的.NET场景.