Rx.net 中 SelectMany 的反向结果

Reverse result from SelectMany in Rx.net

我对响应式 linq(也称为 Rx.NET)做了一些研究。我已经看到 SelectMany 方法的行为作为以下​​最终结果:展平可观察对象中的内容,以便它从多个流变为单个流。我的问题是,是否有一种方法作用完全相反?这意味着在将多个事件流放入一个可观察对象后(multiplexing), how would one achieve that ? I know that GroupBy 与我所请求的行为非常相似,除了缺少一件事:我想要一个可能尚未发出的键的可观察对象。

我正在考虑自己实施该行为,但如果有机会,我错过了这种方法的存在,请告诉我我错了,它确实存在!

GroupBy 是您要查找的内容,但正如您指出的那样,它确实无法预先创建 "known" 组。但是,我想您可以发挥创造力,并为您的序列播种这些已知组中的每一个的值:

如果来源是这样的

Observable.Interval(TimeSpan.FromMilliseconds(100))
    .GroupBy(i => i % 4)

您可以将其修改为

var seed = new[] { 0L, 1L, 2L, 3L }.ToObservable();
Observable.Concat(seed, Observable.Interval(TimeSpan.FromMilliseconds(100)))
    .GroupBy(i => i % 4)
    .Select(grp=>grp.Skip(1))   //Ignore the first/seed value.

你说得对,GroupBy 就是你想要的。很容易将其设置为提前订阅任何密钥。只需这样做:

IObservable<int> oddNumbers =
    Observable
        .Range(0, 10)
        .GroupBy(x => x % 2)
        .Where(gx => gx.Key == 1)
        .Merge();

如果我订阅,我将获得:

1 
3 
5 
7 
9 

但是,这有点浪费时间,因为这直接相当于:

IObservable<int> oddNumbers =
    Observable
        .Range(0, 10)
        .Where(x => x % 2 == 1);

我认为它可能有用的唯一方法是如果您这样做:

IConnectableObservable<IGroupedObservable<int, int>> groupedNumbers =
    Observable
        .Range(0, 10)
        .GroupBy(x => x % 2)
        .Publish();

Func<IConnectableObservable<IGroupedObservable<int, int>>, int, IObservable<int>> anyProject =
    (source, key) =>
        source
            .Where(gx => gx.Key == key)
            .Merge();

IObservable<int> oddNumbers = anyProject(groupedNumbers, 1);

oddNumbers.Subscribe(x => Console.WriteLine(x));

groupedNumbers.Connect();

这仍然给我奇数,但我现在可以在不开始新分组的情况下创建可观察的偶数。