Skip to content

Commit 61579bc

Browse files
danielmarbachdanielmarbach
committed
Wip upgrade
1 parent eb816a4 commit 61579bc

26 files changed

+239
-231
lines changed

‎src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/NServiceBus.Transport.RabbitMQ.AcceptanceTests.csproj‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020

2121
<ItemGroup>
2222
<PackageReference Include="BitFaster.Caching" Version="2.5.1" />
23-
<PackageReference Include="NServiceBus.AcceptanceTests.Sources" Version="9.2.2" GeneratePathProperty="true" />
24-
<PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
23+
<PackageReference Include="NServiceBus.AcceptanceTests.Sources" Version="9.0.2" GeneratePathProperty="true" />
24+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.8" />
2525
</ItemGroup>
2626

2727
<ItemGroup>

‎src/NServiceBus.Transport.RabbitMQ.CommandLine/BrokerConnection.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,10 @@ public BrokerConnection(RabbitMQ.ConnectionFactory connectionFactory)
1010
this.connectionFactory = connectionFactory;
1111
}
1212

13-
public IConnection Create()
13+
public async Task<IConnection> Create(CancellationToken cancellationToken = default)
1414
{
15-
var connection = connectionFactory.CreateAdministrationConnection();
16-
connection.VerifyBrokerRequirements();
15+
var connection = await connectionFactory.CreateAdministrationConnection(cancellationToken).ConfigureAwait(false);
16+
await connection.VerifyBrokerRequirements(cancellationToken: cancellationToken).ConfigureAwait(false);
1717

1818
return connection;
1919
}

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

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,18 +26,16 @@ public DelaysCreateCommand(BrokerConnection brokerConnection, IConsole console)
2626
this.console = console;
2727
}
2828

29-
public Task Run(CancellationToken cancellationToken = default)
29+
public async Task Run(CancellationToken cancellationToken = default)
3030
{
3131
console.WriteLine($"Creating v2 delay infrastructure queues and exchanges...");
3232

33-
using var connection = brokerConnection.Create();
34-
using var channel = connection.CreateModel();
33+
using var connection = await brokerConnection.Create(cancellationToken).ConfigureAwait(false);
34+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
3535

36-
DelayInfrastructure.Build(channel);
36+
await DelayInfrastructure.Build(channel, cancellationToken).ConfigureAwait(false);
3737

3838
console.WriteLine("Queues and exchanges created successfully");
39-
40-
return Task.CompletedTask;
4139
}
4240

4341
readonly BrokerConnection brokerConnection;

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -46,21 +46,19 @@ public DelaysMigrateCommand(BrokerConnection brokerConnection, IRoutingTopology
4646
this.console = console;
4747
}
4848

49-
public Task Run(CancellationToken cancellationToken = default)
49+
public async Task Run(CancellationToken cancellationToken = default)
5050
{
51-
using var connection = brokerConnection.Create();
52-
using var channel = connection.CreateModel();
53-
channel.ConfirmSelect();
51+
using var connection = await brokerConnection.Create(cancellationToken).ConfigureAwait(false);
52+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
53+
await channel.ConfirmSelectAsync(cancellationToken).ConfigureAwait(false);
5454

5555
for (int currentDelayLevel = DelayInfrastructure.MaxLevel; currentDelayLevel >= 0 && !cancellationToken.IsCancellationRequested; currentDelayLevel--)
5656
{
5757
MigrateQueue(channel, currentDelayLevel, cancellationToken);
5858
}
59-
60-
return Task.CompletedTask;
6159
}
6260

