Merge pull request #1468 from Particular/set-native-content-type

Allow setting customizing the outgoing messages by setting AMQP attributes just before dispatching the message to the broker

Author: SzymonPobiega

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/When_customizing_outgoing_messages.cs

@@ -0,0 +1,67 @@
+namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
+{
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using global::RabbitMQ.Client.Events;
+    using NServiceBus.AcceptanceTests;
+    using NServiceBus.AcceptanceTests.EndpointTemplates;
+    using NUnit.Framework;
+
+    class When_customizing_outgoing_messages : NServiceBusAcceptanceTest
+    {
+        [Test]
+        public async Task Should_set_value()
+        {
+            var scenario = await Scenario.Define<Context>()
+                .WithEndpoint<Receiver>(b => b.When((bus, c) => bus.SendLocal(new Message())))
+                .Done(c => c.MessageReceived)
+                .Run();
+
+            Assert.That(scenario.BasicDeliverEventArgs.BasicProperties.AppId, Is.EqualTo("MyValue"));
+        }
+
+        public class Receiver : EndpointConfigurationBuilder
+        {
+            public Receiver()
+            {
+                EndpointSetup<DefaultServer>(endpointConfiguration =>
+                {
+                    endpointConfiguration.ConfigureRabbitMQTransport().OutgoingNativeMessageCustomization =
+                        (operation, properties) =>
+                        {
+                            properties.AppId = "MyValue";
+                        };
+                });
+            }
+
+            class MyEventHandler : IHandleMessages<Message>
+            {
+                Context testContext;
+
+                public MyEventHandler(Context testContext)
+                {
+                    this.testContext = testContext;
+                }
+
+                public Task Handle(Message message, IMessageHandlerContext context)
+                {
+                    testContext.BasicDeliverEventArgs = context.Extensions.Get<BasicDeliverEventArgs>();
+                    testContext.MessageReceived = true;
+
+                    return Task.CompletedTask;
+                }
+            }
+        }
+
+        public class Message : IMessage
+        {
+        }
+
+        class Context : ScenarioContext
+        {
+            public bool MessageReceived { get; set; }
+
+            public BasicDeliverEventArgs BasicDeliverEventArgs { get; set; }
+        }
+    }
+}

src/NServiceBus.Transport.RabbitMQ.Tests/ApprovalFiles/APIApprovals.Approve.approved.txt

@@ -22,6 +22,7 @@ namespace NServiceBus
         public System.TimeSpan HeartbeatInterval { get; set; }
         public System.Func<RabbitMQ.Client.Events.BasicDeliverEventArgs, string> MessageIdStrategy { get; set; }
         public System.TimeSpan NetworkRecoveryInterval { get; set; }
+        public System.Action<NServiceBus.Transport.IOutgoingTransportOperation, RabbitMQ.Client.IBasicProperties> OutgoingNativeMessageCustomization { get; set; }
         public NServiceBus.PrefetchCountCalculation PrefetchCountCalculation { get; set; }
         public System.TimeSpan TimeToWaitBeforeTriggeringCircuitBreaker { get; set; }
         public bool UseExternalAuthMechanism { get; set; }

src/NServiceBus.Transport.RabbitMQ/RabbitMQTransport.cs

@@ -5,6 +5,7 @@
     using System.Security.Cryptography.X509Certificates;
     using System.Threading;
     using System.Threading.Tasks;
+    using RabbitMQ.Client;
     using RabbitMQ.Client.Events;
     using Transport;
     using Transport.RabbitMQ;
@@ -91,6 +92,21 @@ public TimeSpan TimeToWaitBeforeTriggeringCircuitBreaker
             }
         }
 
+        /// <summary>
+        /// Gets or sets the action that allows customization of the native <see cref="BasicProperties"/> 
+        /// just before it is dispatched to the rabbitmq client. 
+        /// </summary>
+        /// <remarks>
+        /// <para>
+        /// This customization is applied after any configured transport customizations, meaning that 
+        /// any changes made here may override or conflict with previous transport-level adjustments. 
+        /// Exercise caution, as modifying the message at this stage can lead to unintended behavior 
+        /// downstream if the message structure or properties are altered in ways that do not align 
+        /// with expectations elsewhere in the system.
+        /// </para>
+        /// </remarks>
+        public Action<IOutgoingTransportOperation, IBasicProperties> OutgoingNativeMessageCustomization { get; set; }
+
         /// <summary>
         /// The calculation method for the prefetch count. The default is 3 times the maximum concurrency value.
         /// </summary>
@@ -183,17 +199,35 @@ public override async Task<TransportInfrastructure> Initialize(HostSettings host
                 certCollection = new X509Certificate2Collection(ClientCertificate);
             }
 
