Added ability to set a custom routing key per message

src/Bindings/RabbitMQAsyncCollector.cs

@@ -9,7 +9,7 @@
 
 namespace Microsoft.Azure.WebJobs.Extensions.RabbitMQ
 {
-    internal class RabbitMQAsyncCollector : IAsyncCollector<byte[]>
+    internal class RabbitMQAsyncCollector : IAsyncCollector<RabbitMQMessage>
     {
         private readonly RabbitMQContext _context;
         private readonly IBasicPublishBatch _batch;
@@ -28,9 +28,9 @@ public RabbitMQAsyncCollector(RabbitMQContext context, ILogger logger)
             _batch = _context.Service.BasicPublishBatch;
         }
 
-        public Task AddAsync(byte[] message, CancellationToken cancellationToken = default)
+        public Task AddAsync(RabbitMQMessage message, CancellationToken cancellationToken = default)
         {
-            _batch.Add(exchange: string.Empty, routingKey: _context.ResolvedAttribute.QueueName, mandatory: false, properties: null, body: message);
+            _batch.Add(exchange: string.Empty, routingKey: message.RoutingKey ?? _context.ResolvedAttribute.QueueName, mandatory: false, properties: null, body: message.Body);
             _logger.LogDebug($"Adding message to batch for publishing...");
 
             return Task.CompletedTask;

src/Config/RabbitMQExtensionConfigProvider.cs

@@ -43,7 +43,7 @@ public void Initialize(ExtensionConfigContext context)
 
             var rule = context.AddBindingRule<RabbitMQAttribute>();
             rule.AddValidator(ValidateBinding);
-            rule.BindToCollector<byte[]>((attr) =>
+            rule.BindToCollector((attr) =>
             {
                 return new RabbitMQAsyncCollector(CreateContext(attr), _logger);
             });

src/RabbitMQMessage.cs

@@ -0,0 +1,14 @@
+namespace Microsoft.Azure.WebJobs.Extensions.RabbitMQ
+{
+    public class RabbitMQMessage
+    {
+        public RabbitMQMessage(byte[] body)
+        {
+            Body = body;
+        }
+
+        public byte[] Body { get; set; }
+
+        public string RoutingKey { get; set; }
+    }
+}

test/WebJobs.Extensions.RabbitMQ.Samples/RabbitMQSamples.cs

@@ -51,12 +51,12 @@ public static void TimerTrigger_PocoOutput(
         // So you can add items to the queue while the sample is running, and the trigger will be called until the queue is empty.
         public static async Task ProcessMessage_RabbitMQAsyncCollector(
             [QueueTrigger(@"samples-rabbitmq-messages")] string message,
-            [RabbitMQ(QueueName = "queue")] IAsyncCollector<byte[]> messages,
+            [RabbitMQ(QueueName = "queue")] IAsyncCollector<RabbitMQMessage> messages,
             ILogger logger)
         {
             logger.LogInformation($"Received queue trigger");
             byte[] messageInBytes = Encoding.UTF8.GetBytes(message);
-            await messages.AddAsync(messageInBytes);
+            await messages.AddAsync(new RabbitMQMessage(messageInBytes) { RoutingKey = "custom-routing-key" });
         }
 
         // To run:

test/WebJobs.Extensions.RabbitMQ.Tests/RabbitMQAsyncCollectorTests.cs

@@ -40,7 +40,7 @@ public async Task AddAsync_AddsMessagesToQueue()
             var collector = new RabbitMQAsyncCollector(context, logger);
 
             byte[] body = Encoding.UTF8.GetBytes("hi");
-            await collector.AddAsync(body);
+            await collector.AddAsync(new RabbitMQMessage(body));
 
             mockBatch.Verify(m => m.Add(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<bool>(), It.IsAny<IBasicProperties>(), body), Times.Exactly(1));
         }