63-
void MigrateQueue(IModel channel, int delayLevel, CancellationToken cancellationToken)
61+
void MigrateQueue(IChannel channel, int delayLevel, CancellationToken cancellationToken)
6462
{
6563
var currentDelayQueue = $"nsb.delay-level-{delayLevel:00}";
6664
var messageCount = channel.MessageCount(currentDelayQueue);

‎src/NServiceBus.Transport.RabbitMQ.CommandLine/Commands/Queue/QueueMigrateCommand.cs‎

Lines changed: 24 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ public QueueMigrateCommand(string queueName, BrokerConnection brokerConnection,
4141
migrationState = new MigrationState();
4242
}
4343

44-
public Task Run(CancellationToken cancellationToken = default)
44+
public async Task Run(CancellationToken cancellationToken = default)
4545
{
4646
console.WriteLine($"Starting migration of '{queueName}'");
4747

@@ -54,7 +54,7 @@ public Task Run(CancellationToken cancellationToken = default)
5454
switch (migrationState.CurrentStage)
5555
{
5656
case MigrationStage.Starting:
57-
migrationState.CurrentStage = MoveMessagesToHoldingQueue(connection, cancellationToken);
57+
migrationState.CurrentStage = await MoveMessagesToHoldingQueue(connection, cancellationToken).ConfigureAwait(false);
5858
break;
5959
case MigrationStage.MessagesMovedToHoldingQueue:
6060
migrationState.CurrentStage = DeleteMainQueue(connection);
@@ -73,18 +73,16 @@ public Task Run(CancellationToken cancellationToken = default)
7373
break;
7474
}
7575
}
76-
77-
return Task.CompletedTask;
7876
}
7977

80-
MigrationStage MoveMessagesToHoldingQueue(IConnection connection, CancellationToken cancellationToken)
78+
async Task<MigrationStage> MoveMessagesToHoldingQueue(IConnection connection, CancellationToken cancellationToken)
8179
{
82-
using var channel = connection.CreateModel();
80+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
8381

8482
console.WriteLine($"Migrating messages from '{queueName}' to '{holdingQueueName}'");
8583

8684
// does the holding queue need to be quorum?
87-
channel.QueueDeclare(holdingQueueName, true, false, false, quorumQueueArguments);
85+
await channel.QueueDeclareAsync(holdingQueueName, true, false, false, quorumQueueArguments).ConfigureAwait(false);
8886
console.WriteLine($"Created queue '{holdingQueueName}'");
8987

9088
// bind the holding queue to the exchange of the queue under migration
@@ -140,25 +138,25 @@ MigrationStage CreateMainQueueAsQuorum(IConnection connection)
140138
return MigrationStage.QuorumQueueCreated;
141139
}
142140

143-
MigrationStage RestoreMessages(IConnection connection, CancellationToken cancellationToken)
141+
async Task<MigrationStage> RestoreMessages(IConnection connection, CancellationToken cancellationToken)
144142
{
145-
using var channel = connection.CreateModel();
143+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
146144

147-
channel.QueueBind(queueName, queueName, string.Empty);
145+
await channel.QueueBindAsync(queueName, queueName, string.Empty, cancellationToken: cancellationToken).ConfigureAwait(false);
148146
console.WriteLine($"Re-bound '{queueName}' to exchange '{queueName}'");
149147

150-
channel.QueueUnbind(holdingQueueName, queueName, string.Empty);
148+
await channel.QueueUnbindAsync(holdingQueueName, queueName, string.Empty, cancellationToken: cancellationToken).ConfigureAwait(false);
151149
console.WriteLine($"Unbound '{holdingQueueName}' from exchange '{queueName}'");
152150

153151
var messageIds = new Dictionary<string, string>();
154152

155153
// move all messages in the holding queue back to the main queue
156-
channel.ConfirmSelect();
154+
await channel.ConfirmSelectAsync(cancellationToken).ConfigureAwait(false);
157155

158-
var numMessageMovedBackToMain = ProcessMessages(
156+
var numMessageMovedBackToMain = await ProcessMessages(
159157
channel,
160158
holdingQueueName,
161-
message =>
159+
async (message, token) =>
162160
{
163161
string? messageIdString = null;
164162

@@ -175,55 +173,55 @@ MigrationStage RestoreMessages(IConnection connection, CancellationToken cancell
175173
}
176174
}
177175

178-
channel.BasicPublish(string.Empty, queueName, message.BasicProperties, message.Body);
179-
channel.WaitForConfirmsOrDie();
176+
await channel.BasicPublishAsync(string.Empty, queueName, new BasicProperties(message.BasicProperties), message.Body, cancellationToken: token).ConfigureAwait(false);
177+
await channel.WaitForConfirmsOrDieAsync(token).ConfigureAwait(false);
180178

181179
if (messageIdString != null)
182180
{
183181
messageIds.Add(messageIdString, string.Empty);
184182
}
185183
},
186-
cancellationToken);
184+
cancellationToken).ConfigureAwait(false);
187185

