Working on message bus subscriptions options

src/Foundatio.TestHarness/Messaging/MessageBusTestBase.cs

@@ -522,7 +522,7 @@ await messageBus.SubscribeAsync<SimpleMessageA>(msg => {
                     Interlocked.Increment(ref messageCount);
                     cancellationTokenSource.Cancel();
                     countdown.Signal();
-                }, cancellationTokenSource.Token);
+                }, cancellationToken: cancellationTokenSource.Token);
 
                 await messageBus.SubscribeAsync<object>(msg => countdown.Signal());
 

src/Foundatio.TestHarness/Queue/QueueTestBase.cs

@@ -553,6 +553,7 @@ public virtual async Task WillWaitForItemAsync() {
                 return;
 
             try {
+                Log.MinimumLevel = LogLevel.Trace;
                 await queue.DeleteQueueAsync();
                 await AssertEmptyQueueAsync(queue);
 
@@ -564,14 +565,16 @@ public virtual async Task WillWaitForItemAsync() {
                 Assert.InRange(sw.Elapsed, TimeSpan.FromMilliseconds(50), TimeSpan.FromMilliseconds(5000));
 
                 _ = Task.Run(async () => {
-                    await SystemClock.SleepAsync(500);
+                    await Task.Delay(500);
+                    _logger.LogInformation("Enqueuing async message");
                     await queue.EnqueueAsync(new SimpleWorkItem {
                         Data = "Hello"
                     });
                 });
 
                 sw.Restart();
-                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(1));
+                _logger.LogInformation("Dequeuing message with timeout");
+                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(10));
                 sw.Stop();
                 if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                 Assert.True(sw.Elapsed > TimeSpan.FromMilliseconds(400));

src/Foundatio/Foundatio.csproj

@@ -5,5 +5,6 @@
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="6.0" />
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="6.0" />
+    <PackageReference Include="System.Threading.Channels" Version="6.0.0" />
   </ItemGroup>
 </Project>

src/Foundatio/Messaging/IMessageBus.cs

@@ -7,6 +7,8 @@ public interface IMessageBus : IMessagePublisher, IMessageSubscriber, IDisposabl
     public class MessageOptions {
         public string UniqueId { get; set; }
         public string CorrelationId { get; set; }
+        public string Topic { get; set; }
+        public string MessageType { get; set; }
         public TimeSpan? DeliveryDelay { get; set; }
         public IDictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
     }

src/Foundatio/Messaging/IMessageSubscriber.cs

@@ -1,37 +1,68 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace Foundatio.Messaging {
     public interface IMessageSubscriber {
-        Task SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken = default) where T : class;
+        Task<IMessageSubscription> SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) where T : class;
     }
 
     public static class MessageBusExtensions {
-        public static Task SubscribeAsync<T>(this IMessageSubscriber subscriber, Func<T, Task> handler, CancellationToken cancellationToken = default) where T : class {
-            return subscriber.SubscribeAsync<T>((msg, token) => handler(msg), cancellationToken);
+        public static Task<IMessageSubscription> SubscribeAsync<T>(this IMessageSubscriber subscriber, Func<T, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) where T : class {
+            return subscriber.SubscribeAsync<T>((msg, token) => handler(msg), options, cancellationToken);
         }
 
-        public static Task SubscribeAsync<T>(this IMessageSubscriber subscriber, Action<T> handler, CancellationToken cancellationToken = default) where T : class {
+        public static Task<IMessageSubscription> SubscribeAsync<T>(this IMessageSubscriber subscriber, Action<T> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) where T : class {
             return subscriber.SubscribeAsync<T>((msg, token) => {
                 handler(msg);
                 return Task.CompletedTask;
-            }, cancellationToken);
+            }, options, cancellationToken);
         }
 
-        public static Task SubscribeAsync(this IMessageSubscriber subscriber, Func<IMessage, CancellationToken, Task> handler, CancellationToken cancellationToken = default) {
-            return subscriber.SubscribeAsync<IMessage>((msg, token) => handler(msg, token), cancellationToken);
+        public static Task<IMessageSubscription> SubscribeAsync(this IMessageSubscriber subscriber, Func<IMessage, CancellationToken, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) {
+            return subscriber.SubscribeAsync<IMessage>((msg, token) => handler(msg, token), options, cancellationToken);
         }
 
-        public static Task SubscribeAsync(this IMessageSubscriber subscriber, Func<IMessage, Task> handler, CancellationToken cancellationToken = default) {
-            return subscriber.SubscribeAsync((msg, token) => handler(msg), cancellationToken);
+        public static Task<IMessageSubscription> SubscribeAsync(this IMessageSubscriber subscriber, Func<IMessage, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) {
+            return subscriber.SubscribeAsync((msg, token) => handler(msg), options, cancellationToken);
         }
 
-        public static Task SubscribeAsync(this IMessageSubscriber subscriber, Action<IMessage> handler, CancellationToken cancellationToken = default) {
+        public static Task<IMessageSubscription> SubscribeAsync(this IMessageSubscriber subscriber, Action<IMessage> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) {
             return subscriber.SubscribeAsync((msg, token) => {
                 handler(msg);
                 return Task.CompletedTask;
-            }, cancellationToken);
+            }, options, cancellationToken);
         }
     }
+
+    public interface IMessageSubscription : IAsyncDisposable {
+        string SubscriptionId { get; }
+        IDictionary<string, string> Properties { get; }
+    }
+
+    public class MessageSubscription : IMessageSubscription {
+        private readonly Func<ValueTask> _disposeSubscriptionFunc;
+
+        public MessageSubscription(string subscriptionId, Func<ValueTask> disposeSubscriptionFunc) {
+            SubscriptionId = subscriptionId;
+            _disposeSubscriptionFunc = disposeSubscriptionFunc;
+        }
+
+        public string SubscriptionId { get; }
+        public IDictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
+
+        public ValueTask DisposeAsync() {
+            return _disposeSubscriptionFunc?.Invoke() ?? new ValueTask();
+        }
+    }
+
+    public class MessageSubscriptionOptions {
+        /// <summary>
+        /// The topic name
+        /// </summary>
+        public string Topic { get; set; }
+
+        public IDictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
+    }
 }

