Skip to content

Commit a76da88

Browse files
bordingbording
committed
Move shared code into helper class
1 parent 6fa0582 commit a76da88

4 files changed

Lines changed: 145 additions & 259 deletions

File tree

‎src/NServiceBus.Transport.RabbitMQ.CommandLine.Tests/MigrateDelayedMessages/When_migrating_a_delayed_message.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,14 @@ public void Should_generate_the_correct_routing_key(int originalDelayInSeconds,
1919
var originalTimeSent = GetDateTimeOffsetFromValues(originalTimeSentValues);
2020
var utcNow = GetDateTimeOffsetFromValues(utcNowValues);
2121

22-
var (destinationQueue, newRoutingKey, newDelayLevel) = DelaysMigrateCommand.GetNewRoutingKey(originalDelayInSeconds, originalTimeSent, originalRoutingKey, utcNow);
22+
var (destinationQueue, newRoutingKey, newDelayLevel) = DelayCommandHelpers.GetNewRoutingKey(originalDelayInSeconds, originalTimeSent, originalRoutingKey, utcNow);
2323

24-
Assert.Multiple(() =>
24+
using (Assert.EnterMultipleScope())
2525
{
2626
Assert.That(destinationQueue, Is.EqualTo(expectedDestination));
2727
Assert.That(newRoutingKey, Is.EqualTo(expectedRoutingKey));
2828
Assert.That(newDelayLevel, Is.EqualTo(expectedDelayLevel));
29-
});
29+
}
3030
}
3131

3232
DateTimeOffset GetDateTimeOffsetFromValues(int[] values)

