System.Reactive.Joins 指南

Question

我正在寻找 System.Reactive.Joins 的介绍/一些文档，其中包括 Pattern、Plan、QueryablePattern 和 QueryablePlan 类。 Google 什么都没有出现（“System.Reactive.Joins”），MSDN 什么都没有，here 没有示例，this question 的优秀资源也没有覆盖这个命名空间。

有人指点一下吗？

Answer 1

找到了一个很好的 SO question 来显示用法，但对我来说，Plan and Pattern 的总体目的是创建一个可观察的组合单元，而不是组合的可观察单元。语义，我知道，但对我来说，使用这种语法似乎比其他各种“加入”方法更容易一些。它允许您将连接与投影完全分开，因此您可以存储中间计划并随时将它们与其他可观察对象组合。

例如：

// Suppose we have observables o1, o2, ..., o9. 
// All IObservable<int>.

var o1and2 = o1.And(o2);  // Store this bad boy for later use. Pattern<int, int>

var o5and6and9 = o5
                .And(o6)
                .And(o9)
                .Then((t1, t2, t3) => t1 + t2 + t3);  // Plan<int>

var o3and7 = o3
            .And(o7)
            .Then((t1, t2) => string.Format("Result: {0}", t1 + t2)); // Plan<string>

var o12ando8and6 = o1and2
                  .And(o8)
                  .And(o6)
                  .Then((t1, t2, t3, t4) => ((decimal) t1, t2, t3.ToString(), t4));   
                  // Plan<(decimal, int, string, int)>

// "When" groups similar results together.
// It will fire when any of the Patterns give a result.

var obs1 = Observable
          .When(o1and2.Then((t1,t2) => t1+t2), o5and6and9); // IObservable<int>

var obs2 = Observable.When(o3and7);       // IObservable<string>
var obs3 = Observable.When(o12ando8and6); // IObservable<(decimal, int, string,int)>

SO 文章： Reactive Extensions for .NET (Rx): Take action once all events are completed

另外，找到了一个 RX 文档，它实际上有助于理解如何使用它：http://www.clipcode.net/mentoring/RxReferenceLibrary.pdf [dead]

Answer 2

这是我找到的唯一内容：Join Patterns in Rx。我还想查看有关这些主题的其他资源。

Answer 3

Reactive Extensions for .NET (Rx) ^{blogs.msdn.com}

...然后使用连接模式等待三个结果中的前两个返回。

How to join multiple IObservable sequences? ^{stackoverflow.com}

如何在 v.2838 中使用新的 Join 运算符...

Introduction to the Reactive Extensions for JavaScript – New Release and Joins ^{weblogs.asp.net}

... JavaScript 的响应式扩展，其中包括我最近一直在谈论的许多更改，包括第三方库集成、我在之前的帖子中介绍的聚合以及今天帖子主题的连接。

System.Reactive.Joins Namespace ^{msdn.microsoft.com}

Microsoft 对 System.Reactive.Joins 命名空间的定义。

Answer 4

除了我提到的示例here（现在更新为使用 Observable.When 而不是 Observable.Join），对我来说，一般的概念似乎是当你想要组成事件本身，而不是事件的内容。因此，当我想做与正在观察的数据相关的事情时，我会使用其他查询运算符，例如 SelectMany。当我想纯粹为了响应哪些事件以及哪些事件模式以特定顺序触发时做某事，那么我会使用连接模式。现在，这并不是说您不能结合其他查询运算符，例如 Where 和 Select 来过滤和投影 IObservable 中的一些值，这些值稍后将用于连接模式，但我认为这是一个很好的通用方式来看看。

我确实尝试了一些例子，但我并没有真正想出更好的......只是相同概念的更复杂的案例。您希望对可能具有复杂关系的相关事件模式进行分组，这些关系在事件级别而不是 IObservable 的内容中有意义。

Answer 5

使用 Reflector（或类似工具）也不错，
目前，浏览源代码对于 Rx 来说是非常宝贵的工具。您可以看到它实现了什么、如何实现 - 并与其他标准 Rx 模式进行比较。
此外，也许您可​​以发布一个问题，即您要解决的问题 - 然后从那里开始（其他人可能有更多想法）。
编辑： Rx 没有源代码因此。然而，即使是 Rx 的人也建议通过使用 Reflector 或类似工具反汇编来分析库。例如。您可以免费使用 ILSpy，下载 Rx-Providers Nuget 或者我猜你有，将 System.Reactive.Providers.dll 和 System.Reactive.dll 加载到其中并享受:)。它既不完美也不是官方的，也不编译，至少很容易 - 但你可以很好地了解事情的运作方式，它对我帮助很大，非常宝贵！

Answer 6

我也找不到太多。但我自己的尝试取得了一些成功。

cotacha 是：delayObservable 的类型参数没有任何意义。它们只是被声明为具有don't care 返回值。似乎（*我不知道）喜欢 IObservable<> 的目的只是监听 OnCompleted 事件。

因此我在实验中使用了这个 Observable：

IObservable<long> timingobs = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(1);

如果您为快速结束的 Observable 找到更简单的解决方案，请告诉我。

如果您使用不同的毫秒数（如 600），结果会有所不同。

代码：

            var apples = Observable.Create<Apple>(o => Task.Run(() =>
            {
                o.OnNext(Apple.boskop);
                Thread.Sleep(500);
                o.OnNext(Apple.boskop);
                o.OnNext(Apple.granny);
                Thread.Sleep(500);
                o.OnNext(Apple.pink);
                o.OnNext(Apple.pink);
                Thread.Sleep(500);
                o.OnNext(Apple.granny);
                o.OnCompleted();
                return Disposable.Empty;
            }));

            var pears = Observable.Create<Pear>(o => Task.Run(()=>
            {
                o.OnNext(Pear.AA);
                Thread.Sleep(300);
                o.OnNext(Pear.BB);
                o.OnNext(Pear.CC);
                Thread.Sleep(300);
                o.OnNext(Pear.BB);
                Thread.Sleep(300);
                o.OnNext(Pear.DD);
                o.OnCompleted();
                return Disposable.Empty;
            }));



            IObservable<long> timingobs = Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(1);
            pears.Join(apples, a => timingobs, p => timingobs, (pear, apple) => $"{apple} + {pear}")
                .Subscribe(Console.WriteLine);

            Console.ReadKey();

结果（100 毫秒）：

boskop + AA
pink + DD
pink + DD

结果（600 毫秒）：

boskop + AA
boskop + BB
boskop + CC
boskop + AA
boskop + BB
boskop + CC
granny + AA
granny + BB
granny + CC
boskop + BB
boskop + BB
granny + BB
boskop + DD
granny + DD
pink + BB
pink + DD
pink + BB
pink + DD
granny + DD