在 Rx 中，如何按 id 对事件进行分组并按多个时间跨度限制每个组？

可以这么说，我陷入了 Rx 热潮，这个问题与我的相关here https://stackoverflow.com/questions/19425965/rx-how-to-group-by-a-key-a-complex-object-and-later-do-selectmany-without-stopp and here https://stackoverflow.com/questions/19313360/how-to-partition-groupby-a-stream-and-monitor-absence-of-elements-in-rx-within。尽管如此，也许这些对某人有帮助，因为我可以将它们视为同一主题的有用变体。

问题：如何将随机流分组int（例如，在随机间隔生成的间隔 [0, 10] 上）将对象分组并为每个组提供可变数量的缺少事件警报（由于缺乏更好的定义，有关更多背景信息，请参阅链接的帖子）。更具体地说，通过代码，如何为每组定义多个节流设置，如下所示：

var idAlarmStream = idStream
.Select(i => i)
.GroupByUntil(key => key.Id, grp => grp.Throttle(Timespan.FromMilliseconds(1000))
.SelectMany(grp => grp.TakeLast(1))
.Subscribe(i => Console.WriteLine(i));

如果每组 ID 缺失时间超过一秒，订阅函数将被调用。如果想为没有事件定义三个不同的值（例如，一秒、五秒和十秒）并在事件到达时全部取消，该怎么办？我能想到的是：

• 将每个 ID 拆分为idStream分成几个合成 ID，并提供真实 ID 和合成 ID 之间的双射映射。例如本例中 ID：1 -> 100, 101, 102； ID: 2 -> 200, 201, 203 然后定义一个选择器函数Throttle像这样Func<int, Timespan>(i => /* switch(i)...*/)然后当Subscribe将被调用，将 ID 映射回来。另请参阅链接的问题以了解更多背景信息。
• 创建一个嵌套分组，其中 ID 被分组，然后 ID 组将根据限制值被复制/复制/分叉（我不知道正确的术语）到组中。我认为这种方法相当复杂，我不确定它是否是最好的方法。尽管如此，我肯定有兴趣看到这样的查询。

在更一般的设置中，我怀疑，这是每个组有多个处理程序的情况，尽管我还没有找到与此相关的任何内容。

作为（希望澄清）一个例子idStream推送一个 ID: 1，三个不同的计数器将启动，每个计数器等待下一个事件发生，如果没有及时检测到新的 ID 1，则发出警报。计数器 1 (C1) 等待 5 秒，计数器 2 (C2) 等待 7 秒，计数器 3 (C3) 等待 10 秒。如果在间隔 [0, 5] 秒内收到新的 ID 1，则所有计数器将使用上述值重新初始化，并且不会发送警报。如果在间隔 [0, 7) 秒内收到新的 ID，C1 报警，C2 和 C3 将重新初始化。类似地，如果在 [0, 10) 秒间隔内收到新 ID，C1 和 C2 就会触发，但 C3 会重新初始化。

也就是说，在给定某些条件的情况下，针对一个ID将存在多个“缺席警报”或一般而言所采取的操作。我不确定什么是一个好的模拟......也许将“警报灯”堆叠在塔中，以便首先是绿色，然后是黄色，最后是红色。随着 ID 缺失的时间越来越长，一种颜色将会亮起（在这种情况下，红色是最后一种）。然后当检测到一个ID时，所有的灯都会关闭。

将詹姆斯的代码修改为如下示例并保留其余部分后，我发现了Subscribe将在两个警报级别上的第一个事件发生时直接调用。

const int MaxLevels = 2;
var idAlarmStream = idStream
    .Select(i => i)
    .AlarmSystem(keySelector, thresholdSelector, MaxLevels, TaskPoolScheduler.Default)
    .Subscribe(i =>
    {
        Debug.WriteLine("Alarm on id \"{0}\" at {1}", i, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture));
    });

让我们看看这里发生了什么，如果MaxLevels可以动态提供...

.FromTicks我花了几分钟才意识到这一点。

我认为这是可行的 - 我稍后会尝试添加更完整的解释。每个警报级别都有一个定义的阈值（每个信号组）。预计这些活动的持续时间将越来越长。

基本思想是将所有先前级别的信号馈送到当前级别。第一级是信号本身的“零”级，在返回警报流之前将其过滤掉。请注意，TSignal 键需要支持值标识。

我确信还有简化的空间！

单元测试示例：

public class AlarmTests : ReactiveTest
{
    [Test]
    public void MultipleKeyMultipleSignalMultipleLevelTest()
    {
        var threshold1 = TimeSpan.FromTicks(300);
        var threshold2 = TimeSpan.FromTicks(800);

        var scheduler = new TestScheduler();
        var signals = scheduler.CreateHotObservable(
            OnNext(200, 1),
            OnNext(200, 2),
            OnNext(400, 1),
            OnNext(420, 2),
            OnNext(800, 1),
            OnNext(1000, 1),
            OnNext(1200, 1));

        Func<int, int> keySelector = i => i;
        Func<int, int, TimeSpan> thresholdSelector = (key, level) =>
        {
            if (level == 1) return threshold1;
            if (level == 2) return threshold2;
            return TimeSpan.MaxValue;
        };

        var results = scheduler.CreateObserver<Alarm<int>>();

        signals.AlarmSystem(
            keySelector,
            thresholdSelector,
            2,
            scheduler).Subscribe(results);

        scheduler.Start();

        results.Messages.AssertEqual(
            OnNext(700, new Alarm<int>(1, 1)),
            OnNext(720, new Alarm<int>(2, 1)),
            OnNext(1220, new Alarm<int>(2, 2)),
            OnNext(1500, new Alarm<int>(1, 1)),
            OnNext(2000, new Alarm<int>(1, 2)));
    }

