FileSystemWatcherでのイベント二重起動を防止しつつ、同時更新にも対応したい

Question

実現したいこと

使用言語:C#
開発環境:Visual Studio 2022

指定フォルダを監視し、対象フォルダ内にファイルが投入された場合に、それを入力としてバッチファイルを作成・起動するWindowsFormsアプリケーションを開発しています。バッチ起動後は、エラーコードを取得しています。
バッチファイルでは更に別のアプリケーションを起動しており、そちらがスレッドにより同時起動できるため、非同期処理をしたいと考えています。

発生している問題・分からないこと

FileSystemWatcherを使用すると、ファイルの新規作成や更新が一度だけでもイベントが複数回発生することがあります。それに対処するため、Reactive ExtensionsのThrottleにて指定秒数待機し、イベントをまとめてしまう方法をとっています。

この方法で1ファイルずつ投入されるパターンには対処できました。
しかし、複数のファイルが一度に投入された場合に、指定秒数以内にイベントがまとめて発生するためか、イベントが一つしか発生しません。

まとめると、以下のようなことを実現したいと思っています。
・FileSystemWatcherでのChangedイベント重複発生を回避したい
・ただし、複数ファイルがまとめて投入された(イベント発生は同時だがe.FullPathの値が異なっている)場合は、各ファイルで発生したイベントを全て実行させたい

該当のソースコード

using Microsoft.VisualBasic.ApplicationServices;
using Microsoft.VisualBasic.Logging;
using System.Diagnostics;
using System.Formats.Tar;
using System.Text;
using System.IO;
using static System.Net.Mime.MediaTypeNames;
using static System.Runtime.CompilerServices.RuntimeHelpers;
using System;
using System.Reactive.Linq;

namespace マルチスレッドSample {

    public partial class Form1 : Form {
        public Form1() {
            InitializeComponent();
        }

        private FileSystemWatcher watcher = null;

        public string createBat(string inputFile) {
            DateTime productionDT = DateTime.Now;
            string thNo = "-" + Thread.CurrentThread.ManagedThreadId.ToString();

            Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);

            //バッチ作成
            string batFileName = "マルチスレッドsample_" + productionDT.ToString("yyyyMMddHHmmss") + thNo + ".bat";
            using(var sw = new StreamWriter(batFileName,false,Encoding.GetEncoding("shift_jis"))) {
                sw.WriteLine("説明用サンプル");
                sw.WriteLine("@echo errorlevel:%errorlevel%");
            }

            //ProcessStartInfoインスタンスを生成
            ProcessStartInfo processInfo = new ProcessStartInfo(batFileName){
                CreateNoWindow = true,
                RedirectStandardOutput = true,
                UseShellExecute = false
            };
            //バッチ実行
            Process process = Process.Start(processInfo);
            //バッチファイルの実行結果を変数outputに格納する
            string output = process.StandardOutput.ReadToEnd();
            process.WaitForExit();

            //エラーレベルを取得するための配列
            string[] arr = output.Split("errorlevel:");
            //呼び出し元にエラーコードを返す
            return arr[1].Substring(0,1);
        }

        private void button1_Click(object sender,EventArgs e) {
            watcher = new FileSystemWatcher();
            
            watcher.Path = @"C:\Users\UserName\test\Input";
            watcher.NotifyFilter = 
                NotifyFilters.FileName
                | NotifyFilters.DirectoryName
                | NotifyFilters.LastWrite
                | NotifyFilters.LastAccess;

            watcher.Filter = "*.csv";
            watcher.SynchronizingObject = this;

            //Rx使用
            watcher.ChangedAsObservable()
                //1秒待機
                .Throttle(TimeSpan.FromSeconds(1))
                .Subscribe(e => {  
                    Task.Run(() => {
                        string rtnCode = createBat(e.FullPath);  
                    });
                });

            //監視開始
            watcher.EnableRaisingEvents = true;
        }
    } 
        static class FileSystemWatcherExtensions{
        //ChangedイベントをIObservable<TEventArgs>にする
        public static IObservable<FileSystemEventArgs> ChangedAsObservable(this FileSystemWatcher self){
            return Observable.FromEvent<FileSystemEventHandler, FileSystemEventArgs>(
                h => (_, e) => h(e),
                h => self.Changed += h,
                h => self.Changed -= h);
            }
        }
 }

試したこと・調べたこと

• teratailやGoogle等で検索した
• ソースコードを自分なりに変更した
• 知人に聞いた
• その他

上記の詳細・結果

自分では、一度createBatを実行した後にe.FullPathを変数に保存し、次回実行時にその変数とe.FullPathを比較する方法を考えました。両者が一致していなければ待機せずそのままcreateBatを実行し、一致している場合のみ待機するといった具合です。
しかし、(考えてみれば当然なのですが)、この方法ですとThrottleによる待機をしない場合はやはり重複してイベントが発生します。

あくまでもThrottleでの待機は必ず行い、その中で更新されたファイル名が変わった場合と重複している場合で、処理を分岐しないといけないのだと推測しています。

不足している情報がございましたら、お手数ですがご指摘ください。
非同期処理やRxの使い方に不慣れなため、基本的な内容であればご容赦いただけますと幸いです。

補足

特になし

Answer 1

GroupByすればいいんじゃないですかね？
Reactive Extensions再入門 その２０「GroupByメソッドでグルーピングしてみよう」 - かずきのBlog@hatena

FileSystemWatcherだと確認しにくいので雑なSubjectで^^;

using System.Diagnostics;
using System.Reactive.Linq;
using System.Reactive.Subjects;


var sw = new Stopwatch();

