如何在 rx.net 中实现我自己的操作符

Question

我需要 RX 中的磁滞滤波器功能。只有当先前发出的值和当前输入值相差一定量时，它才应该从源流中发出一个值。作为一个通用的扩展方法，它可以有如下签名：

public static IObservable<T> HysteresisFilter<T>(this IObservable<t> source, Func<T/*previously emitted*/, T/*current*/, bool> filter)

我无法弄清楚如何使用现有的运营商来实现这一点。我一直在寻找来自 RxJava 的 lift 之类的东西，以及任何其他创建我自己的运算符的方法。我看过这个checklist，但是我在网上没有找到任何例子。

以下方法（实际上都是相同的）对我来说似乎可以解决问题，但是是否有更多的 Rx 方法 可以做到这一点，例如不包装 subject 或实际实现运算符?

async Task Main()
{
    var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));

    var rnd = new Random();
    var s = Observable.Interval(TimeSpan.FromMilliseconds(10))
            .Scan(0d, (a,_) => a + rnd.NextDouble() - 0.5)
            .Publish()
            .AutoConnect()
            ;

    s.Subscribe(Console.WriteLine, cts.Token);

    s.HysteresisFilter((p, c) => Math.Abs(p - c) > 1d).Subscribe(x => Console.WriteLine($"1> {x}"), cts.Token);
    s.HysteresisFilter2((p, c) => Math.Abs(p - c) > 1d).Subscribe(x => Console.WriteLine($"2> {x}"), cts.Token);

    await Task.Delay(Timeout.InfiniteTimeSpan, cts.Token).ContinueWith(_=>_, TaskContinuationOptions.OnlyOnCanceled);
}

public static class ReactiveOperators
{
    public static IObservable<T> HysteresisFilter<T>(this IObservable<T> source, Func<T, T, bool> filter)
    {
        return new InternalHysteresisFilter<T>(source, filter).AsObservable; 
    }

    public static IObservable<T> HysteresisFilter2<T>(this IObservable<T> source, Func<T, T, bool> filter)
    {
        var subject = new Subject<T>();
        T lastEmitted = default;
        bool emitted = false;

        source.Subscribe(
            value =>
            {
                if (!emitted || filter(lastEmitted, value))
                {
                    subject.OnNext(value);
                    lastEmitted = value;
                    emitted = true;
                }
            } 
            , ex => subject.OnError(ex)
            , () => subject.OnCompleted()
        );

        return subject;
    }

    private class InternalHysteresisFilter<T>: IObserver<T>
    {
        Func<T, T, bool> filter;
        T lastEmitted;
        bool emitted;

        private readonly Subject<T> subject = new Subject<T>();

        public IObservable<T> AsObservable => subject;

        public InternalHysteresisFilter(IObservable<T> source, Func<T, T, bool> filter)
        {
            this.filter = filter;
            source.Subscribe(this);
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            return subject.Subscribe(observer);
        }

        public void OnNext(T value)
        {
            if (!emitted || filter(lastEmitted, value))
            {
                subject.OnNext(value);
                lastEmitted = value;
                emitted = true;
            }
        }

        public void OnError(Exception error)
        {
            subject.OnError(error);
        }

        public void OnCompleted()
        {
            subject.OnCompleted();
        }
    }
}

旁注：将有数千个此类过滤器应用于尽可能多的流。我需要吞吐量超过延迟，因此我正在寻找 CPU 和内存中开销最小的解决方案，即使其他人看起来更漂亮。

Answer 1

我在Introduction to Rx 一书中看到的大多数示例都使用Observable.Create 方法来创建新的运算符。

Create 工厂方法是实现自定义可观察序列的首选方法。主题的使用应主要停留在样本和测试领域。 (citation)

public static IObservable<T> HysteresisFilter<T>(this IObservable<T> source,
    Func<T, T, bool> predicate)
{
    return Observable.Create<T>(observer =>
    {
        T lastEmitted = default;
        bool emitted = false;
        return source.Subscribe(value =>
        {
            if (!emitted || predicate(lastEmitted, value))
            {
                observer.OnNext(value);
                lastEmitted = value;
                emitted = true;
            }
        }, observer.OnError, observer.OnCompleted);
    });
}

Answer 2

这个答案与@Theodor 的答案相同，但它避免使用我通常会避免的Observable.Create。

public static IObservable<T> HysteresisFilter2<T>(this IObservable<T> source,
    Func<T, T, bool> predicate)
{
    return source
        .Scan((emitted: default(T), isFirstItem: true, emit: false), (state, newItem) => state.isFirstItem || predicate(state.emitted, newItem)
            ? (newItem, false, true)
            : (state.emitted, false, false)
        )
        .Where(t => t.emit)
        .Select(t => t.emitted);
}

.Scan 是您在跟踪 observable 中项目的状态时想要使用的。