Skip to content

Commit 4aa5203

Browse files
VulthilVulthil
committed
refactor: dedupe outbox type resolution and RabbitMQ wire-message construction
1 parent f340a3b commit 4aa5203

15 files changed

Lines changed: 373 additions & 228 deletions

File tree

‎src/Vulthil.Messaging.Outbox/BrokerOutboxDispatcher.cs‎

Lines changed: 1 addition & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ internal sealed class BrokerOutboxDispatcher(
3030
&& method.IsGenericMethodDefinition
3131
&& method.GetParameters().Length == 3);
3232

33-
private static readonly ConcurrentDictionary<string, Type> TypeCache = [];
3433
private static readonly ConcurrentDictionary<Type, MethodInfo> PublishByType = [];
3534
private static readonly ConcurrentDictionary<Type, MethodInfo> SendByType = [];
3635

@@ -39,7 +38,7 @@ public bool Handles(OutboxDestination destination) =>
3938

4039
public async Task DispatchAsync(OutboxMessageData message, CancellationToken cancellationToken)
4140
{
42-
var messageType = ResolveType(message.Type);
41+
var messageType = OutboxMessageTypeResolver.Resolve(message.Type, "message");
4342
var payload = JsonSerializer.Deserialize(message.Content, messageType, messageConfigurationProvider.JsonSerializerOptions)!;
4443
var metadata = message.Metadata is null
4544
? null
@@ -93,16 +92,4 @@ private static void Apply(BrokerOutboxMetadata? metadata, IPublishContext contex
9392
context.AddHeaders(headers);
9493
}
9594
}
96-
97-
private static Type ResolveType(string typeName) => TypeCache.GetOrAdd(typeName, static name =>
98-
{
99-
var type = Type.GetType(name)
100-
?? AppDomain.CurrentDomain.GetAssemblies()
101-
.Select(assembly => assembly.GetType(name))
102-
.FirstOrDefault(found => found is not null);
103-
104-
return type ?? throw new InvalidOperationException(
105-
$"Unable to resolve the message type '{name}' for an outbox relay. " +
106-
"Ensure the assembly that defines the type is loaded in the relay process.");
107-
});
10895
}

‎src/Vulthil.Messaging.RabbitMq/Consumers/MessageContextFactory.cs‎

Lines changed: 75 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -66,46 +66,20 @@ public static MessageContext<TMessage> CreateContext<TMessage>(
6666
/// </summary>
6767
public static MessageContextSnapshot CreateSnapshot(BasicDeliverEventArgs ea)
6868
{
69-
var context = BuildMetadata(ea);
69+
var metadata = ExtractMetadata(ea);
7070
return new MessageContextSnapshot
7171
{
72-
MessageId = context.MessageId,
73-
RequestId = context.RequestId,
74-
CorrelationId = context.CorrelationId,
75-
ConversationId = context.ConversationId,
76-
InitiatorId = context.InitiatorId,
77-
SourceAddress = context.SourceAddress,
78-
DestinationAddress = context.DestinationAddress,
79-
ResponseAddress = context.ResponseAddress,
80-
FaultAddress = context.FaultAddress,
81-
RoutingKey = context.RoutingKey,
82-
RetryCount = context.RetryCount,
83-
};
84-
}
85-
86-
private static MessageContext BuildMetadata(BasicDeliverEventArgs ea)
87-
{
88-
var props = ea.BasicProperties;
89-
var headers = props.Headers ?? new Dictionary<string, object?>();
90-
var sentTime = GetSentTime(props);
91-
return new MessageContext
92-
{
93-
MessageId = props.MessageId,
94-
CorrelationId = props.CorrelationId,
95-
RequestId = props.CorrelationId,
96-
RoutingKey = ea.RoutingKey,
97-
Headers = AmqpHeaderValueNormalizer.Normalize(headers),
98-
Redelivered = ea.Redelivered,
99-
RetryCount = RabbitMqConstants.GetRetryCount(headers),
100-
ConversationId = RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
101-
InitiatorId = RabbitMqConstants.GetHeaderString(headers, "InitiatorId"),
102-
SourceAddress = RabbitMqConstants.GetHeaderUri(headers, "SourceAddress"),
103-
DestinationAddress = RabbitMqConstants.GetHeaderUri(headers, "DestinationAddress"),
104-
ResponseAddress = RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
105-
?? (string.IsNullOrEmpty(props.ReplyTo) ? null : new Uri($"queue:{props.ReplyTo}")),
106-
FaultAddress = RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
107-
SentTime = sentTime,
108-
ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime)
72+
MessageId = metadata.MessageId,
73+
RequestId = metadata.RequestId,
74+
CorrelationId = metadata.CorrelationId,
75+
ConversationId = metadata.ConversationId,
76+
InitiatorId = metadata.InitiatorId,
77+
SourceAddress = metadata.SourceAddress,
78+
DestinationAddress = metadata.DestinationAddress,
79+
ResponseAddress = metadata.ResponseAddress,
80+
FaultAddress = metadata.FaultAddress,
81+
RoutingKey = metadata.RoutingKey,
82+
RetryCount = metadata.RetryCount,
10983
};
11084
}
11185