‎src/NServiceBus.Transport.RabbitMQ.CommandLine/Commands/Delays/DelayCommandHelpers.cs‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
namespace NServiceBus.Transport.RabbitMQ.CommandLine;
2+
3+
using System.Globalization;
4+
using System.Text;
5+
using global::RabbitMQ.Client;
6+
7+
static class DelayCommandHelpers
8+
{
9+
const string timeSentHeader = "NServiceBus.TimeSent";
10+
const string dateTimeOffsetWireFormat = "yyyy-MM-dd HH:mm:ss:ffffff Z";
11+
const int indexStartOfDestinationQueue = DelayInfrastructure.MaxNumberOfBitsToUse * 2;
12+
13+
static bool poisonQueueCreated = false;
14+
static readonly HashSet<string> declaredDestinationQueues = [];
15+
16+
public static async Task TransferMessages(bool sourceIsV1, IChannel sourceChannel, IChannel destinationChannel, string poisonMessageQueue, IRoutingTopology routingTopology, TextWriter output, CancellationToken cancellationToken = default)
17+
{
18+
for (int currentDelayLevel = DelayInfrastructure.MaxLevel; currentDelayLevel >= 0 && !cancellationToken.IsCancellationRequested; currentDelayLevel--)
19+
{
20+
await TransferQueue(sourceIsV1, sourceChannel, destinationChannel, currentDelayLevel, poisonMessageQueue, routingTopology, output, cancellationToken);
21+
}
22+
}
23+
24+
static async Task TransferQueue(bool sourceIsV1, IChannel sourceChannel, IChannel destinationChannel, int delayLevel, string poisonMessageQueue, IRoutingTopology routingTopology, TextWriter output, CancellationToken cancellationToken)
25+
{
26+
var currentDelayQueue = sourceIsV1 ? $"nsb.delay-level-{delayLevel:00}" : DelayInfrastructure.LevelName(delayLevel);
27+
var messageCount = await sourceChannel.MessageCountAsync(currentDelayQueue, cancellationToken);
28+
29+
if (messageCount > 0)
30+
{
31+
output.Write($"Processing {messageCount} messages at delay level {delayLevel:00}. ");
32+
33+
int skippedMessages = 0;
34+
int processedMessages = 0;
35+
36+
for (int i = 0; i < messageCount && !cancellationToken.IsCancellationRequested; i++)
37+
{
38+
var message = await sourceChannel.BasicGetAsync(currentDelayQueue, false, cancellationToken);
39+
40+
if (message is null)
41+
{
42+
// Queue is empty
43+
break;
44+
}
45+
46+
if (MessageIsInvalid(message))
47+
{
48+
skippedMessages++;
49+
50+
if (!poisonQueueCreated)
51+
{
52+
await sourceChannel.QueueDeclareAsync(poisonMessageQueue, true, false, false, cancellationToken: cancellationToken);
53+
poisonQueueCreated = true;
54+
}
55+
56+
await sourceChannel.BasicPublishAsync(string.Empty, poisonMessageQueue, false, new BasicProperties(message.BasicProperties), message.Body, cancellationToken: cancellationToken);
57+
await sourceChannel.BasicAckAsync(message.DeliveryTag, false, cancellationToken);
58+
59+
continue;
60+
}
61+
62+
var messageHeaders = message.BasicProperties.Headers;
63+
64+
var delayInSeconds = 0;
65+
66+
if (messageHeaders is not null)
67+
{
68+
var headerValue = messageHeaders[DelayInfrastructure.DelayHeader];
69+
70+
if (headerValue is not null)
71+
{
72+
delayInSeconds = (int)headerValue;
73+
}
74+
}
75+
76+
var timeSent = GetTimeSent(message);
77+
78+
var (destinationQueue, newRoutingKey, newDelayLevel) = GetNewRoutingKey(delayInSeconds, timeSent, message.RoutingKey, DateTimeOffset.UtcNow);
79+
80+
// Make sure the destination queue is bound to the delivery exchange to ensure delivery
81+
if (!declaredDestinationQueues.Contains(destinationQueue))
82+
{
83+
await routingTopology.BindToDelayInfrastructure(destinationChannel, destinationQueue, DelayInfrastructure.DeliveryExchange, DelayInfrastructure.BindingKey(destinationQueue), cancellationToken);
84+
declaredDestinationQueues.Add(destinationQueue);
85+
}
86+
87+
var publishExchange = DelayInfrastructure.LevelName(newDelayLevel);
88+
89+
if (messageHeaders is not null)
90+
{
91+
//These headers need to be removed so that they won't be copied to an outgoing message if this message gets forwarded
92+
messageHeaders.Remove(DelayInfrastructure.XDeathHeader);
93+
messageHeaders.Remove(DelayInfrastructure.XFirstDeathExchangeHeader);
94+
messageHeaders.Remove(DelayInfrastructure.XFirstDeathQueueHeader);
95+
messageHeaders.Remove(DelayInfrastructure.XFirstDeathReasonHeader);
96+
}
97+
98+
await destinationChannel.BasicPublishAsync(publishExchange, newRoutingKey, false, new BasicProperties(message.BasicProperties), message.Body, cancellationToken: cancellationToken);
99+
await sourceChannel.BasicAckAsync(message.DeliveryTag, false, cancellationToken);
100+
processedMessages++;
101+
}
102+
103+
output.WriteLine($"{processedMessages} successful, {skippedMessages} skipped.");
104+
}
105+
else
106+
{
107+
output.WriteLine($"No messages to process at delay level {delayLevel:00}.");
108+
}
109+
}
110+
111+
public static (string DestinationQueue, string NewRoutingKey, int NewDelayLevel) GetNewRoutingKey(int delayInSeconds, DateTimeOffset timeSent, string currentRoutingKey, DateTimeOffset utcNow)
112+
{
113+
var originalDeliveryDate = timeSent.AddSeconds(delayInSeconds);
114+
var newDelayInSeconds = Convert.ToInt32(originalDeliveryDate.Subtract(utcNow).TotalSeconds);
115+
var destinationQueue = currentRoutingKey[indexStartOfDestinationQueue..];
116+
var newRoutingKey = DelayInfrastructure.CalculateRoutingKey(newDelayInSeconds, destinationQueue, out int newDelayLevel);
117+
118+
return (destinationQueue, newRoutingKey, newDelayLevel);
119+
}
120+
121+
static DateTimeOffset GetTimeSent(BasicGetResult message)
122+
{
123+
var timeSentHeaderValue = message.BasicProperties.Headers?[timeSentHeader];
124+
var timeSentHeaderBytes = Array.Empty<byte>();
125+
126+
if (timeSentHeaderValue is not null)
127+
{
128+
timeSentHeaderBytes = (byte[])timeSentHeaderValue;
129+
}
130+
131+
var timeSentString = Encoding.UTF8.GetString(timeSentHeaderBytes);
132+
return DateTimeOffset.ParseExact(timeSentString, dateTimeOffsetWireFormat, CultureInfo.InvariantCulture);
133+
}
134+
135+
static bool MessageIsInvalid(BasicGetResult? message) =>
136+
message?.BasicProperties?.Headers is null
137+
|| !message.BasicProperties.Headers.ContainsKey(DelayInfrastructure.DelayHeader)
138+
|| !message.BasicProperties.Headers.ContainsKey(timeSentHeader);
139+
}

