inline subscriptions for messages

zero/Communication/Messages/MessageAggregator.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Linq.Expressions;
 
 namespace zero.Communication;
 
@@ -26,6 +27,13 @@ public class MessageAggregator : IMessageAggregator
   }
 
 
+  /// <inheritdoc />
+  public void Subscribe<TMessage>(Expression<Func<TMessage, Task>> handle) where TMessage : class, IMessage
+  {
+    Subscription.Add(new InlineMessageSubscription<TMessage>(handle));
+  }
+
+
   /// <inheritdoc />
   public void Activate<TMessage, TMessageHandler>()
     where TMessage : class, IMessage
@@ -43,6 +51,11 @@ public interface IMessageAggregator
   /// </summary>
   Task Publish<TMessage>(TMessage message) where TMessage : class, IMessage;
 
+  /// <summary>
+  /// Subscribe to a message
+  /// </summary>
+  void Subscribe<TMessage>(Expression<Func<TMessage, Task>> handle) where TMessage : class, IMessage;
+
   /// <summary>
   /// Subscribes the specified handler to the spified message type
   /// </summary>

zero/Communication/Messages/MessageSubscription.cs

@@ -1,4 +1,5 @@
 using Microsoft.Extensions.DependencyInjection;
+using System.Linq.Expressions;
 using System.Reflection;
 
 namespace zero.Communication;
@@ -33,6 +34,38 @@ internal class MessageSubscription<TMessage, TMessageHandler> : IMessageSubscrip
 }
 
 
+internal class InlineMessageSubscription<TMessage> : IMessageSubscription
+  where TMessage : class, IMessage
+{
+  private Expression<Func<TMessage, Task>> _handler;
+
+  public InlineMessageSubscription(Expression<Func<TMessage, Task>> handler)
+  {
+    _handler = handler;
+  }
+
+
+  public bool CanDeliver<T>()
+  {
+    return typeof(TMessage).GetTypeInfo().IsAssignableFrom(typeof(T));
+  }
+
+  public async Task Deliver(IServiceProvider serviceProvider, object message)
+  {
+    if (message == null)
+    {
+      throw new ArgumentNullException(nameof(message));
+    }
+    if (!(message is TMessage))
+    {
+      throw new ArgumentException($"{nameof(message)} must be of type '{typeof(TMessage).FullName}'");
+    }
+
+    await _handler.Compile()((TMessage)message);
+  }
+}
+
+
 public interface IMessageSubscription
 {
   bool CanDeliver<TMessage>();