逆Observable.Throttle

Question

我正在寻找一个可观察的扩展方法来做一个反向节流。我的意思是让第一项通过，然后忽略在适当时间内跟随这些项的项。

input - due time 2
|*.*.*..*..|
output
|*......*..|

请注意，这是一个与下面的问题不同的问题(这些问题都是相同的)。下面的问题需要一个固定的抑制持续时间，而我需要一个抑制持续时间增加，每次新的项目到达太早。从视觉上看，下列解决方案的输出如下：

input - due time 2
|*.*.*..*..|
output
|*...*..*..|

• How to take first occurrence and then supress events for 2 seconds (RxJS)
• How to throttle event stream using RX?
• Rx: How can I respond immediately, and throttle subsequent requests

更新

我想出了以下解决方案，但是我对调度程序和并发性还不太了解，无法确定锁定是否足够好。当Scheduler参数被添加到该方法时，我也不知道如何实现该方法。

    public static IObservable<T> InverseThrottle<T>(this IObservable<T> source, TimeSpan dueTime)
    {
        IDisposable coolDownSupscription = null;
        object subscriptionLock = new object();

        return source
            .Where(i =>
            {
                lock (subscriptionLock)
                {
                    bool result;
                    if (coolDownSupscription == null)
                    {
                        result = true;
                    }
                    else
                    {
                        coolDownSupscription.Dispose();
                        result = false;
                    }

                    coolDownSupscription = Observable
                            .Interval(dueTime)
                            .Take(1)
                            .Subscribe(_ =>
                            {
                                lock (subscriptionLock)
                                {
                                    coolDownSupscription = null;
                                }
                            });
                    return result;
                }
            });
    }

Answer 1

你可以用这样的东西..。

source
    .GroupByUntil(
        x => Unit.Default,
        x => x.Throttle(TimeSpan.FromSeconds(100))
    )
    .SelectMany(
        x => x.ToList().Take(1) // yields first item on completion of the observable.
    );

Answer 2

我建议这个。

public static class IObservable_FirstThenThrottle
{
    public static IObservable<TSource> FirstThenThrottle<TSource>(this IObservable<TSource> source, TimeSpan dueTime)
    {
        var first = source.Take(1);
        var second = source.Skip(1).Throttle(dueTime);
        return first.Merge(second);
    }
}

当第一个项目出现时，它就会触发。然后用dueTime节流剩余的序列。

这是一个大理石图，显示了dueTime = 2发生了什么。

source 0-1-2--3--|
result 0------2--3--|