@@ -115,35 +89,76 @@ private static MessageContext<TMessage> BuildTypedMetadata<TMessage>(
11589
IPublisher? publisher,
11690
ISendEndpointProvider? sendEndpointProvider,
11791
CancellationToken cancellationToken)
92+
{
93+
var metadata = ExtractMetadata(ea);
94+
return MessageContext.BuildTyped(
95+
message,
96+
publisher,
97+
sendEndpointProvider,
98+
metadata.MessageId,
99+
metadata.CorrelationId,
100+
metadata.RequestId,
101+
metadata.RoutingKey,
102+
metadata.Headers,
103+
metadata.Redelivered,
104+
metadata.RetryCount,
105+
metadata.ConversationId,
106+
metadata.InitiatorId,
107+
metadata.SourceAddress,
108+
metadata.DestinationAddress,
109+
metadata.ResponseAddress,
110+
metadata.FaultAddress,
111+
metadata.SentTime,
112+
metadata.ExpirationTime,
113+
cancellationToken);
114+
}
115+
116+
/// <summary>
117+
/// Extracts the transport metadata common to both the fault snapshot and the bare-JSON typed context from a
118+
/// single AMQP delivery: the property/header paths, the response-address fallback to <c>ReplyTo</c>, and the
119+
/// TTL anchored to the delivery's timestamp.
120+
/// </summary>
121+
private static DeliveryMetadata ExtractMetadata(BasicDeliverEventArgs ea)
118122
{
119123
var props = ea.BasicProperties;
120124
var headers = props.Headers ?? new Dictionary<string, object?>();
121125
var sentTime = GetSentTime(props);
122-
return new MessageContext<TMessage>
123-
{
124-
Message = message,
125-
Publisher = publisher,
126-
SendEndpointProvider = sendEndpointProvider,
127-
CancellationToken = cancellationToken,
128-
MessageId = props.MessageId,
129-
CorrelationId = props.CorrelationId,
130-
RequestId = props.CorrelationId,
131-
RoutingKey = ea.RoutingKey,
132-
Headers = AmqpHeaderValueNormalizer.Normalize(headers),
133-
Redelivered = ea.Redelivered,
134-
RetryCount = RabbitMqConstants.GetRetryCount(headers),
135-
ConversationId = RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
136-
InitiatorId = RabbitMqConstants.GetHeaderString(headers, "InitiatorId"),
137-
SourceAddress = RabbitMqConstants.GetHeaderUri(headers, "SourceAddress"),
138-
DestinationAddress = RabbitMqConstants.GetHeaderUri(headers, "DestinationAddress"),
139-
ResponseAddress = RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
126+
return new DeliveryMetadata(
127+
MessageId: props.MessageId,
128+
CorrelationId: props.CorrelationId,
129+
RequestId: props.CorrelationId,
130+
RoutingKey: ea.RoutingKey,
131+
Headers: AmqpHeaderValueNormalizer.Normalize(headers),
132+
Redelivered: ea.Redelivered,
133+
RetryCount: RabbitMqConstants.GetRetryCount(headers),
134+
ConversationId: RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
135+
InitiatorId: RabbitMqConstants.GetHeaderString(headers, "InitiatorId"),
136+
SourceAddress: RabbitMqConstants.GetHeaderUri(headers, "SourceAddress"),
137+
DestinationAddress: RabbitMqConstants.GetHeaderUri(headers, "DestinationAddress"),
138+
ResponseAddress: RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
140139
?? (string.IsNullOrEmpty(props.ReplyTo) ? null : new Uri($"queue:{props.ReplyTo}")),
141-
FaultAddress = RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
142-
SentTime = sentTime,
143-
ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime)
144-
};
140+
FaultAddress: RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
141+
SentTime: sentTime,
142+
ExpirationTime: RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime));
145143
}
146144

