如何使用Observable或async/await包装异步回调？

Question

我正在使用与回调一起工作的网络API。所以基本上，我有一大堆的方法调用，我需要这个第三方库，像这样使用：

void SendNetworkRequest(string requestType, Action<Response> callback) 

我找到的代码变得有点古怪，因为我所有的方法依赖于网络来自第三方API的资源也需要自行实现回调。举例来说，在我的主场景我可能想要得到的球员资料和我的代码看起来是这样的：

void InitMainScene() 
{ 
     _networkHelper.SendNetworkRequest("GetPlayersInfo",OnPlayerInfoResponse); 
} 

void OnPlayerInfoResponse(Response response) 
{ 
    _playerInfo = response.Info; 
} 

我最近卷入了RX和我的代码abundently使用它我。我读过一些关于异步/等待的内容。我已经尝试了很多，尤其是使用RX，并尝试使用Observable.FromAsync（），但无法让它工作。

我在这里错过了什么？我如何编写更清晰的代码，并且不需要使用RX或异步/等待的回调？以下是我正在寻找的psuedocode：

 void InitMainScene() 
     { 
      _playerInfo = Overservable.DoMagicThing<Response>(() => { 
_networkHelper.SendNetworkRequest("GetPlayersInfo",(r) =>{return r;}); }); 
     } 

Answer 1

我不确定RX。但是，您可以基于回调的SendNetworkRequest转换为这样的awaitable任务：

void SendNetworkRequest(string requestType, Action<Response> callback) 
     { 
      // This code by third party, this is here just for testing 
      if (callback != null) 
      { 
       var result = new Response(); 
       callback(result); 
      } 
     } 

     Task<Response> SendNetworkRequestAsync(string requestType) 
     { 
      return Task.Run(() => 
      { 
       var t = new TaskCompletionSource<Response>(); 

       SendNetworkRequest(requestType, s => t.TrySetResult(s)); 

       return t.Task; 
      }); 
     } 

现在你可以消耗SendNetworkRequestAsync与异步/更自然地等待

Answer 2

这里是如何，其中Rx做到这一点。我换出Response为string，这样我可以编译和测试，但琐碎换回来：

public IObservable<string> SendNetworkRequestObservable(string requestType) 
{ 
    return Observable.Create<string>(observer => 
    { 
     SendNetworkRequest(requestType, s => observer.OnNext(s)); 
     observer.OnCompleted(); 
     return Disposable.Empty; 
    }); 
} 

// Define other methods and classes here 
public void SendNetworkRequest(string requestType, Action<string> callback) 
{ 
    callback(requestType); // implementation for testing purposes 
} 

Answer 3

这里是我采取的其中Rx这样做。

里面的NetworkHelper类添加这个方法：

public IObservable<Response> SendNetworkRequestObservable(string requestType) 
{ 
    return 
     Observable 
      .Create<Response>(observer => 
      { 
       Action<Response> callback = null; 
       var callbackQuery = 
        Observable 
         .FromEvent<Response>(h => callback += h, h => callback -= h) 
         .Take(1); 
       var subscription = callbackQuery.Subscribe(observer); 
       this.SendNetworkRequest(requestType, callback); 
       return subscription; 
      }); 
} 

这将创建一个可观察的是内部使用Observable.FromEvent(...).Take(1)创建基于期待的Action<Response> callback传递给SendNetworkRequest方法的单一调用可观察到的。

唯一的问题是，该调用发生在当前线程的完成权。如果你想这个代码在后台线程运行，但结果返回到当前线程，那么你可以这样做：

public IObservable<Response> SendNetworkRequestObservable(string requestType) 
{ 
    return 
     Observable 
      .Start(() => 
       Observable 
        .Create<Response>(observer => 
        { 
         Action<Response> callback = null; 
         var callbackQuery = 
          Observable 
           .FromEvent<Response>(h => callback += h, h => callback -= h) 
           .Take(1); 
         var subscription = callbackQuery.Subscribe(observer); 
         this.SendNetworkRequest(requestType, callback); 
         return subscription; 
        })) 
      .Merge(); 
} 

无论哪种方式，你会这样称呼它：

var _playerInfo = 
    _networkHelper 
     .SendNetworkRequestObservable("GetPlayersInfo") 
     .Wait(); 

只是一个侧面说明：您的DoMagicThing是一个同步呼叫。在Rx中，这是一个.Wait()呼叫，但最好是在可以时使用正常的.Subscribe。

---

按照注释，你可以用async做到这一点：

async void InitMainScene() 
{ 
    _playerInfo = await _networkHelper.SendNetworkRequestObservable("GetPlayersInfo"); 
}