Как выбрать возрастающую подпоследовательность значений из IObservable <T>

Question

Как написать этот метод?

public static IObservable<T> IncreasingSubsequence<T>(this IObservable<T> observable, IComparer<T> comparer)
{
    // ???
}

Получаемая наблюдаемая должна выдвигать только те значения, которые превышают максимум всех предыдущих значений.

Answer 1

Другой подход заключается в использовании Scan и DistinctUnitChanged. Вот пример использования int для простоты

IObservable<int> xs;
xs.Scan((last,cur) => cur > last ? cur : last).DistinctUntilChanged()

и более общая форма

public static IObservable<T> IncreasingSubsequence<T>(this IObservable<T> xs, IComparer<T> comp)
{
    return xs.Scan((last,cur) => comp.Compare(cur, last) == 1 ? cur : last)
                     .DistinctUntilChanged();
}

Answer 2

Я думаю, что самый простой способ - использовать Where() и тот факт, что замыкания изменчивы:

public static IObservable<T> IncreasingSubsequence<T>(
    this IObservable<T> observable, IComparer<T> comparer = null)
{
    if (observable == null)
        throw new ArgumentNullException("observable");
    if (comparer == null)
        comparer = Comparer<T>.Default;
    T max = default(T);
    bool first = true;
    return observable.Where(x =>
    {
        if (first)
        {
            first = false;
            max = x;
            return true;
        }
        if (comparer.Compare(x, max) > 0)
        {
            max = x;
            return true;
        }
        return false;
    });
}