Getting Started with Reactive Extensions

In today’s section, we will talk about Rx library. Basically, Rx is a library for composing asynchronous and event based programs using observable collections. This is very useful in cases wherein you are pulling data asynchronously from different sources and then manipulating the same and finally printing the result. In these kind of scenarios, you have to write lots of glue code and of-course, these codes will be error prone as one of the sources if just throws an error, then what will happen?

This way, you really need to do lots of stuff for the things working fine. Hence, Rx is the answer to this kind of situation, which keeps the thing simple yet lightweight. Rx also uses LINQ query on the observable collections.

But, it would be nice to talk a little about collections before starting Rx. IEnumerables are one of the most widely used Pull Based collection which is synchronous in nature. Below is the sample snippet for the same.

interface IEnumerable<out T>
    {
        IEnumerator<T> GetEnumerator();
    }

    interface IEnumerator<out T>:IDisposable
    {
        bool moveNext();
        T currennt { get; }
        void Reset();
    }

And, let us suppose due to any reason, your datasource went down for some time, then what will happen? It will keep waiting until the database comes online. And, you will land in an embarrassing situation which is something like shown below:

However, you can convert the same Pull based interface to Push based interface. Below is the sample for the same.

//Observables:- Push based 
    interface IObservable<out T>
    {
        IDisposable subscribe(IObserver<T> observer);
    }

    interface IObserver<in T>
    {
        void onNext(T value);
        void onError(Exception ex);
        void onCompleted();
    }

Very complete, precise and stable. We will see a few examples around the same in a moment. Below in the sample console app, you can see that IObservable and IObserver are available by default in .NET 4.0 and higher.

Now, let me go ahead and install Rx extension from nuget.

On successful installation, you can verify the assemblies.

Now, these are three phases of getting observables.

Let me explain the same with a simple demo. Now, as you can see in the below screen shot, as soon as I started creating observable, it started giving a bunch of overloads which I can make use of.

using System;
using System.Reactive.Linq;

namespace ReactiveExtensions
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            IObservable<string> obj = Observable.Generate(
                0, //Sets the initial value like for loop
                _ => true,               //Don't stop till I say so, infinite loop
                i => i + 1,              //Increment the counter by 1 everytime
                i => new string('#', i), //Append #
                i => TimeSelector(i));   //delegated this to private method which just calculates time

            //Subscribe here
            using (obj.Subscribe(Console.WriteLine))
            {
                Console.WriteLine("Press any key to exit!!!");
                Console.ReadLine();
            }
        }

        //Returns TimeSelector
        private static TimeSpan TimeSelector(int i)
        {
            return TimeSpan.FromSeconds(i);
        }
    }
}

In the above snippet, I have mentioned comment on each line, what it is doing. The best thing with this is its asynchronous nature which you can see below in the output.

Here, my program is executing on one thread and I am also typing on there on the same console, which means UI is not blocked. It's free for any activity. I hope you liked this small example around Observables. We will delve more in coming sections. Till then, stay tuned and happy coding!

CodeProject