如何在 Rx 中处理多个事件（具有依赖关系）

Question

我有一些 WCF 调用，比如说 A、B、C、D、E。它们位于 SL 应用程序中（所以我想我必须小心线程等）。
我希望 B 在 A 完成后运行，而 C 可以与它​​们同时运行。我希望D在他们三个都完成后就跑。另外，基于一个条件（简单的 if...）我希望 E 也运行（与上述任何一个同时运行）。

我刚刚下载了 Rx，感觉它就是完成这项工作的工具。然而我还没有掌握所有的操作符和技巧等等。

这可以做到吗？如何？请解释为此所需的运算符。

编辑添加：为了简单起见，假设所有这些服务都采用字符串（或整数）作为参数，并返回实现复杂接口的东西，因此事件是简单的事件处理程序（非通用），然后我将它们转换为处理程序代码中的特殊接口。
有些输入在开始时就可用，有些来自先前服务的结果（例如，B 的输入字符串是根据 A 的结果计算的）

Answer 1

我已经考虑了编辑并制定了解决方案。

我不得不说 Microsoft 使得在 Silverlight 中编写简单的 WCF 调用变得非常困难。处理事件以从异步调用获取返回数据非常痛苦。

话虽如此，通过一些“复制粘贴”样板代码，Rx 在消除复杂性方面做得很好。

这是我想出的最终客户端代码：

        var service = new Service1Client();

        var abs =
            from a in service.AObservable("a")
            from b in service.BObservable(a)
            select new { a, b };

        var cs = service.CObservable("c");

        var abcs = abs.Zip(cs, (ab, c) => new { ab.a, ab.b, c });

        var ds =
            from abc in abcs
            from d in service.DObservable(abc.a, abc.b, abc.c)
            select String.Format(
                "A={0} B={1} C={2} D={3}",
                abc.a, abc.b, abc.c, d);

        var condition = true;

        var es = from e in condition
                    ? service.EObservable("e")
                    : Observable.Empty<string>()
                 select String.Format("E={0}", e);

        ds.Merge(es).ObserveOnDispatcher()
                    .Subscribe(r => this.label1.Content += " " + r);

一方面，这个代码并不像我希望的那么漂亮，但另一方面，我认为我无法让它变得更好。希望您能看到您的要求是如何得到满足的。

现在有坏消息 - 必须为每个服务代理创建扩展方法。根据 EventArgs 的定义方式，如果不使用特定服务引用类作为扩展方法的 this 参数，就无法获取结果。

从好的方面来说，虽然代码主要是“复制和粘贴”，几乎没有修改。

此外，您确实需要创建与您使用的参数一样多的扩展方法加一。

以下是我需要为示例代码定义的扩展方法：

public static class Service1ClientEx
{
    public static Func<IObservable<R>> ToObservableFunc<EA, R>(this Service1Client service, Action<Service1Client> async, Action<Service1Client, EventHandler<EA>> ah, Action<Service1Client, EventHandler<EA>> rh, Func<EA, R> resultSelector)
        where EA : System.ComponentModel.AsyncCompletedEventArgs
    {
        return () => Observable.Create<R>(o =>
        {
            var response =
                from ep in Observable.FromEventPattern<EA>(h => ah(service, h), h => rh(service, h))
                select resultSelector(ep.EventArgs);

            var subscription = response.Take(1).Subscribe(o);

            async(service);

            return subscription;
        }).ObserveOn(Scheduler.ThreadPool);
    }

    public static Func<P0, IObservable<R>> ToObservableFunc<P0, EA, R>(this Service1Client service, Action<Service1Client, P0> async, Action<Service1Client, EventHandler<EA>> ah, Action<Service1Client, EventHandler<EA>> rh, Func<EA, R> resultSelector)
        where EA : System.ComponentModel.AsyncCompletedEventArgs
    {
        return p0 => Observable.Create<R>(o =>
        {
            var response =
                from ep in Observable.FromEventPattern<EA>(h => ah(service, h), h => rh(service, h))
                select resultSelector(ep.EventArgs);

            var subscription = response.Take(1).Subscribe(o);

            async(service, p0);

            return subscription;
        }).ObserveOn(Scheduler.ThreadPool);
    }

