如何使用Silverlight中的Reactive Extensions(Rx)组织这些调用？

Question

我有一些必须按顺序执行的调用.考虑具有Query和Load方法的IService.Query提供了一个小部件列表,并且load提供了一个"默认"小部件.因此,我的服务看起来像这样.

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，而不是将回调作为参数。在这种情况下，可以使用SelectMany来实现顺序工作流程，如下所示......

        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# 异步看起来会更加清晰（在示例中，我假设服务方法分别返回 Async> 和 Async）。请注意，示例没有考虑哪个线程正在修改数据字段，在实际代码中您应该注意这一点：

    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
        )

编辑

事实上，可以将技术与您在问题中提到的迭代器一起使用：

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