Реагирующие события буферов до тех пор, пока не будет запрошен

Question

Есть демо-приложение, которое я подготовил.

using System.Collections.Concurrent; 
using System.Reactive.Linq; 

class Program 
{ 
    static void Main(string[] args) 
    { 
     var stored = new ConcurrentQueue<long>(); 

     Observable.Interval(TimeSpan.FromMilliseconds(20)) 
      .Subscribe(it => stored.Enqueue(it)); 

     var random = new Random(); 

     Task.Run(async() => 
     { 
      while (true) 
      { 
       await Task.Delay((int)(random.NextDouble() * 1000)); 
       var currBatch = stored.ToArray(); 
       for (int i = 0; i < currBatch.Length; i++) 
       { 
        long res; 
        stored.TryDequeue(out res); 
       } 
       Console.WriteLine("[" + string.Join(",", currBatch) + "]"); 
      } 
     }); 

     Console.ReadLine(); 
    } 
} 

Он моделирует независимого потребителя, который срабатывает случайным интервалом времени. В реальном случае источник событий будет поступать из файловой системы, хотя может быть и всплеск.

Что это значит - это хранить неопределенное количество событий в параллельной очереди, пока потребитель не решит использовать собранные события.

У меня есть сильное чувство, что этот код небезопасен. Можно ли воспроизвести такое поведение чисто ручным образом?

Если нет, можете ли вы предложить лучший/более безопасный подход?

Answer 1

Здесь вы идете:

var producer = Observable.Interval(TimeSpan.FromMilliseconds(20)); 
var random = new Random(); 

Task.Run(async() => 
{ 
    var notify = new Subject<int>(); 
    producer.Window(() => notify) 
     .SelectMany(ev => ev.ToList()) 
     .Subscribe(currBatch => Console.WriteLine("[" + string.Join(",", currBatch) + "]")); 

    while (true) 
    { 
     await Task.Delay((int)(random.NextDouble() * 1000)); 
     notify.OnNext(1); 
    } 
}); 

Console.ReadLine();