#271 Allow for publishing of messages to a request/response queue. D…

…rop response as there is no listener. Signed-off-by: Richard Pringle <richardpringle@gmail.com>
zarusz · Jun 25, 2024 · 0981d8d · 0981d8d
1 parent 11c7c48
commit 0981d8d
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 6 deletions.
diff --git a/src/SlimMessageBus.Host/MessageBusBase.cs b/src/SlimMessageBus.Host/MessageBusBase.cs
@@ -610,20 +610,21 @@ public virtual Task ProduceResponse(string requestId, object request, IReadOnlyD
         if (consumerInvoker == null) throw new ArgumentNullException(nameof(consumerInvoker));
 
         var responseType = consumerInvoker.ParentSettings.ResponseType;
+        if (!requestHeaders.TryGetHeader(ReqRespMessageHeaders.ReplyTo, out object replyTo))
+        {
+            _logger.LogDebug($$"""Skipping sending response {Response} of type {MessageType} as the header {{ReqRespMessageHeaders.ReplyTo}} is missing for RequestId: {RequestId}""", response, responseType, requestId);
+            return Task.CompletedTask;
+        }
+
         _logger.LogDebug("Sending the response {Response} of type {MessageType} for RequestId: {RequestId}...", response, responseType, requestId);
-
+
         var responseHeaders = CreateHeaders();
         responseHeaders.SetHeader(ReqRespMessageHeaders.RequestId, requestId);
         if (responseException != null)
         {
             responseHeaders.SetHeader(ReqRespMessageHeaders.Error, responseException.Message);
         }
 
-        if (!requestHeaders.TryGetHeader(ReqRespMessageHeaders.ReplyTo, out object replyTo))
-        {
-            throw new MessageBusException($"The header {ReqRespMessageHeaders.ReplyTo} was missing on the message");
-        }
-
         _headerService.AddMessageTypeHeader(response, responseHeaders);
 
         return ProduceToTransport(response, responseType, (string)replyTo, responseHeaders, null);

diff --git a/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs b/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs
@@ -1,4 +1,9 @@
 namespace SlimMessageBus.Host.Test;
+
+using System.Threading;
+
+using Moq.Protected;
+
 using SlimMessageBus.Host.Test.Common;
 
 public class MessageBusBaseTests : IDisposable
@@ -655,4 +660,50 @@ public async Task When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce(
         bus._startedCount.Should().Be(1);
         bus._stoppedCount.Should().Be(1);
     }
+
+    public class ProduceResponseTests
+    {
+        [Fact]
+        public async Task When_Given_NoReplyToHeader_DoNothing()
+        {
+            // arrange
+            var requestId = "req-123";
+            var request = new object();
+            var response = new object();
+
+            object value;
+            var mockRequestHeaders = new Mock<IReadOnlyDictionary<string, object>>();
+            mockRequestHeaders.Setup(x => x.TryGetValue(ReqRespMessageHeaders.ReplyTo, out value)).Returns(false).Verifiable(Times.Once);
+
+            var mockMessageTypeResolver = new Mock<IMessageTypeResolver>();
+
+            var mockServiceProvider = new Mock<IServiceProvider>();
+            mockServiceProvider.Setup(x => x.GetService(typeof(IMessageTypeResolver))).Returns(mockMessageTypeResolver.Object);
+
+            var mockMessageTypeConsumerInvokerSettings = new Mock<IMessageTypeConsumerInvokerSettings>();
+            mockMessageTypeConsumerInvokerSettings.SetupGet(x => x.ParentSettings).Returns(() => new ConsumerSettings() { ResponseType = response.GetType() });
+
+            var settings = new MessageBusSettings { ServiceProvider = mockServiceProvider.Object };
+
+            var mockMessageBus = new Mock<MessageBusBase>(settings) { CallBase = true };
+            mockMessageBus.Protected().Setup<Task<ProduceToTransportBulkResult<BulkMessageEnvelope>>>(
+                "ProduceToTransportBulk",
+                [typeof(BulkMessageEnvelope)],
+                false,
+                ItExpr.IsAny<IReadOnlyCollection<BulkMessageEnvelope>>(),
+                ItExpr.IsAny<string>(),
+                ItExpr.IsAny<IMessageBusTarget>(),
+                ItExpr.IsAny<CancellationToken>())
+                .Verifiable(Times.Never);
+
+            var target = mockMessageBus.Object;
+
+            // act
+            await target.ProduceResponse(requestId, request, mockRequestHeaders.Object, response, null, mockMessageTypeConsumerInvokerSettings.Object);
+
+            // assert
+            mockRequestHeaders.VerifyAll();
+            mockMessageBus.VerifyAll();
+        }
+    }
 }