Solving FileSystemWatcher Multiple Changed Events with System.Reactive

FileEvent.cs

  public class FileEvent
  {
    public FileEvent(string fullPath, WatcherChangeTypes changeTypes)
    {
      FullPath = fullPath;
      ChangeType = changeTypes;
    }
    public string FullPath { get; set; }
    public WatcherChangeTypes ChangeType { get; set; }
 }

Program.cs

using FileWatcher;

await Host.CreateDefaultBuilder(args)
  .ConfigureLogging((_, logging) => logging.ClearProviders().AddSimpleConsole(opt => {
    opt.IncludeScopes = false;
    opt.SingleLine = true;
    opt.TimestampFormat = "[hh:mm:ss.fff] ";
  }))
    .ConfigureServices((context, services) => {
      var configRoot = context.Configuration;
      services.Configure<AppSettings>(configRoot.GetSection(nameof(AppSettings))); // put AppSettings into the DI container
      services.AddHostedService<ReactiveWorker>();
    })
    .RunConsoleAsync();

SampleWorker.cs

public class ReactiveWorker : BackgroundService
{
  private readonly ILogger<ReactiveWorker> logger;
  private readonly AppSettings appSettings;

  public ReactiveWorker(ILogger<ReactiveWorker> logger, IOptions<AppSettings> appSettingsAccessor)
  {
    this.logger = logger;
    appSettings = appSettingsAccessor.Value;
  }
  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
    if (string.IsNullOrWhiteSpace(appSettings.WatchDirectory))
    {
      logger.LogError("Configured watch directory is not valid");
      return;
    }

    logger.LogInformation("Watching {watchDirectory} for changes", appSettings.WatchDirectory);

    using var watcher = new FileSystemWatcher(appSettings.WatchDirectory, "*.txt");
    watcher.EnableRaisingEvents = true;

    // We only care about renames and writes.  File creations are notifications about 0-byte files.
    watcher.NotifyFilter = NotifyFilters.FileName | NotifyFilters.LastWrite;
    watcher.IncludeSubdirectories = false;

    var renameEvents = Observable.FromEventPattern<RenamedEventArgs>(watcher, "Renamed").Select(e => new FileEvent(e.EventArgs.FullPath, e.EventArgs.ChangeType));
    var changeEvents = Observable.FromEventPattern<FileSystemEventArgs>(watcher, "Changed").Select(e => new FileEvent(e.EventArgs.FullPath, e.EventArgs.ChangeType));

    changeEvents
      .GroupBy(x => x.FullPath) // splits the stream into sub-streams per file
      .Subscribe(
        group =>
          // we are now dealing with a stream of events for just one file
          group
            .Throttle(TimeSpan.FromSeconds(8)) // on the network large file transfers will file change events ~8s apart!
            .Subscribe(HandleFile)
      );

    // Renames are instant and don't trigger mulitple events
    renameEvents.Subscribe(HandleFile);

    await Task.Delay(TimeSpan.FromDays(1), stoppingToken);
  }

  private void HandleFile(FileEvent e)
  {
    logger.LogInformation("** {file} from {type}", e.FullPath, e.ChangeType);
  }
}