Observable Throttled Buffer?

Question

I am developing a file monitoring system. What I need is to record each change, collect them to buffer, and throttled emit. In other words, when there is a change, the buffer start recording. And if there's no change in 200ms, then emit all the changes (either one by one or collection is OK). The Throttle method does not meet my requirement, since it only return the last element in each group (the green and purple point in the following diagram). Buffer

I've coded following (using GroupByUntil) as suggested in this answer by cwharris:

using var watcher = new FileSystemWatcher("some_path")
{
    Filter = "*.*",
    NotifyFilter = NotifyFilters.FileName | NotifyFilters.Size,
    IncludeSubdirectories = true,
    EnableRaisingEvents = true,
};

var Files = Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
    h => watcher.Created += h, h => watcher.Created -= h)
    .Select(x => x.EventArgs.FullPath)
    .GroupByUntil(x => true, g => g.Throttle(TimeSpan.FromMilliseconds(200)))
    .SelectMany(x => x.ToArray());


using IDisposable handle = Files.Subscribe(x => Console.WriteLine(String.Join("\n", x)));

Is there any other way to implement this? e.g. Using Buffer and bufferClosingSelector?

I think he means debouncing. Since many people confuse debouncing and throttling (including the Rx operator), it’s a pretty fair mistake. — Shlomo, Oct 28 '22 at 11:00
@Shlomo both debouncing and throttling cause elements to be dropped, and I am not seeing any dropped element in the marble diagram. My guess is that the OP is searching for this: [Reactive Throttle returning all items added within the TimeSpan](https://stackoverflow.com/questions/8849810/reactive-throttle-returning-all-items-added-within-the-timespan). — Theodor Zoulias, Oct 28 '22 at 11:41

score 0 · Answer 1 · answered Oct 29 '22 at 02:05

As commented by @Theodor Zoulias, since it's a hot observable, the solution becomes:

var Files = Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
    h => watcher.Created += h, h => watcher.Created -= h)
    .Select(x => x.EventArgs.FullPath);

var emits = Files.Throttle(TimeSpan.FromMilliseconds(200));
var FinalFiles = Files.Window(() => emits).SelectMany(x => x.ToArray());

Observable Throttled Buffer?

1 Answers1