自定义 Rx 运算符，仅在有最近值时进行节流

Question

我正在尝试创建一个看起来非常有用的 Rx 运算符，但令人惊讶的是，我在 Stackoverflow 上没有发现任何精确匹配的问题。我想在Throttle 上创建一个变体，如果有一段时间不活动，它会让值立即通过。我想象的用例是这样的：

我有一个下拉菜单，当值更改时会启动 Web 请求。如果用户按住箭头键并快速循环浏览这些值，我不想启动对每个值的请求。但是，如果我限制流，那么用户每次只需以正常方式从下拉列表中选择一个值时，就必须等待限制持续时间。

因此，普通的Throttle 看起来像这样：

我想创建如下所示的ThrottleSubsequent：

请注意，弹珠 1、2 和 6 会毫无延迟地通过，因为它们每个都经过一段时间的不活动。

我的尝试如下所示：

public static IObservable<TSource> ThrottleSubsequent<TSource>(this IObservable<TSource> source, TimeSpan dueTime, IScheduler scheduler)
{
    // Create a timer that resets with each new source value
    var cooldownTimer = source
        .Select(x => Observable.Interval(dueTime, scheduler)) // Each source value becomes a new timer
        .Switch(); // Switch to the most recent timer

    var cooldownWindow = source.Window(() => cooldownTimer);

    // Pass along the first value of each cooldown window immediately
    var firstAfterCooldown = cooldownWindow.SelectMany(o => o.Take(1));

    // Throttle the rest of the values 
    var throttledRest = cooldownWindow
        .SelectMany(o => o.Skip(1))
        .Throttle(dueTime, scheduler);

    return Observable.Merge(firstAfterCooldown, throttledRest);
}

这似乎可以工作，但我很难对此进行推理，而且我觉得这里有一些边缘情况可能会因为重复值或其他东西而变得混乱。我想从更有经验的 Rx-ers 那里得到一些反馈，看看这段代码是否正确，和/或是否有更惯用的方式来做这件事。

Answer 1

好吧，这是一个测试套件（使用 nuget Microsoft.Reactive.Testing）：

var ts = new TestScheduler();
var source = ts.CreateHotObservable<char>(
    new Recorded<Notification<char>>(200.MsTicks(), Notification.CreateOnNext('A')),
    new Recorded<Notification<char>>(300.MsTicks(), Notification.CreateOnNext('B')),
    new Recorded<Notification<char>>(500.MsTicks(), Notification.CreateOnNext('C')),
    new Recorded<Notification<char>>(510.MsTicks(), Notification.CreateOnNext('D')),
    new Recorded<Notification<char>>(550.MsTicks(), Notification.CreateOnNext('E')),
    new Recorded<Notification<char>>(610.MsTicks(), Notification.CreateOnNext('F')),
    new Recorded<Notification<char>>(760.MsTicks(), Notification.CreateOnNext('G'))
);

var target = source.ThrottleSubsequent(TimeSpan.FromMilliseconds(150), ts);
var expectedResults = ts.CreateHotObservable<char>(
    new Recorded<Notification<char>>(200.MsTicks(), Notification.CreateOnNext('A')),
    new Recorded<Notification<char>>(450.MsTicks(), Notification.CreateOnNext('B')),
    new Recorded<Notification<char>>(500.MsTicks(), Notification.CreateOnNext('C')),
    new Recorded<Notification<char>>(910.MsTicks(), Notification.CreateOnNext('G'))
);

var observer = ts.CreateObserver<char>();
target.Subscribe(observer);
ts.Start();

ReactiveAssert.AreElementsEqual(expectedResults.Messages, observer.Messages);

并使用

public static class TestingHelpers
{
    public static long MsTicks(this int i)
    {
        return TimeSpan.FromMilliseconds(i).Ticks;
    }
}

似乎通过了。如果你想减少它，你可以把它变成这样：

public static IObservable<TSource> ThrottleSubsequent2<TSource>(this IObservable<TSource> source, TimeSpan dueTime, IScheduler scheduler)
{
    return source.Publish(_source => _source
        .Window(() => _source
            .Select(x => Observable.Interval(dueTime, scheduler))
            .Switch()
        ))
        .Publish(cooldownWindow =>
            Observable.Merge(
                cooldownWindow
                    .SelectMany(o => o.Take(1)),
                cooldownWindow
                    .SelectMany(o => o.Skip(1))
                    .Throttle(dueTime, scheduler)
            )
        );
}

---

编辑：

Publish 强制共享订阅。如果你有一个糟糕的（或昂贵的）源可观察到订阅副作用，Publish 确保你只订阅一次。这是Publish 提供帮助的示例：

void Main()
{
    var source = UglyRange(10);
    var target = source
        .SelectMany(i => Observable.Return(i).Delay(TimeSpan.FromMilliseconds(10 * i)))
        .ThrottleSubsequent2(TimeSpan.FromMilliseconds(70), Scheduler.Default) //Works with ThrottleSubsequent2, fails with ThrottleSubsequent
        .Subscribe(i => Console.WriteLine(i));
}
static int counter = 0;
public IObservable<int> UglyRange(int limit)
{
    var uglySource = Observable.Create<int>(o =>
    {
        if (counter++ == 0)
        {
            Console.WriteLine("Ugly observable should only be created once.");
            Enumerable.Range(1, limit).ToList().ForEach(i => o.OnNext(i));
        }
        else
        {
            Console.WriteLine($"Ugly observable should only be created once. This is the {counter}th time created.");
            o.OnError(new Exception($"observable invoked {counter} times."));
        }
        return Disposable.Empty;
    });
    return uglySource;
}