How to debounce until value changed (with timeout)?

EDIT: Thanks for the marble diagram. I turned that into some test cases. You were correct, my previous solution was lacking a timer. I added that here int he form of doubling each message, but one put through immediately, the other with a delay. Here's the solution:

public static IObservable<TSource> ThrottleBy4<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, TimeSpan timeout,
    IEqualityComparer<TKey> comparer = null, IScheduler scheduler = null)
{
    comparer = comparer ?? EqualityComparer<TKey>.Default;
    scheduler = scheduler ?? DefaultScheduler.Instance;

    return source
        .Timestamp(scheduler)
        .Publish(_val => Observable.Merge(  // For every incoming item, create two items: One immediate, one delayed by the timeout time.
            _val.Select(v => (value: v, isOriginal: true)),
            _val.Select(v => (value: v, isOriginal: false)).Delay(timeout, scheduler)
        ))
        .StateSelect(Timestamped.Create(default(TSource), DateTimeOffset.MinValue),
            (prevVal, t) =>         // Result function
            {
                // special handling for the initial state
                if (prevVal.Timestamp == DateTimeOffset.MinValue)
                    return (prevVal, false);

                if (t.isOriginal)   // If an original value, only emit if the value changed.
                    return (prevVal, !comparer.Equals(keySelector(t.value.Value), keySelector(prevVal.Value)));
                else                // If a repeat value, only emit if the prevVal state is the same timestamp and value.
                    return (prevVal, comparer.Equals(keySelector(t.value.Value), keySelector(prevVal.Value)) && t.value.Timestamp == prevVal.Timestamp);
            },
            (prevVal, t) => t.isOriginal ? t.value : prevVal        // State function. Only change state if the incoming item is an original value.
        )
        .Where(t => t.Item2)
        .Select(t => t.Item1.Value);
}

Here's the test code:

TestScheduler ts = new TestScheduler();
var source = ts.CreateHotObservable<string>(
    new Recorded<Notification<string>>(200.MsTicks(), Notification.CreateOnNext("A1")),
    new Recorded<Notification<string>>(300.MsTicks(), Notification.CreateOnNext("A2")),
    new Recorded<Notification<string>>(500.MsTicks(), Notification.CreateOnNext("B1")),
    new Recorded<Notification<string>>(800.MsTicks(), Notification.CreateOnNext("B2"))
);

var comparer = new FirstLetterComparer();
var target = source
    .ThrottleBy4(s => s, TimeSpan.FromSeconds(1), comparer: comparer, scheduler: ts);

var expectedResults = ts.CreateHotObservable<string>(
    new Recorded<Notification<string>>(500.MsTicks(), Notification.CreateOnNext("A2")),
    new Recorded<Notification<string>>(1800.MsTicks(), Notification.CreateOnNext("B2"))
);

var observer = ts.CreateObserver<string>();
target.Subscribe(observer);
ts.Start();

ReactiveAssert.AreElementsEqual(expectedResults.Messages, observer.Messages);

and with these helper classes:

public class FirstLetterComparer : IEqualityComparer<string>
{
    public bool Equals(string s1, string s2)
    {
        if (s1 == null && s2 == null) 
            return true;
        if (s1 == null || s2 == null)
            return false;
        return (s1[0] == s2[0]);
    }

    public int GetHashCode(string s)
    {
        return s == null ? 0 : s[0].GetHashCode();
    }
}

public static class X
{
    public static long MsTicks(this int i)
    {
        return TimeSpan.FromMilliseconds(i).Ticks;
    }
}

Previous answer:

I see two possible problems with your solution (aside from the Timeout issue):

1. Using default(T) as token-value will trip you up at some point. That would not allow 0 to get through in an IObservable<int> for example.
2. You may have a multiple-subscription problem because of your use of the field prev. Multiple subscribers would share the field, which could lead to a race condition and incorrect behavior.

