Rx.Net：处理订阅时执行异步副作用

Question

我有一个通过串口与计算机通信的设备。发送一个“START”命令后，设备回应一个确认，并开始监控一些外部活动。然后它根据外部活动异步传输串行端口上的一些消息。当设备收到“停止”命令时，它会回应确认，然后停止发送更多消息（表示外部活动）。

我已经实现了带有冷观察功能的启动/停止命令，这些命令执行副作用（串口发送命令），并且如果在串口上收到一个ACKckowledge，则会发出一个单一的Unit.Default。我想构建一个IObservable，它发出与外部活动相关的消息，并在订阅时执行“START”副作用，并且在订阅处理时执行“STOP”副作用。 “START”很简单，我只需要做一个“SelectMany”，但我不知道如何执行“STOP”。

class MonitoringDevice 
{ 
    private SerialPort _sp; 
    private IObservable<byte> _receivedBytes; 

    public IObservable<ExternalActivity> ActivityStream { get; } 

    public MonitoringDevice() 
    { 
     _sp = new SerialPort("COM1"); 
     _receivedBytes = Observable 
         .FromEventPattern<SerialDataReceivedEventHandler, SerialDataReceivedEventArgs>(
          h => 
          { 
           _sp.DiscardInBuffer(); 
           _sp.DataReceived += h; 
          }, 
          h => 
          { 
           _sp.DataReceived -= h; 
          }) 
         .SelectMany(x => 
         { 

          byte[] buffer = new byte[1024]; 
          var ret = new List<byte>(); 
          int bytesRead = 0; 
          do 
          { 
           bytesRead = _sp.Read(buffer, 0, buffer.Length); 
           ret.AddRange(buffer.Take(bytesRead)); 
          } while ((bytesRead >= buffer.Length)); 
          return ret; 

         }) 
         .Publish() 
         .RefCount(); 


     ActivityStream = StartMonitoringAsync() 
         .SelectMany(_receivedBytes.ToActivity()); 
         // we need to execute StopMonitoringAsync 
         // when a subscription to ActivityStream is disposed 

     _sp.Open(); 
    } 



    private IObservable<Unit> StartMonitoringAsync() 
    { 
     return Observable 
       .Create<Unit>(
       obs => 
       { 
        _sp.Write("START"); 
        return _receivedBytes 
          .ToAcknowledge() 
          .FirstAsync() 
          .Timeout(TimeSpan.FromMilliseconds(1000)) 
          .Subscribe(obs); 
       }); 
    } 


    private IObservable<Unit> StopMonitoringAsync() 
    { 
     return Observable 
       .Create<Unit>(
       obs => 
       { 
        _sp.Write("STOP"); 
        return _receivedBytes 
          .ToAcknowledge() 
          .FirstAsync() 
          .Timeout(TimeSpan.FromMilliseconds(1000)) 
          .Subscribe(obs); 
       }); 
    } 


} 

ExternalActivity只是一个POCO。

ToAcknowledge是一种扩展方法，返回IObservable，当设备传输确认时发出Unit.Default。 - 这是按预期工作的;

ToActivity是一种扩展方法，返回IObservable，它解析输入的串行数据并发出ExternalActivity对象。 - 这是按预期工作的;

---

编辑：增加了对ToAcknowledge和ToActivity扩展方法实现。

public static IObservable<Unit> ToAcknowledge(this IObservable<byte> source) 
    { 
     return source.Buffer(3, 1) 
       .Where(bfr => bfr.SequenceEqual(new byte[] { 65, 67, 75 })) // ACK 
       .Select(x => Unit.Default); 

    } 

    public static IObservable<ExternalActivity> ToActivity(this IObservable<byte> source) 
    { 
     return source 
       .Publish(ps => ps.Buffer(ps.Where(x => x == 1),    // SOH 
              bo => ps.Where(x => x == 4))) // EOT 
       .Select(bfr => bfr.Take(bfr.Count - 1).Skip(1)) 
       .Where(bfr => bfr.Count() == 12) 
       .Select(bfr => 
       { 
        var timestamp = BitConverter.ToInt64(bfr.Take(8).ToArray(), 0); 
        var id = BitConverter.ToInt32(bfr.ToArray(), 8); 
        return new ExternalActivity(timestamp, id); 
       });   
    } 

Answer 1

您可以通过修改StartAsync是这样的：

private IObservable<Unit> StartAsync(Action unsubscribe) 
{ 
    return 
     Observable 
      .Create<Unit>(o => 
      { 
       var subscription = 
        Observable 
         .Timer(TimeSpan.FromSeconds(1)) 
         .Select(_=> Unit.Default) 
         .Subscribe(o); 
       return new CompositeDisposable(
        subscription, 
        Disposable.Create(unsubscribe)); 
      });; 
} 

然后你就可以在你喜欢的任何Action unsubscribe注入。与此代码

尝试测试：

var subscription = 
    StartAsync(() => Console.WriteLine("Done")) 
    .Subscribe(); 

Thread.Sleep(3000); 

subscription.Dispose(); 

你会看到“完成”后3秒写入控制台。