Reactive .NET once again

Photo by Karan Mandre on Unsplash

Reactive programming, aka Rx:

is a library for composing asynchronous and event-based programs by using observable sequences.

There is a lot of to it, but despite being around in .NET for long time it still feels “weird” (as last week my colleague called it).

Why?

I think reason is simple Object Oriented is so deep inside every developer out there, that we assume everything else is “weird”.

Rx is popular eg. in Angular and JS world, but in .NET it always surprise to see Rx code.

Why and what

I wanted to share some example how Rx in .NET works using quasi real world example. I modified .NET tutorial on SignalR to use Rx. Obviously it is not real and finished product but still is able to demonstrate some aspects of Rx:

Rx in 1min

In Rx there are two fundamental items:

Observable — something one can observe.

Observer — someone who observe.

To not duplicate, most of information can be found at: ReactiveX.io

.NET System

Built in System namespace interfaces:

IObservable<T>

IObserver<T>

are not good starting point, while you need to implement everything yourself. All good stuff is actually in:

GitHub - dotnet/reactive: The Reactive Extensions for .NET

dotnet add package System.Reactive --version 6.0.0

Example

Whole logic in oneliner

Everything starts with stream which is supplied from some source. In this example it is data from chat, which contains objects:

First step is to limit how often user can send information:

.Sample(TimeSpan.FromSeconds(5))

SendTo is implemented using Do operator:

.Do(data => {<side effect>})

According to Reactive assumptions there should be no side effects, but in this case, sending data is an side effect, so Do is good fit.

There are additional steps which looks very similar to LINQ, because they are LINQ counterpart in Rx world

.GroupBy()
.Select()
.SelectMany()
.Scan()

CountData is an wrapper for Scan which is an accumulation operator:

 .Scan((x, y) => (x.Username, x.Count + y.Count));

DistinctUntilChanged()

is very interesting operator, while it encapsulate logic to get distinct elements only when they change so in case of following stream:

A A A B B B A A C C -> DistinctUntilChanged()-> A B A C

.Subscribe()

Subscribe without any observe is kind of hack, while it starts whole stream. And while all processing is done in pipeline, there is no need to supply any observer at all :)