Rx.NET не запускает onError или onComplete

Question

Я изучаю Rx.NET и в примере кода ниже. Я пытаюсь выполнить событие изменения времени выбора времени асинхронной задачи. OnNext работает отлично, но onError и onComplete этого не делают. Что я делаю не так?

private void frmReporting_Load(object sender, EventArgs e){ 
    dtPickerValueChanged = Observable.FromEventPattern(
     ev => dtPickerFrom.ValueChanged += ev, 
     ev => dtPickerFrom.ValueChanged -= ev); 

    dtPickerValueChanged = dtPickerValueChanged.Merge(Observable.FromEventPattern(
     ev => dtPickerTo.ValueChanged += ev, 
     ev => dtPickerTo.ValueChanged -= ev)); 

    dtPickerValueChanged  
     .Where(x => dtPickerFrom.Value <= dtPickerTo.Value) 
     .Finally(() => { tsslStatus.Text = "Finally"; })  
     .ObserveOn(SynchronizationContext.Current) 
     .Subscribe(
      onNext: async x => { 
       var result = await Test(dtPickerFrom.Value, dtPickerTo.Value); 
       MessageBox.Show(result); 
      }, 
      onError: exception => { MessageBox.Show(exception.Message); }, 
      onCompleted:() => { MessageBox.Show("Finished loading data"); }); 
} 

public Task<string> Test(DateTime start, DateTime end){ 
    return Task.Run(
     async() => 
     { 
      await Task.Delay(3000); 
      return start.ToString("dd/MM/yyyy - HH:mm:ss") + " | " + end.ToString("dd/MM/yyyy - HH:mm:ss"); 
     }); 
} 

Answer 1

Когда вы не используете Rx и просто присоединяете обработчик событий, это событие не может сказать вам, что оно завершено. Обработчик предположит, что всегда будет больше событий, даже если он будет вызываться только один раз (подумайте onLoaded event в js land). При преобразовании события в наблюдаемое вы увидите такое же поведение; библиотека Rx не имеет возможности узнать, закончился ли поток событий.

Вы всегда можете использовать методы Take, чтобы ограничить количество предметов, получаемых вашим подписчиком. Надеюсь, что это помогает ответить, почему наблюдаемые никогда не заканчиваются

MouseMove.TakeUntil(MouseUp).Subscribe(...); 
DocLoad.Take(1).Subscribe(...); 

Как для OnError никогда не будет вызван - если ваш абонент выдает сообщение об ошибке, то это проблема с вашим абонентом, а не наблюдаемая сама и не имело бы никакого смысла для пусть все остальные подписчики знают, что один из подчиненных был изгоем.

OnError обычно используется для указания того, что произошла ошибка при получении значения в скворе. например

Observable.Create(sub => { 
    try { 
     var res = SomethingThatCanThrow(); 
     sub.OnNext(res); 
     sub.OnCompleted(); 
    } catch (Exception ex) { 
     sub.OnError(ex); 
    } 

    return Disposable.Empty; 
});