SubscribeFilter Include MediumMessage in ConsumerContext (#1464)

By including the MediumMessage, we give consumers more options for handling in
the subscribe filter, for example to implement rules based on the number of
retries already done, or to set disable retrying a message under some
circumstances.

Co-authored-by: Benjamin Schwehn <[email protected]>

Author: bschwehn

src/DotNetCore.CAP/Internal/ConsumerContext.cs

@@ -3,6 +3,7 @@
 
 using System;
 using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Persistence;
 
 namespace DotNetCore.CAP.Internal;
 
@@ -12,7 +13,7 @@ namespace DotNetCore.CAP.Internal;
 public class ConsumerContext
 {
     public ConsumerContext(ConsumerContext context)
-        : this(context.ConsumerDescriptor, context.DeliverMessage)
+        : this(context.ConsumerDescriptor, context.MediumMessage)
     {
     }
 
@@ -21,10 +22,10 @@ public ConsumerContext(ConsumerContext context)
     /// </summary>
     /// <param name="descriptor">consumer method descriptor. </param>
     /// <param name="message"> received message.</param>
-    public ConsumerContext(ConsumerExecutorDescriptor descriptor, Message message)
+    public ConsumerContext(ConsumerExecutorDescriptor descriptor, MediumMessage message)
     {
         ConsumerDescriptor = descriptor ?? throw new ArgumentNullException(nameof(descriptor));
-        DeliverMessage = message ?? throw new ArgumentNullException(nameof(message));
+        MediumMessage = message ?? throw new ArgumentNullException(nameof(message));
     }
 
     /// <summary>
@@ -35,5 +36,10 @@ public ConsumerContext(ConsumerExecutorDescriptor descriptor, Message message)
     /// <summary>
     /// consumer received message.
     /// </summary>
-    public Message DeliverMessage { get; }
+    public Message DeliverMessage => MediumMessage.Origin;
+
+    /// <summary>
+    /// consumer received medium message.
+    /// </summary>
+    public MediumMessage MediumMessage { get; }
 }

src/DotNetCore.CAP/Internal/ISubscribeExector.Default.cs

@@ -175,7 +175,7 @@ private bool UpdateMessageForRetry(MediumMessage message)
     private async Task InvokeConsumerMethodAsync(MediumMessage message, ConsumerExecutorDescriptor descriptor,
         CancellationToken cancellationToken)
     {
-        var consumerContext = new ConsumerContext(descriptor, message.Origin);
+        var consumerContext = new ConsumerContext(descriptor, message);
         var tracingTimestamp = TracingBefore(message.Origin, descriptor.MethodInfo);
         try
         {

test/DotNetCore.CAP.Test/SubscribeInvokerTest.cs

@@ -4,6 +4,7 @@
 using System.Threading.Tasks;
 using DotNetCore.CAP.Internal;
 using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Persistence;
 using DotNetCore.CAP.Serialization;
 using Microsoft.Extensions.DependencyInjection;
 using Xunit;
@@ -45,7 +46,8 @@ public async Task InvokeTest()
                 [Headers.MessageName] = "fake.output.integer"
             };
             var message = new Message(header, null);
-            var context = new ConsumerContext(descriptor, message);
+            var mediumMessage = new MediumMessage() { Origin = message };
+            var context = new ConsumerContext(descriptor, mediumMessage);
 
             var ret = await SubscribeInvoker.InvokeAsync(context);
             Assert.Equal(int.MaxValue, ret.Result);

test/DotNetCore.CAP.Test/SubscribeInvokerWithCancellation.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using DotNetCore.CAP.Internal;
 using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Persistence;
 using DotNetCore.CAP.Serialization;
 using Microsoft.Extensions.DependencyInjection;
 using Xunit;
@@ -55,7 +56,8 @@ public async Task InvokeTest()
                 [Headers.MessageName] = "fake.output.withcancellation"
             };
             var message = new Message(header, null);
-            var context = new ConsumerContext(descriptor, message);
+            var mediumMessage = new MediumMessage() { Origin = message };
+            var context = new ConsumerContext(descriptor, mediumMessage);
 
             var cancellationToken = new CancellationToken();
             var ret = await SubscribeInvoker.InvokeAsync(context, cancellationToken);