Write clean code with Reactive Extensions

If you have a process that can run for a long time and return several intermediate results over time, then the .NET Framework Reactive Extensions will allow you to simplify the code and better manage it.



In most cases, you simply call the method and get the result at the output. But some processes are different. For example, the method can be performed for a long period of time. Or, worse, the method not only runs for a long time, but also irregularly returns some intermediate results during execution. Of course, including for this, there are events in the .NET Framework, with the help of events a single object can call the method of the second object, passing some information, at that moment in time when it is needed.
')
But there is a solution to this problem better than using events - Reactive Extensions. If you have a process that works for a long time and returns intermediate results from time to time, then Jet Extensions will help you handle such results whenever they come. The code from using Reaction Extensions instead of events not only becomes simpler, but you also get richer functionality (for example, you can use LINQ to filter out any unnecessary data).

The documentation for Reactive Extensions is described as a way to handle data flow. When working with Extensions, it is not difficult to imagine any process that iterates over data in a collection, periodically searches for something interesting and sends to the application what it found - this causes the application to react (to react) , hence the name “Reactive” (reactive) ). In our example, we will assume that we want to do something while our application performs a number of changes in the sales order. To do this, we need to write the StatusChanged method and call it every time an order changes in the application. Or, we can add the StatusChanged event to our SalesOrder class and call it every time the status changes — we can simply connect our code to this event.

Solving this problem with the help of Jet Extensions, in my opinion, is not only simpler than the ones proposed above, but also — thanks for their integration with LINQ — more flexible. Definitely, when it comes to connecting or disconnecting a source of information, a solution using Reactive Extensions is simpler than similar code using events.

Installation of Reactive Extensions

First you need to connect the Reactive Extensions to the project using NuGet. There are several packages of Reactive Extensions, including implementations for JavaScript, Android, and handling LINQ requests for web services. To find the package you need, just run the search for the phrase “Reactive Extensions” and add the Main Library package to the project.

The second step is to decide which data we want to return with each change in the order. Together with the current order status, it makes sense to also return the order ID. Let's make for this a class with the necessary properties:

public class StatusChange { public int OrderId { get; set; } public string OrderStatus { get; set; } } 

In the next step, we will define the Reactive Extensions Subject , which will work with this type, initialize it when our application starts and subscribe, using the Subscribe method, to calls of the Subject's methods to be aware of the changes:

 ISubject<StatusChange> statChange = new Subject<StatusChange>(); statChange.Subscribe(sc => MessageBox.Show(sc.OrderStatus)); 

In the Subscribe method, we pass a lambda expression indicating what we want to do when there are any changes in the order. In our example, we simply show the value of the OrderStatus property of our class.

Now everywhere in our application, when we change the status of an order, we will create an instance of the StatusChange class, fill in its properties and call the OnNext method of the created Subject. A typical notification code for initial order creation might look like this:

 statChange.OnNext(new StatusChange() { OrderId = 1, OrderStatus = "New" }); 

Each time the OnNext method is called, a message will be displayed with the value of the StatusChange.OrderStatus property, which we defined in the lambda expression.

Solution expansion

Of course, in a real project, handling status changes may require more than one line of code, even more than we want to fit in a lambda expression. Instead of lambda expressions, we can always pass a pointer to another method to the Subscribe method, for example:

 statChange.Subscribe(StatusChanged); 

The method may not take any parameters, but if it accepts as a parameter an object of the same type with which the Subject is associated, then the object specified in the OnNext call will be passed to it . Such a method, for example, can simply output each new state to the console:

 public static void StatusChanged(StatusChange status) { Console.WriteLine(status.OrderStatus); } 

If we need to perform several different methods when changing status, then there is no need to pack them into one method. Instead, we can call Subscribe our Subject several times, passing the necessary methods in turn.

 statChange.Subscribe(StatusChanged); statChange.Subscribe(StatusAudit); 

With this approach, we have a weak linking of the process that makes changes to the order (our application), with processes that react to these changes (methods StatusChanged , StatusAudit ). Only one thing links these processes together - the definition of the StatusChange class, which we can safely extend with additional properties without breaking other processes. Until now, it almost did not differ from the use of events, except that the code needed to write a little less.

But the use of Reactive Extensions not only simplifies our code, it allows us to rise above events. To begin with, let's say we don’t want to handle every change in order status. For example, we want to catch only those orders whose state has been changed to “In progress”. We may use LINQ to clarify what results we want to get from the Subject. Before trying this out, we need to add the System.Reactive.LINQ namespace to our code.

After connecting this namespace, we will see that we can write LINQ expressions or use LINQ extension methods to choose which results we want to receive and process. All three examples below show that our method will be called only for changes with the “Processing” status:

 statChange.Where(c => c.OrderStatus == "Processing").Subscribe(ReportStatusChange); var scs = from sc in statChange where sc.OrderStatus == "Processing" select sc; scs.Subscribe(ReportStatusChange); var sub = (from sc in statChange where sc.OrderStatus == "Processing" select sc).Subscribe(ReportStatusChange); 

We may also want to have special handlers when the status changes to erroneous or when placing an order. We could well define this by order status, but the Reactive Extensions provide a better solution: the OnError and OnCompleted methods of the Subject. When we call the Subscribe Subject method, we can pass parameters on the methods (or lambda expressions) that must be executed when the Subject OnError and OnCompleted methods are called . In the example below, the names of the methods are changed to make the code more visual:

 statChange.Subscribe(OnNext, OnError, OnCompleted); 

The OnError method must accept an exception as a parameter, and the OnCompleted method must be without parameters. A typical example would be:

 public static void OnError(Exception ex) { Console.Error.WriteLine(ex.Message); } public static void OnCompleted() { Console.WriteLine("order processing completed"); } 

Now, in case something went wrong, our application should call the OnError method of the Subject. When calling this method, you need to pass an exception to the parameter, which contains information about the problem (in a real project there will be something better than the example below):

 statChange.OnError(new Exception("Something has gone horribly wrong!")); 

When an application finishes working with an order, it should call the OnCompleted method of the Subject. In addition to calling handlers for this state change, this method also instructs the Subject that he should not send any more notifications (by the way, another thing that cannot be done with events is to disable subscribers on the side of the event). Also, the Subject can free himself from all listeners by calling the Dispose method.

Notification Encapsulation

In our application, there is one problem - in every place where we change the status of the order, we must remember to call the Subject's OnNext method. It would be good to automate it. Ideally, we can hide this call inside the setter of the Status property of the order class. This eliminates both duplication of code and the ability to forget to call the OnNext method.

In the code listing below, the SalesOrder class contains a property of the type ISubject , which is initialized in the constructor by an instance of the Subject. Now, the Subject's OnNext method will be called wherever the Status property changes (additional code can be added to this class to also support the OnError and OnCompleted Subject methods):

 public class SalesOrder { string _status; public ISubject<StatusChange> StatChange { get; private set; } public int Id { get; set; } public string Status { get { return _status; } set { _status = value; var sc = new StatusChange() { OrderId = this.Id, OrderStatus = this.Status }; StatChange.OnNext(sc); } } public SalesOrder() { StatChange = new Subject<StatusChange>(); } }