Friday, December 23, 2011

IDisposable implementations in Reactive Extensions

System.Reactive.Disposable namespace contains several implementations of the IDisposable interface in conjunction with RX.
public interface IDisposable
{
    void Dispose();
}
The main reason for the existence of these classes is that they can be used in your implementations of the observable's Subscribe method which returns an IDisposable.
public interface IObservable<out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}
For example you can return an Empty disposable which actually does nothing:
var source = Observable.Create<int>(
    observer => {
        observer.OnNext(1);
        return Disposable.Empty;
    }
);
Or you can use the static Disposable.Create method and provide it with an action which will be called during disposal:
var source = Observable.Create<int>(
    observer => {
        observer.OnNext(1);
        return Disposable.Create(() => "Dispose".Dump());
    }
);

using (source.Subscribe(v => v.Dump()))
{
}
================
1
Dispose
================
In the following example we will return instead of an IDisposable an action which will be wrapped to an IDisposable:
var source = Observable.Create<int>(
    observer => {
        observer.OnNext(1);
        return () => "Dispose".Dump();
    }
);
A BooleanDisposable will change it's IsDisposed property to false after disposal:
var booleanDisposable = new BooleanDisposable();
if (!booleanDisposable.IsDisposed)
{
    booleanDisposable.Dispose();
}
The SingleAssignmentDisposable's Disposable property can be set only ones as the name suggest, for the second attempt it will raise an InvalidOperationException with the message "Disposable has already been assigned.". You can use it in two different ways. You can first assign your IDisposable to the Disposable property and then dispose the SingleAssignmentDisposable:
var sad = new SingleAssignmentDisposable();

sad.Disposable = 
    Disposable.Create(() => "Single disposed".Dump()); 
    
sad.Dispose();
================
Single disposed
================
Or you can first dispose the SingleAssignmentDisposable and then assign your IDisposable to it. In this case your IDisposable will be automatically disposed during the assignment to the Disposable property:
var sad = new SingleAssignmentDisposable();

sad.Dispose();

sad.Disposable = 
    Disposable.Create(() => "Single disposed".Dump()); 
================
Single disposed
================
SerialDisposable allows you to set multiple times the Disposable property. In that case the current Disposable will be disposed and then change to the new one.
var serialDisposable = new SerialDisposable();
serialDisposable.Disposable = Disposable.Create(() => "Disposed 1".Dump());
"Set a new disposable.".Dump();
serialDisposable.Disposable = Disposable.Create(() => "Disposed 2".Dump());
"Call serial disposable's dispose.".Dump();
serialDisposable.Dispose();
================
Set a new disposable.
Disposed 1
Call serial disposable dispose.
Disposed 2
================
In case that you want to dispose multiple disposables at once you can use the CompositeDisposable class which implements the ICollection<T> interface:
IDisposable disposable1 = 
            Observable.Return(1)
            .Subscribe(Console.WriteLine);
                            
IDisposable disposable2 = 
            Observable.Return(2)
            .Subscribe(Console.WriteLine);
                            
using(new CompositeDisposable(disposable1, disposable2))
{
};
================
1
2
================
or
IDisposable disposable1 = Disposable.Create(() => "Disposed 1".Dump());
IDisposable disposable2 = Disposable.Create(() => "Disposed 2".Dump());
IDisposable disposable3 = Disposable.Create(() => "Disposed 3".Dump());
var compositeDisposable = new CompositeDisposable()
    {
        disposable1,
        disposable2
    };
compositeDisposable.Add(disposable3);
compositeDisposable.Dispose();
================
Disposed 1
Disposed 2
Disposed 3
================
The final example will use a CancellationDisposable which is disposed when you call the Cancel method on the injected CancellationTokenSource:
var cts = new CancellationTokenSource();

var cd = new CancellationDisposable(cts);

cts.Cancel();

cd.IsDisposed.Dump();
================
True
================
There are also other implementations of the IDisposable interface in this namespace such as MultipleAssignmentDisposable, SchedulerDisposable and so on, but I'm not going to cover these in this topic.