188186
console.WriteLine($"Moved {numMessageMovedBackToMain} messages from '{holdingQueueName}' to '{queueName}'");
189187

190188
return MigrationStage.MessagesMovedToQuorumQueue;
191189
}
192190

193-
MigrationStage CleanUpHoldingQueue(IConnection connection)
191+
async Task<MigrationStage> CleanUpHoldingQueue(IConnection connection, CancellationToken cancellationToken)
194192
{
195-
using var channel = connection.CreateModel();
193+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
196194

197-
if (channel.MessageCount(holdingQueueName) != 0)
195+
if (await channel.MessageCountAsync(holdingQueueName, cancellationToken).ConfigureAwait(false) != 0)
198196
{
199197
throw new Exception($"'{holdingQueueName}' is not empty and was not deleted. Run the command again to retry message processing.");
200198
}
201199

202-
channel.QueueDelete(holdingQueueName);
200+
await channel.QueueDeleteAsync(holdingQueueName, cancellationToken: cancellationToken).ConfigureAwait(false);
203201
console.WriteLine($"Removed '{holdingQueueName}'");
204202

205203
return MigrationStage.CleanUpCompleted;
206204
}
207205

208-
uint ProcessMessages(IModel channel, string sourceQueue, Action<BasicGetResult> onMoveMessage, CancellationToken cancellationToken)
206+
async Task<uint> ProcessMessages(IChannel channel, string sourceQueue, Func<BasicGetResult, CancellationToken, Task> onMoveMessage, CancellationToken cancellationToken)
209207
{
210-
var messageCount = channel.MessageCount(sourceQueue);
208+
var messageCount = await channel.MessageCountAsync(sourceQueue, cancellationToken).ConfigureAwait(false);
211209

212210
for (var i = 0; i < messageCount; i++)
213211
{
214212
cancellationToken.ThrowIfCancellationRequested();
215213

216-
var message = channel.BasicGet(sourceQueue, false);
214+
var message = await channel.BasicGetAsync(sourceQueue, false, cancellationToken).ConfigureAwait(false);
217215

218216
if (message == null)
219217
{
220218
// Queue is empty
221219
break;
222220
}
223221

224-
onMoveMessage(message);
222+
await onMoveMessage(message, cancellationToken).ConfigureAwait(false);
225223

226-
channel.BasicAck(message.DeliveryTag, false);
224+
await channel.BasicAckAsync(message.DeliveryTag, false, cancellationToken).ConfigureAwait(false);
227225
}
228226

229227
return messageCount;

‎src/NServiceBus.Transport.RabbitMQ.CommandLine/NServiceBus.Transport.RabbitMQ.CommandLine.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616

1717
<ItemGroup>
1818
<PackageReference Include="NServiceBus" Version="9.1.1" />
19-
<PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
19+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.8" />
2020
<PackageReference Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />
2121
</ItemGroup>
2222

‎src/NServiceBus.Transport.RabbitMQ.Tests/NServiceBus.Transport.RabbitMQ.Tests.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
<PackageReference Include="NServiceBus" Version="9.2.2" />
2424
<PackageReference Include="Particular.Approvals" Version="1.0.0" />
2525
<PackageReference Include="PublicApiGenerator" Version="11.1.0" />
26-
<PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
26+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.8" />
2727
</ItemGroup>
2828

2929
</Project>

‎src/NServiceBus.Transport.RabbitMQ.TransportTests/NServiceBus.Transport.RabbitMQ.TransportTests.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
<ItemGroup>
2020
<PackageReference Include="BitFaster.Caching" Version="2.5.1" />
2121
<PackageReference Include="NServiceBus.TransportTests.Sources" Version="9.2.2" />
22-
<PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
22+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.8" />
2323
</ItemGroup>
2424

2525
<ItemGroup>

‎src/NServiceBus.Transport.RabbitMQ/Administration/QueuePurger.cs‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
namespace NServiceBus.Transport.RabbitMQ
22
{
3+
using System.Threading;
4+
using System.Threading.Tasks;
5+
36
class QueuePurger
47
{
58
readonly ConnectionFactory connectionFactory;
@@ -9,13 +12,11 @@ public QueuePurger(ConnectionFactory connectionFactory)
912
this.connectionFactory = connectionFactory;
1013
}
1114

12-
public void Purge(string queue)
15+
public async Task Purge(string queue, CancellationToken cancellationToken = default)
1316
{
14-
using (var connection = connectionFactory.CreateAdministrationConnection())
15-
using (var channel = connection.CreateModel())
16-
{
17-
channel.QueuePurge(queue);
18-
}
17+
using var connection = await connectionFactory.CreateAdministrationConnection(cancellationToken).ConfigureAwait(false);
18+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
19+
await channel.QueuePurgeAsync(queue, cancellationToken).ConfigureAwait(false);
1920
}
2021
}
2122
}

