Skip to content

Commit 56af06b

Browse files
lukebakkenlukebakken
committed
Update to RC13
1 parent 2215d20 commit 56af06b

File tree

21 files changed

+153
-107
lines changed

21 files changed

+153
-107
lines changed

‎dotnet/EmitLog/EmitLog.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/EmitLogDirect/EmitLogDirect.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/EmitLogTopic/EmitLogTopic.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/NewTask/NewTask.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/PublisherConfirms/PublisherConfirms.cs

Lines changed: 89 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,22 @@
11
using RabbitMQ.Client;
2+
using RabbitMQ.Client.Exceptions;
23
using System.Diagnostics;
34
using System.Text;
45

5-
const int MESSAGE_COUNT = 50_000;
6+
const int MessageCount = 50_000;
7+
const int MaxOutstandingConfirms = 128;
8+
9+
var channelOptions = new CreateChannelOptions
10+
{
11+
PublisherConfirmationsEnabled = true,
12+
PublisherConfirmationTrackingEnabled = true,
13+
OutstandingPublisherConfirmationsRateLimiter = new ThrottlingRateLimiter(MaxOutstandingConfirms)
14+
};
15+
16+
var props = new BasicProperties
17+
{
18+
Persistent = true
19+
};
620

721
await PublishMessagesIndividuallyAsync();
822
await PublishMessagesInBatchAsync();
@@ -14,110 +28,140 @@ static Task<IConnection> CreateConnectionAsync()
1428
return factory.CreateConnectionAsync();
1529
}
1630

17-
static async Task PublishMessagesIndividuallyAsync()
31+
async Task PublishMessagesIndividuallyAsync()
1832
{
19-
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages individually " +
20-
"and handling confirms all at once");
33+
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MessageCount:N0} messages individually " +
34+
"and handling confirms individually (i.e., the slowest way)");
2135

2236
using IConnection connection = await CreateConnectionAsync();
23-
using IChannel channel = await connection.CreateChannelAsync();
37+
using IChannel channel = await connection.CreateChannelAsync(channelOptions);
2438

2539
// declare a server-named queue
2640
QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
2741
string queueName = queueDeclareResult.QueueName;
28-
await channel.ConfirmSelectAsync();
2942

3043
var sw = new Stopwatch();
3144
sw.Start();
3245

33-
for (int i = 0; i < MESSAGE_COUNT; i++)
46+
for (int i = 0; i < MessageCount; i++)
3447
{
3548
byte[] body = Encoding.UTF8.GetBytes(i.ToString());
36-
await channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName, body: body);
49+
try
50+
{
51+
await channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName, body: body,
52+
mandatory: true, basicProperties: props);
53+
}
54+
catch (PublishException pubEx)
55+
{
56+
Console.Error.WriteLine("{0} [ERROR] publish exception: {1}", DateTime.Now, pubEx);
57+
}
58+
catch (Exception ex)
59+
{
60+
Console.Error.WriteLine("{0} [ERROR] other exception: {1}", DateTime.Now, ex);
61+
}
3762
}
3863

39-
await channel.WaitForConfirmsOrDieAsync();
40-
4164
sw.Stop();
4265

43-
Console.WriteLine($"{DateTime.Now} [INFO] published {MESSAGE_COUNT:N0} messages individually " +
66+
Console.WriteLine($"{DateTime.Now} [INFO] published {MessageCount:N0} messages individually " +
4467
$"in {sw.ElapsedMilliseconds:N0} ms");
4568
}
4669

47-
static async Task PublishMessagesInBatchAsync()
70+
async Task PublishMessagesInBatchAsync()
4871
{
49-
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and handling " +
72+
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MessageCount:N0} messages and handling " +
5073
$"confirms in batches");
5174

5275
using IConnection connection = await CreateConnectionAsync();
53-
using IChannel channel = await connection.CreateChannelAsync();
76+
using IChannel channel = await connection.CreateChannelAsync(channelOptions);
5477

5578
// declare a server-named queue
5679
QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
5780
string queueName = queueDeclareResult.QueueName;
58-
await channel.ConfirmSelectAsync();
5981

60-
int batchSize = 100;
82+
/*
83+
* Note: since throttling happens when 50% of the outstanding confirms are reached,
84+
* each batch size should not be greater than this value
85+
*/
86+
int batchSize = MaxOutstandingConfirms / 2;
6187
int outstandingMessageCount = 0;
6288

