Tworzenie razem PLINQ i BlockingCollection

Question

Połączyłem prostą aplikację, która monitoruje zdarzenia tworzenia plików, tworzy pewne obiekty z zawartości plików i wykonuje pewne przetwarzanie. Oto przykładowy kod:

class Program 
{ 
    private const string Folder = "C:\\Temp\\InputData"; 

    static void Main(string[] args) 
    { 
     var cts = new CancellationTokenSource(); 
     foreach (var obj in Input(cts.Token)) 
      Console.WriteLine(obj); 
    } 

    public static IEnumerable<object> Input(CancellationToken cancellationToken) 
    { 
     var fileList = new BlockingCollection<string>(); 

     var watcher = new FileSystemWatcher(Folder); 
     watcher.Created += (source, e) => 
     { 
      if (cancellationToken.IsCancellationRequested) 
       watcher.EnableRaisingEvents = false; 
      else if (Path.GetFileName(e.FullPath) == "STOP") 
      { 
       watcher.EnableRaisingEvents = false; 
       fileList.CompleteAdding(); 
       File.Delete(e.FullPath); 
      } 
      else 
       fileList.Add(e.FullPath); 
     }; 
     watcher.EnableRaisingEvents = true; 

     return from file in 
        fileList.GetConsumingEnumerable(cancellationToken) 
       //.AsParallel() 
       //.WithCancellation(cancellationToken) 
       //.WithDegreeOfParallelism(5) 
       let obj = CreateMyObject(file) 
       select obj; 
    } 

    private static object CreateMyObject(string file) 
    { 
     return file; 
    } 
} 

To wszystko działa poprawnie, ale kiedy odkomentowaniu AsParallel (i kolejne dwie linie) to nie przynosi rezultatów od razu. To opóźnienie jest prawdopodobnie spowodowane partycją PLINQ? Oczekuję jednak, że zapytanie wygeneruje elementy, gdy tylko zostaną dodane do BlockingCollection. Czy można to osiągnąć za pomocą PLINQ?

Answer 1

Do tego należy zaprojektować .WithMergeOptions(ParallelMergeOptions.NotBuffered).