8

イベントの処理を補助するオブザーバブルにファイルウォッチャーをラップすることを検討してきましたが、必要な動作を取得する方法を理解するのに苦労しています。ファイル ウォッチャーは、ファイルが配置されるディレクトリを監視します。ファイルが最初にそのディレクトリに配置されると、ファイル ウォッチャーで Created イベントが発生します。ただし、ファイルが大きい場合やネットワーク接続が遅い場合は、ファイルの更新時に一連の Changed イベントが発生します。書き込みが完了するまでファイルを処理したくないので、本当に必要なのはこのタイムラインです

|Created    |Changed   |Changed   |Changed                      
________________________________________________ 
^Write starts       ^Write finishes       ^Processing Starts

Rx でイベントをフィルタリングする方法をいくつか調べましたが、「ファイル ファイルが X 秒間変更されていない場合に関数を起動する」という必要なものを取得できませんでした。スロットルは途中でイベントを失うのでダメです。バッファ境界でイベントが発生する可能性があるため、バッファは適切ではありません。

タイムアウトを使用することを考えていましたが、例外がスローされたことに夢中ではなく、ファイルが書き込まれているときに処理を開始し、イベントがまったくなくなったときに処理を開始したかったのです。

Reactive Extensions vs FileSystemWatcherには、実際には解決されなかった同様の質問があります。

これを簡単に行う方法はありますか？これは珍しい使用例ではないと確信しています。

4

4 に答える 4

3

ObservableFileSystemWatcherFileSystemWatcher-型の監視可能なラッパー- 完全に機能します。という名前の NuGet パッケージを追加し、ReactiveFileSystemWatcher次のようにテストするコンソール アプリケーションを作成します。

class Program
{
  static void Main(string[] args)
  {
    using (var watcher = new ObservableFileSystemWatcher(c => { c.Path = @"C:\FolderToWatch\"; c.IncludeSubdirectories = true; }))
    {
      watcher.Created.Select(x => $"{x.Name} was {x.ChangeType}").Subscribe(Console.WriteLine);
      watcher.Changed.Select(x => $"{x.Name} was {x.ChangeType}").Subscribe(Console.WriteLine);
      watcher.Renamed.Select(x => $"{x.OldName} was {x.ChangeType} to {x.Name}").Subscribe(Console.WriteLine);
      watcher.Deleted.Select(x => $"{x.Name} was {x.ChangeType}").Subscribe(Console.WriteLine);
      watcher.Errors.Subscribe(Console.WriteLine);
      watcher.Start();
      Console.ReadLine();
    }
  }
}
于 2016-04-17T15:09:12.597 に答える
2

編集：レビューの後、これが欲しいとは思わないでください...

少し単純化しすぎているかもしれませんが、Throttleここでは理想的ではないでしょうか?

これは決して「単純」ではありませんが、私の以前のアイデアよりも、あなたが望んでいることを実行できると思います:

(おまけ: テストケース付き! ;) )

void Main()
{
    var pathToWatch = @"c:\temp\";
    var fsw = new FileSystemWatcher(pathToWatch);

    // set up observables for create and changed
    var changedObs = 
       Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
          dlgt => fsw.Changed += dlgt, 
          dlgt => fsw.Changed -= dlgt);
    var createdObs = 
       Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>( 
          dlgt => fsw.Created += dlgt, 
          dlgt => fsw.Created -= dlgt);

    // the longest we'll wait between last file write and calling it "changed"
    var maximumTimeBetweenWrites = TimeSpan.FromSeconds(1);

    // A "pulse" ticking off every 10ms (adjust this as desired)
    var timer = Observable
        .Timer(TimeSpan.Zero, TimeSpan.FromMilliseconds(10))
        .Select(i => DateTime.Now);

    var watcher = 
        from creation in createdObs
        from change in changedObs
            // we only care about changes matching a create
            .Where(changeEvt => changeEvt.EventArgs.Name == creation.EventArgs.Name)
            // take latest of (pulse, changes) and select (event, time since last file write)
            .CombineLatest(timer, (evt, now) => new {
                    Change = evt, 
                    DeltaFromLast = now.Subtract(new FileInfo(evt.EventArgs.FullPath).LastWriteTime)})
            // skip all until we trigger than "time before considered changed" threshold
            .SkipWhile(evt => evt.DeltaFromLast < maximumTimeBetweenWrites)
            // Then lock on that until we change a diff file
            .Distinct(evt => evt.Change.EventArgs.FullPath)
        select change.Change;

    var disp = new CompositeDisposable();

    // to show creates
    disp.Add(
        createdObs.Subscribe(
           evt => Console.WriteLine("New file:{0}", 
                evt.EventArgs.FullPath)));

    // to show "final changes"
    disp.Add(
        watcher.Subscribe(
           evt => Console.WriteLine("{0}:{1}:{2}", 
                 evt.EventArgs.Name, 
                 evt.EventArgs.ChangeType, 
                 evt.EventArgs.FullPath)));

    fsw.EnableRaisingEvents = true;

    var rnd = new Random();
    Enumerable.Range(0,10)
        .AsParallel()
        .ForAll(i => 
            {
                var filename = Path.Combine(pathToWatch, "foo" + i + ".txt");
                if(File.Exists(filename))
                    File.Delete(filename);

                foreach(var j in Enumerable.Range(0, 20))
                {
                    var writer = File.AppendText(filename);
                    writer.WriteLine(j);
                    writer.Close();
                    Thread.Sleep(rnd.Next(500));
                }
            });

    Console.WriteLine("Press enter to quit...");
    Console.ReadLine();
    disp.Dispose();        
}
于 2013-01-04T19:41:53.807 に答える