var query = from method in typeof (Observable).GetMethods() from parameter in method.GetParameters() where typeof (IScheduler).IsAssignableFrom(parameter.ParameterType) group method by method.Name into m orderby m.Key select m.Key; query.Run(Console.WriteLine); /* BufferWithTime, Catch, Concat, Delay, Empty, Generate, GenerateWithTime, Interval Merge, ObserveOn, OnErrorResumeNext, Prune, Publish, Range, Repeat, Replay Retry, Return, Sample, Start, StartWith, Subscribe, SubscribeOn, Take, Throttle Throw, TimeInterval, Timeout, Timer, Timestamp, ToAsync, ToObservable */There are some methods that we already will be familiar with such as ObserveOn and SubscribeOn. Then there are others that will optionally take an IScheduler in one of the method overloads. TDD/TestFirst coders will want to opt for the overload that takes the IScheduler so that we can have some control over scheduling in our tests.
var interval = Observable
.Interval(TimeSpan.FromSeconds(1))
.Take(5);
If we to write a test that ensured that we received 5 values and they were each 1 second apart it would take 5 seconds. That would be no good, I want hundreds if not thousands of tests to run in 5 seconds. A more common time related example that would need tests is setting a timeout.var stream = Observable.Never<int>(); var exceptionThrown = false; stream.Timeout(TimeSpan.FromMinutes(1)) .Run( i => Console.WriteLine("This will never run."), ex => exceptionThrown = true); Assert.IsTrue(exceptionThrown);This test would take 1 minute to run. However if we did some test first code in this style, where we added the timeout after the test, running the test initially to fail (Red-Green-Refactor) would never complete. Hmmmm…..
var scheduler = new TestScheduler(); var wasExecuted = false; scheduler.Schedule(() => wasExecuted = true); Assert.IsFalse(wasExecuted); scheduler.RunTo(1); //execute 1 tick of queued actions Assert.IsTrue(wasExecuted);The TestScheduler type is actually an implementation of the abstract VirtualScheduler<TAbsolute, TRelative> where the TestSchuduler specifies long for both TAbsolute and TRelative. The net result is that the TestScheduler interface looks something like this
public class TestScheduler { public TestScheduler(); // Methods public long FromTimeSpan(TimeSpan timeSpan); public long Increment(long absolute, long relative); public void Run(); public void RunTo(long time); public IDisposable Schedule(Action action); public IDisposable Schedule(Action action, TimeSpan dueTime); public IDisposable Schedule(Action action, long dueTime); public void Sleep(long ticks); public DateTimeOffset ToDateTimeOffset(long absolute); // Properties public DateTimeOffset Now { get; } public long Ticks { get; } }We should already be familiar with what the Schedule methods would do, the other method of interest for this post are the Run() and RunTo(long) methods. Run() will just execute all the actions that have been scheduled. RunTo(long) will execute all the actions that have been scheduled up to the time specified by the long value which represents ticks in this case. Having a quick look into the implementations for Schedule, RunTo and Run give us the insight we need to really understand the virtual schedulers.
public IDisposable Schedule(Action action) { return this.Schedule(action, TimeSpan.Zero); } public IDisposable Schedule(Action action, TimeSpan dueTime) { return this.Schedule(action, this.FromTimeSpan(dueTime)); } public IDisposable Schedule(Action action, TRelative dueTime) { BooleanDisposable disposable = new BooleanDisposable(); TAbsolute local = this.Increment(this.Ticks, dueTime); Action action2 = delegate { if (!disposable.IsDisposed) { action(); } }; ScheduledItem<TAbsolute> item = new ScheduledItem<TAbsolute>(action2, local); this.queue.Enqueue(item); return disposable; } public void RunTo(TAbsolute time) { while ((this.queue.Count > 0) && (this.queue.Peek().DueTime.CompareTo(time) <= 0)) { ScheduledItem<TAbsolute> item = this.queue.Dequeue(); this.Ticks = item.DueTime; item.Action(); } } public void Run() { while (this.queue.Count > 0) { ScheduledItem<TAbsolute> item = this.queue.Dequeue(); this.Ticks = item.DueTime; item.Action(); } }Obviously this is the internal implementation and would be subject to change, however as the documentation currently is quite weak for the VirtualScheduler and TestScheduler I think this use of reflection is helpful. This example should clear up what happens when items are scheduled at the same time.
var scheduler = new TestScheduler(); long dueTime = 4L; scheduler.Schedule(() => Console.WriteLine("1"), dueTime); scheduler.Schedule(() => Console.WriteLine("2"), dueTime); scheduler.Schedule(() => Console.WriteLine("3"), dueTime+1); scheduler.Schedule(() => Console.WriteLine("4"), dueTime+1); Console.WriteLine("RunTo(dueTime)"); scheduler.RunTo(dueTime); Console.WriteLine("Run()"); scheduler.Run(); /* Output: RunTo(dueTime) 1 2 Run() 3 4 */
[TestMethod] public void Testing_with_test_scheduler() { var scheduler = new TestScheduler(); var interval = Observable .Interval(TimeSpan.FromSeconds(1), scheduler) .Take(5); bool isComplete = false; interval.Subscribe(Console.WriteLine, () => isComplete = true); scheduler.Run(); Assert.IsTrue(isComplete); //Executes in less than 0.01s "on my machine" }While this is mildly interesting, what I think is more important is how we would test a real piece of code. Imagine if you will a Presenter that subscribes to a stream of prices. As prices are published it adds them to a ViewModel. Assuming this is a WPF or Silverlight implementation we take the liberty of enforcing that the subscription be done on the ThreadPool and the observing is executed on the Dispatcher.
public class MyPresenter { ... public void Show(string symbol) { _myService.PriceStream(symbol) .SubscribeOn(Scheduler.ThreadPool) .ObserveOn(Scheduler.Dispatcher) .Subscribe(price=>_viewModel.Prices.Add(price)); } ... }While this snippet of code may do what we want it to do, it will be hard to test as it is accessing the schedulers via static properties. To help my testing, I have created my own interface that exposes the same IScheduler implementations that the Scheduler type does.
public interface ISchedulerService { IScheduler CurrentThread { get; } IScheduler Dispatcher { get; } IScheduler Immediate { get; } IScheduler NewThread { get; } IScheduler ThreadPool { get; } //IScheduler TaskPool { get; } } public sealed class SchedulerService : ISchedulerService { public IScheduler CurrentThread { get { return Scheduler.CurrentThread; } } public IScheduler Dispatcher { get { return Scheduler.Dispatcher; } } public IScheduler Immediate { get { return Scheduler.Immediate; } } public IScheduler NewThread { get { return Scheduler.NewThread; } } public IScheduler ThreadPool { get { return Scheduler.ThreadPool; } } //public IScheduler TaskPool { get { return Scheduler.TaskPool; } } }This now allows me to inject an ISchedulerService which gives me a seam to help my testing. I can now write some tests for my Presenter.
[TestInitialize] public void SetUp() { _myServiceMock = new Mock<IMyService>(); _viewModelMock = new Mock<IViewModel>(); _schedulerService = new TestSchedulers(); var prices = new ObservableCollection<decimal>(); _viewModelMock.SetupGet(vm => vm.Prices).Returns(prices); _viewModelMock.SetupProperty(vm => vm.IsConnected); } [TestMethod] public void Should_pass_symbol_to_MyService_PriceStream() { var expected = "SomeSymbol"; var priceStream = new Subject<decimal>(); _myServiceMock.Setup(svc => svc.PriceStream(It.Is<string>(symbol=>symbol==expected))).Returns(priceStream); var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService); sut.Show(expected); _myServiceMock.Verify(); } [TestMethod] public void Should_add_to_VM_Prices_when_MyService_publishes_price() { decimal expected = 1.23m; var priceStream = new Subject<decimal>(); _myServiceMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream); var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService); sut.Show("SomeSymbol"); _schedulerService.ThreadPool.Schedule(() => priceStream.OnNext(expected)); //Schedule the OnNext _schedulerService.ThreadPool.RunTo(1); //Execute the OnNext action _schedulerService.Dispatcher.RunTo(1); //Execute the OnNext Handler (ie adding to the Prices collection) Assert.AreEqual(1, _viewModelMock.Object.Prices.Count); Assert.AreEqual(expected, _viewModelMock.Object.Prices.First()); }These two tests show first a simple expectation that a string value passed to my Show(string) method will be passed to the underlying service. This is not at all relevant to Rx. The next test shows the usage of my implementation of the ISchedulerService specific for testing. It exposes all of the IScheduler properties as instances of TestSchedulers. This now allows me to inject TestSchedulers for testing which in-turn allows me to control the rate at which things are scheduled.
[TestMethod] public void Should_timeout_if_no_prices_for_10_seconds() { var timeoutPeriod = TimeSpan.FromSeconds(10); var priceStream = Observable.Never<decimal>(); _myServiceMock.Setup(svc => svc.PriceStream(It.IsAny<string>())).Returns(priceStream); var sut = new MyPresenter(_myServiceMock.Object, _viewModelMock.Object, _schedulerService); sut.Show("SomeSymbol"); _schedulerService.ThreadPool.RunTo(timeoutPeriod.Ticks - 1); Assert.IsTrue(_viewModelMock.Object.IsConnected); _schedulerService.ThreadPool.RunTo(timeoutPeriod.Ticks); Assert.IsFalse(_viewModelMock.Object.IsConnected); }The key points to note are:
public class MyPresenter { private readonly IMyService _myService; private readonly IViewModel _viewModel; private readonly ISchedulerService _schedulerService; public MyPresenter(IMyService myService, IViewModel viewModel, ISchedulerService schedulerService) { _myService = myService; _schedulerService = schedulerService; _viewModel = viewModel; } public void Show(string symbol) { _myService.PriceStream(symbol) .SubscribeOn(_schedulerService.ThreadPool) .ObserveOn(_schedulerService.Dispatcher) .Timeout(TimeSpan.FromSeconds(10), _schedulerService.ThreadPool) .Subscribe(OnPriceUpdate, ex => { if (ex is TimeoutException) _viewModel.IsConnected = false; }); _viewModel.IsConnected = true; } private void OnPriceUpdate(decimal price) { _viewModel.Prices.Add(price); } }I hope this post on testing helps you bring you skills that you have been developing in Rx to a level where you may be comfortable considering them for production use.
public sealed class TestSchedulers : ISchedulerService { private readonly TestScheduler _currentThread = new TestScheduler(); private readonly TestScheduler _dispatcher = new TestScheduler(); private readonly TestScheduler _immediate = new TestScheduler(); private readonly TestScheduler _newThread = new TestScheduler(); private readonly TestScheduler _threadPool = new TestScheduler(); #region Implementation of ISchedulerService IScheduler ISchedulerService.CurrentThread { get { return _currentThread; } } IScheduler ISchedulerService.Dispatcher { get { return _dispatcher; } } IScheduler ISchedulerService.Immediate { get { return _immediate; } } IScheduler ISchedulerService.NewThread { get { return _newThread; } } IScheduler ISchedulerService.ThreadPool { get { return _threadPool; } } #endregion public TestScheduler CurrentThread { get { return _currentThread; } } public TestScheduler Dispatcher { get { return _dispatcher; } } public TestScheduler Immediate { get { return _immediate; } } public TestScheduler NewThread { get { return _newThread; } } public TestScheduler ThreadPool { get { return _threadPool; } } } public sealed class ImmediateSchedulers : ISchedulerService { public IScheduler CurrentThread { get { return Scheduler.Immediate; } } public IScheduler Dispatcher { get { return Scheduler.Immediate; } } public IScheduler Immediate { get { return Scheduler.Immediate; } } public IScheduler NewThread { get { return Scheduler.Immediate; } } public IScheduler ThreadPool { get { return Scheduler.Immediate; } } }The full source code is now available either via svn at http://code.google.com/p/rx-samples/source/checkout or as a zip file.
Do(LazyEvaluation()); Do(EagerEvaluation()); private void Do(IEnumerable<int> list) { foreach (var i in list) { Console.WriteLine("Read out first value of {0}", i); break; } } public IEnumerable<int> LazyEvaluation() { Console.WriteLine("About to return 1"); yield return 1; Console.WriteLine("About to return 2");//Never called in this example yield return 2; } public IEnumerable<int> EagerEvaluation() { var result = new List<int>(); Console.WriteLine("About to return 1"); result.Add(1); Console.WriteLine("About to return 2");//executed but not used. result.Add(2); return result; }Implementations of IObservable<T> can exhibit similar variations in style.
private static IObservable<string> GetProducts() { return Observable.CreateWithDisposable<string>( o => { using(var conn = new SqlConnection(@"Data Source=.\SQLSERVER;Initial Catalog=AdventureWorksLT2008;Integrated Security=SSPI;")) using (var cmd = new SqlCommand("Select Name FROM SalesLT.ProductModel", conn)) { conn.Open(); SqlDataReader reader = cmd.ExecuteReader(CommandBehavior.CloseConnection); while (reader.Read()) { o.OnNext(reader.GetString(0)); } o.OnCompleted(); return Disposable.Create(()=>Console.WriteLine("--Disposed--")); } }); }This style of API would allow for a non blocking call to fetch the list of products and would inform the consumer of when the list was complete. This is fairly common stuff, but note that every time this is called, the database will be accessed again.
public void ColdSample() { var productStream = GetProducts().Take(3); productStream.Subscribe(Console.WriteLine); Console.ReadLine(); }
public void SimpleColdSample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period); observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i)); Thread.Sleep(period); observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i)); Console.ReadKey(); /* Ouput: first subscription : 0 first subscription : 1 second subscription : 0 first subscription : 2 second subscription : 1 first subscription : 3 second subscription : 2 */ }
public void SimpleConnectSample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period).Publish(); observable.Connect(); observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i)); Thread.Sleep(period); observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i)); Console.ReadKey(); /* Ouput: first subscription : 0 first subscription : 1 second subscription : 1 first subscription : 2 second subscription : 2 */ }In the example above the observable variable is an IConnectableObservable<T>, and by calling Connect() it will subscribe to the underlying (the Observable.Interval). In this case we are quick enough to subscribe before the first item is published but only on the first subscription. The second subscription subscribes late and misses the first publication. We could move the invocation of the Connect() method until after each of the subscriptions have been made so that even with the Thread.Sleep we wont really subscribe to the underlying until after both subscriptions are made. This would be done as follows:
public void SimpleConnectSample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period).Publish(); observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i)); Thread.Sleep(period); observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i)); observable.Connect(); Console.ReadKey(); /* Ouput: first subscription : 0 second subscription : 0 first subscription : 1 second subscription : 1 first subscription : 2 second subscription : 2 */ }You can probably imagine how this could be quite useful where an application had the need to share streams of data. In a trading application if you wanted to consume a price stream for a certain asset in more than one place, you would want to reuse that stream and not have to make another subscription to the server providing that data. Publish() and Connect() are perfect solutions for this.
public void ConnectAndDisposeSample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period).Publish(); observable.Subscribe(i => Console.WriteLine("subscription : {0}", i)); var exit = false; while (!exit) { Console.WriteLine("Press enter to connect, esc to exit."); var key = Console.ReadKey(true); if(key.Key== ConsoleKey.Enter) { var connection = observable.Connect(); //--Connects here-- Console.WriteLine("Press any key to dispose of connection."); Console.ReadKey(); connection.Dispose(); //--Disconnects here-- } if(key.Key==ConsoleKey.Escape) { exit = true; } } /* Ouput: Press enter to connect, esc to exit. Press any key to dispose of connection. subscription : 0 subscription : 1 subscription : 2 Press enter to connect, esc to exit. Press any key to dispose of connection. subscription : 0 subscription : 1 subscription : 2 Press enter to connect, esc to exit. */ }Let us finally consider automatic disposal of a connection. It would be common place for a single stream to be shared between subscriptions, as per the price stream example mentioned above. It would however also be common place for the developer to want to only have the stream running hot if there are subscriptions to it. Therefore it seems not only obvious that there should be a mechanism for automatically connecting (once a subscription has been made), but also a mechanism for disconnecting (once there are no more subscriptions) from a stream. First let us look at what happens to a stream when we connect with no subscribers, and then later unsubscribe:
public void OrphanedStreamExample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period) .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running. .Publish(); observable.Connect(); Console.WriteLine("Press any key to subscribe"); Console.ReadKey(); var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i)); Console.WriteLine("Press any key to unsubscribe."); Console.ReadKey(); subscription.Dispose(); Console.WriteLine("Press any key to exit."); Console.ReadKey(); /* Ouput: Press any key to subscribe Publishing 0 Publishing 1 Press any key to unsubscribe. Publishing 2 subscription : 2 Publishing 3 subscription : 3 Press any key to exit. Publishing 4 Publishing 5 */ }A few things to note here:
public void RefCountExample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period) .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running. .Publish() .RefCount(); //observable.Connect(); Use RefCount instead now Console.WriteLine("Press any key to subscribe"); Console.ReadKey(); var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i)); Console.WriteLine("Press any key to unsubscribe."); Console.ReadKey(); subscription.Dispose(); Console.WriteLine("Press any key to exit."); Console.ReadKey(); /* Ouput: Press any key to subscribe Press any key to unsubscribe. Publishing 0 subscription : 0 Publishing 1 subscription : 1 Publishing 2 subscription : 2 Press any key to exit. */ }
public void PruneExample() { var period = TimeSpan.FromSeconds(1); var observable = Observable.Interval(period) .Take(5) .Do(l => Console.WriteLine("Publishing {0}", l)) //produce Side effect to show it is running. .Prune(); observable.Connect(); Console.WriteLine("Press any key to subscribe"); Console.ReadKey(); var subscription = observable.Subscribe(i => Console.WriteLine("subscription : {0}", i)); Console.WriteLine("Press any key to unsubscribe."); Console.ReadKey(); subscription.Dispose(); Console.WriteLine("Press any key to exit."); Console.ReadKey(); /* Ouput: Press any key to subscribe Publishing 0 Publishing 1 Press any key to unsubscribe. Publishing 2 Publishing 3 Publishing 4 subscription : 4 Press any key to exit. */ }
public void ReplayOnHotExample() { var period = TimeSpan.FromSeconds(1); var hot = Observable.Interval(period) .Take(3) .Publish(); hot.Connect(); Thread.Sleep(period); //Run hot and ensure a value is lost. var observable = hot.Replay(); observable.Connect(); observable.Subscribe(i => Console.WriteLine("first subscription : {0}", i)); Thread.Sleep(period); observable.Subscribe(i => Console.WriteLine("second subscription : {0}", i)); Console.ReadKey(); observable.Subscribe(i => Console.WriteLine("third subscription : {0}", i)); Console.ReadKey(); /* Ouput: first subscription : 1 second subscription : 1 first subscription : 2 second subscription : 2 third subscription : 1 third subscription : 2 */ }
public static class Observable { public static IObservable<TSource> ObserveOn<TSource>( this IObservable<TSource> source, IScheduler scheduler) {...} public static IObservable<TSource> SubscribeOn<TSource>( this IObservable<TSource> source, IScheduler scheduler) {...} } public interface IScheduler { IDisposable Schedule(Action action); IDisposable Schedule(Action action, TimeSpan dueTime); DateTimeOffset Now { get; } }The IScheduler interface is of less interest to me than the types that implement the interface. Depending on your platform* (Silverlight3, Silverlight4, .Net 3.5, .Net 4.0) you will be exposed appropriate implementations via a static class Scheduler. These are the static properties that you can find on the Scheduler type that expose different schedulers.
private static void ScheduleTasks(IScheduler scheduler) { Action leafAction = () => Console.WriteLine("leafAction."); Action innerAction = () => { Console.WriteLine("innerAction start."); scheduler.Schedule(leafAction); Console.WriteLine("innerAction end."); }; Action outerAction = () => { Console.WriteLine("outer start."); scheduler.Schedule(innerAction); Console.WriteLine("outer end."); }; scheduler.Schedule(outerAction); } public void CurrentThreadExample() { ScheduleTasks(Scheduler.CurrentThread); Console.ReadLine(); /*Output: * outer start. * outer end. * innerAction start. * innerAction end. * leafAction. */ } public void ImmediateExample() { ScheduleTasks(Scheduler.Immediate); Console.ReadLine(); /*Output: * outer start. * innerAction start. * leafAction. * innerAction end. * outer end. */ }*Sorry Rx for JavaScript, I have not even opened the box on you and don’t know anything about scheduling in JavaScript.
public void LoadCustomers() { _customerService.GetCustomers() .SubscribeOn(Scheduler.NewThread) .ObserveOn(Scheduler.Dispatcher) .Subscribe(Customers.Add); }So all of the schedulers just offer a nice abstraction to us to utilise the various ways we can write concurrent code. Besides saving me from having to write the tedious code to get code onto a new thread or thread pool it also makes Rx threading easy. Oh Rx, you thought I had forgotten. I didn’t think that any of the schedulers except Current & Immediate warranted a further explanation but, I do think it is worth pointing out some of the “fun” threading problems you can face even though the scheduling has been abstracted away from you.
var stream = new Subject<int>(); Console.WriteLine("Next line should deadlock the system."); var value = stream.First(); stream.OnNext(1); Console.WriteLine("I can never execute....");Hopefully we wont ever write code that silly, and if we did our tests would give us fairly quick feed back that things were wrong. What lets deadlocks slip into the system is when they manifest themselves at integration points. This example may be a little harder to find but is only small step away from the silly 1st example. Here we block in the constructor on a UI element which will always be created on the dispatcher. The blocking call is waiting for an event, that can only be raised from the dispatcher – deadlock.
public Window1() { InitializeComponent(); DataContext = this; Value = "Default value"; //Deadlock! We need the dispatcher to continue // to allow me to click the button to produce a value. Value = _subject.First(); //This will give same result but will not be blocking(deadlocking). _subject.Take(1).Subscribe(value => Value = value); } private void MyButton_Click(object sender, RoutedEventArgs e) { _subject.OnNext("New Value"); } public string Value { get { return _value; } set { _value = value; var handler = PropertyChanged; if (handler != null) handler(this, new PropertyChangedEventArgs("Value")); } }In this example we start seeing things that can become more sinister. This example has a Button that the click command will try to get the first value from an Observable exposed via an interface.
public partial class Window1 : INotifyPropertyChanged { private readonly IMyService _service = new MyService(); //Imagine DI here. private int _value2; public Window1() { InitializeComponent(); DataContext = this; } public int Value2 { get { return _value2; } set { _value2 = value; var handler = PropertyChanged; if (handler != null) handler(this, new PropertyChangedEventArgs("Value2")); } } #region INotifyPropertyChanged Members public event PropertyChangedEventHandler PropertyChanged; #endregion private void MyButton2_Click(object sender, RoutedEventArgs e) { Value2 = _service.GetTemperature().First(); } }There is only a small problem here in that we block on the Dispatcher thread (.First() is a blocking call), however this manifest's itself into a deadlock if the service code is written incorrectly.
class MyService : IMyService { public IObservable<int> GetTemperature() { return Observable.Create<int>( o => { o.OnNext(27); o.OnNext(26); o.OnNext(24); return () => { }; }) .SubscribeOnDispatcher(); } }This odd implementation with explicit scheduling will cause the 3 OnNext calls to be scheduled once the .First() call has finished, which is waiting for an OnNext to be called – Deadlock.
//Generate values 0,1,2 var stream1 = Observable.Generate(0, i => i < 3, i => i, i => i + 1); //Generate values 100,101,102,103,104 var stream2 = Observable.Generate(100, i => i < 105, i => i, i => i + 1); stream1 .Concat(stream2) .Subscribe(Console.WriteLine); Console.ReadLine(); /* Returns: * * stream1 --0--0--0--| * stream2 -----------0--0--0--0--| * * result --0--0--0--0--0--0--0--| */If either stream was to OnError then the result stream would OnError too. This means that if stream1 produced an OnError then stream2 would never be used. If you wanted stream2 to be used regardless of if stream1 produced an OnError or not then the extension method OnErrorResumeNext would be your best option.
//Generate values 0,1,2 var stream1 = Observable.Range(0,3); //Generate values 100,101,102,103,104 var stream2 = Observable.Range(100,5); stream1 .Amb(stream2) .Subscribe(Console.WriteLine); Console.ReadLine(); /* Returns: * if stream 1 produces a value first... * stream1 --0--0--0--| * stream2 ---0--0--0--0--0--| * * result --0--0--0--| //All from stream1 */ stream1.Delay(TimeSpan.FromMilliseconds(100)) .Amb(stream2) .Subscribe(Console.WriteLine); Console.ReadLine(); /* Returns: * stream1 ---0--0--0--| * stream2 --0--0--0--0--0--| * * result --0--0--0--0--0--| //All from stream2 */The Merge extension method does a primitive combination of multiple streams where they implement the same type of T. The result will also be an IObservable<T> but will have the values produced to the result stream as the occur in the source streams. The stream will complete when all of the source streams complete or when an OnError is published by any stream.
//Generate values 0,1,2 var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3); //Generate values 100,101,102,103,104 var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(5).Select(i => i + 100); stream1 .Merge(stream2) .Subscribe(Console.WriteLine); Console.ReadLine(); /* * Returns: * stream1 ----0----0----0| * stream2 --0--0--0--0--0| * * result --0-00--00-0--00-| * Output: * 100 * 0 * 101 * 102 * 1 * 103 * 104 //Note this is a race condition. 2 could be * 2 // published before 104. */Merge also provides other overloads that allow you to pass more than 2 source observables via an IEnumerable or params arrays. The Overload that take a params array it great for when we know how many streams we want to merge at compile time, and the IEnumerable overload is better for when we dont know at compile time how many streams we need to merge.
//Create a third stream var stream3 = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(10).Select(i => i + 200); //Number of streams known at compile time. Observable.Merge(stream1, stream2, stream3) .Subscribe(Console.WriteLine); Console.ReadLine(); //We can dynamically create a list at run time with this overload. var streams = new List<IObservable<long>>(); streams.Add(stream1); streams.Add(stream2); streams.Add(stream3); Observable.Merge(streams).Subscribe(Console.WriteLine); Console.ReadLine();A quick video on Merge on Channe9.
//Generate values 0,1,2 var stream1 = Enumerable.Range(0, 3).ToObservable(); //Generate values 100,101,102,103,104 var stream2 = Enumerable.Range(100, 5).ToObservable(); stream1 .SelectMany(i => stream2, (lhs, rhs) => new { Left = lhs, Right = rhs }) .Subscribe(Console.WriteLine); Console.ReadLine(); /* * Output. * { Left = 0, Right = 100 } * { Left = 0, Right = 101 } * { Left = 1, Right = 100 } * { Left = 0, Right = 102 } * { Left = 1, Right = 101 } * { Left = 2, Right = 100 } * { Left = 0, Right = 103 } * { Left = 1, Right = 102 } * { Left = 2, Right = 101 } * { Left = 0, Right = 104 } * { Left = 1, Right = 103 } * { Left = 2, Right = 102 } * { Left = 1, Right = 104 } * { Left = 2, Right = 103 } * { Left = 2, Right = 104 } */A quick Video on SelectMany on channel9
//Generate values 0,1,2 var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3); //Generate values a,b,c,d,e,f var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(6).Select(i => Char.ConvertFromUtf32((int)i + 97)); stream1 .Zip(stream2, (lhs, rhs) => new { Left = lhs, Right = rhs }) .Subscribe(Console.WriteLine); Console.ReadLine(); /* Returns: * stream1 ----0----1----2| stream 1 values represented as ints * stream2 --a--b--c--d--e--f| s2 values represented as chars * * result ----0----1----2| * a b c */Here are two short videos on Zip (first, second)on Channel9. Note the second video is actually incorrect, can you spot why?
//Generate values 0,1,2 var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3); //Generate values a,b,c,d,e,f var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(6).Select(i => Char.ConvertFromUtf32((int)i + 97)); stream1 .CombineLatest(stream2, (s1, s2) => new { Left = s1, Right = s2 }) .Subscribe(Console.WriteLine, () => Console.WriteLine("Complete")); Console.ReadLine(); /* * Returns: * stream1 ----0----1----2| stream 1 values represented as ints * stream2 --a--b--c--d--e--f| stream 2 values represented as chars * * result ----00--01-1--22-2| the result as pairs. * ab cc d de f */Quick video on CombineLatest on Channel9
//Generate values 0,1,2 var stream1 = Observable.Interval(TimeSpan.FromMilliseconds(250)).Take(3); //Generate values a,b,c,d,e var stream2 = Observable.Interval(TimeSpan.FromMilliseconds(150)).Take(5).Select(i => Char.ConvertFromUtf32((int)i + 97)); stream1 .ForkJoin(stream2, (s1, s2) => new { Left = s1, Right = s2 }) .Subscribe(Console.WriteLine); Console.ReadLine(); /* * Returns: * stream1 ----0----1----2| stream 1 values represented as ints * stream2 --a--b--c--d--e| s2 values represented as chars * * result --------------2| the result as pairs. * e */One thing to note with most of the extension methods discussed is that they generally have a matching static method that takes a params array or IEnumerable as discussed in the Merge chapter.
try { var subject = new Subject<int>(); subject.Subscribe(Console.WriteLine); //Or explicitly rethrow as per this line. //subject.Subscribe(Console.WriteLine, ex => { throw ex.PrepareForRethrow(); }); subject.OnNext(1); subject.OnError(new IOException("test exception")); } catch (IOException ioEx) { Console.WriteLine("Caught IO exception: {0}", ioEx.Message); }Throwing published OnError exceptions does not allow for a very composable style of coding. It may be useful in cases like the example above, but the power of Rx as you will see in future posts is the ability to create compositions of streams together. Once we are running compositions of streams then SEH is not very helpful or useful. Rx provides several methods to provide a composition friendly way to handle errors and exceptions.
--0--0--0—|This is a sample of a stream that publishes 4 values then errors.
--0--0--0--0--XWhile these examples may seem too simple to warrant a visual representation, the simplicity of marble diagrams are great once we using multiple streams.
S1--0--0--X S2-----------0--0--0--0 R --0--0-----0--0--0--0In the example above we could recreate this with several methods.
public static void RetrySample<T>(IObservable<T> stream) { stream.Retry().Subscribe(t=>Console.WriteLine(t)); //Will always retry the stream Console.ReadKey(); } /* Given stream that will produce 0,1,2 then error; the output would be 0 1 2 0 1 2 0 1 2 ..... */which would look like this as a marble diagram
S--0--0--0--x--0--0--0--x--0--0--0-- R--0--0--0-----0--0--0-----0--0--0--Alternatively we can specify the max number of times to retry. In this example we only retry once, therefore the error that gets published on the second subscription will be passed up to the final subscription. Note that to retry once you pass a value of 2. Maybe the method should be called Try?
public static void RetryOnceSample<T>(IObservable<T> stream) { stream.Retry(2) .Subscribe(t=>Console.WriteLine(t), ex=>Console.WriteLine("Gave up on 2nd Error")); Console.ReadKey(); } /*Ouput: 0 1 2 0 1 2 Gave up on 2nd Error */As a marble diagram this would look like
S--0--0--0--x--0--0--0--x R--0--0--0-----0--0--0--xOnErrorResumeNext may cause some old VB developers to shudder but it offers a different route to use than Retry. While retry will always try to re-subscribe to the same stream; OnErrorResumeNext takes another IObservable<T> as a parameter to use when the original stream publishes and error. In this example when the stream1 publishes and error we re-subscribe to stream2.
public static void FailoverSample<T>(IObservable<T> stream, IObservable<T> failover) { stream .OnErrorResumeNext(failover) .Subscribe(t=>Console.WriteLine(t), ex => Console.WriteLine(ex)); } /* stream --"1"--"2"--"3"--X failover--------------------"a"--"b"--"c"--X result --"1"--"2"--"3"-----"a"--"b"--"c"--| */An important thing to note here is that when the second stream publishes an error the result stream just completes and ignores the error.
public static void CatchSample<T>(IObservable<T> stream, IObservable<T> failover) { stream .Catch((InvalidOperationException ex) => failover) .Subscribe(t => Console.WriteLine(t), ex => Console.WriteLine(ex), () => Console.WriteLine("Completed")); }Catch has other overloads that allow you to pass a params array of Observable<T> or an Enumerable<Observable<T>> instead of just specifying an intial and a failover stream. This means you can effectively have a large list of streams to try when a previous one fails. Of course if any of them actually complete then the next one will not be used. If the last one publishes an OnError then that error will then be published to the IObservalbe that the Catch method returned.
public abstract class Notification<T> : IEquatable<Notification<T>> { // Properties public abstract Exception Exception { get; } public abstract bool HasValue { get; } public abstract NotificationKind Kind { get; } public abstract T Value { get; } // Methods public abstract TResult Accept<TResult>(IObserver<T, TResult> observer); public abstract void Accept(IObserver<T> observer); public abstract void Accept(Action<T> onNext, Action<Exception> onError, Action onCompleted); public abstract TResult Accept<TResult>(Func<T, TResult> onNext, Func<Exception, TResult> onError, Func<TResult> onCompleted); // ... }By inspecting the Kind property you can identify which type of publication you have received. If the Kind is OnError you can access its published Exception via the Exception property. Accessing the Value property will get you either
public static class ExampleExtensions { /// <summary> /// Logs implicit notifications to console. /// </summary> /// <example> /// <code>myStream.Log().Subscribe(....);</code> /// </example> public static IObservable<T> Log<T>(this IObservable<T> stream) { return stream.Materialize() .Do(n => Console.WriteLine(n)) .Dematerialize(); } }Note that here we use Dematerialize to take our stream of Notification<T> and transform it back to our original stream. You could also use Materialize to create your own more powerful Catch methods, but we will look at applications of Rx later.
var interval = Observable.Interval(TimeSpan.FromMilliseconds(100));
var firstSubscription =
interval.Subscribe(value => Console.WriteLine("1st subscription recieved {0}", value));
var secondSubscription =
interval.Subscribe(value => Console.WriteLine("2nd subscription recieved {0}", value));
Thread.Sleep(500);
firstSubscription.Dispose();
Console.WriteLine("Disposed of 1st subscription");
Console.ReadKey();
/*Outputs:
1st subscription recieved 0
2nd subscription recieved 0
1st subscription recieved 1
2nd subscription recieved 1
1st subscription recieved 2
2nd subscription recieved 2
1st subscription recieved 3
2nd subscription recieved 3
2nd subscription recieved 4
Disposed of 1st subscription
2nd subscription recieved 5
2nd subscription recieved 6
2nd subscription recieved 7
etc....
*/
In the above example, it looks like the values are being produced by the interval Observable by a single OnNext call, however these are independent and work similarly to how a Observable.Create<T> method would work. In this sample we just pause a bit before making our second subscription. Note that the output is different to the above example.var interval = Observable.Interval(TimeSpan.FromMilliseconds(100));
var firstSubscription =
interval.Subscribe(value => Console.WriteLine("1st subscription recieved {0}", value));
Thread.Sleep(500);
var secondSubscription =
interval.Subscribe(value => Console.WriteLine("2nd subscription recieved {0}", value));
Thread.Sleep(500);
firstSubscription.Dispose();
Console.WriteLine("Disposed of 1st subscription");
/*
Ouput:
1st subscription recieved 0
1st subscription recieved 1
1st subscription recieved 2
1st subscription recieved 3
1st subscription recieved 4
2nd subscription recieved 0
1st subscription recieved 5
2nd subscription recieved 1
1st subscription recieved 6
1st subscription recieved 7
2nd subscription recieved 2
1st subscription recieved 8
2nd subscription recieved 3
Disposed of 1st subscription
2nd subscription recieved 4
2nd subscription recieved 5
2nd subscription recieved 6
etc...
*/
The benefits of using the IDisposable Type instead of creating a new ISubscription/IUnsubscription interface or amending the IObservable<T> interface to have an Unsubscribe method is that you get all of these things for free:var subject = new Subject<int>(); subject.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed")); subject.OnCompleted(); subject.OnNext(2);Of course you could implement your own IObservable<T> that did allow publishing after an OnComplete or an OnError, however it would not follow the precedence of the current Subject types and would be a non standard implementation. I think it would be safe to say that the inconsistent behaviour would cause unpredictable behaviour in the applications that consumed your code.