Merge pull request #242 from WeihanLi/dev

1.0.74

Directory.Packages.props

@@ -2,6 +2,13 @@
   <PropertyGroup>
     <!-- Enable central package management -->
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
+    <!-- Enable Transitive Package Pinning -->
+    <CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
+    <!-- https://learn.microsoft.com/en-us/nuget/concepts/auditing-packages -->    
+    <NuGetAudit>true</NuGetAudit>    
+    <NuGetAuditMode>all</NuGetAuditMode>
+    <!-- https://learn.microsoft.com/en-us/nuget/reference/errors-and-warnings/nu1901-nu1904 -->
+    <WarningsAsErrors>NU1901;NU1902;NU1903;NU1904</WarningsAsErrors>
   </PropertyGroup>
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0' OR '$(TargetFramework)' == 'net8.0'">
     <PackageVersion Include="Microsoft.Extensions.Configuration" Version="8.0.0" />

README.md

@@ -1,11 +1,11 @@
 # WeihanLi.Common
 
-## Build status
-
 [![WeihanLi.Common Latest Stable](https://img.shields.io/nuget/v/WeihanLi.Common.svg)](https://www.nuget.org/packages/WeihanLi.Common/)
 
 [![WeihanLi.Common Latest Preview](https://img.shields.io/nuget/vpre/WeihanLi.Common)](https://www.nuget.org/packages/WeihanLi.Common/absoluteLatest)
 
+## Build status
+
 [![Azure Pipelines Build Status](https://weihanli.visualstudio.com/Pipelines/_apis/build/status/WeihanLi.WeihanLi.Common?branchName=master)](https://weihanli.visualstudio.com/Pipelines/_build/latest?definitionId=16&branchName=master)
 
 [![Github Actions Build Status](https://github.com/WeihanLi/WeihanLi.Common/actions/workflows/default.yml/badge.svg)](https://github.com/WeihanLi/WeihanLi.Common/actions/workflows/default.yml)

build/version.props

@@ -2,7 +2,7 @@
   <PropertyGroup>
     <VersionMajor>1</VersionMajor>
     <VersionMinor>0</VersionMinor>
-    <VersionPatch>73</VersionPatch>
+    <VersionPatch>74</VersionPatch>
     <VersionPrefix>$(VersionMajor).$(VersionMinor).$(VersionPatch)</VersionPrefix>
   </PropertyGroup>
 </Project>

samples/AspNetCoreSample/Events/EventConsumer.cs

@@ -1,4 +1,5 @@
-using WeihanLi.Common.Event;
+using WeihanLi.Common;
+using WeihanLi.Common.Event;
 using WeihanLi.Extensions;
 
 namespace AspNetCoreSample.Events;
@@ -7,25 +8,24 @@ public class EventConsumer
   (IEventQueue eventQueue, IEventHandlerFactory eventHandlerFactory)
   : BackgroundService
 {
-    private readonly IEventQueue _eventQueue = eventQueue;
-    private readonly IEventHandlerFactory _eventHandlerFactory = eventHandlerFactory;
-
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         while (!stoppingToken.IsCancellationRequested)
         {
-            var queues = await _eventQueue.GetQueuesAsync();
+            var queues = await eventQueue.GetQueuesAsync();
             if (queues.Count > 0)
             {
                 await queues.Select(async q =>
                         {
-                            if (await _eventQueue.TryDequeueAsync(q, out var @event, out var properties))
+                            await foreach (var e in eventQueue.ReadAllAsync(q, stoppingToken))
                             {
-                                var handlers = _eventHandlerFactory.GetHandlers(@event.GetType());
+                                var @event = e.Data;
+                                Guard.NotNull(@event);
+                                var handlers = eventHandlerFactory.GetHandlers(@event.GetType());
                                 if (handlers.Count > 0)
                                 {
                                     await handlers
-                                            .Select(h => h.Handle(@event, properties))
+                                            .Select(h => h.Handle(@event, e.Properties))
                                             .WhenAll()
                                         ;
                                 }

samples/DotNetCoreSample/LoggerTest.cs

@@ -37,10 +37,24 @@ public static void MicrosoftLoggingTest()
         var services = new ServiceCollection()
             .AddLogging(builder =>
                 // builder.AddConsole()
-                builder.AddFile()
+                builder.AddFile(options => options.LogFormatter = (category, level, exception, msg, timestamp) => 
+                    $"{timestamp} - [{category}] {level} - {msg}\n{exception}")
                 )
             .AddSingleton(typeof(GenericTest<>))
             .BuildServiceProvider();
+
+        var logger = services.GetRequiredService<ILoggerFactory>()
+            .CreateLogger("test");
+        while (!ApplicationHelper.ExitToken.IsCancellationRequested)
+        {
+            logger.LogInformation("Echo time: {Time}", DateTimeOffset.Now);
+            Thread.Sleep(500);
+        }
+
+        ConsoleHelper.ReadKeyWithPrompt();
+        services.GetRequiredService<ILoggerFactory>()
+            .CreateLogger("test")
+            .LogInformation("test 123");
         services.GetRequiredService<GenericTest<int>>()
             .Test();
         services.GetRequiredService<GenericTest<string>>()
@@ -60,8 +74,6 @@ public static void MicrosoftLoggingTest()
 
     private class GenericTest<T>(ILogger<GenericTest<T>> logger)
     {
-        private readonly ILogger<GenericTest<T>> _logger = logger;
-
-        public void Test() => _logger.LogInformation("test");
+        public void Test() => logger.LogInformation("test");
     }
 }

samples/DotNetCoreSample/Program.cs

@@ -347,8 +347,8 @@
 
 // InvokeHelper.TryInvoke(() => throw null, 3);
 
-// InvokeHelper.TryInvoke(LoggerTest.MicrosoftLoggingTest);
-await InvokeHelper.TryInvokeAsync(InMemoryStreamTest.MainTest);
+InvokeHelper.TryInvoke(LoggerTest.MicrosoftLoggingTest);
+// await InvokeHelper.TryInvokeAsync(InMemoryStreamTest.MainTest);
 
 ConsoleHelper.ReadKeyWithPrompt("Press any key to exit");
 

src/Directory.Build.props

@@ -25,6 +25,7 @@
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <LangVersion>preview</LangVersion>
     <NoWarn>$(NoWarn);CS9216;</NoWarn>
+    <NuGetAuditMode>direct</NuGetAuditMode>
   </PropertyGroup>
   <ItemGroup>
     <Using Include="System.Object" Alias="Lock" Condition="!$([MSBuild]::IsTargetFrameworkCompatible('$(TargetFramework)', 'net9.0'))" />

src/WeihanLi.Common/Event/AckQueue.cs

@@ -0,0 +1,109 @@
+using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+using WeihanLi.Common.Helpers;
+
+namespace WeihanLi.Common.Event;
+
+public sealed class AckQueueOptions
+{
+    public TimeSpan AckTimeout { get; set; } = TimeSpan.FromMinutes(1);
+
+    public bool AutoRequeue { get; set; }
+
+    public TimeSpan RequeuePeriod { get; set; } = TimeSpan.FromMinutes(1);
+}
+
+public sealed class AckQueue : DisposableBase
+{
+    private readonly AckQueueOptions _options;
+    private readonly ConcurrentQueue<IEvent> _queue = new();
+    private readonly ConcurrentDictionary<string, IEvent> _unAckedMessages = new();
+    private readonly Timer? _timer;
+
+    public AckQueue() : this(new()) { }
+
+    public AckQueue(AckQueueOptions options)
+    {
+        _options = options;
+        if (options.AutoRequeue)
+        {
+            _timer = new Timer(_ => RequeueUnAckedMessages(), null, options.RequeuePeriod, options.RequeuePeriod);
+        }
+    }
+
+    public Task EnqueueAsync<TEvent>(TEvent @event, EventProperties? properties = null)
+    {
+        properties ??= new EventProperties();
+        if (string.IsNullOrEmpty(properties.EventId))
+        {
+            properties.EventId = Guid.NewGuid().ToString();
+        }
+
+        if (properties.EventAt == default)
+        {
+            properties.EventAt = DateTimeOffset.Now;
+        }
+
+        var internalEvent = new EventWrapper<TEvent>
+        {
+            Data = @event,
+            Properties = properties
+        };
+
+        _queue.Enqueue(internalEvent);
+        return Task.CompletedTask;
+    }
+
+    public Task<IEvent<TEvent>?> DequeueAsync<TEvent>()
+    {
+        if (_queue.TryDequeue(out var eventWrapper))
+        {
+            _unAckedMessages.TryAdd(eventWrapper.Properties.EventId, eventWrapper);
+            return Task.FromResult((IEvent<TEvent>?)eventWrapper);
+        }
+
+        return Task.FromResult<IEvent<TEvent>?>(null);
+    }
+
+    public Task AckMessageAsync(string eventId)
+    {
+        _unAckedMessages.TryRemove(eventId, out _);
+        return Task.CompletedTask;
+    }
+
+    public void RequeueUnAckedMessages()
+    {
+        foreach (var message in _unAckedMessages)
+        {
+            if (DateTimeOffset.Now - message.Value.Properties.EventAt > _options.AckTimeout)
+            {
+                if (_unAckedMessages.TryRemove(message.Key, out var eventWrapper)
+                    && eventWrapper != null)
+                {
+                    _queue.Enqueue(eventWrapper);
+                }
+            }
+        }
+    }
+
+    public async IAsyncEnumerable<IEvent> ReadAllAsync(
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            while (_queue.TryDequeue(out var eventWrapper))
+            {
+                _unAckedMessages.TryAdd(eventWrapper.Properties.EventId, eventWrapper);
+                yield return eventWrapper;
+            }
+
+            await Task.Delay(200, cancellationToken);
+        }
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        _timer?.Dispose();
+        base.Dispose(disposing);
+    }
+}

src/WeihanLi.Common/Event/EventBase.cs

@@ -64,29 +64,65 @@ public interface IEvent<out T>
     T Data { get; }
 }
 
-internal sealed class EventWrapper<T> : IEvent, IEvent<T>
+public class EventWrapper<T> : IEvent, IEvent<T>
 {
     public required T Data { get; init; }
-    object? IEvent.Data => (object?)Data;
+    object? IEvent.Data => Data;
     public required EventProperties Properties { get; init; }
 }
 
-public static class EventBaseExtensions
+public static class EventExtensions
 {
-    private static readonly JsonSerializerSettings EventSerializerSettings = JsonSerializeExtension.SerializerSettingsWith(s =>
-                {
-                    s.TypeNameHandling = TypeNameHandling.Objects;
-                });
+    private static readonly JsonSerializerSettings EventSerializerSettings = JsonSerializeExtension
+        .SerializerSettingsWith(s => 
+        {
+            s.NullValueHandling = NullValueHandling.Ignore;
+            s.TypeNameHandling = TypeNameHandling.Objects;
+        });
 
-    public static string ToEventMsg<TEvent>(this TEvent @event) where TEvent : class, IEventBase
+    public static string ToEventMsg<TEvent>(this TEvent @event)
+    {
+        Guard.NotNull(@event);
+        return GetEvent(@event).ToJson(EventSerializerSettings);
+    }
+
+    public static string ToEventRawMsg<TEvent>(this TEvent @event)
     {
         Guard.NotNull(@event);
         return @event.ToJson(EventSerializerSettings);
     }
 
-    public static IEventBase ToEvent(this string eventMsg)
+    private static IEvent GetEvent<TEvent>(this TEvent @event)
+    {
+        if (@event is IEvent eventEvent)
+            return eventEvent;
+
+        if (@event is IEventBase eventBase)
+            return new EventWrapper<TEvent>()
+            {
+                Data = @event, 
+                Properties = new()
+                {
+                    EventAt = eventBase.EventAt, 
+                    EventId = eventBase.EventId
+                }
+            };
+
+        return new EventWrapper<TEvent> 
+        {
+            Data = @event,
+            Properties = new EventProperties
+            {
+                EventAt = DateTimeOffset.Now
+            }
+        };
+    }
+
+    public static TEvent ToEvent<TEvent>(this string eventMsg)
     {
         Guard.NotNull(eventMsg);
-        return eventMsg.JsonToObject<IEventBase>(EventSerializerSettings);
+        return eventMsg.JsonToObject<TEvent>(EventSerializerSettings);
     }
+
+    public static IEvent ToEvent(this string eventMsg) => ToEvent<IEvent>(eventMsg);
 }

src/WeihanLi.Common/Event/EventHandlerFactory.cs

@@ -7,12 +7,10 @@ namespace WeihanLi.Common.Event;
 
 public sealed class DefaultEventHandlerFactory(IEventSubscriptionManager subscriptionManager) : IEventHandlerFactory
 {
-    private readonly IEventSubscriptionManager _subscriptionManager = subscriptionManager;
-
     [RequiresUnreferencedCode("Unreferenced code may be used")]
     public ICollection<IEventHandler> GetHandlers(Type eventType)
     {
-        var eventHandlers = _subscriptionManager.GetEventHandlers(eventType);
+        var eventHandlers = subscriptionManager.GetEventHandlers(eventType);
         return eventHandlers;
     }
 }