Thursday, December 22, 2011

The Aggregate and Scan RX extension methods.

This post is about how to sum all numbers in a range from an observable sequence with the help of the Aggregate method. Then the FirstOrDefault is used to convert the resulting IObservable to an int.
var sumOfNumbers = Observable.Range(1, 10)
                   .Aggregate(0, (x, y) => x + y)
                   .FirstOrDefault();

Console.WriteLine("Sum of numbers from 1 to 10 is {0}", sumOfNumbers);
==========
Sum of numbers from 1 to 10 is 55
==========
Or you can you the Scan method to display the intermediate results with the help of the Do method and finally store the result with the LastOrDefault.
int sumOfNumbers = Observable.Range(1, 10)
.Scan(0, (x, y) => x + y)
.Do(Console.WriteLine)
.LastOrDefault();

Console.WriteLine("Sum of numbers from 1 to 10 is {0}", sumOfNumbers);
==========
1
3
6
10
15
21
28
36
45
55
Sum of numbers from 1 to 10 is 55
==========
You can use the Aggregate method to fill a collection with received values from the observable:
IList<int> collection = 
    Observable.Range(1, 5)
              .Aggregate(new List<int>(), (list, value) => {
                               list.Add(value);
                            return list;
                         })
              .FirstOrDefault();

foreach (var element in collection)
{
    Console.WriteLine(element);
}
==========
1
2
3
4
5
==========

Advancing the time with TestScheduler in Reactive Extensions.

In this post I'm going to show you how you can programmatically advance the time with the test scheduler so you don't have to wait 5 days for an event in your test environment. First we will prepare a testable cold observable. It will schedule us 4 OnNext notifications every 100 virtual ticks and finally after 500 ticks an OnCompleted notification kind.
var testScheduler = new TestScheduler();            
    
var records = new Recorded<Notification<int>>[] {
    ReactiveTest.OnNext(100, 1),
    ReactiveTest.OnNext(200, 2),
    ReactiveTest.OnNext(300, 3),
    ReactiveTest.OnNext(400, 4),
    ReactiveTest.OnCompleted<int>(500)
};

ITestableObserver<int> testableObserver = 
    testScheduler.CreateObserver<int>();
    
ITestableObservable<int> testableObservable = 
    testScheduler.CreateColdObservable(records);

IDisposable d = testableObservable.Subscribe(testableObserver)
Now you can use AdvanceBy which advances the test schedulers Clock by this relative time (testScheduler.Clock+time) and runs all the scheduled work within it. You can also use the AdvanceTo method to run all scheduled items up to this absolute time and again advance the schedulers Clock to this point. The Start method runs the remaining work from the actual Clock time.
int subscrCount = testableObservable.Subscriptions.Count();
Console.WriteLine("Number of subscriptions to the test observable: {0}"
                  , subscrCount);

testScheduler.AdvanceBy(200);
Console.WriteLine("Messages sent({0}) until {1}"
                  , testableObserver.Messages.Count 
                  , testScheduler.Clock);

testScheduler.AdvanceTo(400);
Console.WriteLine("Messages sent({0}) until {1}"
                  , testableObserver.Messages.Count 
                  , testScheduler.Clock);

testScheduler.Start();
Console.WriteLine("Messages sent({0}) until {1}"
                  , testableObserver.Messages.Count 
                  , testScheduler.Clock);
                      
foreach (var message in testableObserver.Messages)
{
    Console.WriteLine("Value {0} at {1}", message.Value, message.Time);
}
At the end you can check that all scheduled items were run and sent to the mock observer.
================
Number of subscriptions to the test observable: 1
Messages sent(2) until 200
Messages sent(4) until 400
Messages sent(5) until 500
Value OnNext(1) at 100
Value OnNext(2) at 200
Value OnNext(3) at 300
Value OnNext(4) at 400
Value OnCompleted() at 500

