Come organizzo queste chiamate usando Reactive Extensions (Rx) in Silverlight?

Question

Ho alcune chiamate che devono essere eseguite in sequenza. Prendi in considerazione un IService con metodo Query e Load. La query fornisce un elenco di widget e il carico fornisce un widget "predefinito". Quindi, il mio servizio è simile a questo.

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

Con questo in mente, ecco un abbozzo del modello di visualizzazione:

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

Quello che mi piacerebbe fare è semplificare il flusso di lavoro sequenziale di chiamare query e quindi il valore di default. Forse il modo migliore per farlo è annidato con espressioni lambda come ho mostrato, ma ho pensato che potrebbe esserci un modo più elegante con Rx. Non voglio usare Rx per motivi di Rx, ma se mi può permettere di organizzare la logica sopra, quindi è più facile leggere/mantenere nel metodo, ne approfitterò. Idealmente, qualcosa come:

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

Con la biblioteca di potenza threading, mi piacerebbe fare qualcosa di simile:

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

che lo rende molto più evidente che il flusso di lavoro è sequenziale ed elimina nidificazione (le rese sono parte dell'enumeratore asincrono e sono limiti che bloccano fino a quando i risultati non tornano).

Qualsiasi cosa simile avrebbe senso per me.

Quindi l'essenza della domanda: sto tentando di inserire un piolo quadrato in un buco rotondo, o c'è un modo per ridefinire le chiamate asincrone annidate usando Rx?

Answer 1

È possibile convertire i metodi di servizio in modo che restituiscano IObservable anziché richiamare come parametro. In questo caso il flusso di lavoro sequenziale può essere implementato usando SelectMany, qualcosa di simile ...

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

Tuttavia IMO F # asyncs osserveranno molto più chiaro (nel campione di presumo che i metodi di servizio restituisce Async> e Async rispettivamente). Si noti che il campione non prende in considerazione ciò che sta modificando filo campi di dati, nel codice mondo reale si dovrebbe prestare attenzione a questo:

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 
     ) 

a cura

In realtà è possibile utilizzare la tecnica con iteratori lei ha citato nella sua domanda:

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

si potrebbe anche farlo usando ReactiveXaml, anche se da quando il vostro CurrentWidget e widget sono entrambi mutevole, non è possibile effettuare il più pulito (c'è una classe cal portato ObservableAsPropertyHelper che aggiornerà una proprietà basata su un IObservable e fuoco la 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; } 
}