时间戳

首页 > 解决方案 > 条件延迟+油门算子

c# - 条件延迟+油门算子

问题描述

我正在编写一个自定义 RX 运算符，它结合了 Throttle 和 Delay 的功能，具有以下签名

public static IObservable<T> DelayWhen(this IObservable<T> self, TimeSpan delay, Func<T, bool> condition);

规则如下：

  1. 如果condition(t)返回false，立即发出。
  2. 如果condition(t)返回true，延迟 delay时间。
  3. 如果self在延迟期间发出一个值，则执行以下操作：
    1. 如果condition(t)返回false，取消/跳过计划延迟发射的值并发射新值
    2. 如果condition(t)返回true，则跳过/忽略这个新值（即，如果self在延迟期间不再发出任何值，则延迟值将发出）。

从规则中可以看出，这里有一些让人想起节流的行为。

我解决这个问题的各种尝试包括一些async刚刚变得复杂的方法。我真的觉得这应该可以使用现有的运营商来解决。例如，请参阅https://stackoverflow.com/a/16290788/2149075，它使用Amb得非常简洁，我觉得它非常接近我想要实现的目标。

标签： c#system.reactive

解决方案

这个问题并不完全清楚，因此使用以下测试用例作为场景：

Observable.Interval(TimeSpan.FromSeconds(1))
    .Take(10)
    .DelayWhen(TimeSpan.FromSeconds(1.5), i => i % 3 == 0 || i % 2 == 0)

这应该导致以下结果：

//        T: ---1---2---3---4---5---6---7---8---9---0---1----
// original: ---0---1---2---3---4---5---6---7---8---9
//   delay?: ---Y---N---Y---Y---Y---N---Y---N---Y---Y
// expected: -------1---------2-----5-------7-------------8
//
// 0: Delayed, but interrupted by 1, 
// 1: Non-delayed, emit immediately
// 2: Delayed, emit after 1.5 seconds
// 3: Delayed, since emitted during a delay, ignored
// 4: Delayed, but interrupted by 5.
// 5: Non-delayed, emit immediately
// 6: Delayed, but interrupted by 7.
// 7: Non-delayed, emit immediately
// 8: Delayed, but interrupted by 9
// 9: Delayed, since emitted during a delay, ignored

如果这不符合要求，请澄清问题。@Theodore 的解决方案获得了正确的时间，但发出了 3 和 9，忽略了“取消/跳过计划延迟发射的值并发出新值”子句。

这在功能上等同于 Theodore 的代码，但 (IMO) 更易于使用和理解：

public static IObservable<T> DelayWhen2<T>(this IObservable<T> source, TimeSpan delay, Func<T, bool> condition, IScheduler scheduler)
{
    return source
        .Select(x => (Item: x, WithDelay: condition(x)))
        .Publish(published => published
            .SelectMany(t => t.WithDelay 
                ? Observable.Return(t)
                    .Delay(delay, scheduler)
                    .TakeUntil(published.Where(t2 => !t2.WithDelay))
                : Observable.Return(t)
            )
        )
        .Select(e => e.Item);
}

从那里，我不得不嵌入你是否延迟的状态.Scan：

public static IObservable<T> DelayWhen3<T>(this IObservable<T> source, TimeSpan delay, Func<T, bool> condition)
{
    return DelayWhen3(source, delay, condition, Scheduler.Default);
}

public static IObservable<T> DelayWhen3<T>(this IObservable<T> source, TimeSpan delay, Func<T, bool> condition, IScheduler scheduler)
{
    return source
        .Select(x => (Item: x, WithDelay: condition(x)))
        .Publish(published => published
            .Timestamp(scheduler)
            .Scan((delayOverTime: DateTimeOffset.MinValue, output: Observable.Empty<T>()), (state, t) => {
                if(!t.Value.WithDelay)  
                    //value isn't delayed, current delay status irrelevant, emit immediately, and cancel previous delay.
                    return (DateTimeOffset.MinValue, Observable.Return(t.Value.Item));
                else
                    if (state.delayOverTime > t.Timestamp)
                        //value should be delayed, but current delay already in progress. Ignore value.
                        return (state.delayOverTime, Observable.Empty<T>());
                    else
                        //value should be delayed, no delay in progress. Set delay state, and return delayed observable.
                        return (t.Timestamp + delay, Observable.Return(t.Value.Item).Delay(delay, scheduler).TakeUntil(published.Where(t2 => !t2.WithDelay)));
            })
        )
        .SelectMany(t => t.output);
}

在.Scan运算符中，您嵌入了前一个Delay过期的时间。这样您就知道可以处理应该在现有延迟内延迟的值。我向scheduler时间敏感函数添加了参数以启用测试：

var ts = new TestScheduler();

var target = Observable.Interval(TimeSpan.FromSeconds(1), ts)
    .Take(10)
    .DelayWhen3(TimeSpan.FromSeconds(1.5), i => i % 3 == 0 || i % 2 == 0, ts);

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

var expected = new List<Recorded<Notification<long>>> {
    new Recorded<Notification<long>>(2000.MsTicks(), Notification.CreateOnNext<long>(1)),
    new Recorded<Notification<long>>(4500.MsTicks(), Notification.CreateOnNext<long>(2)),
    new Recorded<Notification<long>>(6000.MsTicks(), Notification.CreateOnNext<long>(5)),
    new Recorded<Notification<long>>(8000.MsTicks(), Notification.CreateOnNext<long>(7)),
    new Recorded<Notification<long>>(10500.MsTicks(), Notification.CreateOnNext<long>(8)),
    new Recorded<Notification<long>>(10500.MsTicks() + 1, Notification.CreateOnCompleted<long>()),
};

ReactiveAssert.AreElementsEqual(expected, observer.Messages);

MsTicks 的代码：

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

推荐阅读