c# - 使用 RX 观察 WCF 服务调用

Question

我有一个 WCF 心跳服务接受来自客户端的保持事件。如果 keep alive 没有在指定时间内到达，则会引发超时并采取一些措施。现在，我正在尝试简化使用 System.Threading.Timer 的现有代码，并将其替换为基于 RX 的解决方案。我缺少的是如何将 WCF 服务方法类转换为可观察的。我考虑过以下内容:

• 在 WCF 服务和缓存观察器中实现 IObservable 作为订阅参数传递，可能不是一个好主意和难看的代码

• 像这样使用主题:

  
   Subject s1 = new Subject();
   s1.Timeout(TimeSpan.FromSeconds(3))
     .Subscribe(_ => { }, ex => Console.WriteLine("Timeout at " + DateTime.Now));<p></p>
  
  <p>Console.WriteLine("Hertbeat at: " + DateTime.Now);
   s1.OnNext(Unit.Default);
   Thread.Sleep(TimeSpan.FromSeconds(2));
   Console.WriteLine("Hertbeat at: " + DateTime.Now);
   s1.OnNext(Unit.Default);
  </p>

  这可行，但在生产代码中使用主题可能不是一个好主意。
  • 在服务实现中使用事件:

  
  class HeartbeatService
  {
    public void Heartbeat()
    {
      OnHeartbeatArrived();
    }<p></p>
  
  <p>public event Action HeartbeatArrived;</p>
  
  <p>protected virtual void OnHeartbeatArrived()
    {
  Action handler = HeartbeatArrived;
  if (handler != null) handler(Unit.Default);
    }
  }</p>

然后使用RX

HeartbeatService heartbeatService = new HeartbeatService();
Observable.FromEvent<Unit>(action => heartbeatService.HeartbeatArrived += action,
                           action => heartbeatService.HeartbeatArrived -= action)
  .Timeout(TimeSpan.FromSeconds(3))
  .Subscribe(_ => { },
             ex => Console.WriteLine("Timeout " + ex + ", at " + DateTime.Now));

Console.WriteLine("Heartbeat at: " + DateTime.Now);
heartbeatService.Heartbeat();
Thread.Sleep(TimeSpan.FromSeconds(2));
Console.WriteLine("Heartbeat at: " + DateTime.Now);
heartbeatService.Heartbeat();
Console.WriteLine("Should timeout now");

我也不喜欢这个解决方案，WCF 服务中的事件似乎是多余的。在 RX 中有这样做的好方法还是我想在这里滥用 RX？

Answer 1

Throttle 运算符是 Rx 中心跳算法的关键部分。基本思想是设置一个包含警报的流，只要心跳到达，该警报就会被抑制。有关更多详细信息，请参阅我的博客文章:http://www.zerobugbuild.com/?p=230