如何在Silverlight中使用Reactive Extensions（Rx）組織這些調用？

Question

我有一些必須按順序執行的調用。考慮具有Query和Load方法的IService。查詢給出了一個小部件列表，並且加載提供了一個「默認」小部件。因此，我的服務看起來像這樣。

void IService.Query(Action<IEnumerable<Widget>,Exception> callback); 
void IService.Load(Action<Widget,Exception> callback); 

考慮到這一點，這裏是視圖模型的草圖：

public class ViewModel : BaseViewModel 
{ 
    public ViewModel() 
    { 
     Widgets = new ObservableCollection<Widget>(); 

     WidgetService.Query((widgets,exception) => 
     { 
      if (exception != null) 
      { 
       throw exception; 
      } 

      Widgets.Clear(); 

      foreach(var widget in widgets) 
      { 
      Widgets.Add(widget); 
      } 

      WidgetService.Load((defaultWidget,ex) => 
      { 
      if (ex != null) 
      { 
       throw ex; 
      } 
      if (defaultWidget != null) 
      { 
       CurrentWidget = defaultWidget; 
      } 
      } 
     }); 
    } 

    public IService WidgetService { get; set; } // assume this is wired up 

    public ObservableCollection<Widget> Widgets { get; private set; } 

    private Widget _currentWidget; 

    public Widget CurrentWidget 
    { 
     get { return _currentWidget; } 
     set 
     { 
     _currentWidget = value; 
     RaisePropertyChanged(()=>CurrentWidget); 
     } 
    } 
} 

我想要做的是簡化調用查詢，然後默認的順序工作流。也許最好的方法是用lambda表達式嵌套，但我認爲Rx可能更優雅。我不想爲了Rx而使用Rx，但是如果它可以讓我組織上面的邏輯，以便在方法中讀取/維護更容易，我會利用它。理想的情況下，是這樣的：

Observable.Create(
    ()=>firstAction(), 
    ()=>secondAction()) 
.Subscribe(action=>action(),error=>{ throw error; }); 

隨着電力線程庫，我願意做這樣的事情：

Service.Query(list=>{result=list}; 
yield return 1; 
ProcessList(result); 
Service.Query(widget=>{defaultWidget=widget}; 
yield return 1; 
CurrentWidget = defaultWidget; 

這使得它更明顯的是，工作流是順序，並消除嵌套（的收益率異步枚舉器的一部分，並且是阻止直到結果返回的邊界）。

任何類似的東西對我來說都是有意義的。

所以這個問題的本質：我是否試圖將一個方形釘嵌入圓孔中，或者有沒有一種方法可以使用Rx重新定義嵌套的異步調用？

Answer 1

您可以轉換服務的方法，以便它們返回IObservable而不是將回調作爲參數。在這種情況下連續工作流可以使用

 WidgetService.Query() 
      .SelectMany(
       widgets => 
       { 
        Widgets.Clear(); 
        foreach (var w in widgets) 
        { 
         Widgets.Add(w); 
        } 

        return WidgetService.Load(); 
       } 
      ) 
      .Do(
       defaultWidget => 
       { 
        if (defaultWidget != null) 
         Default = defaultWidget; 
       } 
      ) 
      .Subscribe(
       _ => { }, 
       e => { throw e; } 
      ); 

分別

然而IMO F＃異步操作會顯得更加清晰（樣品我認爲服務的方法返回異步>和異步 的SelectMany，這樣的事情...實施）。需要注意的是樣品中的帳戶沒有采取什麼線程正在修改的數據字段，在現實世界中的代碼，你應該注意這一點：

let load = async { 
      let! widgets = WidgetService.Query() 

      Widgets.Clear() 
      for w in widgets do 
       Widgets.Add(w) 

      let! defaultWidget = WidgetService.Load() 
      if defaultWidget <> null then 
       Default <- defaultWidget 

      return() 
     } 

    Async.StartWithContinuations(
     load, 
     ignore, // success continuation - ignore result 
     raise, // error continuation - reraise exception 
     ignore // cancellation continuation - ignore 
     ) 

EDITED

事實上，它是可以使用的技術

private IEnumerable<IObservable<object>> Intialize() 
    { 
     var widgetsList = WidgetService.Query().Start(); 
     yield return widgetsList; 

     Widgets.Clear(); 
     foreach (var w in widgetsList[0]) 
     { 
      Widgets.Add(w); 
     } 

     var defaultWidgetList = WidgetService.Load().Start(); 
     yield return defaultWidgetList; 

     if (defaultWidgetList[0] != null) 
      Default = defaultWidgetList[0]; 
    } 

    Observable 
     .Iterate(Intialize) 
     .Subscribe(
     _ => { }, 
     ex => { throw ex; } 
     ); 

Answer 2

你也可以做到這一點使用ReactiveXaml，但因爲你的CurrentWidget和窗口小部件都是可變的，你不能讓它乾淨（有一類CAL：迭代器，你在你的問題中提到導致ObservableAsPropertyHelper這將更新基於一個一個的IObservable財產和火RaisePropertyChanged）：

public class ViewModel 
{ 
    public ViewModel() 
    { 
     // These return a Func that wraps an async call in an IObservable<T> 
     // that always yields only one item (the result of the call) 
     var QueryAsObservable = Observable.FromAsyncCommand<IEnumerable<Widget>>(WebService.BeginQuery, WebService.EndQuery); 
     var LoadAsObservable = Observable.FromAsyncCommand<Widget>(WebService.BeginLoad, WebService.EndLoad); 

     // Create a new command 
     QueryAndLoad = new ReactiveAsyncCommand(); 

     // QueryAndLoad fires every time someone calls ICommand.Execute 
     // The .Do is the hacky part, for sync calls it's hidden by RegisterAsyncFunction 
     var async_results = QueryAndLoad.SelectMany(_ => QueryAsObservable()) 
             .Do(_ => DoTranslate.AsyncCompletedNotification.OnNext(new Unit())); 

     // Query up the Widgets 
     async_results.Subscribe(x => x.Run(Widgets.Add)); 

     // Now execute the Load 
     async_results.SelectMany(_ => LoadAsObservable()) 
        .Subscribe(x => CurrentWidget = x); 

     QueryAndLoad.Execute(); 
    } 

    public ReactiveAsyncCommand QueryAndLoad {get; private set; } 

    public ObservableCollection<Widget> Widgets {get; private set; } 

    public Widget CurrentWidget {get; set; } 
}