6389
var sw = new Stopwatch();
6490
sw.Start();
6591

66-
var publishTasks = new List<Task>();
67-
for (int i = 0; i < MESSAGE_COUNT; i++)
92+
static async Task AwaitPublishTasks(IEnumerable<ValueTask> publishTasks)
93+
{
94+
foreach (ValueTask pt in publishTasks)
95+
{
96+
try
97+
{
98+
await pt;
99+
}
100+
catch (PublishException pubEx)
101+
{
102+
Console.Error.WriteLine("{0} [ERROR] publish exception: {1}", DateTime.Now, pubEx);
103+
}
104+
catch (Exception ex)
105+
{
106+
Console.Error.WriteLine("{0} [ERROR] other exception: {1}", DateTime.Now, ex);
107+
}
108+
}
109+
}
110+
111+
var publishTasks = new List<ValueTask>();
112+
for (int i = 0; i < MessageCount; i++)
68113
{
69114
byte[] body = Encoding.UTF8.GetBytes(i.ToString());
70-
var pt = channel.BasicPublishAsync(exchange: string.Empty,
71-
routingKey: queueName, body: body);
72-
publishTasks.Add(pt.AsTask());
115+
116+
var pt0 = channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName, body: body,
117+
mandatory: true, basicProperties: props);
118+
publishTasks.Add(pt0);
119+
73120
outstandingMessageCount++;
74121

75122
if (outstandingMessageCount == batchSize)
76123
{
77-
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
78-
await Task.WhenAll(publishTasks).WaitAsync(cts.Token);
124+
await AwaitPublishTasks(publishTasks);
79125
publishTasks.Clear();
80-
81-
await channel.WaitForConfirmsOrDieAsync(cts.Token);
82126
outstandingMessageCount = 0;
83127
}
84128
}
85129

86-
if (outstandingMessageCount > 0)
130+
if (publishTasks.Count > 0)
87131
{
88-
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
89-
await channel.WaitForConfirmsOrDieAsync(cts.Token);
132+
await AwaitPublishTasks(publishTasks);
90133
}
91134

92135
sw.Stop();
93-
Console.WriteLine($"{DateTime.Now} [INFO] published {MESSAGE_COUNT:N0} messages in batch in " +
136+
Console.WriteLine($"{DateTime.Now} [INFO] published {MessageCount:N0} messages in batch in " +
94137
$"{sw.ElapsedMilliseconds:N0} ms");
95138
}
96139

