Skip to content

Commit ecf6512

Browse files
zaruszzarusz
committed
Handle RabbitMQ RPC deserialization failures
Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>
1 parent 417fc63 commit ecf6512

6 files changed

Lines changed: 175 additions & 64 deletions

File tree

‎src/Host.Plugin.Properties.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,6 @@
33
<Import Project="Common.NuGet.Properties.xml" />
44

55
<PropertyGroup>
6-
<Version>3.5.0</Version>
6+
<Version>3.6.0-rc100</Version>
77
</PropertyGroup>
88
</Project>

‎src/SlimMessageBus.Host.RabbitMQ/Consumers/RabbitMqConsumer.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -53,12 +53,12 @@ IMessageProcessor<BasicDeliverEventArgs> CreateMessageProcessor(IEnumerable<Cons
5353
messageProcessor = new RabbitMqAutoAcknowledgeMessageProcessor(messageProcessor, Logger, _acknowledgementMode, this);
5454

5555
// pick the maximum number of instances
56-
var instances = consumers.Max(x => x.Instances);
57-
// For a given rabbit channel, there is only 1 task that dispatches messages. We want to be able to let each SMB consume process within its own task (1 or more)
58-
messageProcessor = new ConcurrentMessageProcessorDecorator<BasicDeliverEventArgs>(instances, loggerFactory, messageProcessor);
59-
60-
return messageProcessor;
61-
}
56+
var instances = consumers.Max(x => x.Instances);
57+
// For a given rabbit channel, there is only 1 task that dispatches messages. We want to be able to let each SMB consume process within its own task (1 or more)
58+
messageProcessor = new ConcurrentMessageProcessorDecorator<BasicDeliverEventArgs>(instances, loggerFactory, messageProcessor, reportBackgroundExceptions: false);
59+
60+
return messageProcessor;
61+
}
6262

6363
var routingKeyProcessors = consumers
6464
.GroupBy(x => x.GetBindingRoutingKey() ?? string.Empty)

