当源流和目标流在C# Reactive中相同时，C#是如何工作的？

Question

我有两个Subject，一个是带有ID的person对象流，一个是代表与谁为好友的ID的xref流。以下是一些简化的代码。

class Program {
    static void Main() {
        // Set up observables
        var people = new Subject<Person>();
        var friendMap = new Subject<FriendMap>();
        var friendNotices = from p1 in people
                            from p2 in people
                            from pair in friendMap
                            where p1.Id == pair.Id1 && p2.Id == pair.Id2
                            select p1.Name + " befriended " + p2.Name;

        // Subscribe to log
        friendNotices.Subscribe(Console.WriteLine);

        // Add people
        people.OnNext(new Person(1, "Alice"));
        people.OnNext(new Person(2, "Bob"));

        // Add relationships
        friendMap.OnNext(new FriendMap(1, 2)); // => "Alice befriended Bob"
        friendMap.OnNext(new FriendMap(2, 1)); // Doesn't show up!
    }
}

class Person {
    public Person(int id, string name) {
        Id = id;
        Name = name;
    }
    public string Name;
    public int Id;
}

class FriendMap {
    public FriendMap(int id1, int id2) {
        Id1 = id1;
        Id2 = id2;
    }
    public int Id1;
    public int Id2;
}

我遇到的问题是，有时添加xref不会导致friendNotice事件。特别是，如果使用Id2的人是在使用Id1的人之前创建的，那么它似乎就失败了。

这是Rx中的bug还是我的代码中的bug？不管怎样，我该怎么做才能让这件事起作用？

(在我的应用程序中，“交友”是不可交换的- Alice与Bob交友Bob是一种不同的关系，因此“只需交换in并重试”在我的例子中不是一个可用的解决方案)。

Answer 1

这里的问题是对SelectMany函数工作方式的误解(这是由linq from ... from ...映射到的运算符)。

此构造从源流中获取每个元素，并将其投影到目标流。它通过在目标流上为源的每个元素创建订阅以服务于一个投影来做到这一点。

让我们使用一些伪代码来检查这个问题。只考虑查询的这一部分：

from p1 in people
from p2 in people

现在，让我们将人员流分成peopleA和peopleB：

from p1 in peopleA
from p2 in peopleB

如果我们现在呼叫：

peopleA.OnNext(Alice);

实际发生的情况是，将在peopleB上创建一个新的订阅，以满足爱丽丝对peopleB的投影。此时，peopleB中没有元素，因此不会发生投影。

现在如果我们打电话：

peopleB.OnNext(Tom);

Alice -> peopleB的投影将运行，并输出(Alice，Tom)。

现在打电话：

peopleB.OnNext(Dick);

现在，从Alice -> peopleB的投影继续这样(爱丽丝，迪克)将输出。

现在打电话：

peopleA.OnNext(Bob);

现在，在peopleB上为Bob启动了一个新的订阅--但是在peopleB发出之前不会输出任何内容。

现在打电话：

peopleB.OnNext(Harry);

随着Alice和Bob订阅的运行，我们将得到(Alice，Harry)和(Bob，Harry)；

下面是一个简单的例子，供您尝试：

var source = new Subject<string>();
var source2 = new Subject<string>();

var res = from s in source
          from t in source2
          select s + " " + t;

res.Subscribe(Console.WriteLine);

source.OnNext("A"); 
source2.OnNext("1");
source2.OnNext("2");
source.OnNext("B");
source2.OnNext("3");

将给出输出：

A 1
A 2
A 3
B 3

回到“自我SelectMany”。现在一切都变得有点棘手了。关键是正在设置的订阅不会捕获触发其设置的“当前”项。因此，让我们给SelectMany A和B的每个部分贴上标签：

from p1 in people (call this A)
from p2 in people (call this B)

当我们呼叫：

people.OnNext(Alice);

爱丽丝在A上的订阅将在B上进行--但由于它是在Alice发出之后进行的，所以它不会捕获她，也不会输出任何内容。

现在我们呼吁：

people.OnNext(Bob);

对于Alice的订阅将看到Bob on B--这将导致输出(Alice，Bob)。A上的Bob订阅将在B上创建，但同样不会输出任何内容，因为它错过了Bob的输出。

这就是你所看到的。唯一发出的组合是(Alice，Bob)。

但是，您可以通过让people subject在设置新订阅时重放它的内容来解决这个问题。修改示例的第一部分如下：

    // Set up observables
    var people = new Subject<Person>();
    var peopleR = people.Replay().RefCount();
    var friendMap = new Subject<FriendMap>();
    var friendNotices = from p1 in peopleR
                        from p2 in peopleR
                        from pair in friendMap
                        where p1.Id == pair.Id1 && p2.Id == pair.Id2
                        select p1.Name + " befriended " + p2.Name;

..。当然，如果流长时间运行，这是不可取的，因为您要在内存中缓存所有内容。

就你的具体问题而言，我认为你最好改弦更张--但如果不知道你想要达到什么目标，就很难做到规定性，而且这已经是一个相当长的帖子了！一种方法是对FriendMap条目进行简单的订阅，只需查找朋友的名字就可以输出您想要的消息。(我可能漏掉了更大的东西！)

希望您理解您的方法的问题。