================

Wednesday, December 21, 2011

How to use the Notification class in Reactive Extensions

In this topic I will show you how you can use the main three Notification types in RX. They are OnNext, OnError and OnCompleted and you will mainly use them for testing purposes.
public enum NotificationKind
{
    OnNext,
    OnError,
    OnCompleted
}
Let's first create a notification of NotificationKind.OnNext and explore it's properties and methods:
// Create a NotificationKind.OnNext with new Exception
Notification<int> notification = Notification.CreateOnNext(1);

// explore instance properties
Console.WriteLine("Value: {0}" , notification.Value);
Console.WriteLine("HasValue: {0}", notification.HasValue);
Console.WriteLine("Kind: {0}", notification.Kind);
// user friendly ToString
Console.WriteLine(notification.ToString());

var observer = Observer.Create<int>(Console.WriteLine);

// calls the observer's OnNext method 
// with the Value as input parameter
// observer.OnNext(notification.Value);
notification.Accept(observer);

// do the same thing manually
var source = notification.ToObservable();
source.Subscribe(observer.AsObserver());
The code is describe with comments and the output is as follows:
================
Value: 1
HasValue: True
Kind: OnNext
OnNext(1)
1
1
================
A NotificationKind.OnError notification can be created as follows:
// Create a NotificationKind.OnErrorwith value 1
Notification<int> notification = Notification
    .CreateOnError<int>(new Exception("error"));

// explore instance properties
Console.WriteLine("Exception: {0}" , notification.Exception);
Console.WriteLine("HasValue: {0}", notification.HasValue);
Console.WriteLine("Kind: {0}", notification.Kind);
// user friendly ToString
Console.WriteLine(notification.ToString());

var observer = Observer.Create<int>(
        Console.WriteLine,
        error => Console.WriteLine(error.Message)
        );

// calls the observer's OnError method 
// with the Exception as input parameter
// observer.OnError(notification.Exception);
notification.Accept(observer.AsObserver());
================
Exception: System.Exception: error
HasValue: False
Kind: OnError
OnError(System.Exception)
error
================
A NotificationKind.OnCompleted notification can be created as follows:
// Create a NotificationKind.OnCompleted
Notification<int> notification = Notification
    .CreateOnCompleted<int>();

// explore instance properties
Console.WriteLine("HasValue: {0}", notification.HasValue);
Console.WriteLine("Kind: {0}", notification.Kind);
// user friendly ToString
Console.WriteLine(notification.ToString());

var observer = Observer.Create<int>(
        Console.WriteLine,
        () => Console.WriteLine("Completed")
        );

// calls the observer's OnCompleted method 
// observer.OnCompleted();
notification.Accept(observer.AsObserver());
================
HasValue: False
Kind: OnCompleted
OnCompleted()
Completed
================
You can convert an observable to a collection of notifications with the Materialize and ToEnumerable methods. In this case we Concat to observables to produce one sequence and produce an error after the Range ends.
var notifications = Observable.Range(1, 2)
.Concat(Observable.Throw<int>(new InvalidOperationException()))
.Materialize()
.ToEnumerable();

foreach (var notification in notifications)
{
    notification.ToString().Dump();
}
================
OnNext(1)
OnNext(2)
OnError(System.InvalidOperationException)
================
You can achieve the same result with creating an observer from a notification callback. Observer will convert every notification to an appropriate NotificationKind and pass it to the provided callback method.
var notifications = Observable.Range(1, 2)
.Concat(Observable.Throw<int>(new InvalidOperationException()));

Action<Notification<int>> action = notification => {
    notification.ToString().Dump();
};

var observer = action.ToObserver();

notifications.Subscribe(observer);
================
OnNext(1)
OnNext(2)
OnError(System.InvalidOperationException)
================
And finally you can convert back an observer to an Action delegate with the ToNotifier and call it as a method:
var notifier = Observer.Create<int>(
                Console.WriteLine,
                () => "Completed".Dump())
                       .ToNotifier();
                       
