System.Reactive based Bonjour Browser

ObservableZeroconf.cs

public class ObservableZeroconf : IObservableZeroconf
{
    public ObservableZeroconf()
    {
    }

    public IObservable<Service> Search(string serviceType)
    {
        var browser = new NSNetServiceBrowser();
        browser.Schedule(NSRunLoop.Current, NSRunLoopMode.Default.ToString());

        var observableFoundService = Observable
            .FromEventPattern<NSNetServiceEventArgs>(
                h => browser.FoundService += h,
                h => browser.FoundService -= h)
            .Do(e => Console.WriteLine($"Found Service: {e.EventArgs.Service.Name} - More Comming: {e.EventArgs.MoreComing}"))
            .Select(e => e.EventArgs);

        var observableNotSearched = Observable
            .FromEventPattern<NSNetServiceErrorEventArgs>(
                h => browser.NotSearched += h,
                h => browser.NotSearched -= h)
            .Do(e => Console.WriteLine($"Not Searched"))
            .Select(e => e.EventArgs.Errors);

        var observableFoundWithError = observableFoundService
            .Materialize()
            .Merge(
                observableNotSearched
                    .Materialize()
                    .Select(x => Notification.CreateOnError<NSNetServiceEventArgs>(new Exception("Not Searched"))))
            .Dematerialize()
            .Synchronize();

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

                 var observableResolveError = Observable.FromEventPattern<NSNetServiceErrorEventArgs>(
                    h => s.Service.ResolveFailure += h,
                    h => s.Service.ResolveFailure -= h);

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

                s.Service.Resolve(0.0);
                return observableResolvedWithError.Subscribe(o);
            });

        var observableResolvedServices = observableFoundWithError
            .Do(s => Console.WriteLine($"Starting Resolve of {s.Service.Name}"))
            .Select(s => resolveService(s))
            .Merge()
            .TakeWhileInclusive(e => e.MoreComing)
            .Select(s => new Service(s.Service.Name, s.Service.Type, (int)s.Service.Port, GetAddresses(s.Service)));

        return Observable.Create<Service>(o =>
        {
            observableResolvedServices.Subscribe(o);

            Console.WriteLine("browser.SearchForServices");
            browser.SearchForServices(serviceType, "local");
            return Disposable.Create(() => {
                Console.WriteLine("browser.Stop()");
                browser.Stop();  
            });
        });
    }

    public IObservable<Service> Search(params string[] services) => services.Select(Search).Aggregate((o1, o2) => o1.Merge(o2));

    public IObservable<IList<Service[]>> SearchForAll(params string[] services) => services.Select(s => Search(s).ToArray()).Zip();

    private HashSet<IPAddress> GetAddresses(NSNetService service) => 
        service.Addresses
               .Select(CreateIPAddress)
               .Where(ip => !ip.ToString().Equals("0.0.0.0"))
               .ToHashSet();

    private static IPAddress CreateIPAddress(NSData data)
    {
        byte[] address = null;
        using (var ms = new MemoryStream())
        {
            data.AsStream().CopyTo(ms);
            address = ms.ToArray();
        }

        SocketAddress sa = new SocketAddress(AddressFamily.InterNetwork, address.Length);
        // do not overwrite the AddressFamily we provided
        for (int i = 2; i < address.Length; i++)
        {
            sa[i] = address[i];
        }
        IPEndPoint ep = new IPEndPoint(IPAddress.Any, 0);
        return (ep.Create(sa) as IPEndPoint).Address;
    }
}