Feature/unified messaging

Foundatio.sln

@@ -14,6 +14,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		src\Directory.Build.props = src\Directory.Build.props
 		samples\Directory.Build.props = samples\Directory.Build.props
 		NuGet.config = NuGet.config
+		global.json = global.json
+		Dockerfile = Dockerfile
 		README.md = README.md
 	EndProjectSection
 EndProject

src/Foundatio.AppMetrics/AppMetricsClient.cs

@@ -1,5 +1,4 @@
-using System;
-using App.Metrics;
+using App.Metrics;
 using App.Metrics.Counter;
 using App.Metrics.Gauge;
 using App.Metrics.Timer;

src/Foundatio.Extensions.Hosting/Foundatio - Backup.Extensions.Hosting.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
+    <TargetFrameworks>net6.0;net5.0</TargetFrameworks>
+  </PropertyGroup>
+  <ItemGroup>
+    <FrameworkReference Include="Microsoft.AspNetCore.App" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Foundatio\Foundatio.csproj" />
+  </ItemGroup>
+  <ItemGroup>
+    <Compile Include="..\Foundatio\Extensions\TaskExtensions.cs" Link="Extensions\TaskExtensions.cs" />
+  </ItemGroup>
+</Project>

src/Foundatio.Extensions.Hosting/Foundatio.Extensions.Hosting.csproj

@@ -1,10 +1,10 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
-    <TargetFrameworks>net6.0;net5.0</TargetFrameworks>
   </PropertyGroup>
   <ItemGroup>
-    <FrameworkReference Include="Microsoft.AspNetCore.App" />
+    <PackageReference Include="Microsoft.AspNetCore.Diagnostics.HealthChecks" Version="2.2" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Foundatio\Foundatio.csproj" />