notifier(Notification.CreateOnNext(2));    
notifier(Notification.CreateOnCompleted<int>());
================
2
Completed
================

Preparing for unit testing with testable observables

In the following example we are going to explore the basic functionality provided to help us with unit testing of RX observables. First we have to create an instance of the class TestScheduler which implements ISchudeler interface. This will enable us to schedule some work for the future. Then we call the CreateObserver method on it. It will return us an ITestableObserver. There is a property called Messages which is an addition to the classical IObservable interface. It will contain the actions sent to the previously created MockObserver after calling the Start method on the testScheduler. Next we will schedule two notifications. There are three possibilities in the NotificationKind enumeration (OnNext, OnError and OnCompleted). You can create these with Notification.CreateOnNext, Notification.CreateOnError and Notification.CreateOnCompleted static methods. These actions are scheduled to run after 100 and 200 virtual ticks. Schedule contains a callback which will be executed when the virtual time will be advanced to this point. In this case we are calling the notification's Accept method which invokes the mock observer's method corresponding to the notification. In our case it's OnNext for the our first scheduled item and for the second one it's OnCompleted. After we have subscribed to the testable observable we can start the scheduler. This will actually run our scheduled items an send them to the mock observer. This observer will record the received notifications to it's Messages collection.
    var testScheduler = new TestScheduler();
    
    var testableObserver = testScheduler.CreateObserver<int>();
    
    testScheduler.ScheduleAbsolute(Notification.CreateOnNext<int>(2), 100L, (scheduler, state) => {
        state.Accept(testableObserver);
        return Disposable.Empty;
    });
    
    testScheduler.ScheduleAbsolute(Notification.CreateOnCompleted<int>(), 200L, (scheduler, state) => {
        state.Accept(testableObserver);
        return Disposable.Empty;
    });
    
    testScheduler.Start();
    
    foreach (var message in testableObserver.Messages)
    {
        Console.WriteLine("Value {0} at {1}", message.Value, message.Time);
    }
Finally we will send the recorded messages to the output:
================
Value OnNext(2) at 100
Value OnCompleted() at 200
================
We can achieve the same result with the use of CreateColdObservable method. It receives as an input a collection of notification records. Record's Value has to be a Notification which are in this case created with the help of static methods ReactiveTest.OnNext and ReactiveTest.OnCompleted (there also exists a ReactiveTest.OnError). In case of OnNext the first parameter is the relative schedule time and the value of the notification.
    var testScheduler = new TestScheduler();            
    
    var records = new Recorded<Notification<int>>[] {
        ReactiveTest.OnNext(100, 1),
        ReactiveTest.OnCompleted<int>(200)
    };
    
    var testableObserver = testScheduler.CreateObserver<int>();
    var testableObservable = testScheduler.CreateColdObservable(records);
    
    testableObservable.Subscribe(testableObserver);
    
    testScheduler.Start();
    
    foreach (var message in testableObserver.Messages)
    {
        Console.WriteLine("Value {0} at {1}", message.Value, message.Time);
    }
Yet another way how to achieve the same thing is to provide a create observable method and provide creation, subscription and disposal time to the Start method of the test scheduler. In this case the notifications are scheduled relatively to the subscription time. So now we don't have to create a mock observer manually and subscribe to the test observable, it will be managed by the scheduler.
    var testScheduler = new TestScheduler();            
    
    var records = new Recorded<Notification<int>>[] {
        ReactiveTest.OnNext(100, 1),
        ReactiveTest.OnCompleted<int>(200)
    };
    
    var testableObserver = testScheduler.Start(
        () => testScheduler.CreateColdObservable(records),
        0, 50, 300
    );
    
    foreach (var message in testableObserver.Messages)
    {
        Console.WriteLine("Value {0} at {1}", message.Value, message.Time);
    }
The results are:
================
Value OnNext(1) at 150
Value OnCompleted() at 250
================
That's it for now. Next we will explore subjects and the difference between cold and hot observables.

