Merge pull request #29 from WB-Tooling/generic-payload

generic payload

Author: WolfBublitz

Directory.Packages.props

@@ -10,7 +10,7 @@
        │ WB Packages                                                          │
        ╰──────────────────────────────────────────────────────────────────────╯ -->
   <ItemGroup>
-    <PackageVersion Include="WB.Logging.Base" Version="[1.0.0-preview.15, 2.0.0)" />
+    <PackageVersion Include="WB.Logging.Base" Version="[1.0.0-preview.16, 2.0.0)" />
   </ItemGroup>
 
   <!-- ╭──────────────────────────────────────────────────────────────────────╮

src/Extensions/IAsyncLogSinkExtensions.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace WB.Logging;
@@ -16,15 +17,18 @@ internal static class IAsyncLogSinkExtensions
     /// Submits the <paramref name="logMessage"/> to <paramref name="this"/> <see cref="IAsyncLogSink"/> safely, 
     /// catching any exceptions that occur and logging them to the console or a fallback logger instead of throwing.
     /// </summary>
+    /// <typeparam name="TPayload">The <see cref="Type"/> of the payload of the log message.</typeparam>
     /// <param name="this">The <see cref="IAsyncLogSink"/> to submit the log message to.</param>
     /// <param name="logMessage">The log message to submit.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe while waiting for the task to complete.</param>
     /// <returns>A <see cref="Task"/> that represents the asynchronous operation.</returns>
     [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "We want to catch all exceptions to prevent logging failures from crashing the application.")]
