Реактивные расширения: почему этот выход немедленно?

Question

Я читаю IntroToRx, и у меня возникают проблемы с образцом кода. Вот общая сумма моего кода:

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(); 
    } 

    } 
} 

Если я понимаю книгу правильно, это должно написать последовательность чисел на консоль, один раз каждые пять секунд навсегда, так как я никогда не Dispose() последовательности.

Однако, когда я запускаю код, все, что я получаю, это «Готово» в конце. Нет номеров, нет «завершено», ничего, кроме «Готово».

Что я здесь делаю неправильно?

Answer 1

Я предполагаю, что у вас не было терпения ждать 5 секунд, чтобы пройти, иначе вы бы увидели, что код работает.

Основная идея иметь в виду, с Rx что Observable.Subscribe возвращает управление вызовом метода почти сразу. Другими словами, Observable.Subscribe не блокируется до тех пор, пока не будут получены результаты. Таким образом, вызов Console.WriteLine будет вызываться только через пять секунд.

Answer 2

Вам нужно каким-то образом заставить основной поток ждать, что вы делаете. Вы можете использовать семафор, если вам нравится

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(); 
    } 

    } 
} 

Хотя, наверное, лучше в этом случае просто написать

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