✨ Allow applying content-type without having NServiceBus.ContentType header entry

Use global customization function

Author: ramonsmits

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.TransportTests/ConfigureRabbitMQTransportInfrastructure.cs

@@ -16,7 +16,17 @@ public TransportDefinition CreateTransportDefinition()
     {
         var connectionString = Environment.GetEnvironmentVariable("RabbitMQTransport_ConnectionString") ?? "host=localhost";
 
-        var transport = new RabbitMQTransport(RoutingTopology.Conventional(QueueType.Classic), connectionString, false);
+        var transport = new RabbitMQTransport(RoutingTopology.Conventional(QueueType.Classic), connectionString, false)
+        {
+            //Used by When_customizing_outgoing_messages test
+            OutgoingNativeMessageCustomization = (operation, properties) =>
+            {
+                if (operation.Properties.TryGetValue("ContentType", out var overrideContentType))
+                {
+                    properties.ContentType = overrideContentType;
+                }
+            }
+        };
 
         return transport;
     }

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

@@ -0,0 +1,40 @@
+namespace NServiceBus.Transport.RabbitMQ.TransportTests;
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using global::RabbitMQ.Client;
+using global::RabbitMQ.Client.Events;
+using NServiceBus.TransportTests;
+using NUnit.Framework;
+
+public class When_customizing_outgoing_messages : NServiceBusTransportTest
+{
+    [Test]
+    public async Task Should_override_default_values()
+    {
+        var onMessageInvoked = CreateTaskCompletionSource<MessageContext>();
+        IReadOnlyBasicProperties basicProps = null;
+
+        await StartPump(
+            (context, _) =>
+            {
+                var basicEventArgs = context.Extensions.Get<BasicDeliverEventArgs>();
+                basicProps = basicEventArgs.BasicProperties;
+
+                return onMessageInvoked.SetCompleted(context);
+            },
+            (_, __) => Task.FromResult(ErrorHandleResult.RetryRequired),
+            TransportTransactionMode.ReceiveOnly);
+
+        var dispatchProperties = new DispatchProperties { ["ContentType"] = "my-content" };
+
+        await SendMessage(InputQueueName, new Dictionary<string, string>
+        {
+            {"test-header", "original"}
+        }, null, dispatchProperties);
+
+        var messageContext = await onMessageInvoked.Task;
+
+        Assert.That(basicProps.ContentType, Is.EqualTo("my-content"));
+    }
+}

src/NServiceBus.Transport.RabbitMQ/RabbitMQTransport.cs

@@ -5,11 +5,13 @@
     using System.Security.Cryptography.X509Certificates;
     using System.Threading;
     using System.Threading.Tasks;
+    using RabbitMQ.Client;
     using RabbitMQ.Client.Events;
     using Transport;
     using Transport.RabbitMQ;
     using ConnectionFactory = Transport.RabbitMQ.ConnectionFactory;
 
+
     /// <summary>
     /// Transport definition for RabbitMQ.
     /// </summary>
@@ -91,6 +93,11 @@ public TimeSpan TimeToWaitBeforeTriggeringCircuitBreaker
             }
         }
 
+        /// <summary>
+        /// The callback to use when customizing the message before dispatching it to the broker.
+        /// </summary>
+        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>
@@ -192,7 +199,7 @@ public override async Task<TransportInfrastructure> Initialize(HostSettings host
             var converter = new MessageConverter(MessageIdStrategy);
 
             var infra = new RabbitMQTransportInfrastructure(hostSettings, receivers, connectionFactory,
-                RoutingTopology, channelProvider, converter, TimeToWaitBeforeTriggeringCircuitBreaker,
+                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,13 @@
     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 ?? ((_, _) => { });
             this.supportsDelayedDelivery = supportsDelayedDelivery;
         }
 
@@ -55,6 +57,7 @@ Task SendMessage(UnicastTransportOperation transportOperation, ConfirmsAwareChan
 
             var properties = new BasicProperties();
             properties.Fill(message, transportOperation.Properties);
+            messageCustomization(transportOperation, properties);
 
             return channel.SendMessage(transportOperation.Destination, message, properties, cancellationToken);
         }
@@ -67,6 +70,7 @@ Task PublishMessage(MulticastTransportOperation transportOperation, ConfirmsAwar
 
             var properties = new BasicProperties();
             properties.Fill(message, transportOperation.Properties);
+            messageCustomization(transportOperation, properties);
 
             return channel.PublishMessage(transportOperation.MessageType, message, properties, cancellationToken);
         }