Skip to content

Commit c1ab8bb

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

4 files changed

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

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