Reactive Extensions: Why does this exit immediately?

Question

I am reading IntroToRx and I'm having a bit of trouble with the sample code. Here is the sum total of my code:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace LearningReactiveExtensions
{
  public class Program
  {
    static void Main(string[] args)
    {
        var observable = Observable.Interval(TimeSpan.FromSeconds(5));
        observable.Subscribe(
          Console.WriteLine, 
          () => Console.WriteLine("Completed")
        );
        Console.WriteLine("Done");
        Console.ReadKey();
    }

  }
}

If I understand the book correctly, this should write a sequence of numbers to the console, once every five seconds forever since I never Dispose() of the sequence.

However, when I run the code, all I get is the "Done" at the end. No numbers, no "completed", nothing but "Done".

What am I doing wrong here?

score 2 · Accepted Answer · answered Aug 24 '13 at 07:38

2

I am assuming you haven't had the patience to wait for 5 seconds to elapse otherwise you would have seen that the code is working.

The main idea to keep in mind with Rx is that Observable.Subscribe will return control to calling method almost immediately. In other words, Observable.Subscribe does not block until the results are produced. Thus the call to Console.WriteLine will be invoked only after five seconds.

answered Aug 24 '13 at 07:38

RePierre

9,358
2
20
37

It wasn't that I lacked the patience; rather I assumed that "Done" would never display unless the sequence had completed, e.g. never in the example. A fundamental flaw in my understanding. – Scott Baker Aug 26 '13 at 17:19
I was worried for a moment that I had put an example up with the Done clause and the Completed. If you swapped the word "Done" for "Subscribed", then you program would be more accurate. Ideally also capture the subscription and then dispose it after the `ReadKey()`. – Lee Campbell Aug 29 '13 at 08:49

score 0 · Answer 2 · answered Aug 24 '13 at 08:51

You need some way to make the main thread wait for what you are doing. You can use a semaphore if you like

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace LearningReactiveExtensions
{
  public class Program
  {
    static void Main(string[] args)
    {
         SemaphoreSlim ss = new SemaphoreSlim(1);
        var observable = Observable.Interval(TimeSpan.FromSeconds(5));
        observable.Subscribe(
          Console.WriteLine, 
          () => {
               Console.WriteLine("Completed");
               ss.Release();
          }
        );
        ss.Wait();
        Console.WriteLine("Done");
        Console.ReadKey();
    }

  }
}

Though probably better in this case just to write

  static void Main(string[] args)
   {
        SemaphoreSlim ss = new SemaphoreSlim(1);
        Observable.Interval(TimeSpan.FromSeconds(5)).Wait();
        Console.WriteLine("Completed");
        Console.WriteLine("Done");
        Console.ReadKey();
   }

I would argue that you should not be using the primitive wait handles and Rx on such a simple example. IMO, to do so undermines the point of learning Rx. — Lee Campbell, Aug 29 '13 at 08:51
You have to use wait in main. You can't mark it as async and use await. — bradgonesurfing, Aug 29 '13 at 12:14

Reactive Extensions: Why does this exit immediately?

2 Answers2

Linked