The main purpose of Reactive Extensions (Rx) is to enable processing event streams. I set out to use Reactive Extensions for receiving data on the serial port and learned some new things.
Here is the code that uses
Observable.FromEventPattern<T>() to create an
IObservable<T> from the .NET event
The event does not actually contain any information on the data received, it only indicates that there is data available. Reading the data is done inside the lambda expression. Reading serial data will return a list of bytes. This list may contain a complete message or just a part of a message or even multiple messages. To handle this, I want the observable to be an
IObservable<byte>, i.e., it will produce a raw stream of bytes without any indication of where a message begins or ends. This is done through the extension method
public static IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, IEnumerable<TResult>> selector) that is used to flatten the sequence returned by the lambda.
So I now have a stream of bytes. I want these bytes to be chunked into messages. For my particular protocol, messages are separated by a special byte. Separation can be done in two ways:
Here, a new observable is created using
Observable.Create(). This observable subscribes to the byte stream, collects the data in a local collection and fires
OnNext() whenever a message delimiter is encountered.
This version uses the
Scan() operator to achieve the same thing. The output is an
IObservable<IEnumerable<byte>> that fires an
IEnumerable<byte> for every new message.
This code worked well up until the point I started attaching multiple observers to the message stream, one to process the messages and one to just dump received messages to a debug console. What happened then was that the code in the first code sample was called multiple times: once for each subscriber. This meant that each chunk of serial data was only received by one subscriber, not all subscribers. There are two possible solutions to this: Either introduce a
Subject<IEnumerable<byte>> subscribing to
serialPortSource and have consumers subscribe to the subject or use the
Publish() operator that does the work for you.
Creating a new observable that produces deserialized messages from the observable producing lists of bytes is now trivial using a simple
What remains is the question of how to use the received data in a typical workflow of sending out a message and receiving a response in return. Here is an example:
This example uses the
Replay() operator. Replay will capture all events from the observable that are fired after the call to
Connect(). After calling
Connect() the call is sent to the device at the other end of the serial connection. The second
await filters the incoming messages for the desired message (even using a filter criterion that was not known before the request was sent), adds a timeout, uses
FirstAsync() to return an observable that only returns the first element followed by
OnCompleted(), and waits for that
Replay() is capturing all messages, the following
await call on the observable should consider all answers from the target, whether they are received before or after the second call to
3 replies on “Serial communication using Reactive Extensions”
I’ve only been able to get the fragments you provided above to something around half way working.
For the communication I’m having with the serial port, I can have the following two main scenarios:
1. Send a single command – Receive a corresponding response.
2. Send a single command – Receive n number of asynchronous ‘interim’ messages, then the corresponding response.
Obviously scenario 2 looked like a good fit for RX, but I don’t think that’s how you’ve done things. Is that correct? Could you give me any pointers.
How did you fire OnComplete to indicate that you’d received all of the data you were expecting?
Lastly, SerialPort.DataReceived always fires on a different thread from the one used to send the data – how did you work around this?
It seems like a good fit.
The observable produced by ToMessage() will spit out new messages infinitely. I used FirstAsync() to turn it into an observable that will only produce one message and then call OnCompleted(). You can use the Where() before that to only trigger on the final message you are expecting (as I did in the example). This is assuming you don’t need the information in the interim messages in this place.
await will return the message before OnCompleted(), which is what you want here. It will also marshal the response received on a different thread into the context of the awaiting method.
thanks for sharing your knowledge! Can you please provide an example project or more environment stuff of the project to understand better, how to integrate this interesting technique?
Would be very helpful, to have a simple example source code of a whole project…