-            var connectionFactory = new ConnectionFactory(hostSettings.Name, ConnectionConfiguration, certCollection, !ValidateRemoteCertificate,
-                UseExternalAuthMechanism, HeartbeatInterval, NetworkRecoveryInterval, additionalClusterNodes);
+            var connectionFactory = new ConnectionFactory(
+                hostSettings.Name,
+                ConnectionConfiguration,
+                certCollection,
+                !ValidateRemoteCertificate,
+                UseExternalAuthMechanism,
+                HeartbeatInterval,
+                NetworkRecoveryInterval,
+                additionalClusterNodes
+            );
 
             var channelProvider = new ChannelProvider(connectionFactory, NetworkRecoveryInterval, RoutingTopology);
             await channelProvider.CreateConnection(cancellationToken).ConfigureAwait(false);
 
             var converter = new MessageConverter(MessageIdStrategy);
 
-            var infra = new RabbitMQTransportInfrastructure(hostSettings, receivers, connectionFactory,
-                RoutingTopology, channelProvider, converter, TimeToWaitBeforeTriggeringCircuitBreaker,
-                PrefetchCountCalculation, NetworkRecoveryInterval, SupportsDelayedDelivery);
+            var infra = new RabbitMQTransportInfrastructure(
+                hostSettings,
+                receivers,
+                connectionFactory,
+                RoutingTopology,
+                channelProvider,
+                converter,
+                OutgoingNativeMessageCustomization,
+                TimeToWaitBeforeTriggeringCircuitBreaker,
+                PrefetchCountCalculation,
+                NetworkRecoveryInterval,
+                SupportsDelayedDelivery
+            );
 
             if (hostSettings.SetupInfrastructure)
             {

src/NServiceBus.Transport.RabbitMQ/RabbitMQTransportInfrastructure.cs

@@ -5,6 +5,7 @@
     using System.Text;
     using System.Threading;
     using System.Threading.Tasks;
+    using global::RabbitMQ.Client;
 
     sealed class RabbitMQTransportInfrastructure : TransportInfrastructure
     {
@@ -17,6 +18,7 @@ sealed class RabbitMQTransportInfrastructure : TransportInfrastructure
         public RabbitMQTransportInfrastructure(HostSettings hostSettings, ReceiveSettings[] receiverSettings,
             ConnectionFactory connectionFactory, IRoutingTopology routingTopology,
             ChannelProvider channelProvider, MessageConverter messageConverter,
+            Action<IOutgoingTransportOperation, IBasicProperties> messageCustomization,
             TimeSpan timeToWaitBeforeTriggeringCircuitBreaker, PrefetchCountCalculation prefetchCountCalculation,
             TimeSpan networkRecoveryInterval, bool supportsDelayedDelivery)
         {
@@ -26,7 +28,7 @@ public RabbitMQTransportInfrastructure(HostSettings hostSettings, ReceiveSetting
             this.networkRecoveryInterval = networkRecoveryInterval;
             this.supportsDelayedDelivery = supportsDelayedDelivery;
 
-            Dispatcher = new MessageDispatcher(channelProvider, supportsDelayedDelivery);
+            Dispatcher = new MessageDispatcher(channelProvider, messageCustomization, supportsDelayedDelivery);
             Receivers = receiverSettings.Select(x => CreateMessagePump(hostSettings, x, messageConverter, timeToWaitBeforeTriggeringCircuitBreaker, prefetchCountCalculation))
                 .ToDictionary(x => x.Id, x => x);
         }

src/NServiceBus.Transport.RabbitMQ/Sending/MessageDispatcher.cs

@@ -10,11 +10,17 @@
     class MessageDispatcher : IMessageDispatcher
     {
         readonly ChannelProvider channelProvider;
+        readonly Action<IOutgoingTransportOperation, IBasicProperties> messageCustomization;
         readonly bool supportsDelayedDelivery;
 
-        public MessageDispatcher(ChannelProvider channelProvider, bool supportsDelayedDelivery)
+        public MessageDispatcher(
+            ChannelProvider channelProvider,
+            Action<IOutgoingTransportOperation, IBasicProperties> messageCustomization,
+            bool supportsDelayedDelivery
+        )
         {
             this.channelProvider = channelProvider;
+            this.messageCustomization = messageCustomization ?? (static (_, _) => { });
             this.supportsDelayedDelivery = supportsDelayedDelivery;
         }
 
@@ -56,6 +62,7 @@ ValueTask SendMessage(UnicastTransportOperation transportOperation, ConfirmsAwar
 
             var properties = new BasicProperties();
             properties.Fill(message, transportOperation.Properties);
+            messageCustomization(transportOperation, properties);
 
             return channel.SendMessage(transportOperation.Destination, message, properties, cancellationToken);
         }
@@ -68,6 +75,7 @@ ValueTask PublishMessage(MulticastTransportOperation transportOperation, Confirm
 
             var properties = new BasicProperties();
             properties.Fill(message, transportOperation.Properties);
+            messageCustomization(transportOperation, properties);
 
             return channel.PublishMessage(transportOperation.MessageType, message, properties, cancellationToken);
         }