You can fix those two problems by returning a tuple, one value with the boolean of whether new or not, and one value with the value:

    public static IObservable<TSource> ThrottleBy2<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, TimeSpan timeout,
    IEqualityComparer<TKey> comparer = null, IScheduler scheduler = null)
{
    comparer = comparer ?? EqualityComparer<TKey>.Default;
    scheduler = scheduler ?? DefaultScheduler.Instance;

    return source
        .StateSelect(default(TSource), (prevVal, newVal) => (!comparer.Equals(keySelector(prevVal), keySelector(newVal)), newVal), (_, newVal) => newVal)
        .Where(t => t.Item1)
        .Select(t => t.newVal)
        .Timeout(timeout, scheduler)
        .RetryWhen(ex => ex.OfType<TimeoutException>());
}

StateSelect does what you want here: It maintains a state (what you previously had in the prev field), and returns the tuple mentioned previously. It looks like this:

public static IObservable<TResult> StateSelect<TSource, TState, TResult>(this IObservable<TSource> source, TState initialState,
    Func<TState, TSource, TResult> resultSelector, Func<TState, TSource, TState> stateSelector)
{
    return source
        .StateSelectMany(initialState, (state, item) => Observable.Return(resultSelector(state, item)), stateSelector);
}

public static IObservable<TResult> StateSelectMany<TSource, TState, TResult>(this IObservable<TSource> source, TState initialState, 
    Func<TState, TSource, IObservable<TResult>> resultSelector, Func<TState, TSource, TState> stateSelector)
{
    return source
        .Scan(Tuple.Create(initialState, Observable.Empty<TResult>()), (state, item) => Tuple.Create(stateSelector(state.Item1, item), resultSelector(state.Item1, item)))
        .SelectMany(t => t.Item2);
}

This still leaves two small issues:

1. The Timeout issue
2. Using default(TSource) as the initial state can lead to problems if the first true value is default(TSource).

We can fix both by introducing timestamping:

public static IObservable<TSource> ThrottleBy3<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, TimeSpan timeout,
    IEqualityComparer<TKey> comparer = null, IScheduler scheduler = null)
{
    comparer = comparer ?? EqualityComparer<TKey>.Default;
    scheduler = scheduler ?? DefaultScheduler.Instance;

    return source
        .Timestamp(scheduler)
        .StateSelect(Timestamped.Create(default(TSource), DateTimeOffset.MinValue), 
            (prevVal, newVal) => (!comparer.Equals(keySelector(prevVal.Value), keySelector(newVal.Value)) || newVal.Timestamp - prevVal.Timestamp > timeout, newVal), 
            (prevVal, newVal) => !comparer.Equals(keySelector(prevVal.Value), keySelector(newVal.Value)) || newVal.Timestamp - prevVal.Timestamp > timeout ? newVal : prevVal
        )
        .Where(t => t.Item1)
        .Select(t => t.newVal.Value);
}

Here we're storing a Timestamped value as the state, and we alter state if the debounce time is sufficient or the value changes. The result out of that is again, a tuple indicating whether the value should proceed or not, along with the timestamped value.

Hope this helps.

Thanks to very good help from Shlomo, I think I have a good solution to this problem now:

public static IObservable<TSource> ThrottleBy<TSource, TKey>(this IObservable<TSource> source, Func<TSource, TKey> keySelector, TimeSpan timeout,
IEqualityComparer<TKey> comparer = null, IScheduler scheduler = null)
{
    comparer = comparer ?? EqualityComparer<TKey>.Default;
    scheduler = scheduler ?? DefaultScheduler.Instance;

    return source
        .Publish(_val => Observable.Merge(
            _val.Select(v => (value: v, timeout: false)),
            _val.Select(v => (value: v, timeout: true)).Throttle(timeout, scheduler)
        ))
        .Scan((prev: (value: (object)null, timeout: false), emit: (object)null), (state, t) => {
            if (state.prev.value == null) // Initial state
                return (t, null); // Save new state and ignore

            // Emit previous in case of timeout or value changed
            if (t.timeout || (!state.prev.timeout && !comparer.Equals(keySelector(t.value), keySelector((TSource)state.prev.value))))
                return (t, state.prev.value);

            // Save new state and ignore
            return (t, null);
        })
        .Where(x => x.emit != null)
        .Select(x => (TSource)x.emit);
}

It's much like Shlomo's suggestion, but I ended up using Throttle instead of Delay which made it much simpler. Merged in the helper method to make it self contained. I'm boxing the value to avoid default(TSource).