The following class implements a buffer which captures fast changing datapoints or frequently raised events and dispatches them in batches. It will asynchronously call a processing function for the buffered items under any of the following conditions:A configurable timeout has expired since the last time an item was addedA configurable maximum number of items was addedThe buffer is being disposedThere are several buffering strategies available which control how the items are accumulated.It was written in times of .NET 3.5 and still needs to work with it.Looking for general improvements, potential pitfalls I missed, design improvements etc. A couple if things which annoy me but I never got around changing:Violation of open/closed principle in regards of the accumulation strategies. Hasn't been an issue so far but should be fixed at some point I guess.Using a dedicated processing thread. Putting the main method on the thread pool is not really an option because it's long running. Doing it on a timer callback would be nice but it's a bit tricky with the different signals which can trigger a dispatch of the buffered items.Code:public class AccumulationBuffer<TKey, TValue> : IDisposable{    public enum RecordingStrategy    {        AllFifo,        AllLifo,        FirstWins,        LastWins    }    private interface IStorage : IEnumerable<KeyValuePair<TKey, TValue>>    {        void Add(KeyValuePair<TKey, TValue> pair);        void Clear();    }    private class StorageAllFifo : Queue<KeyValuePair<TKey, TValue>>, IStorage    {        public StorageAllFifo(int capacity)            : base(capacity)        {        }        public void Add(KeyValuePair<TKey, TValue> pair)        {            Enqueue(pair);        }    }    private class StorageAllLifo : Stack<KeyValuePair<TKey, TValue>>, IStorage    {        public StorageAllLifo(int capacity)            : base(capacity)        {        }        public void Add(KeyValuePair<TKey, TValue> pair)        {            Push(pair);        }    }    private class StorageMap : Dictionary<TKey, TValue>, IStorage    {        private readonly bool _SkipIfExists;        public StorageMap(int capacity, bool skipIfExists)            : base(capacity)        {            _SkipIfExists = skipIfExists;        }        public void Add(KeyValuePair<TKey, TValue> pair)        {            if (!_SkipIfExists || !ContainsKey(pair.Key))                this[pair.Key] = pair.Value;        }    }    private readonly Action<IEnumerable<KeyValuePair<TKey, TValue>>> _ItemProcessor;    private readonly IStorage _Storage;    private readonly RecordingStrategy _Strategy;    private readonly object _StorageLock = new object();    private Thread _MonitorThread;    private volatile bool _Quit;    private readonly AutoResetEvent _MonitorEvent;    private DateTime _LastAddedTime;    private readonly TimeSpan _InactivityTime;    private readonly int _MaximumAccumulationCount;    private int _CurrentAccumulationCount;    public AccumulationBuffer(RecordingStrategy strategy, TimeSpan inactivityTime, int maximumAccumulationCount, Action<IEnumerable<KeyValuePair<TKey, TValue>>> itemProcessor)    {        if (itemProcessor == null)            throw new ArgumentNullException(itemProcessor);        _ItemProcessor = itemProcessor;        _LastAddedTime = DateTime.MinValue;        _InactivityTime = inactivityTime;        _MaximumAccumulationCount = maximumAccumulationCount;        _MonitorEvent = new AutoResetEvent(false);        _Quit = false;        _CurrentAccumulationCount = 0;        _Strategy = strategy;        switch (_Strategy)        {            case RecordingStrategy.AllFifo:                _Storage = new StorageAllFifo(maximumAccumulationCount);                break;            case RecordingStrategy.AllLifo:                _Storage = new StorageAllLifo(maximumAccumulationCount);                break;            case RecordingStrategy.FirstWins:                _Storage = new StorageMap(maximumAccumulationCount, true);                break;            case RecordingStrategy.LastWins:                _Storage = new StorageMap(maximumAccumulationCount, false);                break;            default:                throw new ArgumentOutOfRangeException(strategy);        }        _MonitorThread = new Thread(MonitorLoop)                             {                                 IsBackground = true,                                 Name = string.Format(Accumulation Buffer Monitor ({0}/{1}), typeof (TKey).Name, typeof (TValue).Name)                             };        _MonitorThread.Start();    }    public void Add(TKey key, TValue value)    {        lock (_StorageLock)        {            _Storage.Add(new KeyValuePair<TKey, TValue>(key, value));            _LastAddedTime = DateTime.Now;            _CurrentAccumulationCount++;            _MonitorEvent.Set();        }    }    private void MonitorLoop()    {        while (!_Quit)        {            bool timeout = false;            while (!timeout && !_Quit)            {                TimeSpan wait = TimeSpan.FromMilliseconds(-1);                lock (_StorageLock)                {                    if (_CurrentAccumulationCount >= _MaximumAccumulationCount)                        break;                    if (_LastAddedTime != DateTime.MinValue)                    {                        wait = _InactivityTime;                    }                }                timeout = !_MonitorEvent.WaitOne(wait); // WaitOne returns false if it timed out            }            // if we are here we got either             // - woken up to quit OR            // - the event has timed out after the inactivity timeout has passed OR            // - the maximum number of items was accumulated            lock (_StorageLock)            {                // always process pending items, whether we quit or not                if (_CurrentAccumulationCount > 0)                {                    var toProcess = new List<KeyValuePair<TKey, TValue>>(_Storage);                    _Storage.Clear();                    _LastAddedTime = DateTime.MinValue;                    _CurrentAccumulationCount = 0;                    ThreadPool.QueueUserWorkItem(x => _ItemProcessor(toProcess));                }                if (_Quit) return;            }        }    }    public void Dispose()    {        Dispose(true);        GC.SuppressFinalize(this);    }    protected virtual void Dispose(bool disposing)    {        if (disposing)        {            if (_MonitorThread != null)            {                _Quit = true;                _MonitorEvent.Set();                if (!_MonitorThread.Join(100))                {                    _MonitorThread.Abort();                }                _MonitorThread = null;            }        }    }}
Buffering of fast changing datapoints or events
c#;.net;asynchronous
Also, there isn't much to say (you have already mentioned the flaws) some nitpickings  You should be consistent in your coding style  if (_CurrentAccumulationCount >= _MaximumAccumulationCount)    break;if (_LastAddedTime != DateTime.MinValue){    wait = _InactivityTime;}  Sometimes you are using braces {} for single if statements (which i preffer) and sometimes not.  I would extract the retrieving of the timeout TimeSpan into a separate method,private TimeSpan GetWaitingTime(){     lock (_StorageLock)     {         if(_CurrentAccumulationCount >= _MaximumAccumulationCount)         {             return TimeSpan.Zero;         }         if (_LastAddedTime != DateTime.MinValue)         {             return _InactivityTime;         }     }     return TimeSpan.FromMilliseconds(-1);}and refactor the while (!timeout && !_Quit) loop to   bool hasTimedOut = false;while (!hasTimedOut && !_Quit){    TimeSpan waitingTime = GetWaitingTime();    if(waitingTime == TimeSpan.Zero)    {        break;    }    hasTimedOut = !_MonitorEvent.WaitOne(waitingTime); // WaitOne returns false if it timed out}The constructor of StorageAllFifo and StorageAllLifo can be removed, as the base constructor is called anyway.