How to select increasing subsequence of values from IObservable<T>

How to write this method?

public static IObservable<T> Increas开发者_JAVA百科ingSubsequence<T>(this IObservable<T> observable, IComparer<T> comparer)
{
    // ???
}

Resulting observable should push only those values that exceed maximum of all previous values.

---

Another approach would be to use Scan and DistinctUnitChanged. Here's a example using ints for simplicity

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

and the more general form

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();
}

---

I think the easiest way is to use Where() and the fact that closures are mutable:

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;
    });
}