Я не считаю, что RX идеально подходит для такого рода последовательной связи. В целом RX, кажется, больше направлен на потоки данных в одном направлении, а не на протоколы туда и обратно. Для таких последовательных соединений я написал класс вокруг последовательного порта, который использует WaitHandles для ожидания ответа на команды. Общая структура:
Приложение вызывает метод для запуска асинхронной операции для отправки последовательности команд. Это запустит поток (я думаю, из пула потоков), который отправит каждую команду по очереди. Как только команда отправлена, операция ожидает в WaitHandle, чтобы получить ответ (или тайм-аут и повторить попытку или не выполнить операцию). Когда ответ обрабатывается, выдается сообщение WaitHandle и отправляется следующая команда.
Событие последовательного приема (которое запускается в фоновых потоках всякий раз, когда поступают данные) создает пакеты данных. Когда полный пакет получен, проверьте, была ли отправлена команда. Если это так, отправьте поток сообщений о новом ответе и дождитесь другого WaitHandle, чтобы обработать ответ (что может быть важно для предотвращения получения получателем данных ответа).
РЕДАКТИРОВАТЬ: Добавлен (несколько большой) образец, показывающий два основных метода отправки и получения.
Не показано свойство 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