Add option to block when the queue is full, instead of dropping events

Also, use `GetConsumingEnumerable` to enumerate the queue, instead of a `while` loop

Author: cocowalla

src/Serilog.Sinks.Async/LoggerConfigurationAsyncExtensions.cs

@@ -16,17 +16,20 @@ public static class LoggerConfigurationAsyncExtensions
         /// <param name="loggerSinkConfiguration">The <see cref="LoggerSinkConfiguration"/> being configured.</param>
         /// <param name="configure">An action that configures the wrapped sink.</param>
         /// <param name="bufferSize">The size of the concurrent queue used to feed the background worker thread. If
-        /// the thread is unable to process events quickly enough and the queue is filled, subsequent events will be
-        /// dropped until room is made in the queue.</param>
+        /// the thread is unable to process events quickly enough and the queue is filled, depending on 
+        /// <paramref name="blockWhenFull"/> the queue will block or subsequent events will be dropped until 
+        /// room is made in the queue.</param>
+        /// <param name="blockWhenFull">Block when the queue is full, instead of dropping events.</param>
         /// <returns>A <see cref="LoggerConfiguration"/> allowing configuration to continue.</returns>
         public static LoggerConfiguration Async(
             this LoggerSinkConfiguration loggerSinkConfiguration,
             Action<LoggerSinkConfiguration> configure,
-            int bufferSize = 10000)
+            int bufferSize = 10000,
+            bool blockWhenFull = false)
         {
             return LoggerSinkConfiguration.Wrap(
                 loggerSinkConfiguration,
-                wrappedSink => new BackgroundWorkerSink(wrappedSink, bufferSize),
+                wrappedSink => new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull),
                 configure);
         }
     }

src/Serilog.Sinks.Async/Sinks/Async/BackgroundWorkerSink.cs

