如何组织与串行端口设备对话的代码?

Question

我正在编写一个。net应用程序，需要与串行端口设备通信。这个设备基本上是老式字母数字传呼机的发射器。偶尔，我的应用程序需要打开一个串行端口并向发送器发送消息。

我知道与设备对话的协议。这是一种来回的"闲聊"协议。发送命令…等待一个特定的响应…再发一个命令…等待另一个特定的响应…发送实际的消息…等待一个"接受"的响应。我可以用一些真正令人讨厌的代码来实现这一点，这些代码涉及一系列对SerialPort对象的Write(…)方法调用，使用Thread。睡眠呼叫。

当然，我不想通过依赖Thread来实际做到这一点。休眠以等待设备响应。看起来响应式扩展框架应该适合这种类型的事情，但是我很难理解它。我从这个开始，但很快就迷路了，不知道下一步该去哪里，或者这是否有意义:

var receivedData = Observable.FromEventPattern<SerialDataReceivedEventArgs>(serialPort, "DataReceived");
receivedData
    .Where(d => d.EventArgs.EventType == SerialData.Chars)
    .Subscribe(args =>
            {
                var response = serialPort.ReadExisting();
                // Now what?
            });

首先，我如何用第一个serialPort.Write()调用来启动这个东西?那么，在发出下一个Write()调用之前，如何通过检查预期的响应来将它们链接在一起呢?当然，如果我没有得到预期的响应，我就会跳出来，抛出异常之类的。我用Rx找对树了吗，还是有另一种模式更适合这个?谢谢!

Answer 1

Rx是对"数据源推送数据"场景的抽象。在您的情况下，您已经将串行端口"读取"方法建模为Rx可观察到的方法，这需要与串行端口写入方法相结合。一个可能的解决方案如下所示，尽管它可能需要根据应用程序的特定需求进行一些其他修改。

            var serialPort = new System.IO.Ports.SerialPort("COM1");
            serialPort.Open();
            var receivedData = Observable.FromEvent<SerialDataReceivedEventArgs>(serialPort, "DataReceived")
                               .Where(d => d.EventArgs.EventType == SerialData.Chars)
                               .Select(_ => serialPort.ReadExisting());
            var replay = new ReplaySubject<string>();
            receivedData.Subscribe(replay);
            var commands = (new List<string>() { "Init", "Hello", "Done" });
            commands.Select((s, i) =>
            {
                serialPort.Write(s);
                var read = replay.Skip(i).First();
                //Validate s command against read response
                return true;//To some value to indicate success or failure
            }).ToList();

Answer 2

我不认为RX非常适合这种串行通信。总的来说，RX似乎更多的是单向数据流，而不是来回协议。对于这样的串行通信，我围绕串行端口编写了一个类，该类使用WaitHandles等待对命令的响应。一般结构为:

应用程序调用一个方法来启动一个异步操作来发送一系列命令。这将启动一个线程(我相信是从线程池中)，它依次发送每个命令。一旦发送了命令，操作将等待WaitHandle来获得响应(或超时并重试或操作失败)。当响应被处理时，接收WaitHandle被发出信号并发送下一个命令。

串行接收事件(当数据传入时在后台线程上运行)构建数据包。当收到完整的报文时，检查是否发送了命令。如果是这样，向发送新响应的线程发出信号，并等待不同的WaitHandle来处理响应(这对于防止接收方丢弃响应数据很重要)。

EDIT:添加了一个(稍大的)示例，显示了两个核心发送和接收方法。

没有显示Me.Receiver属性，它是ISerialReceiver类型的，负责构建数据包，但不确定数据是否是正确的响应。同样没有显示的是CheckResponse和ProcessIncoming，它们是两个抽象方法，由派生类覆盖，分别用于确定响应是否针对刚刚发送的命令和处理"未经请求的"传入数据包。

