Allow setting content-type AMQP attribute via DispatchProperties for the purposes of native integration directly via the transport seam

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

@@ -73,4 +73,8 @@ namespace NServiceBus.Transport.RabbitMQ
         System.Threading.Tasks.ValueTask SetupSubscription(RabbitMQ.Client.IChannel channel, NServiceBus.Unicast.Messages.MessageMetadata type, string subscriberName, System.Threading.CancellationToken cancellationToken = default);
         System.Threading.Tasks.ValueTask TeardownSubscription(RabbitMQ.Client.IChannel channel, NServiceBus.Unicast.Messages.MessageMetadata type, string subscriberName, System.Threading.CancellationToken cancellationToken = default);
     }
+    public static class NativeIntegrationDispatchPropertiesExtensions
+    {
+        public static void SetContentType(this NServiceBus.Transport.DispatchProperties dispatchProperties, string contentType) { }
+    }
 }

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

@@ -0,0 +1,42 @@
+namespace NServiceBus.Transport.RabbitMQ.TransportTests;
+
+using System;
+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_setting_content_type : NServiceBusTransportTest
+{
+    [Test]
+    public async Task Should_use_provided_value()
+    {
+        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();
+        dispatchProperties.SetContentType("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/Sending/BasicPropertiesExtensions.cs

@@ -57,7 +57,11 @@ public static void Fill(this IBasicProperties properties, OutgoingMessage messag
                 }
             }
 
-            if (messageHeaders.TryGetValue(NServiceBus.Headers.ContentType, out var contentType) && contentType != null)
+            if (dispatchProperties.TryGetValue(NativeIntegrationDispatchPropertiesExtensions.ContentTypeAttribute, out var contentType) && contentType != null)
+            {
+                properties.ContentType = contentType;
+            }
+            else if (messageHeaders.TryGetValue(NServiceBus.Headers.ContentType, out contentType) && contentType != null)
             {
                 properties.ContentType = contentType;
             }

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

@@ -0,0 +1,19 @@
+namespace NServiceBus.Transport.RabbitMQ;
+
+/// <summary>
+/// Provides APIs useful for integrating with non-NServiceBus receivers.
+/// </summary>
+public static class NativeIntegrationDispatchPropertiesExtensions
+{
+    internal static readonly string ContentTypeAttribute = "AMQP.content-type";
+
+    /// <summary>
+    /// Sets the content type to be used in the AMQP content-type field, overriding NServiceBus header.
+    /// </summary>
+    /// <param name="dispatchProperties">Dispatch properties object.</param>
+    /// <param name="contentType">Content type value.</param>
+    public static void SetContentType(this DispatchProperties dispatchProperties, string contentType)
+    {
+        dispatchProperties[ContentTypeAttribute] = contentType;
+    }
+}