如何合并不同数据的 Rx 流答案

【问题标题】：How to merge Rx streams of different data如何合并不同数据的 Rx 流
【发布时间】：2014-02-26 01:30:54
【问题描述】：

这是我尝试使用 Rx 实现的简化场景。

public class A
{
    public int Id { get; set; }
    public string Name { get; set; }
};

public class B
{
    public int Id { get; set; }
    public int AId { get; set; }
    public string SomeProperty { get; set; }
};

public class C
{
    public A MyA { get; set; }
    public IObservable<B> MyBObservable { get; set; }
}

这是一个数据序列

A(1, "弗雷德")

A(2, "汤姆")

A(3, "哈利")

B(100, 1, "测试")

B(101, 3, "xxx")

B(102, 1, "test2")

A(4, "简")

...

注意：对应的A实例之前不能有B实例。

我想得到一个看起来像

的结果

C( A(1,"Fred"), IObservable (B(100, 1, "test"), B(102,1, "test2"))

C(A(2, "Tom"), IObservable empty)

C( A(3, "Harry"), IObservable (B(101, 3, "xxx"))

C(A(4, "Jane"), IObservable empty)

结果必须包括 A 的所有实例（即使该 A 还没有 B）

我有一些用户代码

IObservable<A> obsA = Observable.Create<A> { ... };
IObservable<B> obsB = Observable.Create<B> { ... };

obsB.Publish(); // shared

    // what goes here ?

obsB.Connect();

我当前的尝试使用 SelectMany，但并没有完全按照我的意愿进行，因为我只获得出现 A 和 B 的条目。我还需要那些 A 没有 B 的条目。

有没有可能做类似的事情

obsA.SelectMany(a => new C { MyA = a, MyBObservable = obsB.Where(x => x.AId == a.Id) });

如果有帮助，obsA 是一个冷的 observable 并将完成。 obsB 可以是热或冷的 observable。如果 C 是可观察的或列表，我不介意。

非常感谢任何帮助。

问候艾伦

【问题讨论】：

标签： c#-4.0 system.reactive

【解决方案1】：

IObservable<A> obsA = Observable.Create<A> { ... };
IObservable<B> obsB = Observable.Create<B> { ... };

var c = from a in obsA
        join b in obsB on a.Id equals b.Id
        select new C(a, b);

【讨论】：

【解决方案2】：

经过更多的实验，我得出了以下结论

obsA.Subscribe(a => new C (a, obsB.Where(x => x.AId == a.Id)));

然后在 C 的构造函数中

public C (MyA a, IObservable<B> obsB)
{
    MyA = a;
    obsB.Subscribe(o => .. do something ...);
}

在我的版本中，C 类不保存 observable，而是对返回的每个项目执行某些操作的结果。

【讨论】：