‎src/NServiceBus.Transport.RabbitMQ/Administration/SubscriptionManager.cs‎

Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -19,28 +19,22 @@ public SubscriptionManager(ConnectionFactory connectionFactory, IRoutingTopology
1919
this.localQueue = localQueue;
2020
}
2121

22-
public Task SubscribeAll(MessageMetadata[] eventTypes, ContextBag context, CancellationToken cancellationToken = default)
22+
public async Task SubscribeAll(MessageMetadata[] eventTypes, ContextBag context, CancellationToken cancellationToken = default)
2323
{
24-
using (var connection = connectionFactory.CreateAdministrationConnection())
25-
using (var channel = connection.CreateModel())
24+
using var connection = await connectionFactory.CreateAdministrationConnection(cancellationToken).ConfigureAwait(false);
25+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
26+
// TODO: Parallelize?
27+
foreach (var eventType in eventTypes)
2628
{
27-
foreach (var eventType in eventTypes)
28-
{
29-
routingTopology.SetupSubscription(channel, eventType, localQueue);
30-
}
29+
await routingTopology.SetupSubscription(channel, eventType, localQueue, cancellationToken).ConfigureAwait(false);
3130
}
32-
return Task.CompletedTask;
3331
}
3432

35-
public Task Unsubscribe(MessageMetadata eventType, ContextBag context, CancellationToken cancellationToken = default)
33+
public async Task Unsubscribe(MessageMetadata eventType, ContextBag context, CancellationToken cancellationToken = default)
3634
{
37-
using (var connection = connectionFactory.CreateAdministrationConnection())
38-
using (var channel = connection.CreateModel())
39-
{
40-
routingTopology.TeardownSubscription(channel, eventType, localQueue);
41-
}
42-
43-
return Task.CompletedTask;
35+
using var connection = await connectionFactory.CreateAdministrationConnection(cancellationToken).ConfigureAwait(false);
36+
using var channel = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false);
37+
await routingTopology.TeardownSubscription(channel, eventType, localQueue, cancellationToken).ConfigureAwait(false);
4438
}
4539
}
4640
}

0 commit comments

Comments
 (0)