var s = new Subject<int>();
s.GroupBy(i => i) // 実際はi.FullPathとかi.Nameとかに
    .Subscribe(g =>
    {
        g.Throttle(TimeSpan.FromSeconds(1))
         .Subscribe(i => Console.WriteLine($@"{sw.Elapsed:ss\.fff} {i} fire!"));
    });


sw.Start();

foreach (var _ in Enumerable.Range(1, 3))
{
    foreach (var i in Enumerable.Range(1, 3))
    {
        Console.WriteLine($@"{sw.Elapsed:ss\.fff} {i}");
        s.OnNext(i);
        Thread.Sleep(200);
    }
}
foreach (var _ in Enumerable.Range(1, 5))
{
    foreach (var i in Enumerable.Range(1, 2).Reverse())
    {
        Console.WriteLine($@"{sw.Elapsed:ss\.fff} {i}");
        s.OnNext(i);
        Thread.Sleep(200);
    }
}

Console.ReadKey();

00.000 1
00.226 2
00.440 3
00.655 1
00.869 2
01.082 3
01.296 1
01.509 2
01.723 3
01.945 2
02.150 1
02.352 2
02.566 1
02.736 3 fire!
02.781 2
02.994 1
03.208 2
03.422 1
03.637 2
03.851 1
04.641 2 fire!
04.854 1 fire!

---

GroupByUntilの例もありましたが、厳密には意味が違うような気がします（実用上は大差ないかもしれないが）
c# - Rx grouped throttling - Stack Overflow

追記

yajiyajiさんのコードに合わせるとすると、こうですかね？（動作確認はしてません^^;

watcher.ChangedAsObservable()
    .GroupBy(i => i.FullPath) // FullPathでグループ化
    .Subscribe(g =>
    {
        // ここは新たなグループ（FullPath）があったときに来る（Changedイベント毎に来るのではない）
        Debug.WriteLine(g.Key);

        // グループ毎にThrottle（1秒以上同じFullPathのイベントがこなければ最後のイベントを後続に流す）
        g.Throttle(TimeSpan.FromSeconds(1))
        // 購読（つまりグループ毎に別々に購読してる）
         .Subscribe(e =>
         {
             // ここが目的のFullPath毎のThrottle結果
             Task.Run(() =>
             {
                 string rtnCode = createBat(e.FullPath);
             });
         });
    });

Comments

[yajiyaji]

ご回答ありがとうございます。
また、ご丁寧にコードと出力結果まで記載していただきましたこと、重ねてお礼申し上げます。

i.FullPathの値でグループ化できれば、確かに実現できそうに思います。
私がRxに不慣れなため、自身のケースへの応用にお時間をいただいてしまうと思います。
何かしら進展がございましたら、改めてコメントさせていただきます。

[TN8001]

> 私がRxに不慣れなため、自身のケースへの応用にお時間をいただいてしまうと思います。

ああ、すいません。Subscribeがネストしてるんで分かりにくいですかね^^;
回答に追記しておきました。

[yajiyaji]

ご回答いただいた内容で解決しました！

ご丁寧に追記までしていただき、とても感謝しています。
分かりにくいなんてとんでもないです。催促してしまったようで恐縮です。

ご親切にありがとうございました。
ベストアンサーにさせていただきました。

Answer 2

私が話をよくわかってない気がするのですが，

指定フォルダを監視し、対象フォルダ内にファイルが投入された場合に…

という話は，
「何らかの ( FileSystemWatcher ではない別の) 手段で指定フォルダ内のファイルを列挙する」ようなことができるとすれば，その手段でフォルダ内の状態をポーリングすることで実施することができるのではないでしょうか？

で，その「ポーリング」のタイミングというのを FileSystemWatcher のイベントから決めるとかすればどうなのでしょう？

指定秒数待機し、イベントをまとめてしまう

ではなくて，例えば
「指定秒数待機してから　→　その別手段で指定フォルダ内のファイルを列挙（して前回と比較）」
みたくすればどうでしょうか．

「指定秒数」の間は単にイベントは無視してしまえばそれで重複は回避できそうですし，複数のファイルが投入されたとしても（その秒数以内に投入が完了するならば，ですが）問題ないのではないでしょうか．

Comments

[fana]

質問内容はそういう話ではない（FileSystemWatcherだけでやる方法の話をしている）ようにも思える．
そのような場合にはガン無視していただきたく．

[fana]

おっと，「ガン無視」だと，他の人に「違う，そうじゃない」旨が伝わらないので良くないかもですね．
お手数ですが「そういう話をしてるんじゃねぇんだよ，素人はすっこんでろ」的な明言が必要かもです．

[yajiyaji]

ご回答ありがとうございます。
FileSystemWatcherに限定しているわけではないので、大変参考になります。
別の手段で例えばリストにファイル名一覧を保存しておき、FileSystemWatcherのイベント発生時に差分を取得するということですよね。その方向性でも実現できそうに思います。

別の方からもご回答いただいており、両方を試すのにお時間をいただいてしまうと思います。
どういった方法を採用するにせよ、進展がございましたら改めてコメントいたします。

[fana]

> 別の方からもご回答いただいており、両方を試すのにお時間をいただいてしまうと思います

私は「Rxって何？　ガンダム？」みたいな素人なので，他回答の内容に関してはさっぱりわかりませんが，
あなたから見て有望そうな話を試せばよいだけだと思いますよ．
（「全部を実際に試す義務」みたいなのは無いハズなので，ある方法で問題が解決したならばそこで終了して良い）

[yajiyaji]

寛大なお言葉、ありがとうございます。
今回は、TN8001様の方法で解決しましたのでご報告いたします。

FileSystemWatcherでファイル名を取得することに拘泥してしまっていたので、別の方法を検討するという視点はとても参考になりました。

ご親切なコメント、ありがとうございました。
(余談ですが、私もRxと聞くとガンダムを連想します)