Reactive operator for distinct values that must change by amount

Question

I would like feedback on this DistinctUntilChangedBy operator for System.Reactive that only emits when the current value is different than the last by a specified amount. Is there a better implementation?

I want to avoid multiple emits when the stream looks like this: 0.80, 0.81, 0.80, 0.81, etc.

Using the operator like this:

stream.DistinctUntilChangedBy(0.01).Subscribe(x => Console.WriteLine(x));

Should output this:

0.80

The operator looks like this and it is also on GitHub as a gist:

using System;
using System.Collections.Generic;
using System.Reactive.Linq;

namespace Gists.Reactive.Linq
{
    public static partial class ObservableExtensions
    {
        ///<summary>
        ///Only emit when the current value is different than the last by the specified amount.
        ///</summary>    
        public static IObservable<decimal> DistinctUntilChangedBy(this IObservable<decimal> source, decimal amount)
        {
            return source.DistinctUntilChanged(new DistinctUntilChangedByEqualityComparer(amount));
        }

        private class DistinctUntilChangedByEqualityComparer : IEqualityComparer<decimal>
        {
            private readonly decimal amount;

            public DistinctUntilChangedByEqualityComparer(decimal amount)
            {
                this.amount = amount;
            }

            public bool Equals(decimal x, decimal y)
            {
                var diff = Math.Abs(x - y);
                return diff <= Math.Abs(amount);
            }

            public int GetHashCode(decimal obj)
            {
                throw new NotSupportedException();
            }
        }
    }
}

Answer 1

You can do it without the IEqualityComparer but does get a tad bit more complex. Taking the gist of PairWithPrevious

You can create an extension that keeps track of the previous value. The code that makes this look complex is the nullable decimal from the seed data. We do this so that we know it came from seed data and we can skip it.

public static IObservable<decimal> DistinctVariance(this IObservable<decimal> source, decimal offSet)
{
    return source.Scan(Tuple.Create<decimal?, decimal?>(null, null), (tuple, item) => Tuple.Create<decimal?, decimal?>(tuple.Item2, item))
        .Where(tuple =>  tuple.Item2 != null && (tuple.Item1 == null || Math.Abs(tuple.Item1.Value - tuple.Item2.Value) > offSet))
        .Select(tuple => tuple.Item2.Value);

}

This should act like your original function

if we have data like .80m, .81m, .82m, .83m, .80m it would emit .80m, .80m

Now if we wanted to emit once the last publish value has changed more than the variance we could write the function

public static IObservable<decimal> DistinctVariance2(this IObservable<decimal> source, decimal offSet)
{
    return source.Scan((prior, current) => Math.Abs(prior - current) > offSet ? current : prior)
        .DistinctUntilChanged();

}

Since we don't care about the previous value we can just keep storing the last value. Input of .80m, .81m, .82m, .83m, .80m would emit .80, .82, .80

Answer 2

DistinctUntilChangedBy()

This method is public and therefore should do proper argument validation. Since this method is an extension-method, it could be called like:

ObservableExtensions.DistinctUntilChangedBy(someObservable, someDecimal);

If someObservableis null, your method would throw a NullReferenceException where an ArgumentNullException would be a better fit.

Equals(decimal, decimal)

Instead of returning diff <= Math.Abs(amount) you could set the class variable amount in the constructor directly to the Abs(amount) of the passed argument:

this.amount = Math.Abs(amount)

That could speed it up if the Equals() method is called often.

GetHashCode(decimal)

Only if you are 100% sure that this method won't ever be called should you throw a NotSupportedException. See What's the role of GetHashCode in the IEqualityComparer<T> in .NET?