Как я могу надежно перемещать файлы (асинхронно) из каталога по мере их создания?

Question

В моем решении Windows Service есть FileSystemWatcher контроль дерева каталогов для новых файлов, и всякий раз, когда он запускает событие Created, я пытаюсь переместить файлы асинхронно на другой сервер для дальнейшей обработки. Вот код:

foreach (string fullFilePath in 
       Directory.EnumerateFiles(directoryToWatch, "*.*", 
             SearchOption.AllDirectories) 
         .Where(filename => fileTypes.Contains(Path.GetExtension(filename)))) 
      { 
       string filename = Path.GetFileName(fullFilePath); 
       using (FileStream sourceStream = File.Open(filename, FileMode.Open, FileAccess.Read)) 
       { 
        using (FileStream destStream = File.Create(Path.Combine(destination, filename))) 
        { 
         await sourceStream.CopyToAsync(destStream); 
        } 
       } 
      } 

Проблема заключается в том, что, поскольку эти файлы копируются в папку я наблюдаю, они не всегда разблокированы и доступны для меня. Я хочу «повторить попытку», когда я удаляю заблокированный файл, но я не привык думать асинхронно, поэтому я понятия не имею, как вернуть файл с ошибкой в ​​очередь.

Answer 1

Прежде всего, вам нужно «обнаружить» исключения, возникающие в процессе асинхронного выполнения. Это можно сделать что-то вроде этого:

 try 
     { 
      await sourceStream.CopyToAsync(destStream); 
     } 
     catch (Exception copyException) 
     { 
     } 

После исключения обнаружено и надлежащим образом т.е. вы решили, что один конкретное исключением является причиной для повторных попыток, вы должны будете поддерживать свою собственную очередь копирования целей (и пункты назначения), которые должны быть повторены.

Затем вам нужно будет организовать новую точку входа, которая приведет к самой попытке. Такая точка входа может быть вызвана таймером или следующим событием из используемого вами монитора файловой системы (который я бы не рекомендовал). Вам также придется реализовать обнаружение переполнения вашей очереди для случая множественных сбоев. Имейте в виду, что такое обнаружение переполнения также присутствует в мониторе файловой системы, который может просто пропустить уведомление, если слишком много системных событий (многие файлы, как представляется, копируются в контролируемые папки одновременно).

Если эти вопросы вас не сильно беспокоят, я предлагаю вам реализовать таймер или, точнее, тайм-аут, чтобы повторить задачу копирования. Если, с другой стороны, вам нужно надежное решение, я бы сам реализовал монитор файловой системы.

Что касается тайм-аута он может выглядеть следующим образом:

private Queue<myCopyTask> queue; 
    private Timer retryTimeout; 

    public Program() 
    { 
     retryTimeout = new Timer(QueueProcess, null, Timeout.Infinite, Timeout.Infinite); 
    } 

    private void FileSystemMonitorEventhandler() 
    { 
     //New tasks are provided by the file system monitor. 
     myCopyTask newTask = new myCopyTask(); 
     newTask.sourcePath = "..."; 
     newTask.destinationPath = "..."; 

     //Keep in mind that queue is touched from different threads. 
     lock (queue) 
     { 
      queue.Enqueue(newTask); 
     } 

     //Keep in mind that Timer is touched from different threads. 
     lock (retryTimeout) 
     { 
      retryTimeout.Change(1000, Timeout.Infinite); 
     } 
    } 

    //Start this routine only via Timer. 
    private void QueueProcess(object iTimeoutState) 
    { 
     myCopyTask task = null; 

     do 
     { 
      //Keep in mind that queue is touched from different threads. 
      lock (queue) 
      { 
       if (queue.Count > 0) 
       { 
        task = queue.Dequeue(); 
       } 
      } 

      if (task != null) 
      { 
       CopyTaskProcess(task); 
      } 
     } while (task != null); 
    } 

    private async void CopyTaskProcess(myCopyTask task) 
    { 
     FileStream sourceStream = null; 
     FileStream destStream = null; 

     try 
     { 
      sourceStream = File.OpenRead(task.sourcePath); 
      destStream = File.OpenWrite(task.destinationPath); 
      await sourceStream.CopyToAsync(destStream); 
     } 
     catch (Exception copyException) 
     { 
      task.retryCount++; 

      //In order to avoid instant retries on several problematic tasks you probably 
      //should involve a mechanism to delay retries. Keep in mind that this approach 
      //delays worker thread that is implicitly involved by await keyword. 
      Thread.Sleep(100); 

      //Keep in mind that queue is touched from different threads. 
      lock (queue) 
      { 
       queue.Enqueue(task); 
      } 

      //Keep in mind that Timer is touched from different threads. 
      lock (retryTimeout) 
      { 
       retryTimeout.Change(1000, Timeout.Infinite); 
      } 
     } 
     finally 
     { 
      if (sourceStream != null) 
      { 
       sourceStream.Close(); 
      } 

      if (destStream != null) 
      { 
       destStream.Close(); 
      } 
     } 
    } 
} 

internal class myCopyTask 
{ 
    public string sourcePath; 
    public string destinationPath; 
    public long retryCount; 
}