Listening to a single value from an Rx observable sequence

I have an observable sequence and I want to observe only a defined number of values, the number is immaterial, the important part is the fact I don't want to be notified of any further values.

How can this can be achieved?

The answer is easy, use the Rx extension method Take, it'll return the required number of values only. A simple example:

   1:  Observable.FromEventPattern<RoutedEventArgs>(clickButton, "Click")

   2:      .Take(1)

   3:      .ObserveOn(Scheduler.CurrentThread)

   4:      .Subscribe(vt => Debug.WriteLine("{0}: Click called...", ++count));

When the button is clicked in the UI it produces a line in the output window in visual studio:

So using the Take method I can limit the number of observations irrespective of the number of times I click the 'Click Me!' button:

   1:  Observable.FromEventPattern<RoutedEventArgs>(clickButton, "Click")

   2:      .Take(1)

   3:      .ObserveOn(Scheduler.CurrentThread)

   4:      .Subscribe(vt => Debug.WriteLine("{0}: Click called...", ++count));

I was then wondering, and this is the reason for the post:

What's happened to the under lying subscription, has it been disposed?

The seems very similar to a previous post about Rx subscribers and what happens to subscriptions when the OnCompleted is invoked by the disposing the under lying stream. I suspect the answer is 'I don't have anything to worry about, it will all be disposed automatically'.

Lets find out by assigning the subscription to a variable and when the form is closed check the internals of the subscription - yes I am mixing metaphors for .Net event handling ;)

Okay, wasn't expecting that! The under lying composite disposable hasn't been disposed as I was expecting...

I was really expecting it to be automatically disposed at this point, I wonder if adding an OnCompleted implementation for the subscription makes any difference:

So now the subscription is being disposed automatically...

WTF! - How does defining an OnCompleted method mean it can be disposed? Surely what defines it as being eligible for disposing is we're using the Take method with a value of 1.

Now you could say:

'Why worry, I'm sure Rx will take care of it...'

I say anything I create\instantiate which is disposable I am responsible for make sure it is cleaned-up, and this is especially important if you have transient scopes which are regularly being created and destroyed.

Also, the fact I'm using the FromEventPattern extension is not important, the behaviour of the Take method is.

Read More

Applying LINQ principles to business logic

This was inspired by a conversation I was having with @leeoades & @hamishdotnet at work earlier this week, we were discussing the use of Yield keyword to short-cut method execution. Lee has a great post about sequences here. I wanted to get down how my thinking about the Yield keyword has now changed.

Historically I've always thought about using the yield when I'm defining a custom enumerator method, something like GetFooEnumerator - typically I'd have placed this on some kind of model object to allow the easy iteration of some child model instances. Lee was saying why not use yield where ever you have the following usage pattern:

   1:  public IEnumerable<int> Range(int start, int end)

   2:  {

   3:      var numbers = new List<int>();

   4:      for (var i = start; i < end; i++)

   5:      {

   6:          // Do some business logic here...

   7:          numbers.Add(i);

   8:      }

   9:   

  10:      return numbers;

  11:  }

When ever you have a method which internally creates & populates a collection and then returns the collection why not use yield to make it more efficient?

This got me thinking, what kind of improvement efficiency are we talking about?

Time for a quick test, first we need a couple of methods to compare, first one very similar to above and another making use of the yield keyword but both achieving the same logic - returning a collection of numbers to the caller:

   1:  public class Numbers

   2:  {

   3:      public IEnumerable<int> Range(int start, int end)

   4:      {

   5:          var numbers = new List<int>();

   6:          for (var i = start; i < end; i++)

   7:          {

   8:              numbers.Add(i);

   9:          }

  10:   

  11:          return numbers;

  12:      }

  13:   

  14:      public IEnumerable<int> YieldRange(int start, int end)

  15:      {

  16:          for (var i = start; i < end; i++)

  17:          {

  18:              yield return i;

  19:          }

  20:      }

  21:  }

Then a test program - use each method as part of LINQ call to validate the returned sequences contain the number 420:

   1:  static void Main(string[] args)

   2:  {

   3:      Console.WriteLine();

   4:   

   5:      var numbers = new Numbers();

   6:   

   7:      var sw = new Stopwatch();

   8:      sw.Start();

   9:      var found1 = numbers.Range(0, 999).Any(n => n == 420);

  10:      sw.Stop();

  11:   

  12:      Console.WriteLine("Normal Loop: ticks = " + sw.ElapsedTicks);

  13:   

  14:      var sw2 = new Stopwatch();

  15:      sw2.Start();

  16:      var found2 = numbers.YieldRange(0, 999).Any(n => n == 420);

  17:      sw2.Stop();

  18:             

  19:      Console.WriteLine("   Yielding: ticks = " + sw2.ElapsedTicks);

  20:      Console.ReadLine();

  21:  }

Running this up I got the following output!

What!

I wasn't expecting it to be slower, after all the YeildRange method only has to iterate 420 times before finding a match where as the Range method has to iterate over all 1000 numbers before returning...

Then @hamishdotnet pointed out the cost of yielding and I remembered the chapter in Jon Skeet's C# In Depth about iterator blocks and how a set of classes are created by the compiler when ever you use the yield keyword.

Then I had the realisation the methods didn't really have any 'business logic' - DOH!

