将基于事件的 API 转换为 Rx.Net

Question

我正在尝试将现有的基于事件的 API 转换为 Reactive Observable API。我正在使用的具体 API 是 Xamarin.iOS 中的 NSNetServiceBrowser。此 API 让您可以使用 Zeroconf/Bonjour 浏览网络设备。但是，该问题适用于任何此类 API。

NsNetServiceBrowser 提供各种感兴趣的事件： - FoundService - NotSearched - ServiceRemoved

发现服务时引发FoundService事件，搜索失败时引发NotSearched。

我想将 FoundService 和 NotSerched 事件组合成 NSNetService 的 observable。

我当前的实现如下所示：

public IObservable<NSNetService> Search()
{
    var foundObservable = Observable
        .FromEventPattern<NSNetServiceEventArgs>(
            h => serviceBrowser.FoundService += h,
            h => serviceBrowser.FoundService -= h)
        .Select(x => x.EventArgs);

    var notSearchedObservable = Observable
        .FromEventPattern<NSNetServiceErrorEventArgs>(
            h => serviceBrowser.NotSearched += h,
            h => serviceBrowser.NotSearched -= h)
        .Select(x => x.EventArgs);

    var serviceObservable = Observable.Create(
        (IObserver<NSNetServiceEventArgs> observer) =>
        {
            notSearchedObservable.Subscribe(n =>
            {
                string errorMessage = $"Search for {serviceType} failed:";
                foreach (var kv in n.Errors)
                {
                    log.Error($"\t{kv.Key}: {kv.Value}");
                    errorMessage += $" ({kv.Key}, {kv.Value})";
                }
                observer.OnError(new Exception(errorMessage));
            });
            foundObservable.Subscribe(observer);
            return System.Reactive.Disposables.Disposable.Empty;
    }).Select(x => x.Service);

    serviceBrowser.SearchForServices(serviceType, domain);
    return serviceObservable;
}

代码看起来很笨拙，我有一种直觉，我没有正确使用System.Reactive？有没有一种更优雅的方式来组合事件对，其中一个正在产生，另一个是发出错误信号？这是 .NET 中现有的基于事件的 API 中的常见模式。

---

这是一个小型控制台应用程序（仅取决于 System.Reactive），说明了我想要 Reactify 的 API 类型：

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Threading.Tasks;

namespace ReactiveLearning
{
    class Program
    {
        static void Main(string[] args)
        {
            var browser = new ServiceBrowser();

            var observableFound =
                Observable.FromEventPattern<ServiceFoundEventArgs>(
                    h => browser.ServiceFound += h,
                    h => browser.ServiceFound -= h)
                .Select(e => e.EventArgs.Service);

            var observableError =
                Observable.FromEventPattern<ServiceSearchErrorEventArgs>(
                    h => browser.ServiceError += h,
                    h => browser.ServiceError -= h);

            var foundSub = observableFound.Subscribe(s =>
            {
                Console.WriteLine($"Found service: {s.Name}");
            }, () => 
            {
                Console.WriteLine("Found Completed");
            });

            var errorSub = observableError.Subscribe(e =>
            {
                Console.WriteLine("ERROR!");
            }, () => 
            {
                Console.WriteLine("Error Completed");
            });

            browser.Search();

            Console.ReadLine();

            foundSub.Dispose();
            errorSub.Dispose();

            Console.WriteLine();
        }
    }

    class ServiceBrowser
    {
        public EventHandler<ServiceFoundEventArgs> ServiceFound;
        public EventHandler<ServiceSearchErrorEventArgs> ServiceError;

        public void Search()
        {
            Task.Run(async () =>
            {
                for (var i = 0; i < 5; ++i)
                {
                    await Task.Delay(1000);
                    ServiceFound?.Invoke(this, new ServiceFoundEventArgs(new Service($"Service {i}")));
                }

                var r = new Random();
                if (r.NextDouble() > 0.5)
                {
                    ServiceError?.Invoke(this, new ServiceSearchErrorEventArgs());
                }
            });
        }
    }

    class ServiceFoundEventArgs : EventArgs
    {
        public Service Service { get; private set;  }
        public ServiceFoundEventArgs(Service service) => Service = service;
    }

    class ServiceSearchErrorEventArgs : EventArgs {}

    class Service
    {
        public event EventHandler<EventArgs> AddressResolved;
        public event EventHandler<EventArgs> ErrorResolvingAddress;
        public string Name { get; private set; }
        public string Address { get; private set; }
        public Service(string name) => Name = name;