src/Foundatio/Messaging/Message.cs

@@ -1,17 +1,22 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
-using Foundatio.Serializer;
+using System.Threading.Tasks;
 
 namespace Foundatio.Messaging {
-    public interface IMessage {
+    public interface IMessage : IAsyncDisposable {
         string UniqueId { get; }
         string CorrelationId { get; }
+        string Topic { get; }
         string Type { get; }
         Type ClrType { get; }
         byte[] Data { get; }
         object GetBody();
         IDictionary<string, string> Properties { get; }
+
+        Task RenewLockAsync();
+        Task AbandonAsync();
+        Task CompleteAsync();
     }
 
     public interface IMessage<T> : IMessage where T: class {
@@ -29,11 +34,29 @@ public Message(byte[] data, Func<IMessage, object> getBody) {
 
         public string UniqueId { get; set; }
         public string CorrelationId { get; set; }
+        public string Topic { get; set; }
         public string Type { get; set; }
         public Type ClrType { get; set; }
         public IDictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
         public byte[] Data { get; set; }
+
         public object GetBody() => _getBody(this);
+
+        public virtual Task AbandonAsync() {
+            return Task.CompletedTask;
+        }
+
+        public virtual Task CompleteAsync() {
+            return Task.CompletedTask;
+        }
+
+        public virtual Task RenewLockAsync() {
+            return Task.CompletedTask;
+        }
+
+        public virtual ValueTask DisposeAsync() {
+            return new ValueTask(AbandonAsync());
+        }
     }
 
     public class Message<T> : IMessage<T> where T : class {
@@ -44,19 +67,19 @@ public Message(IMessage message) {
         }
 
         public byte[] Data => _message.Data;
-
         public T Body => (T)GetBody();
-
         public string UniqueId => _message.UniqueId;
-
         public string CorrelationId => _message.CorrelationId;
-
+        public string Topic => _message.Topic;
         public string Type => _message.Type;
-
         public Type ClrType => _message.ClrType;
-
         public IDictionary<string, string> Properties => _message.Properties;
 
         public object GetBody() => _message.GetBody();
+
+        public Task AbandonAsync() => _message.AbandonAsync();
+        public Task CompleteAsync() => _message.CompleteAsync();
+        public Task RenewLockAsync() => _message.RenewLockAsync();
+        public ValueTask DisposeAsync() => _message.DisposeAsync();
     }
 }

src/Foundatio/Messaging/MessageBusBase.cs

@@ -6,13 +6,15 @@
 using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Threading.Channels;
 using Foundatio.Serializer;
 using Foundatio.Utility;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 
 namespace Foundatio.Messaging {
     public abstract class MessageBusBase<TOptions> : IMessageBus, IDisposable where TOptions : SharedMessageBusOptions {
+        protected readonly ConcurrentDictionary<string, Channel<Message>> _topics = new();
         private readonly CancellationTokenSource _messageBusDisposedCancellationTokenSource;
         protected readonly ConcurrentDictionary<string, Subscriber> _subscribers = new();
         protected readonly TOptions _options;
@@ -25,7 +27,7 @@ public MessageBusBase(TOptions options) {
             var loggerFactory = options?.LoggerFactory ?? NullLoggerFactory.Instance;
             _logger = loggerFactory.CreateLogger(GetType());
             _serializer = options.Serializer ?? DefaultSerializer.Instance;
-            MessageBusId = _options.Topic + Guid.NewGuid().ToString("N").Substring(10);
+            MessageBusId = _options.DefaultTopic + Guid.NewGuid().ToString("N").Substring(10);
             _messageBusDisposedCancellationTokenSource = new CancellationTokenSource();
         }
 
@@ -59,11 +61,11 @@ protected string GetMappedMessageType(Type messageType) {
         }
 
         private readonly ConcurrentDictionary<string, Type> _knownMessageTypesCache = new();
-        protected virtual Type GetMappedMessageType(string messageType) {
-            if (String.IsNullOrEmpty(messageType))
+        protected virtual Type GetMappedMessageType(IConsumeMessageContext context) {
+            if (context == null || String.IsNullOrEmpty(context.MessageType))
                 return null;
 
-            return _knownMessageTypesCache.GetOrAdd(messageType, type => {
+            return _knownMessageTypesCache.GetOrAdd(context.MessageType, type => {
                 if (_options.MessageTypeMappings != null && _options.MessageTypeMappings.ContainsKey(type))
                     return _options.MessageTypeMappings[type];
 
@@ -86,11 +88,14 @@ protected virtual Type GetMappedMessageType(string messageType) {
                 }
             });
         }
+        protected Type GetMappedMessageType(string messageType) {
+            return GetMappedMessageType(new ConsumeMessageContext { MessageType = messageType });
+        }
 
         protected virtual Task RemoveTopicSubscriptionAsync() => Task.CompletedTask;
         protected virtual Task EnsureTopicSubscriptionAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
-        protected virtual Task SubscribeImplAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken) where T : class {
+        protected virtual Task<IMessageSubscription> SubscribeImplAsync<T>(Func<T, CancellationToken, Task> handler, MessageSubscriptionOptions options, CancellationToken cancellationToken = default) where T : class {
             var subscriber = new Subscriber {
                 CancellationToken = cancellationToken,
                 Type = typeof(T),
@@ -105,14 +110,6 @@ protected virtual Task SubscribeImplAsync<T>(Func<T, CancellationToken, Task> ha
                 }
             };
 
-            if (cancellationToken != CancellationToken.None) {
-                cancellationToken.Register(() => {
-                    _subscribers.TryRemove(subscriber.Id, out _);
-                    if (_subscribers.Count == 0)
-                        RemoveTopicSubscriptionAsync().GetAwaiter().GetResult();
-                });
-            }
-
             if (subscriber.Type.Name == "IMessage`1" && subscriber.Type.GenericTypeArguments.Length == 1) {
                 var modelType = subscriber.Type.GenericTypeArguments.Single();
                 subscriber.GenericType = typeof(Message<>).MakeGenericType(modelType);
@@ -121,15 +118,26 @@ protected virtual Task SubscribeImplAsync<T>(Func<T, CancellationToken, Task> ha
             if (!_subscribers.TryAdd(subscriber.Id, subscriber) && _logger.IsEnabled(LogLevel.Error))
                 _logger.LogError("Unable to add subscriber {SubscriberId}", subscriber.Id);
 
-            return Task.CompletedTask;
+            return Task.FromResult<IMessageSubscription>(new MessageSubscription(Guid.NewGuid().ToString("N"), async () => {
+                _subscribers.TryRemove(subscriber.Id, out _);
+                if (_subscribers.Count == 0)
+                    await RemoveTopicSubscriptionAsync().AnyContext();
+            }));
         }
 
-        public async Task SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken = default) where T : class {
+        public async Task<IMessageSubscription> SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) where T : class {
             if (_logger.IsEnabled(LogLevel.Trace))
                 _logger.LogTrace("Adding subscriber for {MessageType}.", typeof(T).FullName);
 
-            await SubscribeImplAsync(handler, cancellationToken).AnyContext();
+            options ??= new MessageSubscriptionOptions();
+
+            var sub = await SubscribeImplAsync(handler, options).AnyContext();
             await EnsureTopicSubscriptionAsync(cancellationToken).AnyContext();
+
+            if (cancellationToken != CancellationToken.None)
+                cancellationToken.Register(() => sub.DisposeAsync());
+
+            return sub;
         }
 
         protected List<Subscriber> GetMessageSubscribers(IMessage message) {

src/Foundatio/Messaging/NullMessageBus.cs

@@ -10,8 +10,8 @@ public Task PublishAsync(Type messageType, object message, MessageOptions option
             return Task.CompletedTask;
         }
 
-        public Task SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken = default) where T : class {
-            return Task.CompletedTask;
+        public Task<IMessageSubscription> SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, MessageSubscriptionOptions options = null, CancellationToken cancellationToken = default) where T : class {
+            return Task.FromResult<IMessageSubscription>(new MessageSubscription(Guid.NewGuid().ToString(), () => new ValueTask()));
         }
 
         public void Dispose() {}

src/Foundatio/Messaging/SharedMessageBusOptions.cs

@@ -1,26 +1,58 @@
-using System;
+using System;
 using System.Collections.Generic;
 
 namespace Foundatio.Messaging {
     public class SharedMessageBusOptions : SharedOptions {
         /// <summary>
-        /// The topic name
+        /// The default topic name
         /// </summary>
-        public string Topic { get; set; } = "messages";
+        public string DefaultTopic { get; set; } = "messages";
 
         /// <summary>
-        /// Controls which types messages are mapped to.
+        /// Resolves message types
+        /// </summary>
+        public IMessageRouter Router { get; set; }
+
+        /// <summary>
+        /// Statically configured message type mappings. <see cref="Router"/> will be run first and then this dictionary will be checked.
         /// </summary>
         public Dictionary<string, Type> MessageTypeMappings { get; set; } = new Dictionary<string, Type>();
     }
 
+    public interface IMessageRouter {
+        // get topic from bus options, message and message options
+        // get message type from message and options
+        // get .net type from topic, message type and properties (headers)
+        IConsumeMessageContext ToMessageType(Type messageType);
+        Type ToClrType(IConsumeMessageContext context);
+    }
+
+    public interface IConsumeMessageContext {
+        string Topic { get; set; }
+        string MessageType { get; set; }
+        IDictionary<string, string> Properties { get; }
+    }
+
+    public class ConsumeMessageContext : IConsumeMessageContext {
+        public string Topic { get; set; }
+        public string MessageType { get; set; } 
+        public IDictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
+    }
+
     public class SharedMessageBusOptionsBuilder<TOptions, TBuilder> : SharedOptionsBuilder<TOptions, TBuilder>
         where TOptions : SharedMessageBusOptions, new()
         where TBuilder : SharedMessageBusOptionsBuilder<TOptions, TBuilder> {
         public TBuilder Topic(string topic) {
             if (string.IsNullOrEmpty(topic))
                 throw new ArgumentNullException(nameof(topic));
-            Target.Topic = topic;
+            Target.DefaultTopic = topic;
+            return (TBuilder)this;
+        }
+
+        public TBuilder MessageTypeResolver(IMessageRouter resolver) {
+            if (resolver == null)
+                throw new ArgumentNullException(nameof(resolver));
+            Target.Router = resolver;
             return (TBuilder)this;
         }