    [Test]
    public void CheckAlarmIsSuppressed()
    {
        var threshold1 = TimeSpan.FromTicks(300);
        var threshold2 = TimeSpan.FromTicks(500);

        var scheduler = new TestScheduler();
        var signals = scheduler.CreateHotObservable(
            OnNext(200, 1),
            OnNext(400, 1),
            OnNext(600, 1));

        Func<int, int> keySelector = i => i;

        Func<int, int, TimeSpan> thresholdSelector = (signal, level) =>
        {
            if (level == 1) return threshold1;
            if (level == 2) return threshold2;
            return TimeSpan.MaxValue;
        };

        var results = scheduler.CreateObserver<Alarm<int>>();

        signals.AlarmSystem(
            keySelector,
            thresholdSelector,
            2,
            scheduler).Subscribe(results);

        scheduler.Start();

        results.Messages.AssertEqual(
            OnNext(900, new Alarm<int>(1, 1)),
            OnNext(1100, new Alarm<int>(1, 2)));
    }
}



public static class ObservableExtensions
{
    /// <summary>
    /// Create an alarm system that detects signal gaps of length
    /// determined by a signal key and signals alarms of increasing severity.
    /// </summary>
    /// <typeparam name="TSignal">Type of the signal</typeparam>
    /// <typeparam name="TKey">Type of the signal key used for grouping, must implement Equals correctly</typeparam>
    /// <param name="signals">Input signal stream</param>
    /// <param name="keySelector">Function to select a key from a signal for grouping</param>
    /// <param name="thresholdSelector">Function to select a threshold for a given signal key and alarm level.
    /// Should return TimeSpan.MaxValue for levels above the highest level</param>
    /// <param name="levels">Number of alarm levels</param>
    /// <param name="scheduler">Scheduler use for throttling</param>
    /// <returns>A stream of alarms each of which contains the signal and alarm level</returns>
    public static IObservable<Alarm<TSignal>> AlarmSystem<TSignal, TKey>(
        this IObservable<TSignal> signals,
        Func<TSignal, TKey> keySelector,
        Func<TKey, int, TimeSpan> thresholdSelector,
        int levels,
        IScheduler scheduler)
    {
        var alarmSignals = signals.Select(signal => new Alarm<TSignal>(signal, 0))
                                  .Publish()
                                  .RefCount();

        for (int i = 0; i < levels; i++)
        {
            alarmSignals = alarmSignals.CreateAlarmSystemLevel(
                keySelector, thresholdSelector, i + 1, scheduler);
        }

        return alarmSignals.Where(alarm => alarm.Level != 0);

    }

    private static IObservable<Alarm<TSignal>> CreateAlarmSystemLevel<TSignal, TKey>(
        this IObservable<Alarm<TSignal>> alarmSignals,
        Func<TSignal, TKey> keySelector,
        Func<TKey, int, TimeSpan> thresholdSelector,
        int level,
        IScheduler scheduler)
    {
        return alarmSignals
            .Where(alarmSignal => alarmSignal.Level == 0)
            .Select(alarmSignal => alarmSignal.Signal)
            .GroupByUntil(
                keySelector,
                grp => grp.Throttle(thresholdSelector(grp.Key, level), scheduler))
            .SelectMany(grp => grp.TakeLast(1).Select(signal => new Alarm<TSignal>(signal, level)))
            .Merge(alarmSignals);
    }
}

public class Alarm<TSignal> : IEquatable<Alarm<TSignal>>
{
    public Alarm(TSignal signal, int level)
    {
        Signal = signal;
        Level = level;
    }

    public TSignal Signal { get; private set; }
    public int Level { get; private set; }

    private static bool Equals(Alarm<TSignal> x, Alarm<TSignal> y)
    {
        if (ReferenceEquals(x, null))
            return false;
        if (ReferenceEquals(y, null))
            return false;
        if (ReferenceEquals(x, y))
            return true;

        return x.Signal.Equals(y.Signal) && x.Level.Equals(y.Level);
    }

    // Equality implementation added to help with testing.
    public override bool Equals(object other)
    {
        return Equals(this, other as Alarm<TSignal>);
    }

    public override string ToString()
    {
        return string.Format("Signal: {0} Level: {1}", Signal, Level);
    }

    public bool Equals(Alarm<TSignal> other)
    {
        return Equals(this, other);
    }

    public static bool operator ==(Alarm<TSignal> x, Alarm<TSignal> y)
    {
        return Equals(x, y);
    }

    public static bool operator !=(Alarm<TSignal> x, Alarm<TSignal> y)
    {
        return !Equals(x, y);
    }

    public override int GetHashCode()
    {
        return ((Signal.GetHashCode()*37) ^ Level.GetHashCode()*329);
    }
}