Fulfills #20 - add option to block when the queue is full, instead of dropping events

src/Serilog.Sinks.Async/LoggerConfigurationAsyncExtensions.cs

@@ -1,6 +1,6 @@
 using System;
+using System.ComponentModel;
 using Serilog.Configuration;
-
 using Serilog.Sinks.Async;
 
 namespace Serilog
@@ -16,18 +16,40 @@ 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
+        /// 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>
         /// <returns>A <see cref="LoggerConfiguration"/> allowing configuration to continue.</returns>
+        [EditorBrowsable(EditorBrowsableState.Never)]
         public static LoggerConfiguration Async(
             this LoggerSinkConfiguration loggerSinkConfiguration,
             Action<LoggerSinkConfiguration> configure,
-            int bufferSize = 10000)
+            int bufferSize)
+        {
+            return loggerSinkConfiguration.Async(configure, bufferSize, false);
+        }
+
+        /// <summary>
+        /// Configure a sink to be invoked asynchronously, on a background worker thread.
+        /// </summary>
+        /// <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, 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,
+            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,55 +12,63 @@ sealed class BackgroundWorkerSink : ILogEventSink, IDisposable
     {
         readonly ILogEventSink _pipeline;
         readonly int _bufferCapacity;
-        volatile bool _disposed;
-        readonly CancellationTokenSource _cancel = new CancellationTokenSource();
+        readonly bool _blockWhenFull;
         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);
         }
 
         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 (this._queue.IsAddingCompleted)
+                return;
+
+            try
+            {
+                if (_blockWhenFull)
+                {
+                    _queue.Add(logEvent);
+                }
+                else
+                {
+                    if (!_queue.TryAdd(logEvent))
+                        SelfLog.WriteLine("{0} unable to enqueue, capacity {1}", typeof(BackgroundWorkerSink), _bufferCapacity);
+                }
+            }
+            catch (InvalidOperationException)
+            {
+                // Thrown in the event of a race condition when we try to add another event after 
+                // CompleteAdding has been called
+            }
         }
 
         public void Dispose()
         {
-            _disposed = true;
-            _cancel.Cancel();
-            _worker.Wait();            
+            // Prevent any more events from being added
+            _queue.CompleteAdding();
+
+            // Allow queued events to be flushed
+            _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,27 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
 using Serilog.Core;
 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,7 +32,7 @@ public void Dispose()
         [Fact]
         public void WhenCtorWithNullSink_ThenThrows()
         {
-            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000));
+            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000, false));
         }
 
         [Fact]
@@ -80,6 +81,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 = TimeSpan.FromMilliseconds(300);
+
+            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 < 200, "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(2));
+
+            // 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 = TimeSpan.FromMilliseconds(300);
+
+            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 > 200, "Should block the caller when the queue is full");
+                }
+            });
+
+            await Task.Delay(TimeSpan.FromSeconds(2));
+
+            // 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 TimeSpan? DelayEmit { get; set; }
 
         public void Emit(LogEvent logEvent)
         {
+            if (DelayEmit.HasValue)
+                Task.Delay(DelayEmit.Value).Wait();
+
             Events.Add(logEvent);
 
             if (ThrowAfterCollecting)