@@ -12,17 +12,18 @@ sealed class BackgroundWorkerSink : ILogEventSink, IDisposable
     {
         readonly ILogEventSink _pipeline;
         readonly int _bufferCapacity;
+        readonly bool _blockWhenFull;
         volatile bool _disposed;
-        readonly CancellationTokenSource _cancel = new CancellationTokenSource();
         readonly BlockingCollection<LogEvent> _queue;
         readonly Task _worker;
 
-        public BackgroundWorkerSink(ILogEventSink pipeline, int bufferCapacity)
+        public BackgroundWorkerSink(ILogEventSink pipeline, int bufferCapacity, bool blockWhenFull)
         {
             if (pipeline == null) throw new ArgumentNullException(nameof(pipeline));
             if (bufferCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(bufferCapacity));
             _pipeline = pipeline;
             _bufferCapacity = bufferCapacity;
+            _blockWhenFull = blockWhenFull;
             _queue = new BlockingCollection<LogEvent>(_bufferCapacity);
             _worker = Task.Factory.StartNew(Pump, CancellationToken.None, TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
         }
@@ -31,36 +32,35 @@ public void Emit(LogEvent logEvent)
         {
             // The disposed check is racy, but only ensures we don't prevent flush from
             // completing by pushing more events.
-            if (!_disposed && !_queue.TryAdd(logEvent))
-                SelfLog.WriteLine("{0} unable to enqueue, capacity {1}", typeof(BackgroundWorkerSink), _bufferCapacity);
+            if (_disposed)
+                return;
+
+            if (!this._blockWhenFull)
+            {
+                if (!_queue.TryAdd(logEvent))
+                    SelfLog.WriteLine("{0} unable to enqueue, capacity {1}", typeof(BackgroundWorkerSink), _bufferCapacity);
+            }
+            else
+            {
+                this._queue.Add(logEvent);
+            }
         }
 
         public void Dispose()
         {
             _disposed = true;
-            _cancel.Cancel();
+            _queue.CompleteAdding();
             _worker.Wait();            
             (_pipeline as IDisposable)?.Dispose();
-            // _cancel not disposed, because it will make _cancel.Cancel() non-idempotent
         }
 
         void Pump()
         {
             try
             {
-                try
-                {
-                    while (true)
-                    {
-                        var next = _queue.Take(_cancel.Token);
-                        _pipeline.Emit(next);
-                    }
-                }
-                catch (OperationCanceledException)
+                foreach (var next in _queue.GetConsumingEnumerable())
                 {
-                    LogEvent next;
-                    while (_queue.TryTake(out next))
-                        _pipeline.Emit(next);
+                    _pipeline.Emit(next);
                 }
             }
             catch (Exception ex)

test/Serilog.Sinks.Async.Tests/BackgroundWorkerSinkSpec.cs

@@ -1,26 +1,28 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
 using Serilog.Core;
+using Serilog.Debugging;
 using Serilog.Events;
 using Serilog.Parsing;
-using Serilog.Sinks.Async.Tests;
 using Serilog.Sinks.Async.Tests.Support;
 using Xunit;
 
 namespace Serilog.Sinks.Async.Tests
 {
     public class BackgroundWorkerSinkSpec : IDisposable
     {
+        readonly Logger _logger;
         readonly MemorySink _innerSink;
-        readonly BackgroundWorkerSink _sink;
+        BackgroundWorkerSink _sink;
 
         public BackgroundWorkerSinkSpec()
         {
             _innerSink = new MemorySink();
-            var logger = new LoggerConfiguration().WriteTo.Sink(_innerSink).CreateLogger();
-            _sink = new BackgroundWorkerSink(logger, 10000);
+            _logger = new LoggerConfiguration().WriteTo.Sink(_innerSink).CreateLogger();
+            _sink = new BackgroundWorkerSink(_logger, 10000, false);
         }
 
         public void Dispose()
@@ -31,14 +33,15 @@ public void Dispose()
         [Fact]
         public void WhenCtorWithNullSink_ThenThrows()
         {
-            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000));
+            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000, false));
         }
 
         [Fact]
         public async Task WhenEmitSingle_ThenRelaysToInnerSink()
         {
             var logEvent = CreateEvent();
             _sink.Emit(logEvent);
+            _sink.Dispose();
 
             await Task.Delay(TimeSpan.FromSeconds(3));
 
@@ -80,6 +83,77 @@ public async Task WhenEmitMultipleTimes_ThenRelaysToInnerSink()
             Assert.Equal(3, _innerSink.Events.Count);
         }
 
+        [Fact]
+        public async Task WhenQueueFull_ThenDropsEvents()
+        {
+            _sink = new BackgroundWorkerSink(_logger, 1, false);
+
+            // Cause a delay when emmitting to the inner sink, allowing us to fill the queue to capacity 
+            // after the first event is popped
+            _innerSink.DelayEmit = true;
+
+            var events = new List<LogEvent>
+            {
+                CreateEvent(),
+                CreateEvent(),
+                CreateEvent(),
+                CreateEvent(),
+                CreateEvent()
+            };
+            events.ForEach(e =>
+            {
+                var sw = Stopwatch.StartNew();
+                _sink.Emit(e);
+                sw.Stop();
+
+                Assert.True(sw.ElapsedMilliseconds < 2000, "Should not block the caller when the queue is full");
+            });
+
+            // If we *weren't* dropped events, the delay in the inner sink would mean the 5 events would take 
+            // at least 15 seconds to process
+            await Task.Delay(TimeSpan.FromSeconds(18));
+
+            // Events should be dropped
+            Assert.Equal(2, _innerSink.Events.Count);
+        }
+
+        [Fact]
+        public async Task WhenQueueFull_ThenBlocks()
+        {
+            _sink = new BackgroundWorkerSink(_logger, 1, true);
+
+            // Cause a delay when emmitting to the inner sink, allowing us to fill the queue to capacity 
+            // after the first event is popped
+            _innerSink.DelayEmit = true;
+
+            var events = new List<LogEvent>
+            {
+                CreateEvent(),
+                CreateEvent(),
+                CreateEvent()
+            };
+
+            int i = 0;
+            events.ForEach(e =>
+            {
+                var sw = Stopwatch.StartNew();
+                _sink.Emit(e);
+                sw.Stop();
+
+                // Emit should return immediately the first time, since the queue is not yet full. On 
+                // subsequent calls, the queue should be full, so we should be blocked
+                if (i > 0)
+                {
+                    Assert.True(sw.ElapsedMilliseconds > 2000, "Should block the caller when the queue is full");
+                }
+            });
+
+            await Task.Delay(TimeSpan.FromSeconds(12));
+
+            // No events should be dropped
+            Assert.Equal(3, _innerSink.Events.Count);
+        }
+
         private static LogEvent CreateEvent()
         {
             return new LogEvent(DateTimeOffset.MaxValue, LogEventLevel.Error, null,

test/Serilog.Sinks.Async.Tests/BackgroundWorkerSinkTests.cs

@@ -1,9 +1,4 @@
-using System;
-using System.Threading;
-using Serilog.Core;
-using Serilog.Events;
-using Serilog.Parsing;
-using Serilog.Sinks.Async.Tests.Support;
+using Serilog.Sinks.Async.Tests.Support;
 using Xunit;
 
 namespace Serilog.Sinks.Async.Tests

test/Serilog.Sinks.Async.Tests/Support/MemorySink.cs

@@ -2,16 +2,21 @@
 using Serilog.Core;
 using System.Collections.Concurrent;
 using System;
+using System.Threading.Tasks;
 
 namespace Serilog.Sinks.Async.Tests.Support
 {
     public class MemorySink : ILogEventSink
     {
         public ConcurrentBag<LogEvent> Events { get; } = new ConcurrentBag<LogEvent>();
         public bool ThrowAfterCollecting { get; set; }
+        public bool DelayEmit { get; set; }
 
         public void Emit(LogEvent logEvent)
         {
+            if (DelayEmit)
+                Task.Delay(TimeSpan.FromSeconds(3)).Wait();
+
             Events.Add(logEvent);
 
             if (ThrowAfterCollecting)