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

Question

我正在使用与回调一起使用的网络 API。所以基本上，我有一堆方法调用需要在这个 3rd 方库中使用，它们看起来像这样：

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

我发现代码有点古怪，因为我的所有方法都依赖于来自这个 3rd 方 API 的网络资源，也需要自己实现回调。例如，在我的主要场景中，我可能想要获取玩家信息，我的代码如下所示：

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

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

我最近接触了 RX，并在我的代码中大量使用它。我已经阅读了一些关于 async/await 的内容。我已经做了很多实验，尤其是使用 RX，并尝试使用 Observable.FromAsync() 但无法让它工作..

我在这里缺少什么？如何编写更简洁且不需要使用 RX 或 async/await 回调的代码？以下是我正在寻找的伪代码：

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

Answer 1

以下是使用 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 2

我不确定 RX。但是，您可以将基于回调的 SendNetworkRequest 转换为可等待的任务，如下所示：

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

现在您可以更自然地使用带有 async/await 的 SendNetworkRequestAsync

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) 创建一个可观察对象，该可观察对象基于预期对传递给 SendNetworkRequest 方法的 Action<Response> callback 的单个调用。

唯一的问题是调用发生在当前线程上直到完成。如果您希望这段代码在后台线程上运行，但将结果返回给当前线程，那么您可以这样做：

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