¿Cómo organizaría estas llamadas usando Reactive Extensions (Rx) en Silverlight?

Question

Tengo algunas llamadas que se deben ejecutar secuencialmente. Considere un IService que tiene un método Query y un Load. La consulta proporciona una lista de widgets y la carga proporciona un widget "predeterminado". Por lo tanto, mi servicio se ve así.

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

Con esto en mente, aquí es un esbozo del modelo de vista:

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

Lo que me gustaría hacer es simplificar el flujo de trabajo secuencial de llamar a consulta y el valor por defecto. Tal vez la mejor manera de hacerlo es anidando con expresiones lambda como he mostrado, pero pensé que podría haber una manera más elegante con Rx. No quiero usar Rx por Rx, pero si me permite organizar la lógica anterior para que sea más fácil de leer/mantener en el método, lo aprovecharé. Idealmente, algo así como:

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

Con la librería de hilos de potencia, que haría algo como:

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

Eso hace que sea mucho más evidente que el flujo de trabajo es secuencial y elimina anidación (los rendimientos son parte del enumerador asincrónico y son los límites que bloquean hasta que los resultados vuelven).

Cualquier cosa similar tendría sentido para mí.

Entonces, la esencia de la pregunta: ¿estoy tratando de encajar una clavija cuadrada en un agujero redondo, o hay alguna manera de redefinir las llamadas asíncronas anidadas usando Rx?

Answer 1

Puede convertir métodos de servicio para que devuelvan IObservable en lugar de tomar la devolución de llamada como parámetro. En este caso, el flujo de trabajo secuencial puede ser implementado utilizando SelectMany, algo como esto ...

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

Sin embargo OMI F # asyncs se verá mucho más claro (en la muestra Asumo que los métodos de servicio devuelve asíncrono> y asíncrono, respectivamente) Tenga en cuenta que la muestra no toma en cuenta lo que el hilo está modificando los campos de datos, en el código del mundo real se debe prestar atención a esto:

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 
     ) 

editado

De hecho, es posible utilizar la técnica con iteradores que usted ha mencionado en su pregunta:

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

también puede hacerlo utilizando ReactiveXaml, aunque desde su CurrentWidget y widgets son mutables tanto, no se puede hacer lo más limpio (hay una clase de cal conducido ObservableAsPropertyHelper que permitirá actualizar una propiedad basada en un IObservable y despedir al 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; } 
}