Real-Time Stream Processing with Reactive-Extensions (RX)

History

0.jpg

The 1970s is considered to be the beginning of time. In 1991, the public internet was introduced. In 2004, the @ character was added to the morse code. In 2009, the Rx was released and in 2018, Rx is all the things.

Who explored Rx?

Eric Meijer

Eric is the explorer of Rx. He is a Microsoft developer and has a lot of skills.

He is a brilliant teacher, sharp minded. The fascinating thing about him is his colorful t-shirts.

Receipt for a great idea

Raven Tests

 

1.jpg

It contains a set of patterns and the last one of them is completely empty and the person solving need to look at the common features and analyze them in order to find a solution for the problem.

Programming manuals

Eric reads programming manuals for fun in any language he can find.

Reactive extensions

All these hobbies and fun activities led to the idea and creation of the Rx, Reactive Extensions.  

Collections

There is a missing link in the collections. There is pull-based and push-based collections. The pull based collections erase the collections, which are pulled out item by item. The push-based collection when it is difficult to decide when the next item will arrive. For pull based collections, we work with standard operators. This is a first class array functions.

The push-based collections part, there is a set of extra operators which help people to deal with all the asynchrony.

2.jpg

 

The missing link is between the push-based and extras. Eric realized that these two sets or these two collections were very similar. He implemented a standard set of operators into the push-based side. Now, it was convenient to use all the standard operators.

There is still one piece missing in order to interchange between the two collections, to switch or to implement these extra options back into the pull-based side.

3.jpg

Now, it is easy to switch between these two collections and treat them in the exact same way. This standardizes the whole interface working with these collections.

Morse code 101

Samuel F. B. Morse created telegraph and the morse code is named after him.

The morse code consists of dots and dashes and the combination of these dots and dashes represent the letters in the alphabet.

4.jpg

The concept can be simplified using a binary tree concept. Morse code is a binary tree. The left branch is a dot and the right branch is a dash.

5.jpg

The keyboard is required which has one key and it can be used to create long and short signals which go to the screen.

Parts of a message

It is important to know the parts of this messages. A sequence of dots and dashes. There are some very important pieces which are not visible here- the breaks.

6.jpg

There are short and long breaks. The short break is used to distinguish between the single signals and the long breaks to group the signals together.

Message with breaks

So the message with breaks will look like the one shown below.

7.jpg

Translating Morse code with Rx

The button will be clicked, a signal will be produced and when the button will be released the button will stop sending the message.

Create a stream of mors characters

There are two streams where one stream released the time stamps when the button is pushed on and the other one is when the button is released. Some operators can be used to combine the two streams and take only the last values of them.

8.jpg

So, there are timestamps and if they are subtracted, the duration of the signal can be calculated. Then this signal can be mapped into single characters and to morse characters.

Group characters after the break

The second part is grouping the characters, so, all the characters will be grouped and whenever a long break occurs, these values will be emitted in the group.

9.jpg

And then they can be glued together to a morse symbol.

Translate group to letters

10.jpg

In the last part, the groups are translated. Some of these groups are in the set of morse code.

Live coding

There are these marble diagrams.

11.png

And here are the models; time stamps, morse durations, morse characters, symbols, and letters.

12.png

The signals are produced by pushing the button. There is a subject in the service. A subject consists of an observable and an observer. The observer part is exposed to the outside.

So, the next, error and complete functions of this observable can be used. The subject is made private, as the whole subject cannot be exposed. The observable part is made public and the method is exposed.

13.png

Only the observable is reachable from the outside. To secure the next value, the next function is wrapped into another function, where some validations can be done.

14.png

The first stream is called morse character. The start and stop streams need to be combined and the values need to be grouped together. An array of these streams are created first.

The operator is called combine latest.

15.png

The first stream is called morse character. The start and stop streams need to be combined and the values need to be grouped together. An array of these streams are created first.

The operator is called combine latest.

16.png

A new stream, called breaks is created, which is just long breaks.

The buffer function grouped all the values and when the long break occurs, it emits every signal once in the set of an array.

A lot of signals are sent.

The long break is started, and when the signal is not sent, the long breaks are stopped. The arrays are displayed in the new stream.

All these characters are now needed to be glued together and get rid of the long breaks in this array.

The third step is to translate it, which is why a new stream is started, which is called, morse letter stream. Connect the previous stream to make sure everything is working.

Here, a function called, translate symbol to letter is used. Once, the button is pressed, a letter appears, which is, an E.

What if a signal is produced which is not a morse character?

An error is produced. Even if the process is continued, the stream is stationary.

There is an operator called, catch error. It receives the error and let emit another observable.

A stream is emitted. The purpose is to hide the error form the console and instead see the message on the letter screen.

Whenever a new symbol arrives, a new observable must be created, the process must be run and it is restarted over and over again. There is an operator for this, it is called switch map. It takes a value and returns another stream.

17.png

A stream if the symbol is created. The pipe operator is applied.

Now, a letter is received and even after the error, the stream continued to give other letters.

Now the custom pipes are created. This custom pipe is called safe translate. The pipe operator is used and the translation logic is inserted.

It is working, though it does not look very nice.

Now, the configurable custom operator is to be created. The pipe will be wrapped with another function which composes a variable inside the scope of this function. The variable is called error string. And below it, the same error string is applied.  Then, it can be configured on the outside.

It is still working.

A separate process is needed to be created, which drums automatically whenever morsing is stopped, which produces long breaks. Whenever the morsing is started again, the process needs to be stopped.

 

18.png

The console is empty because there was no subscription to these strings. Now, there is a steady stream if long breaks. In the morse code, the maximum no of breaks is 4.

The second part is to start this process whenever the morsing is stopped. And the process ned to be stopped whenever the morsing is started. A new stream is created.

The button is pushed, there is nothing on the screen. But, when it is released, the breaks start.

It needs to be stopped not only after 4 breaks but when the morsing is started again.

The stream of breaks needs to be merged into the morse character stream. If the long breaks are translated, it comes up empty. The empty strings need to be removed.

DEMO: https://biophoton.github.io/angular-morse-code-example/#/encode