Getting Started With Reactive Extensions

In Today's section, we will talk about the Rx library. Basically, Rx is a library for composing asynchronous and event-based programs using observable collections. This is very useful in the case wherein you are pulling data asynchronously from different sources and then manipulating the same and finally printing the result. In these kinds of scenarios, you have to write lots of glue code and of course, these codes will be error-prone. Let's say one of the sources 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 an answer to this kind of situation, which keeps the thing simple yet lightweight. Rx also uses a LINQ query on the observable collections.

 

But, it would be nice to talk a little about collections before starting Rx. IEnumerables is one of the most widely used Pull Based collection which is synchronous in nature. The following 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 something like shown below.

 

 

And, you will end up doing something like this!

 

 

However, you can convert the same Pull based interface to Push based interface. The following 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. The following is 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 the 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 following screenshot, 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 comments on each line that what it is doing. Best thing with this is its asynchronous nature which you can see in the following output:

 

 

Here, my program is executing on one thread and I am also typing there on the same console, which means UI is not blocked. It's free for any activity. I hope, you would have liked this small example around Observables.

 

We will deliver more in the coming sections. Till then, stay tuned! Happy Coding!