Skip to content

Commit 261e095

Browse files
fmattiolifmattioli
authored
Merge pull request #8 from migueltotti/main
FEAT: Add ConsumerMiddleware
2 parents db21d7a + 3ca1050 commit 261e095

10 files changed

Lines changed: 126 additions & 23 deletions

‎src/AzureServiceBusFlow.Sample/Middlewares/AsbSampleConsumerMiddleware.cs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
using Azure.Messaging.ServiceBus;
2+
using AzureServiceBusFlow.Middlewares;
3+
4+
namespace AzureServiceBusFlow.Sample.Middlewares
5+
{
6+
/// <summary>
7+
/// Sample middleware that log the ApplicationProperties
8+
/// before the message reach the MessageHandler
9+
/// </summary>
10+
public class AsbSampleConsumerMiddleware(ILogger<AsbSampleConsumerMiddleware> logger) : IConsumerMiddleware
11+
{
12+
public Task InvokeAsync(ServiceBusReceivedMessage message, Func<Task> next, CancellationToken cancellationToken)
13+
{
14+
var sampleMiddlewareStatus = message.ApplicationProperties["SampleMiddleware"];
15+
16+
logger.LogInformation("Message reached ConsumerMiddleware === Status: {Status}", sampleMiddlewareStatus);
17+
18+
return next();
19+
}
20+
}
21+
}