src/Foundatio.Extensions.Hosting/Startup/IStartupAction.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Threading;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace Foundatio.Extensions.Hosting.Startup {

src/Foundatio.MetricsNET/MetricsNETClient.cs

@@ -1,5 +1,4 @@
-using System;
-using Metrics;
+using Metrics;
 
 namespace Foundatio.Metrics {
     public class MetricsNETClient : IMetricsClient {

src/Foundatio.TestHarness/Caching/CacheClientTestsBase.cs

@@ -411,15 +411,12 @@ public virtual async Task CanSetMinMaxExpirationAsync() {
             using (cache) {
                 await cache.RemoveAllAsync();
 
-                using (TestSystemClock.Install()) {
-                    var now = DateTime.UtcNow;
-                    TestSystemClock.SetFrozenTime(now);
-
-                    var expires = DateTime.MaxValue - now.AddDays(1);
+                using (var clock = TestSystemClock.Install()) {
+                    var expires = DateTime.MaxValue - clock.Now.AddDays(1);
                     Assert.True(await cache.SetAsync("test1", 1, expires));
                     Assert.False(await cache.SetAsync("test2", 1, DateTime.MinValue));
                     Assert.True(await cache.SetAsync("test3", 1, DateTime.MaxValue));
-                    Assert.True(await cache.SetAsync("test4", 1, DateTime.MaxValue - now.AddDays(-1)));
+                    Assert.True(await cache.SetAsync("test4", 1, DateTime.MaxValue - clock.Now.AddDays(-1)));
 
                     Assert.Equal(1, (await cache.GetAsync<int>("test1")).Value);
                     Assert.InRange((await cache.GetExpirationAsync("test1")).Value, expires.Subtract(TimeSpan.FromSeconds(10)), expires);

src/Foundatio.TestHarness/Messaging/MessageBusTestBase.cs

@@ -108,7 +108,7 @@ await messageBus.PublishAsync(new DerivedSimpleMessageA {
 
         public virtual async Task CanSendMappedMessageAsync() {
             var messageBus = GetMessageBus(b => {
-                b.MessageTypeMappings.Add(nameof(SimpleMessageA), typeof(SimpleMessageA));
+                b.TypeNameSerializer = new DefaultTypeNameSerializer(_logger, new Dictionary<string, Type> {{ nameof(SimpleMessageA), typeof(SimpleMessageA) }});
                 return b;
             });
             if (messageBus == null)
@@ -136,7 +136,43 @@ await messageBus.PublishAsync(new SimpleMessageA {
         }
 
         public virtual async Task CanSendDelayedMessageAsync() {
-            const int numConcurrentMessages = 1000;
+            Log.MinimumLevel = LogLevel.Trace;
+            var messageBus = GetMessageBus();
+            if (messageBus == null)
+                return;
+
+            try {
+                var countdown = new AsyncCountdownEvent(1);
+                await messageBus.SubscribeAsync<SimpleMessageA>(msg => {
+                    var msgDelay = TimeSpan.FromMilliseconds(msg.Count);
+                    _logger.LogTrace("Got message delayed by {Delay:g}.", msgDelay);
+
+                    Assert.Equal("Hello", msg.Data);
+                    countdown.Signal();
+                });
+
+                var sw = Stopwatch.StartNew();
+                var delay = TimeSpan.FromMilliseconds(RandomData.GetInt(250, 1500));
+                await messageBus.PublishAsync(new SimpleMessageA {
+                    Data = "Hello",
+                    Count = (int)delay.TotalMilliseconds
+                }, delay);
+                _logger.LogTrace("Published message...");
+
+                await countdown.WaitAsync(TimeSpan.FromSeconds(30));
+                sw.Stop();
+
+                _logger.LogTrace("Got message delayed by {Delay:g} in {Duration:g}", delay, sw.Elapsed);
+                Assert.Equal(0, countdown.CurrentCount);
+                Assert.InRange(sw.Elapsed.TotalMilliseconds, 50, 2000);
+            } finally {
+                await CleanupMessageBusAsync(messageBus);
+            }
+        }
+
+        public virtual async Task CanSendParallelDelayedMessagesAsync() {
+            Log.MinimumLevel = LogLevel.Trace;
+            const int numConcurrentMessages = 100;
             var messageBus = GetMessageBus();
             if (messageBus == null)
                 return;
@@ -158,17 +194,18 @@ await Run.InParallelAsync(numConcurrentMessages, async i => {
                     await messageBus.PublishAsync(new SimpleMessageA {
                         Data = "Hello",
                         Count = i
-                    }, new MessageOptions { DeliveryDelay = TimeSpan.FromMilliseconds(RandomData.GetInt(0, 100)) });
+                    }, TimeSpan.FromMilliseconds(RandomData.GetInt(100, 500)));
                     if (i % 500 == 0)
                         _logger.LogTrace("Published 500 messages...");
                 });
+                _logger.LogTrace("Done publishing {Count} messages.", numConcurrentMessages);
 
-                await countdown.WaitAsync(TimeSpan.FromSeconds(5));
+                await countdown.WaitAsync(TimeSpan.FromSeconds(30));
                 sw.Stop();
 
-                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Processed {Processed} in {Duration:g}", numConcurrentMessages - countdown.CurrentCount, sw.Elapsed);
+                _logger.LogTrace("Processed {Processed} in {Duration:g}", numConcurrentMessages - countdown.CurrentCount, sw.Elapsed);
                 Assert.Equal(0, countdown.CurrentCount);
-                Assert.InRange(sw.Elapsed.TotalMilliseconds, 50, 5000);
+                Assert.InRange(sw.Elapsed.TotalMilliseconds, 100, 2000);
             } finally {
                 await CleanupMessageBusAsync(messageBus);
             }

src/Foundatio.TestHarness/Serializer/SerializerTestsBase.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using BenchmarkDotNet.Attributes;
 using Foundatio.Xunit;
 using Foundatio.Serializer;

src/Foundatio/DeepCloner/Helpers/DeepClonerSafeTypes.cs

@@ -2,15 +2,13 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
-using System.Linq;
 using System.Reflection;
 
-namespace Foundatio.Force.DeepCloner.Helpers
-{
-	/// <summary>
-	/// Safe types are types, which can be copied without real cloning. e.g. simple structs or strings (it is immutable)
-	/// </summary>
-	internal static class DeepClonerSafeTypes
+namespace Foundatio.Force.DeepCloner.Helpers {
+    /// <summary>
+    /// Safe types are types, which can be copied without real cloning. e.g. simple structs or strings (it is immutable)
+    /// </summary>
+    internal static class DeepClonerSafeTypes
 	{
 		internal static readonly ConcurrentDictionary<Type, bool> KnownTypes = new();
 

src/Foundatio/DeepCloner/Helpers/ShallowObjectCloner.cs

@@ -1,8 +1,6 @@
 #define NETCORE
 using System;
 using System.Linq.Expressions;
-using System.Reflection;
-using System.Reflection.Emit;
 
 namespace Foundatio.Force.DeepCloner.Helpers
 {

src/Foundatio/Messaging/Envelope.cs

@@ -0,0 +1,80 @@
+using System;
+using System.Collections.Generic;
+
+namespace Foundatio.Messaging2 {
+    public interface IEnvelope {
+        // trace parent id used for distributed tracing
+        string TraceParentId { get; }
+        // message type
+        string MessageType { get; }
+        // message body
+        object GetMessage();
+        // number of attempts to deliver the message
+        int Attempts { get; }
+        // when the message was originally sent
+        DateTime SentAtUtc { get; }
+        // when the message should expire
+        DateTime? ExpiresAtUtc { get; }
+        // when the message should be delivered when using delayed delivery
+        DateTime? DeliverAtUtc { get; }
+        // additional message data to store with the message
+        IReadOnlyDictionary<string, string> Properties { get; }
+    }
+
+    public class Envelope : IEnvelope {
+        private Lazy<object> _message;
+
+        public Envelope(Func<object> getMessageFunc, string messageType, string coorelationId, DateTime? expiresAtUtc, DateTime? deliverAtUtc, IReadOnlyDictionary<string, string> properties) {
+            _message = new Lazy<object>(getMessageFunc);
+            MessageType = messageType;
+            TraceParentId = coorelationId;
+            ExpiresAtUtc = expiresAtUtc;
+            DeliverAtUtc = deliverAtUtc;
+            Properties = properties;
+        }
+
+        public Message(Func<object> getMessageFunc, MessagePublishOptions options) {
+            _message = new Lazy<object>(getMessageFunc);
+            TraceParentId = options.CorrelationId;
+            MessageType = options.MessageType;
+            ExpiresAtUtc = options.ExpiresAtUtc;
+            DeliverAtUtc = options.DeliverAtUtc;
+            Properties = options.Properties;
+        }
+
+        public string TraceParentId { get; private set; }
+        public string MessageType { get; private set; }
+        public int Attempts { get; private set; }
+        public DateTime SentAtUtc { get; private set; }
+        public DateTime? ExpiresAtUtc { get; private set; }
+        public DateTime? DeliverAtUtc { get; private set; }
+        public IReadOnlyDictionary<string, string> Properties { get; private set; }
+
+        public object GetMessage() {
+            return _message.Value;
+        }
+    }
+
+    public interface IEnvelope<out T> : IEnvelope where T: class {
+        T Message { get; }
+    }
+
+    public class Envelope<T> : IEnvelope<T> where T: class {
+        private readonly IEnvelope _envolope;
+
+        public Envelope(IEnvelope message) {
+            _envolope = message;
+        }
+
+        public T Message => (T)GetMessage();
+
+        public string TraceParentId => _envolope.TraceParentId;
+        public string MessageType => _envolope.MessageType;
+        public int Attempts => _envolope.Attempts;
+        public DateTime SentAtUtc => _envolope.SentAtUtc;
+        public DateTime? ExpiresAtUtc => _envolope.ExpiresAtUtc;
+        public DateTime? DeliverAtUtc => _envolope.DeliverAtUtc;
+        public IReadOnlyDictionary<string, string> Properties => _envolope.Properties;
+        public object GetMessage() => _envolope.GetMessage();
+    }
+}

src/Foundatio/Messaging/IMessageBus.cs

@@ -2,12 +2,7 @@
 using Foundatio.Utility;
 
 namespace Foundatio.Messaging {
-    public interface IMessageBus : IMessagePublisher, IMessageSubscriber, IDisposable {}
-
-    public class MessageOptions {
-        public string UniqueId { get; set; }
-        public string CorrelationId { get; set; }
-        public TimeSpan? DeliveryDelay { get; set; }
-        public DataDictionary Properties { get; set; } = new DataDictionary();
+    public interface IMessageBus : IMessagePublisher, IMessageSubscriber, IDisposable {
+        string MessageBusId { get; }
     }
 }