如何使用Reactive Extensions在域服务中调用Silverlight RIA InvokeOperation？

Question

有谁知道我会如何使用Reactive Extensions调用RIA InvokeOperation？一旦呼叫全部完成，我需要在Silverlight中处理来自多个异步呼叫的一些数据。在我的测试中，我将几个字符串与一个查询的结果组合在一起，现在需要将调用的结果添加到RIA域服务调用方法中，并且被卡住了。

在RIA领域服务端（完成非RX风格）我简单的测试功能如下：

[Invoke] 
    public string DomainServiceFunction() 
    { 
     return "hello"; 
    } 

在客户端的代码，这个老派位调用该方法，而且是我想与RX实现部分：

private void CallDomainServiceFunction(object sender, RoutedEventArgs e) 
    { 
     DomainService1 DomainContext = new DomainService1(); 
     InvokeOperation invokeOp = DomainContext.DomainServiceFunction(OnInvokeCompleted, null); 
    } 

    private void OnInvokeCompleted(InvokeOperation invOp) 
    { 
     Debug.WriteLine(invOp.Value);//This prints "hello". 
    } 

我写了从多个来源的数据组合（这是我想补充的RIA调用的invokeOperation为好）一些测试代码。它使一个元组了几个字符串，并通过查询返回的实体：

private void PrintProperty1() 
    { 
     GISDomainContext DomainContext = new GISDomainContext(); 
     //Query the database to get information for a property using the folio number. 
     var loadProperty = from loadOperation in DomainContextExtensions 
       .LoadAsync(DomainContext, DomainContext.GetPropertyNoOwnersByFolioQuery("19401006.000")) 
          select loadOperation; 
     //Zip up the property entity with a string for testing purposes. 
     var Data = Observable.Return("a bit of text ") 
      .Zip((Observable.Return("some more text") 
      .Zip(loadProperty, (a, b) => Tuple.Create(a, b))), (a,b) => Tuple.Create(a,b)); 
     //Add a bit more stuff just to show it can be done. 

     //THIS IS WHERE I WOULD ALSO ZIP WITH THE VALUE RETURNED FROM AN InvokeOperation. 

     Data.Subscribe 
     (
      //When all the required data are prepared then proceed... 
      r => //On Next 
      { 
       Debug.WriteLine("OnNext: " + r.Item1 + ", " + r.Item2.Item1 + ", " + r.Item2.Item2.Entities.First().folio.ToString()); 
       //results: "OnNext: a bit of text , some more text, 19401006.000" 
       //At this point the data are all now available for further processing. 
      }, 
      r => //On Error 
      { 
       Debug.WriteLine("Error in PrintProperty1: " + r.ToString()); 
      }, 
      () =>//On Completed 
      { 
       Debug.WriteLine("Completed PrintProperty1"); 
      } 
     ); 
    } 

我怀疑FromAsyncPattern是关键，但显然Silverlight中隐藏的开始/结束调用FromAsyncPattern预计作为参数

从here报价：

“Silverlight的一个重要的注意

Silverlight的Web服务！生成的客户端代码有点烦人 - 它隐藏了 BeginXXXX/EndXXXX调用，大概是为了使Intellisense 更清洁。然而，他们不走了，你可以让他们回来的路上是 通过铸造MyCoolServiceClient对象到它的底层接口 （即LanguageServiceClient对象有它实现了一个产生 ILanguageServiceClient界面）”

任何建议？

Answer 1

其实，Silverlight是没有隐瞒任何东西。这些方法只是不通过的RIA Services工具生成的DomainContext来源的代理存在。

但这里有一个包装的扩展方法操作成IObservable：

public static class DomainContextExtensions 
{ 
    // The method takes in an invoke operation proxy method delegate 
    // and returns an observable sequence factory 
    public static Func<T1, IObservable<TResult>> FromInvokeOperation<T1, TResult> 
    (Func<T1, Action<InvokeOperation<TResult>>, object, InvokeOperation<TResult>> operationDelegate) 
    { 
    Contract.Requires<ArgumentNullException>(operationDelegate != null, "operationDelegate"); 
    Contract.Ensures(Contract.Result<Func<T1, IObservable<TResult>>>() != null); 

    return x1 => 
     { 
     // the subject is a storage for the result. 
     var subject = new AsyncSubject<TResult>(); 

     try 
     { 
      var invokeOperation = operationDelegate(x1, operation => 
      { 
       // handle operation results 

       if (operation.IsCanceled) 
       { 
       return; 
       } 

       if (operation.HasError) 
       { 
       subject.OnError(operation.Error); 
       operation.MarkErrorAsHandled(); 
       return; 
       } 

       Contract.Assume(operation.IsComplete); 
       subject.OnNext(operation.Value); 
       subject.OnCompleted(); 
      }, null); 

      // create the operation cancellation object 
      var invokeOperationCancellation = Disposable.Create(() => 
      { 
      // need to check if the operation has completed before the subscription is disposed 
      if (!invokeOperation.IsComplete && invokeOperation.CanCancel) 
      invokeOperation.Cancel(); // this might abort the web call to save bandwidth 
      }); 

      // construct a new observable that adds invoke operation cancellation upon dispose 
      return Observable.Create<TResult>(obs => new CompositeDisposable(invokeOperationCancellation, subject.Subscribe(obs))); 
     } 
     catch (Exception ex) 
     { 
      return Observable.Create<TResult>(obs => 
      { 
       obs.OnError(ex); 
       return Disposable.Empty; 
      }); 
     } 
     }; 
    } 
} 

这应该工作，虽然我没有测试它。

用法：

var context = ... // get your DomainContext 
var param = ... // operation parameter 
// This will create the observable: 
var o = DomainContextExtensions.FromInvokeOperation</*Parameter type goes here*/, /*Result type goes here*/>(context.YourOperationMethod)(param); 
o.Subscribe(...); // subscribe as you wish or build a query 

你必须编写其他的方法来支持调用操作使用不同数量的参数。