Hacer que PLINQ y BlockingCollection funcionen juntos

Question

He creado una aplicación sencilla que supervisa los eventos de creación de archivos, crea algunos objetos a partir del contenido de los archivos y procesa algo. Aquí está el código de ejemplo:

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; 
    } 
} 

Todo funciona bien, pero cuando me Descomentar AsParallel (y las siguientes dos líneas) que no produce los resultados de inmediato. ¿Esta demora es probablemente causada por el particionamiento PLINQ? Sin embargo, espero que esta consulta genere elementos tan pronto como se agreguen a BlockingCollection. ¿Es esto posible de usar PLINQ?

Answer 1

Para eso se debe diseñar .WithMergeOptions(ParallelMergeOptions.NotBuffered).