This core level interface is available within the Base Class Library (BCL) of .NET 4.0 and is available for the older 3.5 as an add-on.

The use is pretty simple and the goal is to provide a standard way of handling the most basic features of any reactive message consumer.

As already seen in the previous chapter, reactive messages flow by a producer and a consumer and subscribe for some messages. The IObserver C# interface is available to construct message receivers that comply with the reactive programming layout by implementing the three main message-oriented events: a message received, an error received, and a task completed message.

The IObserver interface has the following sign and description:

    // Summary: 
    //     Provides a mechanism for receiving push-based notifications. 
    // 
    // Type parameters: 
    //   T: 
    //     The object that provides notification information.This type parameter is 
    //     contravariant. That...

The IObservable interface, the opposite of the IObserver interface, has the task of handling message production and the observer subscription. It routes right messages to the OnNext message handler and errors to the OnError message handler. As its life cycle ends, it acknowledges all the observers on the OnComplete message handler.

To create a valid reactive observable interface, we must write something that is not locking against user input or any other external system input data. The observable object acts as an infinite message generator, something like an infinite enumerable of messages; although in such cases, there is no enumeration.

Once a new message is available somehow, observer routes it to all the subscribers.

In the following example, we will try creating a console application to ask the user for an integer number and then route such a number to all the subscribers. Otherwise, if the given input is not a number, an error will be routed to all the subscribers...

What will happen if we want to stop a single observer from receiving messages from the observable event source? If we change the program Main from the preceding example to the following one, we could experience a wrong observer life cycle design. Here's the code:

//this is the message observable responsible of producing messages 
using (var observer = new ConsoleIntegerProducer()) 
//those are the message observer that consume messages 
using (var consumer1 = observer.Subscribe(new IntegerConsumer(2))) 
using (var consumer2 = observer.Subscribe(new IntegerConsumer(3))) 
{ 
    using (var consumer3 = observer.Subscribe(new IntegerConsumer(5))) 
    { 
        //internal lifecycle 
    } 
 
    observer.Wait(); 
} 
 
Console.WriteLine("END"); 
Console.ReadLine(); 

Here is the result in the output console:

The third observer unable to catch value messages

By using the using construct method...

Sourcing events is the ability to obtain from a particular source where few useful events are usable in reactive programming.

As already pointed out in the previous chapter, reactive programming is all about event message handling. Any event is a specific occurrence of some kind of handleable behavior of users or external systems. We can actually program event reactions in the most pleasant and productive way for reaching our software goals.

In the following example, we will see how to react to CLR events. In this specific case, we will handle filesystem events by using events from the System.IO.FileSystemWatcher class that gives us the ability to react to the filesystem's file changes without the need of making useless and resource-consuming polling queries against the file system status.

Here's the observer and observable implementation:

public sealed class NewFileSavedMessagePublisher...

As said in the previous section, it is time to alter message flow.

The observable interface has the task of producing messages, while conversely observer consumes such messages. To create a message filter, we need to create an object that is both a publisher and a subscriber together.

The implementation must take into consideration the filtering need and the message routing to underlying observers that subscribe to the filter observable object instead of the main one.

Here's an implementation of the filter:

/// <summary> 
/// The filtering observable/observer 
/// </summary> 
public sealed class StringMessageFilter : IObservable<string>, IObserver<string>, IDisposable 
{ 
    private readonly string filter; 
    public StringMessageFilter(string filter) 
    { 
        this.filter = filter; 
    } 
 
    //the observer collection 
    private readonly List<IObserver<string>> observerList...