將基於事件的代碼轉換爲Rx

Question

我有以下代碼（爲發佈目的而簡化）。

public class SomeDataObject 
{ 
    public delegate void ReadyEventHandler; 
    public delegate void ErrorEventHandler; 

    public event ReadyEventHandler Ready; 
    public event ErrorEventHandler Error; 
    ... 
} 

pubic class ConsumerClass 
{ 
    private SomeDataObject dataObject; 

    private Task<List<string>> GetStrings() 
    { 
     List<string> results = new List<string>(); 
     var tcs = new TaskCompletionSource<List<string>>(); 

     SomeDataObject.ReadyEventHandler ReadyHandler = null; 
     SomeDataObject.ErrorEventHandler ErrorHandler = null; 

     ReadyHandler +=() => 
     { 
      for (int i =0; i < dataObject.ItemCount; i++) 
       results.Add(dataObject[i].ToString()); 

      tcs.TrySetResult(results); 
     } 

     ErrorHandler +=() 
     { 
      tcs.TrySetException(new Exception("oops!"); 
     } 

     dataObject.Ready += ReadyHandler; 
     dataObject.Error += ErrorHandler; 

     dataObject.DoRequest(); 
    } 
} 

的想法是，當DoRequest調用時，SomeDataObject會得到一些數據，提高或者就緒或錯誤事件（詳細信息並不重要！）。如果數據可用，則ItemCount指示有多少項目可用。

我是Rx的新手，找不到任何可比的例子。那麼是否有可能將此轉換爲Rx，以便使用Observable.Create以某種方式返回IObservable<string>而不是Task<List<string>>？

問候 艾倫

Answer 1

馬修的答案很接近但有一些問題。首先，它是急切的，這通常不符合Rx /函數式編程的精神。接下來我想你會希望能夠在消費者處置時釋放事件句柄。最後，一個主題的使用應該是一個代碼氣味，這種情況下它指向上述兩個問題:-)

這裏我使用Observable.Create（它應該是工具箱中的＃1 goto工具，與科目是你的最後手段）懶惰地連接，並且當訂閱被處置時也提供斷開/釋放事件。

private IObservable<string> GetStrings() 
{ 
    return Observable.Create<string>(o=> 
    { 
     SomeDataObject.ReadyEventHandler ReadyHandler = null; 
     SomeDataObject.ErrorEventHandler ErrorHandler = null; 

     ReadyHandler +=() => 
     { 
      for (int i =0; i < dataObject.ItemCount; i++) 
       o.OnNext(dataObject[i].ToString()); 

      o.OnCompleted(); 
     } 

     ErrorHandler +=() => 
     { 
      o.OnError(new Exception("oops!")); 
     } 

     dataObject.Ready += ReadyHandler; 
     dataObject.Error += ErrorHandler; 

     dataObject.DoRequest(); 

     return Disposable.Create(()=> 
      { 
       dataObject.Ready -= ReadyHandler; 
       dataObject.Error -= ErrorHandler; 
      }); 
    } 
} 

我也會考慮將dataObject移動到方法的參數。在Async系統中共享狀態是問題的根源。

Answer 2

我認爲下面的代碼會做你想要什麼。 ReplaySubject用於確保調用者獲得所有結果，即使SomeDataObject事件立即開始。

private IObservable<string> GetStrings() 
{ 
    ReplaySubject<string> results = new ReplaySubject<string>(); 

    SomeDataObject.ReadyEventHandler ReadyHandler = null; 
    SomeDataObject.ErrorEventHandler ErrorHandler = null; 

    ReadyHandler +=() => 
    { 
     for (int i =0; i < dataObject.ItemCount; i++) 
      results.OnNext(dataObject[i].ToString()); 

     results.OnCompleted(); 
    } 

    ErrorHandler +=() 
    { 
     results.OnError(new Exception("oops!")); 
    } 

    dataObject.Ready += ReadyHandler; 
    dataObject.Error += ErrorHandler; 

    dataObject.DoRequest(); 

    return results; 
} 

Answer 3

在回答關於李的（完全可愛和蜱值得）回答您的意見，這裏是如何修改他的回答得到一個單一List<string>響應和塊：

private IObservable<List<string>> GetStrings(SomeDataObject dataObject) 
{ 
    return Observable.Create<List<string>>(o=> 
    { 
     SomeDataObject.ReadyEventHandler ReadyHandler = null; 
     SomeDataObject.ErrorEventHandler ErrorHandler = null; 

     ReadyHandler =() => 
     { 
      var results = new List<string>(dataObject.ItemCount); 
      for (int i =0; i < dataObject.ItemCount; i++) 
       results.Add(dataObject[i].ToString()); 

      o.OnNext(results); 
      o.OnCompleted(); 
     }; 

     ErrorHandler =() => 
     { 
      o.OnError(new Exception("oops!")); 
     }; 

     dataObject.Ready += ReadyHandler; 
     dataObject.Error += ErrorHandler; 

     dataObject.DoRequest(); 

     return Disposable.Create(()=> 
      { 
       dataObject.Ready -= ReadyHandler; 
       dataObject.Error -= ErrorHandler; 
      }); 
    }); 
} 

現在你可以阻斷這跟：

var results = GetStrings().Wait(); 

如果使用.NET 4.5，然後在async方法，你也可以做：

var results = await GetStrings();