''' <summary>This field is used by <see cref="SendCommand" /> to wait for a
''' response after sending data.  It is set by <see cref="ReceiveData" />
''' when <see cref="ISerialReceiver.ProcessResponseByte">ProcessResponseByte</see>
''' on the <see cref="Receiver" /> returns true.</summary>
''' <remarks></remarks>
Private ReceiveResponse As New System.Threading.AutoResetEvent(False)
''' <summary>This field is used by <see cref="ReceiveData" /> to wait for
''' the response to be processed after setting <see cref="ReceiveResponse" />.
''' It is set by <see cref="SendCommand" /> when <see cref="CheckResponse" />
''' returns, regardless of the return value.</summary>
''' <remarks></remarks>
Private ProcessResponse As New System.Threading.ManualResetEvent(True)
''' <summary>
''' This field is used by <see cref="SendCommand" /> and <see cref="ReceiveData" />
''' to determine when an incoming packet is a response packet or if it is
''' one of a continuous stream of incoming packets.
''' </summary>
''' <remarks></remarks>
Private responseSolicited As Boolean
''' <summary>
''' Handles the DataReceived event of the wrapped SerialPort.
''' </summary>
''' <param name="sender">The wrapped SerialPort that raised the event.
''' This parameter is ignored.</param>
''' <param name="e">The event args containing data for the event</param>
''' <remarks>This function will process all bytes according to the
''' <see cref="Receiver" /> and allow <see cref="SendCommand" /> to
''' continue or will call <see cref="ProcessIncoming" /> when a complete
''' packet is received.</remarks>
Private Sub ReceiveData(ByVal sender As Object, ByVal e As SerialDataReceivedEventArgs)
    If e.EventType <> SerialData.Chars Then Exit Sub
    Dim input() As Byte
    SyncLock _portLock
        If Not _port.IsOpen OrElse _port.BytesToRead = 0 Then Exit Sub
        input = New Byte(_port.BytesToRead - 1) {}
        _port.Read(input, 0, input.Length)
    End SyncLock
    'process the received data
    If input Is Nothing OrElse input.Length = 0 OrElse Me.Receiver Is Nothing Then Exit Sub
    Dim responseCompleted As Boolean
    For i As Integer = 0 To input.Length - 1
        responseCompleted = Me.Receiver.ProcessResponseByte(input(i))
        'process completed response
        If responseCompleted Then
            responseSolicited = False
            System.Threading.WaitHandle.SignalAndWait(ReceiveResponse, ProcessResponse)
            'the data is not a response to a command sent by the decoder
            If Not responseSolicited Then
                ProcessIncoming(Me.Receiver.GetResponseData())
            End If
        End If
    Next
End Sub
''' <summary>
''' Sends a data command through the serial port.
''' </summary>
''' <param name="data">The data to be sent out the port</param>
''' <returns>The data received from the port or null if the operation
''' was cancelled.</returns>
''' <remarks>This function relies on the Receiver 
''' <see cref="ISerialReceiver.GetResponseData">GetResponseData</see> and 
''' the overriden <see cref="CheckResponse" /> to determine what response 
''' was received and if it was the correct response for the command.
''' <seealso cref="CheckResponse" /></remarks>
''' <exception cref="TimeoutException">The operation timed out.  The packet
''' was sent <see cref="MaxTries" /> times and no correct response was received.</exception>
''' <exception cref="ObjectDisposedException">The SerialTransceiver was disposed before
''' calling this method.</exception>
Private Function SendCommand(ByVal data() As Byte, ByVal ignoreCancelled As Boolean) As Byte()
    CheckDisposed()
    If data Is Nothing Then Return Nothing
    'make a copy of the data to ensure that it does not change during sending
    Dim sendData(data.Length - 1) As Byte
    Array.Copy(data, sendData, data.Length)
    Dim sendTries As Integer = 0
    Dim responseReceived As Boolean
    Dim responseData() As Byte = Nothing
    ReceiveResponse.Reset()
    ProcessResponse.Reset()
    While sendTries < MaxTries AndAlso Not responseReceived AndAlso _
          (ignoreCancelled OrElse Not Me.IsCancelled)
        'send the command data
        sendTries += 1
        If Not Me.WriteData(sendData) Then Return Nothing
        If Me.Receiver IsNot Nothing Then
            'wait for Timeout milliseconds for a response.  If no response is received
            'then waitone will return false.  If a response is received, the AutoResetEvent
            'will be triggered by the SerialDataReceived function to return true.
            If ReceiveResponse.WaitOne(Timeout, False) Then
                Try
                    'get the data that was just received
                    responseData = Me.Receiver.GetResponseData()
                    'check to see if it is the correct response
                    responseReceived = CheckResponse(sendData, responseData)
                    If responseReceived Then responseSolicited = True
                Finally
                    'allow the serial receive function to continue checking bytes
                    'regardless of if this function throws an error
                    ProcessResponse.Set()
                End Try
            End If
        Else
            'when there is no Receiver, assume that there is no response to
            'data sent from the transceiver through this method.
            responseReceived = True
        End If
    End While
    If Not ignoreCancelled AndAlso Me.IsCancelled Then
        'operation was cancelled, return nothing
        Return Nothing
    ElseIf Not responseReceived AndAlso sendTries >= MaxTries Then
        'operation timed out, throw an exception
        Throw New TimeoutException(My.Resources.SerialMaxTriesReached)
    Else
        'operation completed successfully, return the data
        Return responseData
    End If
End Function

Answer 3

通过结合Rx和async/await可以很好地处理发送和接收的工作流。