re #18483 - Making the EMG.Extensions.Logging.Loggly awaitable (#5)

* re #18483 - Making the EMG.Extensions.Logging.Loggly awaitable

Added possibility to flush the messages in the logglyprocessor instead of having to wait for the buffer

* Update README.md

Trigger build

* re #18483 - Making the EMG.Extensions.Logging.Loggly awaitable

Moved buffer to loggly options to allow for customization and fix broken tests

* re #18483 - Making the EMG.Extensions.Logging.Loggly awaitable

Fix for tests

Co-authored-by: Erik Karlsson <[email protected]>

Author: elindanielsson

src/Logging/Loggly/Loggly/LogglyMessage.cs

@@ -23,6 +23,8 @@ public class LogglyMessage
         public LogLevel Level { get; set; }
 
         public Error Error { get; set; }
+
+        public static LogglyMessage Default => new LogglyMessage {Level = LogLevel.Trace};
     }
 
     public class Error

src/Logging/Loggly/Loggly/LogglyOptions.cs

@@ -26,6 +26,8 @@ public class LogglyOptions
         public Action<LogglyMessage> PreProcessMessage { get; set; }
 
         public Encoding ContentEncoding { get; set; } = Encoding.UTF8;
+
+        public TimeSpan Buffer { get; set; } = TimeSpan.FromMilliseconds(50);
     }
 
     public delegate bool FilterDelegate(string categoryName, EventId eventId, LogLevel logLevel);

src/Logging/Loggly/Loggly/LogglyProcessor.cs

@@ -8,26 +8,36 @@ namespace EMG.Extensions.Logging.Loggly
     public interface ILogglyProcessor : IDisposable
     {
         void EnqueueMessage(LogglyMessage message);
+
+        void FlushMessages();
     }
 
     public class LogglyProcessor : ILogglyProcessor
     {
         private readonly ILogglyClient _client;
         private readonly ISubject<LogglyMessage> _messageSubject = new Subject<LogglyMessage>();
+        private readonly ISubject<LogglyMessage> _flush = new Subject<LogglyMessage>();
         private readonly IDisposable _subscription;
 
-        public LogglyProcessor(ILogglyClient client)
+        public LogglyProcessor(ILogglyClient client, LogglyOptions options)
         {
             _client = client ?? throw new ArgumentNullException(nameof(client));
+            _ = options ?? throw new ArgumentNullException(nameof(options));
 
-            _subscription = _messageSubject.Buffer(TimeSpan.FromMilliseconds(50)).Subscribe(ProcessLogQueue);
+            var closing = _messageSubject.Buffer(options.Buffer).Select(i => LogglyMessage.Default).Merge(_flush);
+            _subscription = _messageSubject.Buffer(() => closing).Subscribe(ProcessLogQueue);
         }
 
         public void EnqueueMessage(LogglyMessage message)
         {
             _messageSubject.OnNext(message);
         }
 
+        public void FlushMessages()
+        {
+           _flush.OnNext(LogglyMessage.Default);
+        }
+
         public void Dispose()
         {
             _subscription.Dispose();

tests/Logging/Tests.Loggly/Loggly/LogglyProcessorTests.cs

@@ -3,6 +3,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using AutoFixture;
 using AutoFixture.Idioms;
 using AutoFixture.NUnit3;
 using EMG.Extensions.Logging.Loggly;
@@ -21,8 +22,10 @@ public void Constructor_is_guarded(GuardClauseAssertion assertion)
         }
 
         [Test, AutoMoqData]
-        public async Task Message_is_published_after_adding_to_queue([Frozen] ILogglyClient client, LogglyProcessor sut, LogglyMessage message)
+        public async Task Message_is_published_after_adding_to_queue_with_delay([Frozen] ILogglyClient client, LogglyMessage message, IFixture fixture)
         {
+            var sut = CreateSut(fixture, client, 50);
+
             sut.EnqueueMessage(message);
 
             await Task.Delay(TimeSpan.FromMilliseconds(100));
@@ -37,7 +40,36 @@ public void Message_is_not_published_after_adding_to_queue_when_disposed([Frozen
 
             sut.EnqueueMessage(message);
 
-            Mock.Get(client).Verify(p => p.PublishAsync(message), Times.Never);
+            Mock.Get(client).Verify(p => p.PublishManyAsync(It.Is<IEnumerable<LogglyMessage>>(m => m.Contains(message))), Times.Never);
+        }
+
+        [Test, AutoMoqData]
+        public async Task Message_is_not_published_after_adding_to_queue_without_delay([Frozen] ILogglyClient client, LogglyMessage message, IFixture fixture)
+        {
+            var sut = CreateSut(fixture, client, 1000);
+
+            sut.EnqueueMessage(message);
+
+            Mock.Get(client).Verify(p => p.PublishManyAsync(It.Is<IEnumerable<LogglyMessage>>(m => m.Contains(message))), Times.Never);
+        }
+
+        [Test, AutoMoqData]
+        public async Task Message_is_published_after_adding_to_queue_without_delay_after_flush([Frozen] ILogglyClient client, LogglyMessage message, IFixture fixture)
+        {
+            var sut = CreateSut(fixture, client, 1000);
+
+            sut.EnqueueMessage(message);
+
+            sut.FlushMessages();
+
+            Mock.Get(client).Verify(p => p.PublishManyAsync(It.Is<IEnumerable<LogglyMessage>>(m => m.Contains(message))), Times.Once);
+        }
+
+        private LogglyProcessor CreateSut(IFixture fixture, ILogglyClient client, int buffer)
+        {
+            var options = fixture.Build<LogglyOptions>().With(i => i.Buffer, TimeSpan.FromMilliseconds(buffer))
+                .Create();
+            return new LogglyProcessor(client, options);
         }
     }
 }