c# - Rx 如何按键对一个复杂对象进行分组，然后在没有 "stopping"流的情况下执行 SelectMany？

Question

这与我的另一个问题有关here . James World提出如下解决方案:

// idStream is an IObservable<int> of the input stream of IDs
// alarmInterval is a Func<int, TimeSpan> that gets the interval given the ID
var idAlarmStream = idStream
.GroupByUntil(key => key, grp => grp.Throttle(alarmInterval(grp.Key)))
.SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key)));

<编辑 2:

问题:如何在不等待第一个事件到达的情况下立即启动计时器？我猜这是我问题的根本问题。为此，我计划发送带有我知道应该存在的 ID 的虚拟对象。但是正如我在下面所写的那样，我最终遇到了一些其他问题。不过，我认为解决这个问题也很有趣。

然后转发其他有趣的部分!现在，如果我想像下面这样对一个复杂的对象进行分组，并按如下键进行分组(不会编译)

var idAlarmStream = idStream
    .Select(i => new { Id = i, IsTest = true })
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key)))
    .SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key)));

然后我就惹上麻烦了。我无法修改有关 SelectMany、Concat 和 Observable.Return 的部分，以便查询像以前一样工作。例如，如果我将查询设为

var idAlarmStream = idStream
    .Select(i => new { Id = i, IsTest = true })
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key)))
    .SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key.First())))
    .Subscribe(i => Console.WriteLine(i.Id + "-" + i.IsTest);

然后需要两个事件才能在 Subscribe 中观察到输出。我猜这是调用 First 的结果。此外，我也想在调用 alarmInterval 时使用复杂的对象属性。

有人可以解释发生了什么，甚至是解决方案吗？使用未修改的解决方案的问题在于，分组不仅会查找键值的 ID，还会查找 IsTest 字段。

请注意，问题可能可以首先通过创建显式类或结构然后实现自定义 IEquatable 然后使用 James 来解决' 按原样编写代码，以便仅通过 ID 进行分组。不过，感觉就像黑客一样。

Answer 1

此外，如果您想在闹钟响起之前计算您看到某个项目的次数，您可以这样做，利用 Select 中的计数器过载| .

var idAlarmStream = idStream
    .Select(i => new { Id = i, IsTest = true })
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key))
    .SelectMany(grp => grp.Select((count, alarm) => new { count, alarm }).TakeLast(1));

请注意，第一个(种子)项目将为 0 - 这可能正是您想要的。