src/AzureServiceBusFlow.Sample/Middlewares/AsbSampleMiddleware.cs renamed to src/AzureServiceBusFlow.Sample/Middlewares/AsbSampleProducerMiddleware.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ namespace AzureServiceBusFlow.Sample.Middlewares
77
/// Sample middleware that adds a custom property
88
/// to every message published to Azure Service Bus.
99
/// </summary>
10-
public class AsbSampleMiddleware : IProducerMiddleware
10+
public class AsbSampleProducerMiddleware : IProducerMiddleware
1111
{
1212
public async Task InvokeAsync(ServiceBusMessage message, Func<Task> next)
1313
{

‎src/AzureServiceBusFlow.Sample/Program.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,8 @@
2424

2525
builder.Services.AddAzureServiceBus(cfg => cfg
2626
.ConfigureAzureServiceBus(azureServiceBusConfig)
27-
.UseGlobalProducerMiddleware<AsbSampleMiddleware>()
27+
.UseGlobalProducerMiddleware<AsbSampleProducerMiddleware>()
28+
.UseGlobalConsumerMiddleware<AsbSampleConsumerMiddleware>()
2829
.AddProducer<ExampleCommand1>(p => p
2930
.EnsureQueueExists("command-queue-one")
3031
.WithCommandProducer()

‎src/AzureServiceBusFlow/Abstractions/IServiceBusMessage.cs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22
{
33
public interface IServiceBusMessage
44
{
5-
string RoutingKey { get; }
65
DateTime CreatedDate { get; }
76
}
87
}

‎src/AzureServiceBusFlow/Builders/ServiceBusConfigurationBuilder.cs‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ public ServiceBusConfigurationBuilder ConfigureAzureServiceBus(AzureServiceBusCo
3232
public ServiceBusConfigurationBuilder UseGlobalProducerMiddleware<TMiddleware>()
3333
where TMiddleware : class, IProducerMiddleware
3434
{
35-
// Só registra no DI — já torna global
35+
// Register in DI to make it globally
3636
if (!_services.Any(s =>
3737
s.ServiceType == typeof(IProducerMiddleware) &&
3838
s.ImplementationType == typeof(TMiddleware)))
@@ -43,6 +43,20 @@ public ServiceBusConfigurationBuilder UseGlobalProducerMiddleware<TMiddleware>()
4343
return this;
4444
}
4545

46+
public ServiceBusConfigurationBuilder UseGlobalConsumerMiddleware<TMiddleware>()
47+
where TMiddleware : class, IConsumerMiddleware
48+
{
49+
// Register in DI to make use it globally
50+
if (!_services.Any(s =>
51+
s.ServiceType == typeof(IConsumerMiddleware) &&
52+
s.ImplementationType == typeof(TMiddleware)))
53+
{
54+
_services.AddSingleton<IConsumerMiddleware, TMiddleware>();
55+
}
56+
57+
return this;
58+
}
59+
4660
/// <summary>
4761
/// Adds a producer for messages of type <typeparamref name="TMessage"/> configured via a callback.
4862
/// </summary>

‎src/AzureServiceBusFlow/Builders/ServiceBusConsumerConfigurationBuilder.cs‎

Lines changed: 59 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
using Azure.Messaging.ServiceBus;
22
using AzureServiceBusFlow.Abstractions;
33
using AzureServiceBusFlow.Hosts;
4+
using AzureServiceBusFlow.Middlewares;
45
using AzureServiceBusFlow.Models;
56
using Microsoft.Azure.ServiceBus.Management;
67
using Microsoft.Extensions.DependencyInjection;
@@ -19,6 +20,19 @@ public class ServiceBusConsumerConfigurationBuilder(AzureServiceBusConfiguration
1920
private string? _queueName;
2021
private string? _topicName;
2122
private string? _subscriptionName;
23+
private readonly List<Type> _middlewares = [];
24+
private readonly string _consumerMiddlewareKey = Guid.NewGuid().ToString();
25+
26+
public ServiceBusConsumerConfigurationBuilder UseMiddleware<TMiddleware>()
27+
where TMiddleware : IConsumerMiddleware
28+
{
29+
if (!_middlewares.Contains(typeof(TMiddleware)))
30+
{
31+
_middlewares.Add(typeof(TMiddleware));
32+
}
33+
34+
return this;
35+
}
2236

2337
public ServiceBusConsumerConfigurationBuilder FromQueue(string queueName)
2438
{
@@ -76,15 +90,28 @@ public void Build()
7690
throw new InvalidOperationException("Missing queue or topic configuration!");
7791
}
7892

93+
foreach (var middlewareType in from middlewareType in _middlewares
94+
where !_services.Any(s =>
95+
s.ServiceType == typeof(IConsumerMiddleware) &&
96+
s.ImplementationType == middlewareType)
97+
select middlewareType)
98+
{
99+
_services.AddKeyedSingleton(typeof(IConsumerMiddleware), _consumerMiddlewareKey, middlewareType);
100+
}
101+
79102
_services.AddSingleton<IHostedService>(sp =>
80103
{
81104
var logger = sp.GetRequiredService<ILogger<ServiceBusConsumerHostedService>>();
105+
var localConsumerMiddlewares = sp.GetKeyedServices<IConsumerMiddleware>(_consumerMiddlewareKey) ?? [];
106+
var globalConsumerMiddlewares = sp.GetServices<IConsumerMiddleware>() ?? [];
107+
108+
var consumerMiddlewares = globalConsumerMiddlewares.Union(localConsumerMiddlewares);
82109

83110
if (!string.IsNullOrWhiteSpace(_queueName))
84111
{
85112
return new ServiceBusConsumerHostedService(
86113
(rawMessage, rootProvider, cancellationToken) =>
87-
MessageConsumingHandler(rawMessage, rootProvider, logger, cancellationToken),
114+
MessageConsumingHandler(rawMessage, rootProvider, consumerMiddlewares, logger, cancellationToken),
88115
sp,
89116
logger,
90117
_azureServiceBusConfiguration,
@@ -93,7 +120,7 @@ public void Build()
93120

94121
return new ServiceBusConsumerHostedService(
95122
(rawMessage, rootProvider, cancellationToken) =>
96-
MessageConsumingHandler(rawMessage, rootProvider, logger, cancellationToken),
123+
MessageConsumingHandler(rawMessage, rootProvider, consumerMiddlewares, logger, cancellationToken),
97124
sp,
98125
logger,
99126
_azureServiceBusConfiguration,
@@ -102,7 +129,34 @@ public void Build()
102129
});
103130
}
104131

105-
private async Task MessageConsumingHandler(ServiceBusReceivedMessage rawMessage, IServiceProvider rootProvider, ILogger<ServiceBusConsumerHostedService> logger, CancellationToken cancellationToken)
132+
private async Task MessageConsumingHandler(ServiceBusReceivedMessage rawMessage, IServiceProvider rootProvider, IEnumerable<IConsumerMiddleware> middlewares, ILogger<ServiceBusConsumerHostedService> logger, CancellationToken cancellationToken)
133+
{
134+
Func<Task> finalStep = async () =>
135+
{
136+
await ProcessHandlersAsync(rawMessage, rootProvider, logger, cancellationToken);
137+
};
138+
139+
if (middlewares != null && middlewares.Any())
140+
{
141+
Func<Task> next = finalStep;
142+
143+
foreach (var middleware in middlewares.Reverse())
144+
{
145+
var current = middleware;
146+
var prevNext = next;
147+
148+
next = () => current.InvokeAsync(rawMessage, prevNext, cancellationToken);
149+
}
150+
151+
await next();
152+
}
153+
else
154+
{
155+
await finalStep();
156+
}
157+
}
158+
159+
private async Task ProcessHandlersAsync(ServiceBusReceivedMessage rawMessage, IServiceProvider rootProvider, ILogger<ServiceBusConsumerHostedService> logger, CancellationToken cancellationToken)
106160
{
107161
if (!rawMessage.ApplicationProperties.TryGetValue("MessageType", out var messageTypeNameObj))
108162
{
@@ -169,9 +223,9 @@ private async Task MessageConsumingHandler(ServiceBusReceivedMessage rawMessage,
169223
var elapsed = DateTime.UtcNow - startTime;
170224

171225
logger.LogInformation(
172-
"Message {MessageType} with RoutingKey {RoutingKey} consumed and handled by {HandlerName} at {StartTime} in {ElapsedMilliseconds} ms",
226+
"Message {MessageType} with CorrelationId {CorrelationId} consumed and handled by {HandlerName} at {StartTime} in {ElapsedMilliseconds} ms",
173227
messageTypeName,
174-
rawMessage.Subject,
228+
rawMessage.CorrelationId,
175229
handlerType.Name,
176230
startTime.ToString("o"),
177231
elapsed.TotalMilliseconds

‎src/AzureServiceBusFlow/Builders/ServiceBusProducerConfigurationBuilder.cs‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ public class ServiceBusProducerConfigurationBuilder<TMessage>(AzureServiceBusCon
1717
private string? _topicName;
1818
private string? _queueName;
1919
private readonly List<Type> _middlewares = [];
20+
private readonly string _producerMiddlewareKey = Guid.NewGuid().ToString();
2021

2122
public ServiceBusProducerConfigurationBuilder<TMessage> UseMiddleware<TMiddleware>()
2223
where TMiddleware : IProducerMiddleware
@@ -104,13 +105,18 @@ internal void Build()
104105
s.ImplementationType == middlewareType)
105106
select middlewareType)
106107
{
107-
_services.AddSingleton(typeof(IProducerMiddleware), middlewareType);
108+
_services.AddKeyedSingleton(typeof(IProducerMiddleware), _producerMiddlewareKey, middlewareType);
108109
}
109110

110111
_services.AddSingleton<IServiceBusProducer<TMessage>>(sp =>
111112
{
112113
var logger = sp.GetRequiredService<ILogger<ServiceBusProducer<TMessage>>>();
113-
var middlewares = sp.GetServices<IProducerMiddleware>();
114+
115+
var localProducerMiddlewares = sp.GetKeyedServices<IProducerMiddleware>(_producerMiddlewareKey) ?? [];
116+
var globalProducerMiddlewares = sp.GetServices<IProducerMiddleware>() ?? [];
117+
118+
var middlewares = globalProducerMiddlewares.Union(localProducerMiddlewares);
119+
114120
var name = _queueName ?? _topicName!;
115121
return new ServiceBusProducer<TMessage>(
116122
_azureServiceBusConfiguration,

‎src/AzureServiceBusFlow/Hosts/ServiceBusConsumerHostedService.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -70,19 +70,19 @@ await _retryPolicy.ExecuteAsync(async _ =>
7070
}
7171
catch (Exception ex)
7272
{
73-
logger.LogError(ex, "Error while trying process MessageType {MessageType} with RoutingKey {RoutingKey} with id {MessageId} - MessageBody {Body}",
73+
logger.LogError(ex, "Error while trying process MessageType {MessageType} with CorrelationId {CorrelationId} with id {MessageId} - MessageBody {Body}",
7474
message.GetType().Name,
75-
message.Subject,
75+
message.CorrelationId,
7676
message.MessageId,
7777
message.Body);
7878

7979
if (_processor.ReceiveMode == ServiceBusReceiveMode.PeekLock)
8080
{
8181
await args.AbandonMessageAsync(message, cancellationToken: args.CancellationToken);
8282

83-
logger.LogWarning("Message {MessageType} with RoutingKey {RoutingKey} with id {MessageId} abandoned. Will retry again.",
83+
logger.LogWarning("Message {MessageType} with CorrelationId {CorrelationId} with id {MessageId} abandoned. Will retry again.",
8484
message.GetType().Name,
85-
message.Subject,
85+
message.CorrelationId,
8686
message.MessageId);
8787
}
8888
}

‎src/AzureServiceBusFlow/Middlewares/IConsumerMiddleware.cs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
using Azure.Messaging.ServiceBus;
2+
3+
namespace AzureServiceBusFlow.Middlewares
4+
{
5+
public interface IConsumerMiddleware
6+
{
7+
Task InvokeAsync(ServiceBusReceivedMessage message, Func<Task> next, CancellationToken cancellationToken);
8+
}
9+
10+
}

‎src/AzureServiceBusFlow/Producers/ServiceBusProducer.cs‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -31,22 +31,21 @@ public async Task ProduceAsync(TMessage message, CancellationToken cancellationT
3131
var json = JsonConvert.SerializeObject(message);
3232
var serviceBusMessage = new ServiceBusMessage(json)
3333
{
34-
Subject = message.RoutingKey,
34+
Subject = message.GetType().Name,
3535
ApplicationProperties =
3636
{
3737
{ "MessageType", message.GetType().Name },
3838
{ "CreatedAt", message.CreatedDate.ToString("O") },
39-
{ "RoutingKey", (message as IServiceBusMessage)?.RoutingKey },
4039
}
4140
};
4241

4342
async Task finalStep()
4443
{
4544
await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
4645

47-
_logger.LogInformation("Message {MessageType} with RoutingKey {MessageId} published successfully!",
46+
_logger.LogInformation("Message {MessageType} with CorrelationId {CorrelationId} published successfully!",
4847
message.GetType().Name,
49-
message.RoutingKey);
48+
serviceBusMessage.CorrelationId);
5049
}
5150

5251
// Run middlewares, if it exist
@@ -72,12 +71,11 @@ public async Task ProduceAsync(TMessage message, MessageOptions producerOptions,
7271
var json = JsonConvert.SerializeObject(message);
7372
var serviceBusMessage = new ServiceBusMessage(json)
7473
{
75-
Subject = (message as IServiceBusMessage)?.RoutingKey ?? message.GetType().Name,
74+
Subject = message.GetType().Name,
7675
ApplicationProperties =
7776
{
7877
{ "MessageType", message.GetType().Name },
7978
{ "CreatedAt", (message as IServiceBusMessage)?.CreatedDate.ToString("O") ?? DateTime.UtcNow.ToString("O") },
80-
{ "RoutingKey", (message as IServiceBusMessage)?.RoutingKey }
8179
}
8280
};
8381

@@ -100,9 +98,9 @@ async Task finalStep()
10098
{
10199
await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
102100

103-
_logger.LogInformation("Message {MessageType} with RoutingKey {MessageId} published successfully!",
101+
_logger.LogInformation("Message {MessageType} with CorrelationId {CorrelationId} published successfully!",
104102
message.GetType().Name,
105-
message.RoutingKey);
103+
serviceBusMessage.CorrelationId);
106104
}
107105

108106
if (_middlewares != null && _middlewares.Any())

0 commit comments

Comments
 (0)