BatchedDispatcher.cs

이 파일의 문서화 페이지로 가기

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Windows.Threading;
 
namespace Dreamine.Logging.Wpf.Services
{
    public sealed class BatchedDispatcher<T>
    {
        public const int MaxBatchSize = 256;

        private readonly Dispatcher _dispatcher;
        private readonly Action<IReadOnlyList<T>> _onBatch;
        private readonly DispatcherPriority _priority;
        private readonly ConcurrentQueue<T> _pending = new();
        private int _scheduled;

        public BatchedDispatcher(
            Dispatcher dispatcher,
            Action<IReadOnlyList<T>> onBatch,
            DispatcherPriority priority = DispatcherPriority.Background)
        {
            _dispatcher = dispatcher ?? throw new ArgumentNullException(nameof(dispatcher));
            _onBatch = onBatch ?? throw new ArgumentNullException(nameof(onBatch));
            _priority = priority;
        }

        public void Enqueue(T item)
        {
            _pending.Enqueue(item);
            ScheduleFlushIfNeeded();
        }

        private void ScheduleFlushIfNeeded()
        {
            // Ensure only one DispatcherOperation is pending at any time.
            if (Interlocked.CompareExchange(ref _scheduled, 1, 0) != 0)
            {
                return;
            }
 
            _dispatcher.BeginInvoke(_priority, new Action(Flush));
        }

        private void Flush()
        {
            try
            {
                var buffer = new List<T>(Math.Min(_pending.Count, MaxBatchSize));
 
                while (buffer.Count < MaxBatchSize && _pending.TryDequeue(out var item))
                {
                    buffer.Add(item);
                }
 
                if (buffer.Count > 0)
                {
                    _onBatch(buffer);
                }
            }
            finally
            {
                Volatile.Write(ref _scheduled, 0);
 
                // Re-arm if producers added more while we were running.
                if (!_pending.IsEmpty)
                {
                    ScheduleFlushIfNeeded();
                }
            }
        }
    }
}