How to code simple meta-stream with RxJS

2
votes

Background

It is the first time I am trying reactive programming. I have a stream that receives buffers of data from time to time.

When a buffer starts with 02 it means the start of a message, and when it ends with 03 it means the end of the message.

Example:

- - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - - - >

Objective

My objective is to fire an event every time I detect a message is completed, with who whole message as an argument.

Research

After reading several tutorials on reactive programming and reading RxJS, I know I want to create a hot observable.

With my research, I believe I want to create a "meta-observable" or a "meta-stream", i.e., an observable of observables or a stream of streams (confusing as this may be ... ).

My plan is the following:

1 - create an observable for all buffer messages that come in 2 - subscribe to step 1, and create an observable that detects the beginning of a message ( 02 ) and the end of a message ( 03 ) 3 - subscribe something to step 2 that will take care of the data.

So, my diagram would look like the following:

- 05 34 43 31 - - - 12 23 43 - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - - - > 
                             - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - |

Questions

1 - Since I am receiving this data via a socket in node.js:

socket = net.createConnection( { host: "localhost", port: 8080}, () => {
    socket.on( "data", console.log );
} );

I don't think I can use the EventEmmiter bridge that rxjs offers. I would have to use Observer.create, correct ?

2 - Even if I solve problem 1, I have no idea whatsoever on how I would create a stream of streams for this effect. Could someone post a code snippet as an example?

1
javascriptnode.jsstreamrxjsobservable

1 Answers

1
votes

You are on the right track. However, you want to create an observable so you probably want Observable.create and not Observer.create. It is very confusing because Observable.create needs a function which takes in an Observer. That function can then emit items into that Observer.

You are also keen to recognize that you want a hot observable. However, Observable.create is going to give you a hot observable anyways. You will still want to use some kind of publish to avoid multiple subscriptions. We will do that with share. As for the buffering, you can probably use buffer. At the end of the day I think it would be something like...

  • A - An observable created with Observable.create wrapped around your socket callbacks.
  • B - Wrap A with share
  • C - flatMap B to go from a stream of buffers to a stream of items
  • D - Watch C and use filter/where to only emit on 03.
  • E - buffer C using D as your closing selector

Then expose E as the public API of your service. E will emit the entire message any time a message arrives.

*Note, this method assumes you get messages back-to-back and thus we don't bother with the 02 signal as we know the first byte after the end of a message must be the start of the next message. If this is not the case you will want to handle that better.

** You can share after the flatMap. This will be slightly more performant. You cannot share after the buffer.

*** On a re-read of your question I noticed you want a stream of streams and not a stream of arrays. To get this result you can take the output of buffer and run it through flatMap using Observable.of This will give you hot stream of cold streams.