c# - 使用 Rx 框架使用 void AsyncMethod(Action callback) 模式进行异步调用

Question

我看过很多关于如何在 Rx 框架中使用 Observable.FromAsyncPattern() 来简化异步调用的示例，但我使用的接口(interface)不使用 IAsyncResult BeginXXX/EndXXX 的标准异步模式(IAsyncResult)，所以这对我不起作用。

我正在使用的库公开了带有回调模式的异步函数:

void GetAllObjects(Action<List<Object>> callback)

在一个理想的世界里，我想把这个变成:

var isLoadingUsers = true;
var isLoadingSystems = true;
var isLoadingCustomers = true;
var isLoadingRules = true;

mClient.GetAllUsers(UsersCallback);
mClient.GetAllCustomers(CustomersCallback);
mClient.GetAllRules(RulesCallback);

// set the IsLoadingXXX variables to false in callbacks
// once all are false

mClient.GetAllSystems(SystemsCallback);

像这样:

var o = Observable.ForkJoin(
                     Observable.Start(GetAllUsers()),
                     Observable.Start(GetAllCustomers()),
                     Observable.Start(GetAllRules())
                    ).Finally(() => GetAllSystems);

如何将这种模式转化为返回 IObservable 的东西？

Answer 1

Func<IObservable<TRet>> FromListCallbackPattern(Action<Action<List<TRet>>> function)
{
    return () => {
        // We use a ReplaySubject so that if people subscribe *after* the
        // real method finishes, they'll still get all the items
        ret = new ReplaySubject<TRet>();

        function((list) => {
            // We're going to "rebroadcast" the list onto the Subject
            // This isn't the most Rx'iest way to do this, but it is the most
            // comprehensible :)
            foreach(var v in list) {
                ret.OnNext(v);
            }
            ret.OnCompleted();
        });

        return ret;
    };
}

现在，您可以执行以下操作:

var getAllUsers = FromListCallbackPattern(mClient.GetAllUsers);
getAllUsers().Subscribe(x => /* ... */);