使用 RX 观察 WCF 服务调用

Question

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

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

• 这样使用主题：

  
   Subject s1 = new Subject();
   s1.Timeout(TimeSpan.FromSeconds(3))
     .Subscribe(_ => { }, ex => Console.WriteLine("Timeout at " + DateTime.Now));
  
  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);
  

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

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

然后使用 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