Skip to content

Commit 3b2a545

Browse files
VulthilVulthil
authored
Merge pull request #348 from Vulthil/refactor/messaging-internals-dedup
Dedupe outbox type resolution and RabbitMQ wire-message construction
2 parents c5ec1dc + de3d8f9 commit 3b2a545

9 files changed

Lines changed: 263 additions & 180 deletions

File tree

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

Lines changed: 74 additions & 57 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

@@ -116,34 +90,77 @@ private static MessageContext<TMessage> BuildTypedMetadata<TMessage>(
11690
ISendEndpointProvider? sendEndpointProvider,
11791
CancellationToken cancellationToken)
11892
{
119-
var props = ea.BasicProperties;
120-
var headers = props.Headers ?? new Dictionary<string, object?>();
121-
var sentTime = GetSentTime(props);
93+
var metadata = ExtractMetadata(ea);
12294
return new MessageContext<TMessage>
12395
{
12496
Message = message,
12597
Publisher = publisher,
12698
SendEndpointProvider = sendEndpointProvider,
12799
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")
140-
?? (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)
100+
MessageId = metadata.MessageId,
101+
CorrelationId = metadata.CorrelationId,
102+
RequestId = metadata.RequestId,
103+
RoutingKey = metadata.RoutingKey,
104+
Headers = metadata.Headers,
105+
Redelivered = metadata.Redelivered,
106+
RetryCount = metadata.RetryCount,
107+
ConversationId = metadata.ConversationId,
108+
InitiatorId = metadata.InitiatorId,
109+
SourceAddress = metadata.SourceAddress,
110+
DestinationAddress = metadata.DestinationAddress,
111+
ResponseAddress = metadata.ResponseAddress,
112+
FaultAddress = metadata.FaultAddress,
113+
SentTime = metadata.SentTime,
114+
ExpirationTime = metadata.ExpirationTime,
144115
};
145116
}
146117

118+
/// <summary>
119+
/// Extracts the transport metadata common to both the fault snapshot and the bare-JSON typed context from a
120+
/// single AMQP delivery: the property/header paths, the response-address fallback to <c>ReplyTo</c>, and the
121+
/// TTL anchored to the delivery's timestamp.
122+
/// </summary>
123+
private static DeliveryMetadata ExtractMetadata(BasicDeliverEventArgs ea)
124+
{
125+
var props = ea.BasicProperties;
126+
var headers = props.Headers ?? new Dictionary<string, object?>();
127+
var sentTime = GetSentTime(props);
128+
return new DeliveryMetadata(
129+
MessageId: props.MessageId,
130+
CorrelationId: props.CorrelationId,
131+
RequestId: props.CorrelationId,
132+
RoutingKey: ea.RoutingKey,
133+
Headers: AmqpHeaderValueNormalizer.Normalize(headers),
134+
Redelivered: ea.Redelivered,
135+
RetryCount: RabbitMqConstants.GetRetryCount(headers),
136+
ConversationId: RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
137+
InitiatorId: RabbitMqConstants.GetHeaderString(headers, "InitiatorId"),
138+
SourceAddress: RabbitMqConstants.GetHeaderUri(headers, "SourceAddress"),
139+
DestinationAddress: RabbitMqConstants.GetHeaderUri(headers, "DestinationAddress"),
140+
ResponseAddress: RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
141+
?? (string.IsNullOrEmpty(props.ReplyTo) ? null : new Uri($"queue:{props.ReplyTo}")),
142+
FaultAddress: RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
143+
SentTime: sentTime,
144+
ExpirationTime: RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime));
145+
}
146+
147147
private static DateTimeOffset? GetSentTime(IReadOnlyBasicProperties props)
148148
=> props.Timestamp.UnixTime > 0 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) : null;
149+
150+
private readonly record struct DeliveryMetadata(
151+
string? MessageId,
152+
string? CorrelationId,
153+
string? RequestId,
154+
string RoutingKey,
155+
IReadOnlyDictionary<string, object?> Headers,
156+
bool Redelivered,
157+
int RetryCount,
158+
string? ConversationId,
159+
string? InitiatorId,
160+
Uri? SourceAddress,
161+
Uri? DestinationAddress,
162+
Uri? ResponseAddress,
163+
Uri? FaultAddress,
164+
DateTimeOffset? SentTime,
165+
DateTimeOffset? ExpirationTime);
149166
}

‎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)