Tuesday, December 20, 2011

Creating basic observable collections with RX

In the previous article we created some observables which produced only single notification. This time we will focus on more collection like notifications. We will start with the Repeat method which repeats the first argument n-times based on the second argument:
    var source = Observable.Repeat(1, 3);
    
    IDisposable d = source.Subscribe(
                        value => value.Dump(),
                        error => error.Message.Dump(),
                        () => "Completed".Dump()
    );
The above mentioned code snippet uses an extension method defined in the System.ObservableExtensions static class. It provides you several overloaded versions of ObservableExtensions.Subscribe in case you don't want to create your own observer, but only provide the action methods.
If you are using Visual Studio you have to use the subscribe method with System.Console instead of the Dump method.
    IDisposable d = source.Subscribe(
                    Console.WriteLine,
                    error => Console.WriteLine(error),
                    () => Console.WriteLine("Completed")
    );
The output from LINQPad is as follows:
================
1
1
1
Completed
================
Now we are going to create a Range of notifications. The first argument is the starting value and the second is how many values we will receive together:
    var source = Observable.Range(1, 3);
    
    IDisposable d = source.Subscribe(observer);
================
1
2
3
Completed
================
The last example is the most generic one. The Generate extension method needs these five arguments: an initial state, a condition when to stop, an iterator how to get the next value and a result selector to shape the result:
    var source = Observable.Generate(
                    1, 
                    value => value <= 3,
                    value => value + 1,
                    value => value
                    );                    
    
    IDisposable d = source.Subscribe(observer);
The initial state is set to 1. In every "iteration" the state is incremented by one while the condition is met. The resulting shape of the values won't be changed in this case. This time we have created the exactly same functionality as the Repeat method provides:
================
1
2
3
Completed
================
Next time we will create a testable observable.

How to create an observable with RX

IObservable<T> is defined in the System namespace and defines a provider for push based notifications. It looks like this:
public interface IObservable<out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}
An observable has to implement this interface. Than you can call the Subscribe function on it via which you can inject an observer to obtain notifications from it. The function returns an IDisposable which defines the Dispose method.
You can create a new observable in a lot of different ways. I'm going to show you the basic methods of the Observable static class from System.Reactive.Linq namespace for creating observables. This class also contains a lot of LINQ query operators implemented as extension methods such as Where, Take, etc.
All examples in this article will use the following observer instance:
    var observer = Observer.Create<int>(
                        value => value.Dump(),
                        error => error.Message.Dump(),
                        () => "Completed".Dump()
                        );
The easiest way how to create an observable which notifies us about one value is the Next extension method and subscribe to it:
    var source = Observable.Return(1);
    
    source.Subscribe(observer);
The results are:
================
1
Completed
================  
If you would like to simulate an error condition you can use the Throw extension method. In this case the OnCompleted is not called:
    var source = Observable.Throw<int>(new Exception("Error has occurred."));
    
    source.Subscribe(observer);
The result is:
================
Error has occurred.
================
An empty observable can be created with the Empty extension method. In this case only the OnCompleted method of the observer is called:
    var source = Observable.Empty<int>();
    
    source.Subscribe(observer);
The result is:
================
Completed
================
Ok, so now it's time to create your first own IObservable implementation:
internal class DefaultObservable<T> : IObservable<T>
{
    public IDisposable Subscribe(IObserver<T> observer)
    {
        if (observer == null)
        {
            throw new ArgumentNullException("observer");
        }
        
        observer.OnNext(default(T));
        observer.OnCompleted();
        
        return Disposable.Empty;
    }
}
Now you can create an instance of the DefaultObservable class which notifies our observer with the default value of type T.
    var source = new DefaultObservable<int>();
    IDisposable d = source.Subscribe(observer);
After you subscribe to it you will receive the following output:
================
1
Completed
================
In the next topic we will focus on creating observables which are producing multiple notifications.