C# Rx DistinctUntilChanged() 用于任何结果，包括列表

Question

我正在包装一些 API 调用来假装套接字结果 - 以几秒的间隔循环调用并在更改时发出结果。

使用 DistinctUntilChanged() 运算符时效果很好。但是，当结果是列表时，运算符始终发出结果，因为它与默认比较器不同。

这是我自定义的可观察对象，用于重复任务，无论成功还是失败，都会有一定的延迟。

public class TaskRepeatObservable<T> : IObservable<T>
{
    private readonly Func<Task<T>> _taskFactory;
    private readonly TimeSpan _repeatDelayTimeSpan;
    private readonly ILogger _logger;
    private Func<Exception, bool> _onError;
    public TaskRepeatObservable(Func<Task<T>> taskFactory, TimeSpan repeatDelayTimeSpan = default(TimeSpan), Func<Exception, bool> onError = null)
    {
        _logger = new Log4NetLogger(GetType());
        _logger.IsEnabled = false;

        _taskFactory = taskFactory;
        _repeatDelayTimeSpan = repeatDelayTimeSpan;
        _onError = onError;
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        var tokenSource = new CancellationTokenSource();
        var cancellationToken = tokenSource.Token;
        Task.Factory.StartNew(async () =>
        {
            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        var result = await _taskFactory();
                        observer.OnNext(result);
                    }
                    catch (Exception e)
                    {
                        _logger.Error(e, "Observable Error: " + e.Message);
                        if (_onError != null && !_onError.Invoke(e))
                            throw;
                    }
                    finally
                    {
                        try
                        {
                            if (_repeatDelayTimeSpan > TimeSpan.Zero)
                                await Task.Delay(_repeatDelayTimeSpan, cancellationToken);
                        }
                        catch (TaskCanceledException)
                        {
                            // ignored
                        }
                    }
                }
            }
            catch (Exception e)
            {
                observer.OnError(e);
            }

            _logger.Debug("Observable is cancelled.");
        }, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);

        return Disposable.Create(() =>
        {
            tokenSource.Cancel();
        });
    }
}

这是包装 API 调用的扩展。

public static class ObservableBuilder
{
    ///<summary>
    ///<para>Convert Task to Observable and emit only changed result, it's useful to wrap the api as socket-like.</para>
    ///</summary>
    public static IObservable<T> Repeat<T>(this Func<Task<T>> taskFactory, TimeSpan delay = default(TimeSpan),
        Func<Exception, bool> onError = null)
    {
        return new TaskRepeatObservable<T>(taskFactory, delay, onError).DistinctUntilChanged();
    }

我的问题是 - 如何使 DistinctUntilChanged() 适用于任何结果，包括 List 或 Enumerable。

注意到我尝试实现我自己的比较器。但我仍然不知道如何检查 T 的类型来为 DistinctUntilChanged()

选择正确的比较器

public class IEnumerableComparer<T> : IEqualityComparer<IEnumerable<T>>
{
    public bool Equals(IEnumerable<T> x, IEnumerable<T> y)
    {
        return ReferenceEquals(x, y) || x != null && y != null && x.SequenceEqual(y);
    }

    public int GetHashCode(IEnumerable<T> obj)
    {
        // Will not throw an OverflowException
        unchecked
        {
            return obj.Where(e => e != null).Select(e => e.GetHashCode()).Aggregate(17, (a, b) => 23 * a + b);
        }
    }
}

这是简单的测试代码:

        ObservableBuilder.Repeat(async () =>
        {
            var i = new List<int>() { 1, 2, 3, 4 };
            return i;
        }, TimeSpan.FromSeconds(1)).ToHotObservable().Subscribe(x => Logger.Info($"Result = {x}"));

我希望列表只发出一次结果。

Answer 1

如果你只想拥有一个Repeat函数可以转换所有情况，但对于 IEnumerable 的工作方式有所不同s - 你必须使用一些反射。您需要检查是否输入 T实现IEnumerable<Something>如果是这样 - 使用 DistinctUtilChanged 的特殊比较器- 否则使用默认比较器。

首先，修改你的EnumerableComparer的签名不会有什么坏处。一点，因为您将特别需要 IEqualityComparer<T> ，不是IEqualityComparer<IEnumerable<T>> :

private class EnumerableComparer<T, TItem> : IEqualityComparer<T> where T : IEnumerable<TItem> {
    public bool Equals(T x, T y) {
        return ReferenceEquals(x, y) || x != null && y != null && x.SequenceEqual(y);
    }

    public int GetHashCode(T obj) {
        // Will not throw an OverflowException
        unchecked {
            return obj.Where(e => e != null).Select(e => e.GetHashCode()).Aggregate(17, (a, b) => 23 * a + b);
        }
    }
}

现在我们需要检查是否 T是 IEnumerable并通过反射创建此比较器的实例:

public static class ObservableBuilder {
    public static IObservable<T> Repeat<T>(this Func<Task<T>> taskFactory, TimeSpan delay = default(TimeSpan),
        Func<Exception, bool> onError = null) {
        var ienum = typeof(T).GetInterfaces().FirstOrDefault(c => c.IsGenericType && c.GetGenericTypeDefinition() == typeof(IEnumerable<>));
        if (ienum != null) {
            // implements IEnumerable - create instance of comparer and use
            var comparer = (IEqualityComparer<T>) Activator.CreateInstance(typeof(EnumerableComparer<,>).MakeGenericType(typeof(T), ienum.GenericTypeArguments[0]));
            return new TaskRepeatObservable<T>(taskFactory, delay, onError).DistinctUntilChanged(comparer);
        }
        // otherwise - don't use
        return new TaskRepeatObservable<T>(taskFactory, delay, onError).DistinctUntilChanged();
    }

    private class EnumerableComparer<T, TItem> : IEqualityComparer<T> where T : IEnumerable<TItem> {
        public bool Equals(T x, T y) {
            return ReferenceEquals(x, y) || x != null && y != null && x.SequenceEqual(y);
        }

        public int GetHashCode(T obj) {
            // Will not throw an OverflowException
            unchecked {
                return obj.Where(e => e != null).Select(e => e.GetHashCode()).Aggregate(17, (a, b) => 23 * a + b);
            }
        }
    }
}  

或者，您可以创建 IEqualityComparer<T>每次都会检查 x和y实现IEnumerable<Something>并进行相应的比较，但我预计与构建序列时只进行一次比较相比，每次比较的效率都会较低。