Action的 .net Reactive 扩展中的扩展方法是？

Question

我有两个采用 Action 回调的异步方法。我想知道 Rx 中是否有针对操作的扩展？

我的目标是等到两个回调被调用然后做一些处理？

Answer 1

这是来自 Jesse 和我即将出版的书，但是给你，这是免费赠品:

public Func<T1, IObservable<TRet>> FromCallbackPattern<T1, TRet>(Action<T1, Action<TRet>> originalMethod)
{
    return new Func<T1, IObservable<TRet>>((param1) => {
        var subject = new AsyncSubject<TRet>();

        try {
            return originalMethod(param1, (result) => {
                subject.OnNext(result);
                subject.OnCompleted();
            });
        } catch (Exception ex) {
            subject.OnError(ex);
        }

        return subject;
    });
}

下面是使用方法：

// Here's a sample method that follows the callback pattern
public void DownloadPageTextAsync(string url, Action<string> callback);

var dlPageRx = FromCallbackPattern(DownloadPageTextAsync);

dlPageRx("http://www.jesseliberty.com")
    .Subscribe(pageText => Console.WriteLine(pageText));