Rx IObservable 仅在值发生一定幅度变化时才产生值答案

【问题标题】：Rx IObservable produce value only when value has changed by a certain marginRx IObservable 仅在值发生一定幅度变化时才产生值
【发布时间】：2012-11-21 15:09:26
【问题描述】：

我有一种感觉，这可能是一个非常简单的扩展方法，我错过了但我看不到它......

我基本上想采用一个流来产生一个流，其中值随着每个新值缓慢增加。我想节流/采样这个，不是按时间，而是按“容忍度”。例如

var ob = Enumerable.Range(0, 30).ToObservable(); // 0, 1, 2, 3, 4, 5,....., 30
var largeMovingOb = ob.WhenChangedBy(10); // 0, 10, 20, 30

当我有诸如 [1, 4, 20, 33] 之类的序列并且我想在值更改超过最后一个的 15 时输出 - 这将导致：[1, 20]。值 12 的变化将导致：[1, 20, 33]

是否有内置的 Rx 扩展？理想情况下，它适用于所有数字类型，而无需为每种类型编写重载。

【问题讨论】：

所以你想要一个仅在值变化超过 X 时触发的 Observable？
是的，但不仅来自源产生的最后一个值，还来自最后产生的值。请参阅下面对@jeroenh 的评论。
你知道你可以做var ob = Observable.Range(0, 30);吗？
感谢 Enigmativity。没想到这一点。

标签： c# .net c#-4.0 system.reactive

【解决方案1】：

我认为这很适合Observable.Scan

var ob = Enumerable.Range(0, 30).ToObservable();
var largeMovingOb = ob.Scan((acc, i) => acc + 10 > i ? acc : i)
  .DistinctUntilChanged();

【讨论】：

+1;优雅的解决方案，满足规范，没有丑陋的黑客

【解决方案2】：

您可以重新利用的另一个内置运算符是DistinctUntilChanged，它将为您跟踪最后一个值。这里最大的“hack”是 IEqualityComparer 可能不遵循平等的标准期望（a == b && b == c 并不意味着 a == c 取决于功能）

public static IObservable<T> DistinctUntilChangedBy<T>(
    this IObservable<T> source, Func<T, T, bool> isChanged)
{
    //check arguments
    return source.DistinctUntilChanged(new MarginEqualityComparer<T>(isChanged));
}

class MarginEqualityComparer<T> : IEqualityComparer<T>
{
    MarginEqualityComparer(Func<T, T, bool> comparer)
    {
        _comparer = comparer;
    }

    private readonly Func<T, T, bool> _comparer;

    public bool Equals(T x, T y)
    {
        return _comparer(x, y);
    }

    public int GetHashCode(T obj)
    {
        throw new NotSupportedException("This comparer does not support hashing.");
    }
}

【讨论】：

【解决方案3】：

有一个内置的操作符可以做你想做的事情。

试试这个：

var ob = Observable.Range(0, 30);
var largeMovingOb = ob.DistinctUntilChanged(x => x / 10);

它适用于任何类型，而不仅仅是数字类型，因为签名如下所示：

IObservable<TSource> DistinctUntilChanged<TSource, TKey>(
    this IObservable<TSource> source, Func<TSource, TKey> keySelector)

简单。

【讨论】：

虽然该示例假定值从 0 开始递增 1，但您的解决方案 only 适用于这种情况。如果存在大于 1 的跳跃或种子值不为 0，则这将不起作用。

【解决方案4】：

您可以在IObservable<T> 上使用Where extension method，跟踪您上次产生的内容，并且仅当值超出最后产生的值的容差水平时，谓词才返回真。

这可以被包装到一个扩展方法中，利用闭包来做到这一点，就像这样：

public static IObservable<int> WhenLastObservedChangesByMoreThan(
    this IObservable<int> observable, int tolerance)
{
    // Validate parameters.
    if (observable == null) throw new ArgumentNullException("observable");

    // Tolerance must be positive, so comparisons are correct after
    // addition/subtraction.
    if (tolerance < 0) 
        throw new ArgumentOutOfRangeExeption("tolerance", tolerance,
            "The tolerance parameter must be a non-negative number.");

    // Shortcut: If tolerance is 0, then every value is returned, just
    // return the observable.
    if (tolerance == 0) return observable;

    // The last value yielded.
    int? lastYielded = null;

    // Filter.
    observable = observable.Where(i => {
        // If there is a previous value
        // that was yielded.
        if (lastYielded != null)
        {
            // Is the last value within
            // tolerance?
            if (i - tolerance < i && i < i + tolerance)
            {
                // Do not process.
                return false;
            }
        }

        // This is being yielded, store the value.
        lastYielded = i;

        // Yield the value.
        return true;
    });
}

请注意，以上不是线程安全的，如果您的IObservable<T> 从多个线程调用OnNext，那么您必须锁定对lastYielded 变量的访问（这很容易使用@ 987654324@).

【讨论】：