Added envelop mapping for CorrelationId, added checks to NSB interop …

…unit test
JasperFx · Oct 10, 2023 · 2b18777 · 2b18777
1 parent 776d5c9
commit 2b18777
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 6 deletions.
diff --git a/src/Transports/RabbitMQ/InteropTests/NServiceBus/NServiceBusSpecs.cs b/src/Transports/RabbitMQ/InteropTests/NServiceBus/NServiceBusSpecs.cs
@@ -23,19 +23,21 @@ public async Task nservicebus_sends_message_to_wolverine()
         ResponseHandler.Received.Clear();
 
         var id = Guid.NewGuid();
-
+        var messageId = Guid.NewGuid().ToString();
 
         var session = await theFixture.Wolverine.ExecuteAndWaitAsync(async () =>
         {
+            var options = new PublishOptions();
+            options.SetMessageId(messageId);
             var sender = theFixture.NServiceBus.Services.GetRequiredService<IMessageSession>();
-            await sender.Publish(new ResponseMessage { Id = id });
+            await sender.Publish(new ResponseMessage { Id = id }, options);
         }, 60000);
 
         var envelope = ResponseHandler.Received.FirstOrDefault();
         envelope.Message.ShouldBeOfType<ResponseMessage>().Id.ShouldBe(id);
         envelope.ShouldNotBeNull();
 
-        envelope.CorrelationId.ShouldNotBeNull();
+        envelope.CorrelationId.ShouldBe(messageId);
         envelope.Id.ShouldNotBe(Guid.Empty);
         envelope.ConversationId.ShouldNotBe(Guid.Empty);
     }
@@ -46,19 +48,23 @@ public async Task nservicebus_sends_interface_to_wolverine_who_only_understands_
         ResponseHandler.Received.Clear();
 
         var id = Guid.NewGuid();
-
+        var messageId = Guid.NewGuid().ToString();
 
         var session = await theFixture.Wolverine.ExecuteAndWaitAsync(async () =>
         {
+            var options = new SendOptions();
+            options.SetMessageId(messageId);
+            options.SetDestination("wolverine");
             var sender = theFixture.NServiceBus.Services.GetRequiredService<IMessageSession>();
-            await sender.Send<IInterfaceMessage>("wolverine", x => x.Id = id );
+            await sender.Send<IInterfaceMessage>(x => x.Id = id, options);
         }, 60000);
 
         var envelope = ResponseHandler.Received.FirstOrDefault();
         envelope.Message.ShouldBeOfType<ConcreteMessage>().Id.ShouldBe(id);
         envelope.ShouldNotBeNull();
 
         envelope.CorrelationId.ShouldNotBeNull();
+        envelope.CorrelationId.ShouldBe(messageId);
         envelope.Id.ShouldNotBe(Guid.Empty);
         envelope.ConversationId.ShouldNotBe(Guid.Empty);
     }

diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.NServiceBus.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.NServiceBus.cs
@@ -14,7 +14,8 @@ public void UseNServiceBusInterop()
         {
             m.MapPropertyToHeader(x => x.ConversationId, "NServiceBus.ConversationId");
             m.MapPropertyToHeader(x => x.SentAt, "NServiceBus.TimeSent");
-
+            m.MapPropertyToHeader(x => x.CorrelationId!, "NServiceBus.CorrelationId");
+
             var replyAddress = new Lazy<string>(() =>
             {
                 var replyEndpoint = (RabbitMqEndpoint)_parent.ReplyEndpoint()!;