    public static Func<P0, P1, IObservable<R>> ToObservableFunc<P0, P1, EA, R>(this Service1Client service, Action<Service1Client, P0, P1> async, Action<Service1Client, EventHandler<EA>> ah, Action<Service1Client, EventHandler<EA>> rh, Func<EA, R> resultSelector)
        where EA : System.ComponentModel.AsyncCompletedEventArgs
    {
        return (p0, p1) => Observable.Create<R>(o =>
        {
            var response =
                from ep in Observable.FromEventPattern<EA>(h => ah(service, h), h => rh(service, h))
                select resultSelector(ep.EventArgs);

            var subscription = response.Take(1).Subscribe(o);

            async(service, p0, p1);

            return subscription;
        }).ObserveOn(Scheduler.ThreadPool);
    }

    public static Func<P0, P1, P2, IObservable<R>> ToObservableFunc<P0, P1, P2, EA, R>(this Service1Client service, Action<Service1Client, P0, P1, P2> async, Action<Service1Client, EventHandler<EA>> ah, Action<Service1Client, EventHandler<EA>> rh, Func<EA, R> resultSelector)
        where EA : System.ComponentModel.AsyncCompletedEventArgs
    {
        return (p0, p1, p2) => Observable.Create<R>(o =>
        {
            var response =
                from ep in Observable.FromEventPattern<EA>(h => ah(service, h), h => rh(service, h))
                select resultSelector(ep.EventArgs);

            var subscription = response.Take(1).Subscribe(o);

            async(service, p0, p1, p2);

            return subscription;
        }).ObserveOn(Scheduler.ThreadPool);
    }

    public static Func<P0, P1, P2, P3, IObservable<R>> ToObservableFunc<P0, P1, P2, P3, EA, R>(this Service1Client service, Action<Service1Client, P0, P1, P2, P3> async, Action<Service1Client, EventHandler<EA>> ah, Action<Service1Client, EventHandler<EA>> rh, Func<EA, R> resultSelector)
        where EA : System.ComponentModel.AsyncCompletedEventArgs
    {
        return (p0, p1, p2, p3) => Observable.Create<R>(o =>
        {
            var response =
                from ep in Observable.FromEventPattern<EA>(h => ah(service, h), h => rh(service, h))
                select resultSelector(ep.EventArgs);

            var subscription = response.Take(1).Subscribe(o);

            async(service, p0, p1, p2, p3);

            return subscription;
        }).ObserveOn(Scheduler.ThreadPool);
    }

    public static IObservable<string> AObservable(this Service1Client service, string data)
    {
        return service
            .ToObservableFunc<string, ACompletedEventArgs, string>((s, p0) => s.AAsync(p0), (s, h) => s.ACompleted += h, (s, h) => s.ACompleted -= h, ea => ea.Result)
            .Invoke(data);
    }

    public static IObservable<string> BObservable(this Service1Client service, string data)
    {
        return service
            .ToObservableFunc<string, BCompletedEventArgs, string>((s, p0) => s.BAsync(p0), (s, h) => s.BCompleted += h, (s, h) => s.BCompleted -= h, ea => ea.Result)
            .Invoke(data);
    }

    public static IObservable<string> CObservable(this Service1Client service, string data)
    {
        return service
            .ToObservableFunc<string, CCompletedEventArgs, string>((s, p0) => s.CAsync(p0), (s, h) => s.CCompleted += h, (s, h) => s.CCompleted -= h, ea => ea.Result)
            .Invoke(data);
    }

    public static IObservable<string> DObservable(this Service1Client service, string dataA, string dataB, string dataC)
    {
        return service
            .ToObservableFunc<string, string, string, DCompletedEventArgs, string>((s, p0, p1, p2) => s.DAsync(p0, p1, p2), (s, h) => s.DCompleted += h, (s, h) => s.DCompleted -= h, ea => ea.Result)
            .Invoke(dataA, dataB, dataC);
    }

    public static IObservable<string> EObservable(this Service1Client service, string data)
    {
        return service
            .ToObservableFunc<string, ECompletedEventArgs, string>((s, p0) => s.EAsync(p0), (s, h) => s.ECompleted += h, (s, h) => s.ECompleted -= h, ea => ea.Result)
            .Invoke(data);
    }
}

就像我说的，很痛苦。

您当然不必创建这些类型的扩展方法来在 Silverlight 中将 Rx 与 WCF 一起使用，但我的感觉是，一旦创建了它们，客户端代码就会变得非常非常容易使用。

如果您需要更多解释，请告诉我。