-    internal static async Task SubmitSafeAsync(this IAsyncLogSink @this, LogMessage logMessage)
+    internal static async Task SubmitSafeAsync<TPayload>(this IAsyncLogSink @this, ILogMessage<TPayload> logMessage, CancellationToken cancellationToken = default)
+        where TPayload : notnull
     {
         try
         {
-            await @this.SubmitAsync(logMessage).ConfigureAwait(false);
+            await @this.SubmitAsync(logMessage, cancellationToken).ConfigureAwait(false);
         }
         catch (Exception ex)
         {

src/Extensions/ILogSinkExtensions.cs

@@ -16,11 +16,13 @@ internal static class ILogSinkExtensions
     /// Submits the <paramref name="logMessage"/> to <paramref name="this"/> <see cref="ILogSink"/> safely, 
     /// catching any exceptions that occur and logging them to the console or a fallback logger instead of throwing.
     /// </summary>
+    /// <typeparam name="TPayload">The <see cref="Type"/> of the payload of the log message.</typeparam>
     /// <param name="this">The <see cref="ILogSink"/> to submit the log message to.</param>
     /// <param name="logMessage">The log message to submit.</param>
     /// <returns>A <see cref="Task"/> that represents the asynchronous operation.</returns>
     [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "We want to catch all exceptions to prevent logging failures from crashing the application.")]
-    internal static Task SubmitSafeAsync(this ILogSink @this, LogMessage logMessage)
+    internal static Task WriteSafeAsync<TPayload>(this ILogSink @this, ILogMessage<TPayload> logMessage)
+        where TPayload : notnull
     {
         return Task.Run(() =>
         {

src/LogMessage.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+
+namespace WB.Logging;
+
+internal sealed record LogMessage<TPayload> : ILogMessage<TPayload>
+    where TPayload : notnull
+{
+    // ┌─────────────────────────────────────────────────────────────────────────────┐
+    // │ Private Fields                                                              │
+    // └─────────────────────────────────────────────────────────────────────────────┘
+    private readonly List<string> senders = [];
+
+    // ┌─────────────────────────────────────────────────────────────────────────────┐
+    // │ Public Properties                                                           │
+    // └─────────────────────────────────────────────────────────────────────────────┘
+
+    /// <inheritdoc/>
+    public required TPayload Payload { get; init; }
+
+    /// <inheritdoc/>
+    public required DateTimeOffset Timestamp { get; init; }
+    
+    /// <inheritdoc/>
+    public IReadOnlyList<string> Senders => senders;
+
+    /// <inheritdoc/>
+    public LogLevel? LogLevel { get; init; }
+
+    /// <inheritdoc/>
+    object ILogMessage.Payload => Payload;
+
+    // ┌─────────────────────────────────────────────────────────────────────────────┐
+    // │ Public Methods                                                              │
+    // └─────────────────────────────────────────────────────────────────────────────┘
+
+    /// <summary>
+    /// Adds the <paramref name="sender"/> to the list of senders of this log message.
+    /// </summary>
+    /// <param name="sender">The sender to add.</param>
+    internal void AddSender(string sender)
+        => senders.Add(sender);
+}

src/Logger.cs

@@ -2,12 +2,16 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Linq.Expressions;
+using System.Reflection;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
-using System.Threading.Tasks.Dataflow;
 
 namespace WB.Logging;
 
+internal delegate Task Dispatcher(ILogMessage logMessage, CancellationToken cancellationToken);
+
 /// <inheritdoc/>
 /// <summary>
 /// Initializes a new instance of the <see cref="Logger"/> class.
@@ -19,27 +23,28 @@ public sealed class Logger : ILogger
     // └─────────────────────────────────────────────────────────────────────────────┘
     private readonly Logger? parent;
 
-    private readonly ActionBlock<Func<Task>> logMessageQueue = new(
-            async logMessageAction =>
-            {
-                await logMessageAction().ConfigureAwait(false);
-            },
-            new ExecutionDataflowBlockOptions
-            {
-                MaxDegreeOfParallelism = 1,
-                EnsureOrdered = true
-            });
+    private readonly Channel<ILogMessage> logMessageQueue = Channel.CreateUnbounded<ILogMessage>(new UnboundedChannelOptions
+    {
+        SingleReader = true,
+        SingleWriter = true,
+    });
+
+    private readonly Task logMessageProcessingTask;
+
+    private readonly CancellationTokenSource cancellationTokenSource = new();
 
     private readonly ConcurrentBag<ILogger> childLoggers = [];
 
     private readonly ConcurrentBag<ILogSink> logSinks = [];
 
     private readonly ConcurrentBag<IAsyncLogSink> asyncLogSinks = [];
 
-    private readonly LogMessageFilters logMessageFilterRegistry = new();
+    private readonly LogMessageFilterPipeline logMessageFilterPipeline = new();
 
     private readonly IDisposable minimumLogLevelFilter;
 
+    private readonly ConcurrentDictionary<Type, Dispatcher> dispatchCache = new();
+
     private int isDisposed;
 
     // ┌─────────────────────────────────────────────────────────────────────────────┐
@@ -56,6 +61,7 @@ public Logger(string name)
         Name = name ?? throw new ArgumentNullException(nameof(name));
 
         minimumLogLevelFilter = AddLogMessageFilter(logMessage => logMessage.LogLevel is null || logMessage.LogLevel >= MinimumLogLevel);
+        logMessageProcessingTask = ExecuteAsync(cancellationTokenSource.Token);
     }
 
     // ┌─────────────────────────────────────────────────────────────────────────────┐
@@ -125,8 +131,9 @@ public async ValueTask DisposeAsync()
             return;
         }
 
-        logMessageQueue.Complete();
-        await logMessageQueue.Completion.ConfigureAwait(false);
+        await cancellationTokenSource.CancelAsync().ConfigureAwait(false);
+        cancellationTokenSource.Dispose();
+        await logMessageProcessingTask.ConfigureAwait(false);
 
         foreach (ILogger childLogger in childLoggers)
         {
@@ -181,14 +188,16 @@ public async Task FlushAsync(CancellationToken cancellationToken = default)
     public void Log<TPayload>(LogLevel? logLevel, TPayload payload)
         where TPayload : notnull
     {
-        LogMessage logMessage = new()
+        LogMessage<TPayload> logMessage = new()
         {
             Timestamp = TimestampProvider.CurrentTimestamp,
             LogLevel = logLevel,
-            Payload = payload
+            Payload = payload,
         };
 
         Log(logMessage);
+
+        parent?.Log(logMessage);
     }
 
     /// <inheritdoc/>
@@ -219,43 +228,81 @@ public ILogger CreateChildLogger(string name)
 
     /// <inheritdoc/>
     public IDisposable AddLogMessageFilter(LogMessageFilter filter)
-        => logMessageFilterRegistry.Add(filter);
+        => logMessageFilterPipeline.Add(filter);
 
     // ┌─────────────────────────────────────────────────────────────────────────────┐
     // │ Private Methods                                                             │
     // └─────────────────────────────────────────────────────────────────────────────┘
-    private void Log(LogMessage logMessage)
+
+    private void Log<TPayload>(LogMessage<TPayload> logMessage)
+        where TPayload : notnull
     {
         logMessage.AddSender(Name);
 
-        // Apply filters to the log message
-        if (logMessage.Payload is not FlushItem)
+        logMessageQueue.Writer.TryWrite(logMessage);
+    }
+
+    private async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        try
         {
-            if (!logMessageFilterRegistry.IsMatch(logMessage))
+            await foreach (ILogMessage logMessage in logMessageQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
             {
-                return;
+                if (logMessage.Payload is FlushItem flushItem)
+                {
+                    flushItem.Complete();
+                }
+                else if (logMessageFilterPipeline.IsMatch(logMessage))
+                {
+                    await DispatchAsync(logMessage, cancellationToken).ConfigureAwait(false);
+                }
             }
-
-            parent?.Log(logMessage);
         }
-
-        logMessageQueue.Post(() =>
+        catch (OperationCanceledException)
         {
-            if (logMessage.Payload is FlushItem flushItem)
-            {
-                flushItem.Complete();
+            // normal shutdown
+        }
+    }
 
-                return Task.CompletedTask;
-            }
-            else
-            {
-                Task[] submitTasks = [
-                    .. logSinks.Select(logSink => logSink.SubmitSafeAsync(logMessage)),
-                    .. asyncLogSinks.Select(asyncLogSink => asyncLogSink.SubmitSafeAsync(logMessage))
-                ];
+    private Task DispatchAsync(ILogMessage logMessage, CancellationToken cancellationToken)
+    {
+        Type payloadType = logMessage.Payload.GetType();
 
-                return Task.WhenAll(submitTasks);
-            }
-        });
+        Dispatcher dispatcher = dispatchCache.GetOrAdd(
+            payloadType,
+            static (t, self) => self.CreateDispatcher(t),
+            this);
+
+        return dispatcher(logMessage, cancellationToken);
+    }
+
+    private Dispatcher CreateDispatcher(Type payloadType)
+    {
+        ParameterExpression logMessageParameter = Expression.Parameter(typeof(ILogMessage), "msg");
+        ParameterExpression cancellationTokenParameter = Expression.Parameter(typeof(CancellationToken), "ct");
+
+        UnaryExpression cast = Expression.Convert(logMessageParameter, typeof(ILogMessage<>).MakeGenericType(payloadType));
+
+        MethodCallExpression methodCall = Expression.Call(
+            Expression.Constant(this),
+            typeof(Logger)
+                .GetMethod(nameof(DispatchTyped), BindingFlags.NonPublic | BindingFlags.Instance)!
+                .MakeGenericMethod(payloadType),
+            cast,
+            cancellationTokenParameter);
+
+        Expression<Dispatcher> lambda = Expression.Lambda<Dispatcher>(methodCall, logMessageParameter, cancellationTokenParameter);
+        return lambda.Compile();
+    }
+
+    private Task DispatchTyped<TPayload>(ILogMessage logMessage, CancellationToken cancellationToken)
+        where TPayload : notnull
+    {
+        ILogMessage<TPayload> typedLogMessage = (ILogMessage<TPayload>)logMessage;
+
+        Task[] tasks = [.. logSinks.Select(logSink => logSink.WriteSafeAsync(typedLogMessage)),
+            .. asyncLogSinks.Select(asyncLogSink => asyncLogSink.SubmitSafeAsync(typedLogMessage, cancellationToken))];
+
+        return Task.WhenAll(tasks);
     }
 }

tests/LoggerTests/src/MethodTests/AttachLogMessageFilter.cs

@@ -8,9 +8,10 @@ namespace LoggerTests.MethodTests.AttachLogMessageFilterMethodTests;
 
 internal sealed class TestLogSink : ILogSink
 {
-    public readonly List<LogMessage> ReceivedMessages = [];
+    public readonly List<ILogMessage> ReceivedMessages = [];
 
-    public void Submit(LogMessage logMessage)
+    public void Submit<TPayload>(ILogMessage<TPayload> logMessage)
+        where TPayload : notnull
         => ReceivedMessages.Add(logMessage);
 }
 
@@ -98,6 +99,7 @@ public async Task ShouldDetachFilterWhenDisposableIsDisposed()
 
         // Act
         logger.Log(LogLevel.Info, "INCLUDE message 1");
+        await logger.FlushAsync().ConfigureAwait(false);
         filterDisposable.Dispose();
         logger.Log(LogLevel.Info, "EXCLUDE message 2");
         await logger.FlushAsync().ConfigureAwait(false);

tests/LoggerTests/src/MethodTests/AttachLogSink.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using AwesomeAssertions;
 using WB.Logging;
@@ -8,17 +9,19 @@ namespace LoggerTests.MethodTests.AttachLogSinkMethodTests;
 
 internal sealed class TestLogSink : ILogSink
 {
-    public readonly List<LogMessage> ReceivedMessages = [];
+    public readonly List<ILogMessage> ReceivedMessages = [];
 
-    public void Submit(LogMessage logMessage)
+    public void Submit<TPayload>(ILogMessage<TPayload> logMessage)
+        where TPayload : notnull
         => ReceivedMessages.Add(logMessage);
 }
 
 internal sealed class TestAsyncLogSink : IAsyncLogSink
 {
-    public readonly List<LogMessage> ReceivedMessages = [];
+    public readonly List<ILogMessage> ReceivedMessages = [];
 
-    public ValueTask SubmitAsync(LogMessage logMessage)
+    public ValueTask SubmitAsync<TPayload>(ILogMessage<TPayload> logMessage, CancellationToken cancellationToken)
+        where TPayload : notnull
     {
         ReceivedMessages.Add(logMessage);
 

tests/LoggerTests/src/MethodTests/DisposeAsyncMethodTests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using AwesomeAssertions;
 using WB.Logging;
@@ -12,11 +13,12 @@ internal sealed class TestLogSink : ILogSink, IDisposable
     public void Dispose()
         => IsDisposed = true;
 
-    public void Submit(LogMessage logMessage)
+    public void Submit<TPayload>(ILogMessage<TPayload> logMessage)
+        where TPayload : notnull
         => throw new NotImplementedException("This test log sink should not be used to submit log messages.");
 }
 
-internal sealed class AsyncTestLogSink : ILogSink, IAsyncDisposable
+internal sealed class AsyncTestLogSink : IAsyncLogSink, IAsyncDisposable
 {
     public bool IsDisposed { get; private set; }
 
@@ -26,7 +28,8 @@ public ValueTask DisposeAsync()
         return ValueTask.CompletedTask;
     }
 
-    public void Submit(LogMessage logMessage)
+    public ValueTask SubmitAsync<TPayload>(ILogMessage<TPayload> logMessage, CancellationToken cancellationToken)
+        where TPayload : notnull
         => throw new NotImplementedException("This test log sink should not be used to submit log messages.");
 }
 

tests/LoggerTests/src/MethodTests/FlushAsyncMethodTests.cs

@@ -9,9 +9,10 @@ namespace LoggerTests.MethodTests.FlushAsyncMethodTests;
 
 internal sealed class TestLogSink : ILogSink
 {
-    public readonly List<LogMessage> ReceivedMessages = [];
+    public readonly List<ILogMessage> ReceivedMessages = [];
 
-    public void Submit(LogMessage logMessage)
+    public void Submit<TPayload>(ILogMessage<TPayload> logMessage)
+        where TPayload : notnull
         => ReceivedMessages.Add(logMessage);
 }