        public void ResolveAddress()
        {
            Task.Run(async () =>
            {
                await Task.Delay(500);
                var r = new Random();
                if (r.NextDouble() > 0.5)
                {                    
                    Address = $"http://{Name}.com";
                    AddressResolved?.Invoke(this, EventArgs.Empty);
                }
                else
                {
                    ErrorResolvingAddress?.Invoke(this, EventArgs.Empty);
                }
            });
        }
    }
}

Answer 1

感谢您提供出色的示例代码。您需要使用出色的 Materialize 和 Dematerialize 运算符。方法如下：

var observableFoundWithError =
    observableFound
        .Materialize()
        .Merge(
            observableError
                .Materialize()
                .Select(x =>
                    Notification
                        .CreateOnError<Service>(new Exception("Error"))))
        .Dematerialize()
        .Synchronize();

using (observableFoundWithError.Subscribe(
    s => Console.WriteLine($"Found service: {s.Name}"),
    ex => Console.WriteLine($"Found error: {ex.Message}"),
    () => Console.WriteLine("Found Completed")))
{
    browser.Search();
    Console.ReadLine();
}

Materialize() 运算符将IObservable<T> 转换为IObservable<Notification<T>>，这允许通过OnNext 调用发出标准OnError 和OnCompleted。您可以使用Notification.CreateOnError<T>(new Exception("Error")) 构造可观察的元素，您可以使用Dematerialize() 将其转回IObservable<T>。

我已经抛出 Synchronize() 以确保您创建了一个有效的 observable。 Materialize() 的使用确实让您可以构建不遵循常规可观察合同的可观察对象。 Synchronize() 所做的部分工作就是确保只有一个 OnError 和一个 OnCompleted，并删除出现在两者之一之后的任何 OnNext。

---

试试这个作为在 cmets 中做你想做的事情的一种方式：

static void Main(string[] args)
{
    var browser = new ServiceBrowser();

    var observableFound =
        Observable.FromEventPattern<ServiceFoundEventArgs>(
            h => browser.ServiceFound += h,
            h => browser.ServiceFound -= h)
        .Select(e => e.EventArgs.Service);

    var observableError =
        Observable.FromEventPattern<ServiceSearchErrorEventArgs>(
            h => browser.ServiceError += h,
            h => browser.ServiceError -= h);

    var observableFoundWithError = observableFound
        .Materialize()
        .Merge(
            observableError
                .Materialize()
                .Select(x => Notification.CreateOnError<Service>(new Exception("Error"))))
        .Dematerialize()
        .Synchronize();

    Func<Service, IObservable<Service>> resolveService = s =>
        Observable.Create<Service>(o =>
        {
            var observableResolved = Observable.FromEventPattern<EventArgs>(
                h => s.AddressResolved += h,
                h => s.AddressResolved -= h);

            var observableResolveError = Observable.FromEventPattern<EventArgs>(
                h => s.ErrorResolvingAddress += h,
                h => s.ErrorResolvingAddress -= h);

            var observableResolvedWithError =
                observableResolved
                    .Select(x => s)
                    .Materialize()
                    .Merge(
                        observableResolveError
                        .Do(e => Console.WriteLine($"Error resolving: {s.Name}"))
                        .Materialize()
                        .Select(x => Notification.CreateOnError<Service>(new Exception($"Error resolving address for service: {s.Name}"))))
                    .Dematerialize()
                    .Synchronize();

            s.ResolveAddress();

            return observableResolvedWithError.Subscribe(o);
        });

    using (
        observableFoundWithError
            .Select(s => resolveService(s))
            .Switch()
            .Subscribe(
                s => Console.WriteLine($"Found and resolved service: {s.Name} ({s.Address})"),
                ex => Console.WriteLine($"Found error: {ex.Message}"),
                () => Console.WriteLine("Found Completed")))
    {
        browser.Search();
        Console.ReadLine();
    }
}

public class ServiceBrowser
{
    public event EventHandler<ServiceFoundEventArgs> ServiceFound;
    public event EventHandler<ServiceSearchErrorEventArgs> ServiceError;

    public void Search() { }
}

public class Service
{
    public event EventHandler<EventArgs> AddressResolved;
    public event EventHandler<EventArgs> ErrorResolvingAddress;

    public string Name;
    public string Address;

    public void ResolveAddress() { }
}

public class ServiceFoundEventArgs : EventArgs
{
    public Service Service;
}

public class ServiceSearchErrorEventArgs : EventArgs
{

}

我可能需要进行一些调整——可能是一个Observable.Delay。让我知道它是否有效。