97140
async Task HandlePublishConfirmsAsynchronously()
98141
{
99-
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and " +
142+
Console.WriteLine($"{DateTime.Now} [INFO] publishing {MessageCount:N0} messages and " +
100143
$"handling confirms asynchronously");
101144

145+
// NOTE: setting trackConfirmations to false because this program
146+
// is tracking them itself.
147+
channelOptions.PublisherConfirmationTrackingEnabled = false;
148+
channelOptions.OutstandingPublisherConfirmationsRateLimiter = null;
149+
102150
using IConnection connection = await CreateConnectionAsync();
103-
using IChannel channel = await connection.CreateChannelAsync();
151+
using IChannel channel = await connection.CreateChannelAsync(channelOptions);
104152

105153
// declare a server-named queue
106154
QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
107155
string queueName = queueDeclareResult.QueueName;
108156

109-
// NOTE: setting trackConfirmations to false because this program
110-
// is tracking them itself.
111-
await channel.ConfirmSelectAsync(trackConfirmations: false);
112-
113157
bool publishingCompleted = false;
114158
var allMessagesConfirmedTcs =
115159
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
116160
var outstandingConfirms = new LinkedList<ulong>();
117161
var semaphore = new SemaphoreSlim(1, 1);
118-
void CleanOutstandingConfirms(ulong deliveryTag, bool multiple)
162+
async Task CleanOutstandingConfirms(ulong deliveryTag, bool multiple)
119163
{
120-
semaphore.Wait();
164+
await semaphore.WaitAsync();
121165
try
122166
{
123167
if (multiple)
@@ -155,23 +199,25 @@ void CleanOutstandingConfirms(ulong deliveryTag, bool multiple)
155199
}
156200
}
157201

158-
channel.BasicAcks += (sender, ea) => CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
159-
channel.BasicNacks += (sender, ea) =>
202+
channel.BasicAcksAsync += (sender, ea) =>
203+
CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
204+
205+
channel.BasicNacksAsync += (sender, ea) =>
160206
{
161207
Console.WriteLine($"{DateTime.Now} [WARNING] message sequence number: {ea.DeliveryTag} " +
162208
$"has been nacked (multiple: {ea.Multiple})");
163-
CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
209+
return CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
164210
};
165211

166212
var sw = new Stopwatch();
167213
sw.Start();
168214

169215
var publishTasks = new List<ValueTask>();
170-
for (int i = 0; i < MESSAGE_COUNT; i++)
216+
for (int i = 0; i < MessageCount; i++)
171217
{
172218
string msg = i.ToString();
173219
byte[] body = Encoding.UTF8.GetBytes(msg);
174-
ulong nextPublishSeqNo = channel.NextPublishSeqNo;
220+
ulong nextPublishSeqNo = await channel.GetNextPublishSequenceNumberAsync();
175221
if ((ulong)(i + 1) != nextPublishSeqNo)
176222
{
177223
Console.WriteLine($"{DateTime.Now} [WARNING] i {i + 1} does not equal next sequence " +
@@ -215,6 +261,6 @@ void CleanOutstandingConfirms(ulong deliveryTag, bool multiple)
215261
}
216262

217263
sw.Stop();
218-
Console.WriteLine($"{DateTime.Now} [INFO] published {MESSAGE_COUNT:N0} messages and handled " +
264+
Console.WriteLine($"{DateTime.Now} [INFO] published {MessageCount:N0} messages and handled " +
219265
$"confirm asynchronously {sw.ElapsedMilliseconds:N0} ms");
220266
}

‎dotnet/PublisherConfirms/PublisherConfirms.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/RPCClient/RPCClient.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ public async Task StartAsync()
3030
_replyQueueName = queueDeclareResult.QueueName;
3131
var consumer = new AsyncEventingBasicConsumer(_channel);
3232

33-
consumer.Received += (model, ea) =>
33+
consumer.ReceivedAsync += (model, ea) =>
3434
{
3535
string? correlationId = ea.BasicProperties.CorrelationId;
3636

‎dotnet/RPCClient/RPCClient.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/RPCServer/RPCServer.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ await channel.QueueDeclareAsync(queue: QUEUE_NAME, durable: false, exclusive: fa
1414
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
1515

1616
var consumer = new AsyncEventingBasicConsumer(channel);
17-
consumer.Received += async (object sender, BasicDeliverEventArgs ea) =>
17+
consumer.ReceivedAsync += async (object sender, BasicDeliverEventArgs ea) =>
1818
{
1919
AsyncEventingBasicConsumer cons = (AsyncEventingBasicConsumer)sender;
2020
IChannel ch = cons.Channel;

‎dotnet/RPCServer/RPCServer.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/Receive/Receive.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ await channel.QueueDeclareAsync(queue: "hello", durable: false, exclusive: false
1212
Console.WriteLine(" [*] Waiting for messages.");
1313

1414
var consumer = new AsyncEventingBasicConsumer(channel);
15-
consumer.Received += (model, ea) =>
15+
consumer.ReceivedAsync += (model, ea) =>
1616
{
1717
var body = ea.Body.ToArray();
1818
var message = Encoding.UTF8.GetString(body);

‎dotnet/Receive/Receive.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/ReceiveLogs/ReceiveLogs.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ await channel.ExchangeDeclareAsync(exchange: "logs",
1717
Console.WriteLine(" [*] Waiting for logs.");
1818

1919
var consumer = new AsyncEventingBasicConsumer(channel);
20-
consumer.Received += (model, ea) =>
20+
consumer.ReceivedAsync += (model, ea) =>
2121
{
2222
byte[] body = ea.Body.ToArray();
2323
var message = Encoding.UTF8.GetString(body);

‎dotnet/ReceiveLogs/ReceiveLogs.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

‎dotnet/ReceiveLogsDirect/ReceiveLogsDirect.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
Console.WriteLine(" [*] Waiting for messages.");
3232

3333
var consumer = new AsyncEventingBasicConsumer(channel);
34-
consumer.Received += (model, ea) =>
34+
consumer.ReceivedAsync += (model, ea) =>
3535
{
3636
var body = ea.Body.ToArray();
3737
var message = Encoding.UTF8.GetString(body);

‎dotnet/ReceiveLogsDirect/ReceiveLogsDirect.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.9" />
11+
<PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.13" />
1212
</ItemGroup>
1313

1414
</Project>

0 commit comments

Comments
 (0)