In Rx, how to group latest items after a period of time?

Question

Sorry if the title isn't very clear, I couldn't think of anything better...

I'm receiving user input in the form of an IObservable<char>, and I'd like to transform it to an IObservable<char[]>, by grouping the chars every time the user stops typing for more than 1 second. So, for instance, if the input is as follows:

h
e
l
l
o
(pause)
w
o
r
l
d
(pause)
!
(pause)

I'd like the output observable to be:

['h', 'e', 'l', 'l', 'o']
['w', 'o', 'r', 'l', 'd']
['!']

I suspect the solution is fairly simple, but I can't find the right approach... I tried to use Buffer, GroupByUntil, Throttle and a few others, to no avail.

Any idea would be welcome!

---

EDIT: I've got something that almost works:

        _input.Buffer(() => _input.Delay(TimeSpan.FromSeconds(1)))
              .ObserveOnDispatcher()
              .Subscribe(OnCompleteInput);

But I need the delay to be reset every time a new character is typed...

Answer 1

Buffer and Throttle would be enough, if your source is hot. To make it hot, you can use .Publish().RefCount() to ensure you only end up with one subscription to the source.

IObservable<IList<T>> BufferWithInactivity<T>(this IObservable<T> source,
                                              TimeSpan dueTime)
{
    if (source == null) throw new ArgumentNullException("source");
    //defer dueTime checking to Throttle
    var hot = source.Publish().RefCount();
    return hot.Buffer(() => hot.Throttle(dueTime));
}