Rx.Net 运算符忽略某些值，直到它们在一定时间内相同

Question

我正在尝试创建一个具有以下行为的 Rx.Net 运算符：

• 当事件属于“正常类型”时，直接返回该事件
• 当事件属于“特殊类型”时，等待您在一定时间内收到重复的该类型事件

我想要像下面这样的大理石。

当消息类型为A或B时，我们直接发送。当消息是 C 时，我们要确保它不仅仅是一个传递状态，并且只有在一定时间内是这样的情况下才发送它。这是通过第一个C 和我们收到的当前C 之间的时间来衡量的。在该特定时间之后，所有C 都被“接受”，我们将它们作为正常的传递。

这是我们收到 C 时的样子，它只是传递性的，我们想忽略它。

我尝试用Scan 运算符做一些事情，当我有一个特定的值时，我会尝试返回以前/当前的值，但感觉真的很hacky。

这是我编写的一些代码，用于演示我的尝试。在那种情况下，“特殊类型”只是当值为 999 时，但在运算符中，我想做的可能是另一个测试，甚至是传递给我的运算符的函数。

var oneObservable = Observable.Interval(TimeSpan.FromSeconds(1)).Select(_ => 999);
var intObservable = Observable.Interval(TimeSpan.FromSeconds(1)).Select(value => (int)value);

var myObservable = intObservable.Take(4).Concat(oneObservable.Take(3)).Timestamp().Repeat();

var test = myObservable.Scan(
    (previous: default(Timestamped<int>), current: default(Timestamped<int>)),
    (accumulated, current) =>
    {
        if (current.Value == 999)
        {
            if (accumulated.previous.Value != 999)
            {
                return (accumulated.current, current);
            }
            
            return (accumulated.previous, current);
        }
        else if(accumulated.current.Value == 999){
            return (accumulated.previous, current);
        }

        return (accumulated.current, current);
    })
    .Where(
        value => value.current.Value != 999 
        || (value.previous.Value == 999 && value.current.Timestamp - value.previous.Timestamp > TimeSpan.FromSeconds(1.5)))
    .Select(value => value.current);

Answer 1

以下是自定义 IgnoreNonEstablishedContiguousValue 运算符的简单实现，具有理想的功能：

/// <summary>
/// Ignores elements having a specific value, until this value has
/// been repeated contiguously for a specific duration.
/// </summary>
public static IObservable<T> IgnoreNonEstablishedContiguousValue<T>(
    this IObservable<T> source,
    T value,
    TimeSpan dueTimeUntilEstablished,
    IEqualityComparer<T> comparer = default,
    IScheduler scheduler = default)
{
    // Arguments validation omitted
    comparer ??= EqualityComparer<T>.Default;
    scheduler ??= Scheduler.Default;
    return Observable.Defer(() =>
    {
        IStopwatch stopwatch = null;
        return source.Do(item =>
        {
            if (comparer.Equals(item, value))
                stopwatch ??= scheduler.StartStopwatch();
            else
                stopwatch = null;
        })
        .Where(_ => stopwatch == null || stopwatch.Elapsed >= dueTimeUntilEstablished);
    });
}

此实现基于Do 和Where 运算符。我不太喜欢使用 Scan 运算符作为构建块，因为它会导致代码冗长且可读性较差，恕我直言。 Observable.Defer 包装器的目的是隔离每个订阅的状态。

使用示例：

var oneObservable = Observable.Interval(TimeSpan.FromSeconds(1)).Select(_ => 999);
var intObservable = Observable.Interval(TimeSpan.FromSeconds(1)).Select(v => (int)v);

IObservable<int> myObservable = intObservable.Take(4).Concat(oneObservable.Take(3))
    .Repeat()
    .IgnoreNonEstablishedContiguousValue(999, TimeSpan.FromSeconds(1.5));

Answer 2

这是我想出的：

public static IObservable<T> FilterSpecials<T>(this IObservable<T> source,
    Func<T, bool> specialDetector,
    TimeSpan timeUntilEstablished,
    IScheduler scheduler = default)
{
    return source.FilterSpecials(specialDetector, Observable.Timer(timeUntilEstablished), scheduler);
}

public static IObservable<T> FilterSpecials<T, U>(this IObservable<T> source,
    Func<T, bool> specialDetector,
    IObservable<U> observeSpecialsUntilEstablished,
    IScheduler scheduler = default)
{
    scheduler = scheduler ?? Scheduler.Default;

    return source
        .Select(i => (value: i, isSpecial: specialDetector(i)))
        .StartWith((value: default(T), isSpecial: false))
        .Publish(_source => _source
            .Zip(_source.Skip(1))
            .Select((t, index) => (
                newValue: t.Second.value,
                isNewValueSpecial: t.Second.isSpecial,
                isPreviousValueSpecial: t.First.isSpecial,
                isFirstElement: index == 0)
            )
            .SelectMany((t, index) => t.isNewValueSpecial
                ? (t.isFirstElement || !t.isPreviousValueSpecial)       
                    ? _source.SkipUntil(observeSpecialsUntilEstablished).TakeWhile(i => i.isSpecial).Select(i => i.value)   
                    : Observable.Empty<T>()
                : Observable.Return(t.newValue)
        ));
}

问题的核心在于，对于第一个特殊值，您希望暂时停止收听常规的 source observable，并切换到如下所示的内容：

source.SkipUntil(observeSpecialsUntilEstablished).TakeWhile(i => i.isSpecial).Select(i => i.value)

当你有那个特殊的 observable 监听时，你可以忽略你的常规监听并发出 Observable.Empty<T>()。当您没有特殊值时，您实际上是在执行source.SelectMany(i => Observable.Return(i))，这是返回source 的无操作。

其他一切都是装饰：Zip、Publish 和 StartWith 可以轻松与之前的值进行比较。如果您愿意，可以将其抽象出来。将所有内容放在该命名元组中是为了帮助进行自我记录并防止重新调用 specialDetector，以防万一这是一项昂贵的操作。