‎src/NServiceBus.Transport.RabbitMQ.CommandLine/Commands/Delays/DelaysMigrateCommand.cs‎

Lines changed: 2 additions & 128 deletions
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,12 @@
22
{
33
using System;
44
using System.CommandLine;
5-
using System.Globalization;
6-
using System.Text;
75
using System.Threading.Tasks;
86
using global::RabbitMQ.Client;
97

108
class DelaysMigrateCommand(BrokerConnection brokerConnection, IRoutingTopology routingTopology, TextWriter output)
119
{
1210
const string poisonMessageQueue = "delays-migrate-poison-messages";
13-
const string timeSentHeader = "NServiceBus.TimeSent";
14-
const string dateTimeOffsetWireFormat = "yyyy-MM-dd HH:mm:ss:ffffff Z";
15-
const int indexStartOfDestinationQueue = DelayInfrastructure.MaxNumberOfBitsToUse * 2;
1611

1712
public static Command CreateCommand()
1813
{
@@ -45,132 +40,11 @@ public static Command CreateCommand()
4540
async Task Run(CancellationToken cancellationToken)
4641
{
4742
await using var connection = await brokerConnection.Create(cancellationToken);
43+
4844
var createChannelOptions = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true, outstandingPublisherConfirmationsRateLimiter: null);
4945
await using var channel = await connection.CreateChannelAsync(createChannelOptions, cancellationToken);
5046

51-
for (int currentDelayLevel = DelayInfrastructure.MaxLevel; currentDelayLevel >= 0 && !cancellationToken.IsCancellationRequested; currentDelayLevel--)
52-
{
53-
await MigrateQueue(channel, currentDelayLevel, cancellationToken);
54-
}
47+
await DelayCommandHelpers.TransferMessages(sourceIsV1: true, channel, channel, poisonMessageQueue, routingTopology, output, cancellationToken);
5548
}
56-
57-
async Task MigrateQueue(IChannel channel, int delayLevel, CancellationToken cancellationToken)
58-
{
59-
var currentDelayQueue = $"nsb.delay-level-{delayLevel:00}";
60-
var messageCount = await channel.MessageCountAsync(currentDelayQueue, cancellationToken);
61-
var declaredDestinationQueues = new HashSet<string>();
62-
63-
if (messageCount > 0)
64-
{
65-
output.Write($"Processing {messageCount} messages at delay level {delayLevel:00}. ");
66-
67-
int skippedMessages = 0;
68-
int processedMessages = 0;
69-
70-
for (int i = 0; i < messageCount && !cancellationToken.IsCancellationRequested; i++)
71-
{
72-
var message = await channel.BasicGetAsync(currentDelayQueue, false, cancellationToken);
73-
74-
if (message == null)
75-
{
76-
// Queue is empty
77-
break;
78-
}
79-
80-
if (MessageIsInvalid(message))
81-
{
82-
skippedMessages++;
83-
84-
if (!poisonQueueCreated)
85-
{
86-
await channel.QueueDeclareAsync(poisonMessageQueue, true, false, false, cancellationToken: cancellationToken);
87-
poisonQueueCreated = true;
88-
}
89-
90-
await channel.BasicPublishAsync(string.Empty, poisonMessageQueue, false, new BasicProperties(message.BasicProperties), message.Body, cancellationToken: cancellationToken);
91-
await channel.BasicAckAsync(message.DeliveryTag, false, cancellationToken);
92-
93-
continue;
94-
}
95-
96-
var messageHeaders = message.BasicProperties.Headers;
97-
98-
var delayInSeconds = 0;
99-
100-
if (messageHeaders is not null)
101-
{
102-
var headerValue = messageHeaders[DelayInfrastructure.DelayHeader];
103-
104-
if (headerValue is not null)
105-
{
106-
delayInSeconds = (int)headerValue;
107-
}
108-
}
109-
110-
var timeSent = GetTimeSent(message);
111-
112-
var (destinationQueue, newRoutingKey, newDelayLevel) = GetNewRoutingKey(delayInSeconds, timeSent, message.RoutingKey, DateTimeOffset.UtcNow);
113-
114-
// Make sure the destination queue is bound to the delivery exchange to ensure delivery
115-
if (!declaredDestinationQueues.Contains(destinationQueue))
116-
{
117-
await routingTopology.BindToDelayInfrastructure(channel, destinationQueue, DelayInfrastructure.DeliveryExchange, DelayInfrastructure.BindingKey(destinationQueue), cancellationToken);
118-
declaredDestinationQueues.Add(destinationQueue);
119-
}
120-
121-
var publishExchange = DelayInfrastructure.LevelName(newDelayLevel);
122-
123-
if (messageHeaders != null)
124-
{
125-
//These headers need to be removed so that they won't be copied to an outgoing message if this message gets forwarded
126-
messageHeaders.Remove(DelayInfrastructure.XDeathHeader);
127-
messageHeaders.Remove(DelayInfrastructure.XFirstDeathExchangeHeader);
128-
messageHeaders.Remove(DelayInfrastructure.XFirstDeathQueueHeader);
129-
messageHeaders.Remove(DelayInfrastructure.XFirstDeathReasonHeader);
130-
}
131-
132-
await channel.BasicPublishAsync(publishExchange, newRoutingKey, false, new BasicProperties(message.BasicProperties), message.Body, cancellationToken: cancellationToken);
133-
await channel.BasicAckAsync(message.DeliveryTag, false, cancellationToken);
134-
processedMessages++;
135-
}
136-
137-
output.WriteLine($"{processedMessages} successful, {skippedMessages} skipped.");
138-
}
139-
else
140-
{
141-
output.WriteLine($"No messages to process at delay level {delayLevel:00}.");
142-
}
143-
}
144-
145-
public static (string DestinationQueue, string NewRoutingKey, int NewDelayLevel) GetNewRoutingKey(int delayInSeconds, DateTimeOffset timeSent, string currentRoutingKey, DateTimeOffset utcNow)
146-
{
147-
var originalDeliveryDate = timeSent.AddSeconds(delayInSeconds);
148-
var newDelayInSeconds = Convert.ToInt32(originalDeliveryDate.Subtract(utcNow).TotalSeconds);
149-
var destinationQueue = currentRoutingKey[indexStartOfDestinationQueue..];
150-
var newRoutingKey = DelayInfrastructure.CalculateRoutingKey(newDelayInSeconds, destinationQueue, out int newDelayLevel);
151-
152-
return (destinationQueue, newRoutingKey, newDelayLevel);
153-
}
154-
155-
static DateTimeOffset GetTimeSent(BasicGetResult message)
156-
{
157-
var timeSentHeaderValue = message.BasicProperties.Headers?[timeSentHeader];
158-
var timeSentHeaderBytes = Array.Empty<byte>();
159-
160-
if (timeSentHeaderValue is not null)
161-
{
162-
timeSentHeaderBytes = (byte[])timeSentHeaderValue;
163-
}
164-
165-
var timeSentString = Encoding.UTF8.GetString(timeSentHeaderBytes);
166-
return DateTimeOffset.ParseExact(timeSentString, dateTimeOffsetWireFormat, CultureInfo.InvariantCulture);
167-
}
168-
169-
static bool MessageIsInvalid(BasicGetResult? message) =>
170-
message?.BasicProperties?.Headers == null
171-
|| !message.BasicProperties.Headers.ContainsKey(DelayInfrastructure.DelayHeader)
172-
|| !message.BasicProperties.Headers.ContainsKey(timeSentHeader);
173-
174-
bool poisonQueueCreated = false;
17549
}
17650
}

0 commit comments

Comments
 (0)