Comment organiser ces appels à l'aide de Reactive Extensions (Rx) dans Silverlight?

Question

J'ai certains appels qui doivent s'exécuter séquentiellement. Considérez un IService qui a une méthode Query et une méthode Load. La requête donne une liste de widgets, et la charge fournit un widget "par défaut". Par conséquent, mon service ressemble à ceci.

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

Dans cet esprit, voici une esquisse du modèle de vue:

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

Ce que je voudrais faire est de simplifier le flux de travail séquentiel d'appeler la requête et la valeur par défaut. Peut-être que la meilleure façon de le faire est imbriquée avec les expressions lambda comme je l'ai montré, mais j'ai pensé qu'il pourrait y avoir une manière plus élégante avec Rx. Je ne veux pas utiliser Rx dans l'intérêt de Rx, mais si cela peut me permettre d'organiser la logique ci-dessus pour qu'il soit plus facile à lire/maintenir dans la méthode, j'en profiterai. Idéalement, quelque chose comme:

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

Avec la bibliothèque de threads de puissance, je ferais quelque chose comme:

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

qui le rend beaucoup plus évident que le flux de travail est séquentiel et élimine la nidification (les rendements sont partie de l'énumérateur asynchrone et sont des limites qui bloquent jusqu'à ce que les résultats reviennent).

Tout ce qui est similaire aurait du sens pour moi. Donc, l'essence de la question: est-ce que j'essaie d'insérer une cheville carrée dans un trou rond, ou y a-t-il un moyen de redéfinir les appels asynchrones imbriqués en utilisant Rx?

Answer 1

Vous pouvez convertir des méthodes de service afin qu'elles renvoient IObservable au lieu de prendre le rappel en tant que paramètre. Dans ce cas flux de travail séquentiel peut être mis en œuvre à l'aide SelectMany, quelque chose comme ça ...

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

Cependant l'OMI F # asyncs sera beaucoup plus clair (dans l'échantillon, je suppose que les méthodes de retours Service Async> et Async respectivement). Notez que l'échantillon ne prend pas en compte ce fil est de modifier les champs de données, dans le code monde réel, vous devez faire attention à ceci:

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 
     ) 

ÉDITÉE

En fait, il est possible d'utiliser la technique avec itérateurs vous avez mentionné dans votre question:

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

vous pouvez aussi le faire en utilisant ReactiveXaml, bien que depuis votre CurrentWidget et Widgets sont tous les deux mutable, vous ne pouvez pas le rendre aussi propre (il y a une classe cal conduit ObservableAsPropertyHelper qui mettra à jour une propriété basée sur un IObservable et feu le 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; } 
}