‎src/SlimMessageBus.Host/Consumer/MessageProcessors/ConcurrentMessageProcessorDecorator.cs‎

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -9,27 +9,29 @@ public sealed partial class ConcurrentMessageProcessorDecorator<TMessage> : IMes
99
{
1010
private readonly ILogger _logger;
1111
private SemaphoreSlim _concurrentSemaphore;
12-
private readonly IMessageProcessor<TMessage> _target;
13-
private Exception _lastException;
14-
private TMessage _lastExceptionMessage;
15-
private AbstractConsumerSettings _lastExceptionSettings;
16-
private readonly object _lastExceptionLock = new();
12+
private readonly IMessageProcessor<TMessage> _target;
13+
private Exception _lastException;
14+
private TMessage _lastExceptionMessage;
15+
private AbstractConsumerSettings _lastExceptionSettings;
16+
private readonly bool _reportBackgroundExceptions;
17+
private readonly object _lastExceptionLock = new();
1718

1819
private int _pendingCount;
1920

2021
public int PendingCount => _pendingCount;
2122

2223
public IReadOnlyCollection<AbstractConsumerSettings> ConsumerSettings => _target.ConsumerSettings;
2324

24-
public ConcurrentMessageProcessorDecorator(int concurrency, ILoggerFactory loggerFactory, IMessageProcessor<TMessage> target)
25-
{
26-
if (target is null) throw new ArgumentNullException(nameof(target));
27-
if (loggerFactory is null) throw new ArgumentNullException(nameof(loggerFactory));
25+
public ConcurrentMessageProcessorDecorator(int concurrency, ILoggerFactory loggerFactory, IMessageProcessor<TMessage> target, bool reportBackgroundExceptions = true)
26+
{
27+
if (target is null) throw new ArgumentNullException(nameof(target));
28+
if (loggerFactory is null) throw new ArgumentNullException(nameof(loggerFactory));
2829
if (concurrency <= 0) throw new ArgumentOutOfRangeException(nameof(concurrency));
2930

30-
_logger = loggerFactory.CreateLogger<ConcurrentMessageProcessorDecorator<TMessage>>();
31-
_concurrentSemaphore = new SemaphoreSlim(concurrency);
32-
_target = target;
31+
_logger = loggerFactory.CreateLogger<ConcurrentMessageProcessorDecorator<TMessage>>();
32+
_concurrentSemaphore = new SemaphoreSlim(concurrency);
33+
_target = target;
34+
_reportBackgroundExceptions = reportBackgroundExceptions;
3335
}
3436

3537
#region IDisposable
@@ -90,10 +92,10 @@ private async Task ProcessInBackground(TMessage transportMessage, IReadOnlyDicti
9092
LogEntering(typeof(TMessage));
9193

9294
var r = await _target.ProcessMessage(transportMessage, messageHeaders, consumerContextProperties, currentServiceProvider, cancellationToken).ConfigureAwait(false);
93-
if (r.Exception != null)
94-
{
95-
lock (_lastExceptionLock)
96-
{
95+
if (_reportBackgroundExceptions && r.Exception != null)
96+
{
97+
lock (_lastExceptionLock)
98+
{
9799
// ensure there was no error before this one, in which case forget about this error (the whole event stream will be rewind back).
98100
if (_lastException == null && _lastExceptionMessage == null)
99101
{
@@ -141,4 +143,4 @@ private partial void LogLeaving(Type messageType)
141143
=> _logger.LogDebug("Leaving ProcessMessages for message {MessageType}", messageType);
142144
}
143145

144-
#endif
146+
#endif

‎src/SlimMessageBus.Host/Consumer/MessageProcessors/MessageProcessor.cs‎

Lines changed: 49 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -83,26 +83,26 @@ public async virtual Task<ProcessMessageResult> ProcessMessage(TTransportMessage
8383
{
8484
IMessageTypeConsumerInvokerSettings lastConsumerInvoker = null;
8585
var result = ProcessResult.Success;
86-
Exception lastException = null;
87-
object lastResponse = null;
88-
Type messageType = null;
89-
90-
try
91-
{
92-
messageType = _messageTypeProvider != null
86+
Exception lastException = null;
87+
object lastResponse = null;
88+
Type messageType = null;
89+
IReadOnlyCollection<IMessageTypeConsumerInvokerSettings> consumerInvokers = null;
90+
91+
try
92+
{
93+
messageType = _messageTypeProvider != null
9394
? _messageTypeProvider(transportMessage, messageHeaders)
9495
: GetMessageType(messageHeaders);
95-
96-
if (messageType != null)
97-
{
98-
var message = _messageProvider(messageType, messageHeaders, transportMessage);
99-
try
100-
{
101-
var consumerInvokers = TryMatchConsumerInvoker(messageType, messageHeaders, transportMessage);
102-
103-
foreach (var consumerInvoker in consumerInvokers)
104-
{
105-
lastConsumerInvoker = consumerInvoker;
96+
97+
if (messageType != null)
98+
{
99+
consumerInvokers = [.. TryMatchConsumerInvoker(messageType, messageHeaders, transportMessage)];
100+
var message = _messageProvider(messageType, messageHeaders, transportMessage);
101+
try
102+
{
103+
foreach (var consumerInvoker in consumerInvokers)
104+
{
105+
lastConsumerInvoker = consumerInvoker;
106106

107107
// Skip the loop if it was cancelled
108108
if (cancellationToken.IsCancellationRequested)
@@ -145,12 +145,37 @@ public async virtual Task<ProcessMessageResult> ProcessMessage(TTransportMessage
145145
}
146146
catch (Exception e)
147147
{
148-
LogProcessingMessageFailedTypeKnown(transportMessage, messageType, e);
149-
lastException ??= e;
150-
result = ProcessResult.Failure;
151-
}
152-
return new(result, lastException, lastException != null ? lastConsumerInvoker?.ParentSettings : null, lastResponse);
153-
}
148+
LogProcessingMessageFailedTypeKnown(transportMessage, messageType, e);
149+
lastException ??= e;
150+
result = ProcessResult.Failure;
151+
152+
if (consumerInvokers != null)
153+
{
154+
var consumerInvoker = consumerInvokers.FirstOrDefault(x => x.ParentSettings.ConsumerMode == ConsumerMode.RequestResponse);
155+
if (consumerInvoker != null)
156+
{
157+
lastConsumerInvoker = consumerInvoker;
158+
159+
if (_responseProducer != null && messageHeaders != null)
160+
{
161+
messageHeaders.TryGetHeader(ReqRespMessageHeaders.RequestId, out string requestId);
162+
try
163+
{
164+
await _responseProducer.ProduceResponse(requestId, null, messageHeaders, null, e, consumerInvoker, cancellationToken).ConfigureAwait(false);
165+
166+
// Clear the exception as it will be returned to the sender.
167+
lastException = null;
168+
}
169+
catch
170+
{
171+
// Keep the original exception so the transport can handle the failed message.
172+
}
173+
}
174+
}
175+
}
176+
}
177+
return new(result, lastException, lastException != null ? lastConsumerInvoker?.ParentSettings : null, lastResponse);
178+
}
154179

155180
protected Type GetMessageType(IReadOnlyDictionary<string, object> headers)
156181
{

‎src/Tests/SlimMessageBus.Host.Test/Consumer/ConcurrentMessageProcessorDecoratorTest.cs‎

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,42 @@ public async Task When_ProcessMessage_Given_ExceptionHappensOnTarget_Then_Except
158158
result.Exception.Should().Be(exception);
159159
}
160160

161+
[Fact]
162+
public async Task When_ProcessMessage_Given_BackgroundExceptionReportingDisabled_Then_ExceptionIsNotReportedOnSecondInvocation()
163+
{
164+
// arrange
165+
var subject = new ConcurrentMessageProcessorDecorator<SomeMessage>(1, NullLoggerFactory.Instance, _messageProcessorMock.Object, reportBackgroundExceptions: false);
166+
167+
var exception = new Exception("Boom!");
168+
var firstCallCompleted = new TaskCompletionSource<bool>();
169+
var callCount = 0;
170+
171+
_messageProcessorMock
172+
.Setup(x => x.ProcessMessage(It.IsAny<SomeMessage>(), It.IsAny<IReadOnlyDictionary<string, object>>(), It.IsAny<IDictionary<string, object>>(), It.IsAny<IServiceProvider>(), It.IsAny<CancellationToken>()))
173+
.Returns(() =>
174+
{
175+
if (Interlocked.Increment(ref callCount) == 1)
176+
{
177+
firstCallCompleted.TrySetResult(true);
178+
}
179+
return Task.FromResult(new ProcessMessageResult(ProcessResult.Failure, exception, null, null));
180+
});
181+
182+
var msg = new SomeMessage();
183+
var msgHeaders = new Dictionary<string, object>();
184+
await subject.ProcessMessage(msg, msgHeaders, default);
185+
await firstCallCompleted.Task;
186+
await subject.WaitAll(CancellationToken.None);
187+
188+
// act
189+
var result = await subject.ProcessMessage(msg, msgHeaders, default);
190+
await subject.WaitAll(CancellationToken.None);
191+
192+
// assert
193+
result.Exception.Should().BeNull();
194+
callCount.Should().Be(2);
195+
}
196+
161197
[Fact]
162198
public async Task When_ProcessMessage_Given_ExceptionHappensOnTarget_Then_SemaphoreIsNotLeakedAndProcessingContinues()
163199
{
@@ -214,4 +250,3 @@ public async Task When_ProcessMessage_Given_ExceptionHappensOnTarget_Then_Semaph
214250
subject.PendingCount.Should().Be(0); // All processing should complete
215251
}
216252
}
217-

‎src/Tests/SlimMessageBus.Host.Test/Consumer/MessageProcessorTest.cs‎

Lines changed: 64 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -32,13 +32,14 @@ public MessageProcessorTest()
3232
consumerMethodMock = new Mock<ConsumerMethod>();
3333
consumerInvokerMock.SetupGet(x => x.ConsumerMethod).Returns(consumerMethodMock.Object);
3434

35-
responseProducerMock = new Mock<IResponseProducer>();
36-
messageProviderMock = new Mock<MessageProvider<SomeMessage>>();
37-
38-
consumerBuilder = new ConsumerBuilder<SomeMessage>(new MessageBusSettings());
39-
40-
subject = new Lazy<MessageProcessor<SomeMessage>>(
41-
() => new MessageProcessor<SomeMessage>(
35+
responseProducerMock = new Mock<IResponseProducer>();
36+
messageProviderMock = new Mock<MessageProvider<SomeMessage>>();
37+
38+
consumerBuilder = new ConsumerBuilder<SomeMessage>(new MessageBusSettings());
39+
consumerBuilder.WithConsumer<SomeMessageConsumer>();
40+
41+
subject = new Lazy<MessageProcessor<SomeMessage>>(
42+
() => new MessageProcessor<SomeMessage>(
4243
consumerSettings: [consumerBuilder.ConsumerSettings],
4344
messageBus: busMock.Bus,
4445
messageProvider: messageProviderMock.Object,
@@ -71,13 +72,61 @@ public async Task When_ProcessMessage_Given_MessagePayloadCannotDeserialize_Then
7172
messageProviderMock.VerifyNoOtherCalls();
7273

7374
result.Should().NotBeNull();
74-
result.Result.Should().Be(ProcessResult.Failure);
75-
result.Exception.Should().NotBeNull();
76-
}
77-
78-
[Theory]
79-
[InlineData(null, "UnknownMessage")]
80-
[InlineData(true, "UnknownMessage")]
75+
result.Result.Should().Be(ProcessResult.Failure);
76+
result.Exception.Should().NotBeNull();
77+
}
78+
79+
[Fact]
80+
public async Task When_ProcessMessage_Given_RequestPayloadCannotDeserialize_Then_ErrorResponseIsSent()
81+
{
82+
// arrange
83+
var transportMessageMock = new Mock<SomeMessage>();
84+
var headers = new Dictionary<string, object>
85+
{
86+
{ MessageHeaders.MessageType, typeof(SomeRequest).AssemblyQualifiedName },
87+
{ ReqRespMessageHeaders.RequestId, "request-id" },
88+
{ ReqRespMessageHeaders.ReplyTo, "reply-to" }
89+
};
90+
91+
var deserializationException = new InvalidOperationException("Deserialization failed");
92+
var handlerBuilder = new HandlerBuilder<SomeRequest, SomeResponse>(new MessageBusSettings());
93+
handlerBuilder.WithHandler<SomeRequestMessageHandler>();
94+
95+
var subject = new MessageProcessor<SomeMessage>(
96+
consumerSettings: [handlerBuilder.ConsumerSettings],
97+
messageBus: busMock.Bus,
98+
messageProvider: messageProviderMock.Object,
99+
path: "topic1",
100+
responseProducer: responseProducerMock.Object);
101+
102+
messageProviderMock
103+
.Setup(x => x.Invoke(typeof(SomeRequest), headers, transportMessageMock.Object))
104+
.Throws(deserializationException);
105+
106+
// act
107+
var result = await subject.ProcessMessage(transportMessageMock.Object, headers);
108+
109+
// assert
110+
messageProviderMock.Verify(x => x(typeof(SomeRequest), headers, transportMessageMock.Object), Times.Once);
111+
responseProducerMock.Verify(
112+
x => x.ProduceResponse(
113+
"request-id",
114+
null,
115+
headers,
116+
null,
117+
deserializationException,
118+
handlerBuilder.ConsumerSettings,
119+
It.IsAny<CancellationToken>()),
120+
Times.Once);
121+
122+
result.Should().NotBeNull();
123+
result.Result.Should().Be(ProcessResult.Failure);
124+
result.Exception.Should().BeNull();
125+
}
126+
127+
[Theory]
128+
[InlineData(null, "UnknownMessage")]
129+
[InlineData(true, "UnknownMessage")]
81130
[InlineData(false, "UnknownMessage")]
82131
[InlineData(null, nameof(SomeMessage2))]
83132
[InlineData(true, nameof(SomeMessage2))]
@@ -119,4 +168,4 @@ public async Task When_ProcessMessage_Given_MessageTypeHeaderValueUnknown_Then_R
119168

120169
messageProviderMock.VerifyNoOtherCalls();
121170
}
122-
}
171+
}

0 commit comments

Comments
 (0)