147145
private static DateTimeOffset? GetSentTime(IReadOnlyBasicProperties props)
148146
=> props.Timestamp.UnixTime > 0 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) : null;
147+
148+
private readonly record struct DeliveryMetadata(
149+
string? MessageId,
150+
string? CorrelationId,
151+
string? RequestId,
152+
string RoutingKey,
153+
IReadOnlyDictionary<string, object?> Headers,
154+
bool Redelivered,
155+
int RetryCount,
156+
string? ConversationId,
157+
string? InitiatorId,
158+
Uri? SourceAddress,
159+
Uri? DestinationAddress,
160+
Uri? ResponseAddress,
161+
Uri? FaultAddress,
162+
DateTimeOffset? SentTime,
163+
DateTimeOffset? ExpirationTime);
149164
}

‎src/Vulthil.Messaging.RabbitMq/Publishing/IInternalPublisher.cs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,10 @@ namespace Vulthil.Messaging.RabbitMq.Publishing;
44

55
internal interface IInternalPublisher
66
{
7-
Task InternalPublishAsync<TMessage>(
7+
/// <summary>
8+
/// Publishes the already-serialized message body over the message's fanout/topic exchange.
9+
/// </summary>
10+
Task InternalPublishAsync(
811
byte[] body,
912
BasicProperties props,
1013
string routingKey,

‎src/Vulthil.Messaging.RabbitMq/Publishing/RabbitMqPublisher.cs‎

Lines changed: 13 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
using System.Collections.Concurrent;
22
using System.Diagnostics;
3-
using System.Text.Json;
43
using Microsoft.Extensions.Logging;
54
using RabbitMQ.Client;
65
using Vulthil.Messaging.Abstractions.Publishers;
@@ -35,7 +34,7 @@ public RabbitMqPublisher(
3534
/// Channels come from a bounded pool — each leased channel is used non-concurrently and returned for reuse, or
3635
/// discarded if it faults.
3736
/// </remarks>
38-
public async Task InternalPublishAsync<TMessage>(
37+
public async Task InternalPublishAsync(
3938
byte[] body,
4039
BasicProperties props,
4140
string routingKey,
@@ -102,51 +101,25 @@ public async Task PublishAsync<TMessage>(
102101
?? messageConfiguration.RoutingKeyFormatter?.Invoke(message)
103102
?? string.Empty;
104103

105-
var correlationId = publishContext.CorrelationId
106-
?? messageConfiguration.CorrelationIdFormatter?.Invoke(message)
107-
?? Guid.CreateVersion7().ToString();
108-
109-
var messageId = publishContext.MessageId ?? Guid.CreateVersion7().ToString();
104+
var ids = RabbitMqWireMessageBuilder.ResolveIds(message, publishContext, messageConfiguration);
110105
var exchange = messageConfiguration.Exchange;
111-
var urn = messageConfiguration.Urn;
112-
var urnString = urn.AbsoluteUri;
113106

114-
using var activity = MessagingInstrumentation.ActivitySource.StartActivity(
115-
$"{exchange} publish",
116-
ActivityKind.Producer);
107+
using var activity = RabbitMqWireMessageBuilder.StartProducerActivity(
108+
$"{exchange} publish", "publish", exchange, routingKey, ids.UrnString, ids.MessageId, ids.CorrelationId);
117109

118-
if (activity is not null)
119-
{
120-
activity.SetTag(MessagingInstrumentation.Tags.MessagingSystem, MessagingInstrumentation.SystemValue);
121-
activity.SetTag(MessagingInstrumentation.Tags.MessagingOperation, "publish");
122-
activity.SetTag(MessagingInstrumentation.Tags.MessagingDestination, exchange);
123-
activity.SetTag(MessagingInstrumentation.Tags.MessagingRoutingKey, routingKey);
124-
activity.SetTag(MessagingInstrumentation.Tags.MessageType, urnString);
125-
activity.SetTag(MessagingInstrumentation.Tags.MessagingMessageId, messageId);
126-
activity.SetTag(MessagingInstrumentation.Tags.MessagingCorrelationId, correlationId);
127-
}
110+
var properties = RabbitMqWireMessageBuilder.CreateBaseProperties(ids.UrnString, ids.MessageId, publishContext.Headers);
111+
properties.ReplyTo = RabbitMqAddress.ResolveRoutingKey(publishContext.ResponseAddress);
112+
properties.CorrelationId = ids.CorrelationId;
113+
properties.Persistent = true;
128114

129-
var properties = new BasicProperties()
130-
{
131-
Type = urnString,
132-
MessageId = messageId,
133-
ReplyTo = RabbitMqAddress.ResolveRoutingKey(publishContext.ResponseAddress),
134-
CorrelationId = correlationId,
135-
ContentType = RabbitMqConstants.ContentType,
136-
Headers = new Dictionary<string, object?>(publishContext.Headers),
137-
Persistent = true,
138-
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()),
139-
};
140-
141-
var envelope = MessageEnvelopeFactory.Create(
142-
message, publishContext, messageId, correlationId, urn, _messageConfigurationProvider.JsonSerializerOptions);
143-
var body = JsonSerializer.SerializeToUtf8Bytes(envelope, _messageConfigurationProvider.JsonSerializerOptions);
144-
145-
MessagingLog.Publishing(_logger, urnString, exchange, routingKey, messageId);
115+
var body = RabbitMqWireMessageBuilder.SerializeEnvelope(
116+
message, publishContext, ids.MessageId, ids.CorrelationId, ids.Urn, _messageConfigurationProvider.JsonSerializerOptions);
117+
118+
MessagingLog.Publishing(_logger, ids.UrnString, exchange, routingKey, ids.MessageId);
146119

147120
try
148121
{
149-
await InternalPublishAsync<TMessage>(body, properties, routingKey, messageConfiguration, cancellationToken);
122+
await InternalPublishAsync(body, properties, routingKey, messageConfiguration, cancellationToken);
150123
activity?.SetStatus(ActivityStatusCode.Ok);
151124
}
152125
catch (Exception ex)

‎src/Vulthil.Messaging.RabbitMq/RabbitMqWireMessageBuilder.cs‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
using System.Diagnostics;
2+
using System.Text.Json;
3+
using RabbitMQ.Client;
4+
using Vulthil.Messaging.RabbitMq.Telemetry;
5+
using Vulthil.Messaging.Transport;
6+
7+
namespace Vulthil.Messaging.RabbitMq;
8+
9+
/// <summary>
10+
/// Shared wire-message construction for the RabbitMQ producer paths (<c>RabbitMqPublisher</c>,
11+
/// <c>RabbitMqSendEndpoint</c>, <c>RabbitMqRequester</c>): resolving the correlation/message identifiers, the common
12+
/// <see cref="BasicProperties"/> fields, the serialized <see cref="MessageEnvelope"/>, and the producer activity
13+
/// tags. Each producer applies its own routing/exchange selection and the <see cref="BasicProperties"/> fields that
14+
/// legitimately differ per operation (reply-to resolution, the wire correlation id, persistence, expiration) on top
15+
/// of the shared result, so the extraction covers only the parts the three sites compute identically.
16+
/// </summary>
17+
internal static class RabbitMqWireMessageBuilder
18+
{
19+
/// <summary>The identifiers resolved for a single outgoing message, shared by every producer path.</summary>
20+
/// <param name="CorrelationId">The resolved business correlation identifier.</param>
21+
/// <param name="MessageId">The resolved message identifier.</param>
22+
/// <param name="Urn">The stable wire URN for the message type.</param>
23+
/// <param name="UrnString">The URN's absolute URI string, used as the AMQP <c>Type</c> and activity tag.</param>
24+
internal readonly record struct ResolvedIds(string CorrelationId, string MessageId, Uri Urn, string UrnString);
25+
26+
/// <summary>
27+
/// Resolves the correlation id, message id, and URN for <paramref name="message"/> the same way across every
28+
/// producer path: an explicit value on <paramref name="context"/> wins, then the type's configured formatter
29+
/// (correlation id only), then a fresh id.
30+
/// </summary>
31+
public static ResolvedIds ResolveIds<TMessage>(TMessage message, PublishContext context, MessageConfiguration messageConfiguration)
32+
where TMessage : notnull
33+
{
34+
var correlationId = context.CorrelationId
35+
?? messageConfiguration.CorrelationIdFormatter?.Invoke(message)
36+
?? Guid.CreateVersion7().ToString();
37+
38+
var messageId = context.MessageId ?? Guid.CreateVersion7().ToString();
39+
var urn = messageConfiguration.Urn;
40+
41+
return new ResolvedIds(correlationId, messageId, urn, urn.AbsoluteUri);
42+
}
43+
44+
/// <summary>
45+
/// Builds and serializes the <see cref="MessageEnvelope"/> for the outgoing message. Callers that convert a
46+
/// publish failure into a typed result (rather than letting it propagate) must call this from within their own
47+
/// try/catch, since serialization can fail for the same reasons the send itself can.
48+
/// </summary>
49+
public static byte[] SerializeEnvelope<TMessage>(
50+
TMessage message,
51+
PublishContext context,
52+
string messageId,
53+
string correlationId,
54+
Uri urn,
55+
JsonSerializerOptions jsonOptions,
56+
string? requestId = null)
57+
where TMessage : notnull
58+
{
59+
var envelope = MessageEnvelopeFactory.Create(message, context, messageId, correlationId, urn, jsonOptions, requestId);
60+
return JsonSerializer.SerializeToUtf8Bytes(envelope, jsonOptions);
61+
}
62+
63+
/// <summary>
64+
/// Creates the <see cref="BasicProperties"/> fields common to every RabbitMQ producer path. Callers set the
65+
/// remaining fields that legitimately differ per operation (<c>ReplyTo</c>, the wire <c>CorrelationId</c>,
66+
/// <c>Persistent</c>, <c>Expiration</c>).
67+
/// </summary>
68+
public static BasicProperties CreateBaseProperties(string urnString, string messageId, IReadOnlyDictionary<string, object?> headers) => new()
69+
{
70+
Type = urnString,
71+
MessageId = messageId,
72+
ContentType = RabbitMqConstants.ContentType,
73+
Headers = new Dictionary<string, object?>(headers),
74+
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()),
75+
};
76+
77+
/// <summary>
78+
/// Starts a producer <see cref="Activity"/> and applies the standard Vulthil messaging tags. Returns
79+
/// <see langword="null"/> when no listener is recording the transport's activity source.
80+
/// </summary>
81+
public static Activity? StartProducerActivity(
82+
string name, string operation, string destination, string routingKey, string urnString, string messageId, string correlationId)
83+
{
84+
var activity = MessagingInstrumentation.ActivitySource.StartActivity(name, ActivityKind.Producer);
85+
if (activity is not null)
86+
{
87+
activity.SetTag(MessagingInstrumentation.Tags.MessagingSystem, MessagingInstrumentation.SystemValue);
88+
activity.SetTag(MessagingInstrumentation.Tags.MessagingOperation, operation);
89+
activity.SetTag(MessagingInstrumentation.Tags.MessagingDestination, destination);
90+
activity.SetTag(MessagingInstrumentation.Tags.MessagingRoutingKey, routingKey);
91+
activity.SetTag(MessagingInstrumentation.Tags.MessageType, urnString);
92+
activity.SetTag(MessagingInstrumentation.Tags.MessagingMessageId, messageId);
93+
activity.SetTag(MessagingInstrumentation.Tags.MessagingCorrelationId, correlationId);
94+
}
95+
96+
return activity;
97+
}
98+
}

0 commit comments

Comments
 (0)