Basically the 2 implementations don't do anything, so to simulate this I put a Thread.Sleep in to represent an I/O or processor bound call:

   1:  public class Numbers

   2:  {

   3:      public IEnumerable<int> Range(int start, int end)

   4:      {

   5:          var numbers = new List<int>();

   6:          for (var i = start; i < end; i++)

   7:          {

   8:              System.Threading.Thread.Sleep(10);

   9:              numbers.Add(i);

  10:          }

  11:   

  12:          return numbers;

  13:      }

  14:   

  15:      public IEnumerable<int> YieldRange(int start, int end)

  16:      {

  17:          for (var i = start; i < end; i++)

  18:          {

  19:              System.Threading.Thread.Sleep(10);

  20:              yield return i;

  21:          }

  22:      }

  23:  }

More promising results now...

So what do I mean by 'Applying LINQ principles to business logic'?

By thinking about concepts such as 'lazy evaluation' & 'materialization' when designing your methods you allow for the opportunities for your code to be more efficiently executed when used as part of any linq queries.

Read More

How do I flatten an enumerable IObservable to IObservable?

The title of this post should be 'How do I flatten IObservable<IEnumerable<T>> to IObservable<T>?'

But blogger says 'no'...

This seems like a simple & legitimate Rx question and if you're one of the initiated it's very easy simple.

It came up the other day because a back-end service we're talking to returned an array of stuff and we wanted to flatten it into a stream of single values we could observe.

So what do I replace CONVERT_SOMEHOW with?

voilà...

   1:  IObservable<int[]> numbers = Observable.Return(new[] {1, 2, 3, 4, 5});

   2:   

   3:  IObservable<int> number = numbers.SelectMany(n => n);

In a simple app:

produces the following output:

Read More

Does the Rx subscriber get disposed when OnCompleted is called?

We use an implementation of an observable command - ObservableCommand<T> in our view models not only to wire the actions of controls defined in the XAML but to allow the other interested parties to be notified when the command has executed.

We were wondering what happens when the observable command is disposed, does the OnCompleted method for the subscribers get called and importantly are the subscribers disposed?

Our observable command has an implementation similar to this:

   1:   

   2:  public class ObservableCommand<T> : ICommand, IObservable<T>

   3:  {

   4:      private readonly Subject<T> _subj = new Subject<T>();

   5:   

   6:      public void Execute(object parameter)

   7:      {

   8:          _subj.OnNext((T)parameter);

   9:      }

  10:   

  11:      public bool CanExecute(object parameter)

  12:      {

  13:          return true;

  14:      }

  15:   

  16:      public event EventHandler CanExecuteChanged;

  17:   

  18:      public IDisposable Subscribe(IObserver<T> observer)

  19:      {

  20:          return _subj.Subscribe(observer);

  21:      }

  22:  }

To test question I used a simple class which generates numbers, this exposes 2 methods one to set up a subscription and the other to shut down the under lying stream, this wouldn't be any good for 'real' code but will do for a test:

   1:  public class NumberGenerator

   2:  {

   3:      private readonly TimeSpan _interval = TimeSpan.FromMilliseconds(500);

   4:      private IObserver<int> _stream;

   5:   

   6:      public IObservable<int> Generate()

   7:      {

   8:          var number = 42;

   9:   

  10:          return Observable.Create<int>(o =>

  11:          {

  12:              _stream = o;

  13:              return Observable.Timer(DateTime.Now.Add(_interval), _interval)

  14:              .Subscribe(l =>

  15:              {

  16:                  var currentNumber = number;

  17:                  number = number + 1;

  18:   

  19:                  _stream.OnNext(currentNumber);

  20:              });

  21:          });

  22:      }

  23:   

  24:      public void Complete()

  25:      {

  26:          _stream.OnCompleted();

  27:      }

  28:  }

This was then tested with following program, notice I pass in an Action<T> for the completed action to the Subscribe method:

   1:  internal class Program

   2:  {

   3:      private IObservable<long> _underlying;

   4:   

   5:      private static void Main(string[] args)

   6:      {

   7:          var generator = new NumberGenerator();

   8:   

   9:          var disposable = generator.Generate()

  10:              .Subscribe(n => Console.WriteLine("Number: " + n), () => Console.WriteLine("Completed..."));

  11:   

  12:          Console.ReadLine();

  13:   

  14:          generator.Complete();

  15:   

  16:          Console.ReadLine();

  17:      }

  18:  }

It sets up a subscription on a background thread, the generator then starts producing values every couple of seconds (on another background thread). These are recieved and outputted to the console. Whilst this is happening the main thread is blocked by the first call Console.ReadLine method. Once the enter key is pressed the the internal under lying stream in the generator is shut down and hopefully the OnCompleted method will be called for the subscriber and output "Completed..."

Did OnCompleted get called?

Yes...

So the title question is 'Does the Rx subscriber get disposed when OnComplete is called?'

Yes...

To check this lets look inside the returned IDisposable for the test program:

Debugging the app with a breakpoint on the second Console.ReadLine method, you can see the disposable local variable has been disposed:

So when the under lying stream in Rx is shutdown then all of the subscribers will have their OnCompleted methods called and will be disposed of automatically - perfect!

Read More

Using PostSharp for AOP with Reactive Extensions - update

After digging further into the use of AOP with Reactive Extensions my first post about can be improved by using the Do method instead of the Subscribe method on the IObservable<T> inside the AOP aspect OnExit method.

Why?

The previous implementation I gave would worked well for a hot observable, but for a cold observable this wouldn't apply as it would instantiate a new stream for each subscription setup in the aspect - Lee campbell has a good definition for hot & cold observables.

This means the OnExit method changes from:

to:

There isn't much difference in the implementation, the significant part is re-assigning the ReturnValue for the method execution arguments.

Read More