Skip to content

Commit 67b7f88

Browse files
lukebakkenlukebakken
committed
Merge branch 'main' into stringallocations
2 parents 4e1156b + bf9a35a commit 67b7f88

File tree

9 files changed

+86
-77
lines changed

9 files changed

+86
-77
lines changed

‎projects/RabbitMQ.Client.OAuth2/OAuth2Client.cs‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -232,9 +232,11 @@ public void Dispose()
232232

233233
private Dictionary<string, string> buildRequestParameters()
234234
{
235-
var dict = new Dictionary<string, string>(_additionalRequestParameters);
236-
dict.Add(CLIENT_ID, _clientId);
237-
dict.Add(CLIENT_SECRET, _clientSecret);
235+
var dict = new Dictionary<string, string>(_additionalRequestParameters)
236+
{
237+
{ CLIENT_ID, _clientId },
238+
{ CLIENT_SECRET, _clientSecret }
239+
};
238240
if (_scope != null && _scope.Length > 0)
239241
{
240242
dict.Add(SCOPE, _scope);

‎projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recording.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -170,7 +170,7 @@ private void DoDeleteAutoDeleteExchange(string exchangeName)
170170

171171
bool AnyBindingsOnExchange(string exchange)
172172
{
173-
foreach (var recordedBinding in _recordedBindings)
173+
foreach (RecordedBinding recordedBinding in _recordedBindings)
174174
{
175175
if (recordedBinding.Source == exchange)
176176
{
@@ -400,15 +400,15 @@ await _recordedEntitiesSemaphore.WaitAsync()
400400

401401
private void DoDeleteRecordedConsumer(string consumerTag)
402402
{
403-
if (_recordedConsumers.Remove(consumerTag, out var recordedConsumer))
403+
if (_recordedConsumers.Remove(consumerTag, out RecordedConsumer recordedConsumer))
404404
{
405405
DeleteAutoDeleteQueue(recordedConsumer.Queue);
406406
}
407407
}
408408

409409
private void DeleteAutoDeleteQueue(string queue)
410410
{
411-
if (_recordedQueues.TryGetValue(queue, out var recordedQueue) && recordedQueue.AutoDelete)
411+
if (_recordedQueues.TryGetValue(queue, out RecordedQueue recordedQueue) && recordedQueue.AutoDelete)
412412
{
413413
// last consumer on this connection is gone, remove recorded queue if it is auto-deleted.
414414
if (!AnyConsumersOnQueue(queue))
@@ -420,7 +420,7 @@ private void DeleteAutoDeleteQueue(string queue)
420420

421421
private bool AnyConsumersOnQueue(string queue)
422422
{
423-
foreach (var pair in _recordedConsumers)
423+
foreach (KeyValuePair<string, RecordedConsumer> pair in _recordedConsumers)
424424
{
425425
if (pair.Value.Queue == queue)
426426
{

‎projects/RabbitMQ.Client/client/impl/ChannelBase.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -488,7 +488,7 @@ private void OnChannelShutdown(ShutdownEventArgs reason)
488488
if (_confirmsTaskCompletionSources?.Count > 0)
489489
{
490490
var exception = new AlreadyClosedException(reason);
491-
foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
491+
foreach (TaskCompletionSource<bool> confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
492492
{
493493
confirmsTaskCompletionSource.TrySetException(exception);
494494
}
@@ -635,7 +635,7 @@ protected void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
635635
if (_pendingDeliveryTags.Count == 0 && _confirmsTaskCompletionSources.Count > 0)
636636
{
637637
// Done, mark tasks
638-
foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
638+
foreach (TaskCompletionSource<bool> confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
639639
{
640640
confirmsTaskCompletionSource.TrySetResult(_onlyAcksReceived);
641641
}
@@ -754,7 +754,7 @@ protected async Task<bool> HandleChannelCloseOkAsync(IncomingCommand cmd, Cancel
754754
*/
755755
FinishClose();
756756

757-
if (_continuationQueue.TryPeek<ChannelCloseAsyncRpcContinuation>(out var k))
757+
if (_continuationQueue.TryPeek<ChannelCloseAsyncRpcContinuation>(out ChannelCloseAsyncRpcContinuation k))
758758
{
759759
_continuationQueue.Next();
760760
await k.HandleCommandAsync(cmd)
@@ -1905,7 +1905,7 @@ private static BasicProperties PopulateActivityAndPropagateTraceId<TProperties>(
19051905
props = new BasicProperties();
19061906
}
19071907

1908-
var headers = props.Headers ?? new Dictionary<string, object>();
1908+
IDictionary<string, object> headers = props.Headers ?? new Dictionary<string, object>();
19091909

19101910
// Inject the ActivityContext into the message headers to propagate trace context to the receiving service.
19111911
DistributedContextPropagator.Current.Inject(sendActivity, headers, InjectTraceContextIntoBasicProperties);

‎projects/RabbitMQ.Client/client/impl/Connection.Commands.cs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -212,9 +212,13 @@ await UpdateSecretAsync(_config.CredentialsProvider.Password, "Token refresh", c
212212
private IAuthMechanismFactory GetAuthMechanismFactory(string supportedMechanismNames)
213213
{
214214
// Our list is in order of preference, the server one is not.
215-
foreach (var factory in _config.AuthMechanisms)
215+
foreach (IAuthMechanismFactory factory in _config.AuthMechanisms)
216216
{
217+
#if NET6_0_OR_GREATER
218+
if (supportedMechanismNames.Contains(factory.Name, StringComparison.OrdinalIgnoreCase))
219+
#else
217220
if (supportedMechanismNames.IndexOf(factory.Name, StringComparison.OrdinalIgnoreCase) >= 0)
221+
#endif
218222
{
219223
return factory;
220224
}

‎projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherBase.cs‎

Lines changed: 23 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ protected void AddConsumer(IBasicConsumer consumer, string tag)
2525
{
2626
lock (_consumers)
2727
{
28-
var tagBytes = Encoding.UTF8.GetBytes(tag);
28+
byte[] tagBytes = Encoding.UTF8.GetBytes(tag);
2929
_consumers[tagBytes] = (consumer, tag);
3030
}
3131
}
@@ -39,7 +39,7 @@ protected void AddConsumer(IBasicConsumer consumer, string tag)
3939
return consumerPair;
4040
}
4141

42-
#if !NETSTANDARD
42+
#if NET6_0_OR_GREATER
4343
string consumerTag = Encoding.UTF8.GetString(tag.Span);
4444
#else
4545
string consumerTag;
@@ -60,18 +60,29 @@ public IBasicConsumer GetAndRemoveConsumer(string tag)
6060
{
6161
lock (_consumers)
6262
{
63-
var utf8 = Encoding.UTF8;
64-
var pool = ArrayPool<byte>.Shared;
65-
var buf = pool.Rent(utf8.GetMaxByteCount(tag.Length));
66-
#if NETSTANDARD
67-
int count = utf8.GetBytes(tag, 0, tag.Length, buf, 0);
63+
ArrayPool<byte> pool = ArrayPool<byte>.Shared;
64+
byte[]? buf = null;
65+
try
66+
{
67+
buf = pool.Rent(Encoding.UTF8.GetMaxByteCount(tag.Length));
68+
#if NET6_0_OR_GREATER
69+
int count = Encoding.UTF8.GetBytes(tag, buf);
6870
#else
69-
int count = utf8.GetBytes(tag, buf);
71+
int count = Encoding.UTF8.GetBytes(tag, 0, tag.Length, buf, 0);
7072
#endif
71-
var memory = buf.AsMemory(0, count);
72-
var result = _consumers.Remove(memory, out var consumerPair) ? consumerPair.consumer : GetDefaultOrFallbackConsumer();
73-
pool.Return(buf);
74-
return result;
73+
Memory<byte> memory = buf.AsMemory(0, count);
74+
IBasicConsumer result = _consumers.Remove(memory,
75+
out (IBasicConsumer consumer, string consumerTag) consumerPair) ?
76+
consumerPair.consumer : GetDefaultOrFallbackConsumer();
77+
return result;
78+
}
79+
finally
80+
{
81+
if (buf != null)
82+
{
83+
pool.Return(buf);
84+
}
85+
}
7586
}
7687
}
7788

‎projects/RabbitMQ.Client/client/impl/WireFormatting.Read.cs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ public static object ReadFieldValue(ReadOnlySpan<byte> span, out int bytesRead)
7878
switch ((char)span[0])
7979
{
8080
case 'S':
81-
bytesRead = 1 + ReadLongstr(span.Slice(1), out var bytes);
81+
bytesRead = 1 + ReadLongstr(span.Slice(1), out byte[] bytes);
8282
return bytes;
8383
case 't':
8484
bytesRead = 2;
@@ -96,11 +96,11 @@ public static object ReadFieldValue(ReadOnlySpan<byte> span, out int bytesRead)
9696
// Moved out of outer switch to have a shorter main method (improves performance)
9797
static object ReadFieldValueSlow(ReadOnlySpan<byte> span, out int bytesRead)
9898
{
99-
var slice = span.Slice(1);
99+
ReadOnlySpan<byte> slice = span.Slice(1);
100100
switch ((char)span[0])
101101
{
102102
case 'F':
103-
bytesRead = 1 + ReadDictionary(slice, out var dictionary);
103+
bytesRead = 1 + ReadDictionary(slice, out Dictionary<string, object> dictionary);
104104
return dictionary;
105105
case 'A':
106106
IList arrayResult = ReadArray(slice, out int arrayBytesRead);
@@ -134,10 +134,10 @@ static object ReadFieldValueSlow(ReadOnlySpan<byte> span, out int bytesRead)
134134
bytesRead = 3;
135135
return NetworkOrderDeserializer.ReadUInt16(slice);
136136
case 'T':
137-
bytesRead = 1 + ReadTimestamp(slice, out var timestamp);
137+
bytesRead = 1 + ReadTimestamp(slice, out AmqpTimestamp timestamp);
138138
return timestamp;
139139
case 'x':
140-
bytesRead = 1 + ReadLongstr(slice, out var binaryTableResult);
140+
bytesRead = 1 + ReadLongstr(slice, out byte[] binaryTableResult);
141141
return new BinaryTableValue(binaryTableResult);
142142
default:
143143
bytesRead = 0;

‎projects/RabbitMQ.Client/client/impl/WireFormatting.Write.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -326,7 +326,7 @@ public static int WriteShort(ref byte destination, ushort val)
326326
[MethodImpl(MethodImplOptions.AggressiveInlining)]
327327
public static int WriteShortstr(ref byte destination, ReadOnlySpan<byte> value)
328328
{
329-
var length = value.Length;
329+
int length = value.Length;
330330
if (length <= byte.MaxValue)
331331
{
332332
destination = (byte)length;

‎projects/Test/Integration/TestAsyncConsumer.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -54,11 +54,11 @@ public async Task TestBasicRoundtripConcurrent()
5454
{
5555
QueueDeclareOk q = await _channel.QueueDeclareAsync();
5656

57-
string publish1 = GetUniqueString(1024);
57+
string publish1 = GetUniqueString(512);
5858
byte[] body = _encoding.GetBytes(publish1);
5959
await _channel.BasicPublishAsync("", q.QueueName, body);
6060

61-
string publish2 = GetUniqueString(1024);
61+
string publish2 = GetUniqueString(512);
6262
body = _encoding.GetBytes(publish2);
6363
await _channel.BasicPublishAsync("", q.QueueName, body);
6464

@@ -141,9 +141,9 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
141141
const int publish_total = 4096;
142142
string queueName = GenerateQueueName();
143143

144-
string publish1 = GetUniqueString(32768);
144+
string publish1 = GetUniqueString(512);
145145
byte[] body1 = _encoding.GetBytes(publish1);
146-
string publish2 = GetUniqueString(32768);
146+
string publish2 = GetUniqueString(512);
147147
byte[] body2 = _encoding.GetBytes(publish2);
148148

149149
var publish1SyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

‎projects/Test/Unit/TestTimerBasedCredentialRefresher.cs‎

Lines changed: 35 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
//---------------------------------------------------------------------------
3131

3232
using System;
33+
using System.Threading;
3334
using System.Threading.Tasks;
3435
using RabbitMQ.Client;
3536
using Xunit;
@@ -129,60 +130,51 @@ public void TestDoNotRegisterWhenHasNoExpiry()
129130
[Fact]
130131
public async Task TestRefreshToken()
131132
{
132-
var cbtcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
133-
bool? callbackArg = null;
134-
var credentialsProvider = new MockCredentialsProvider(_testOutputHelper, TimeSpan.FromSeconds(1));
135-
Task cb(bool arg)
133+
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
134+
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)))
136135
{
137-
callbackArg = arg;
138-
cbtcs.SetResult(true);
139-
return Task.CompletedTask;
140-
}
141-
142-
try
143-
{
144-
_refresher.Register(credentialsProvider, cb);
145-
146-
await cbtcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
147-
Assert.True(await cbtcs.Task);
148-
149-
Assert.True(credentialsProvider.RefreshCalled);
150-
Assert.True(callbackArg);
151-
}
152-
finally
153-
{
154-
Assert.True(_refresher.Unregister(credentialsProvider));
136+
using (CancellationTokenRegistration ctr = cts.Token.Register(() => tcs.TrySetCanceled()))
137+
{
138+
var credentialsProvider = new MockCredentialsProvider(_testOutputHelper, TimeSpan.FromSeconds(1));
139+
140+
Task cb(bool arg)
141+
{
142+
tcs.SetResult(arg);
143+
return Task.CompletedTask;
144+
}
145+
146+
_refresher.Register(credentialsProvider, cb);
147+
Assert.True(await tcs.Task);
148+
Assert.True(credentialsProvider.RefreshCalled);
149+
Assert.True(_refresher.Unregister(credentialsProvider));
150+
}
155151
}
156152
}
157153

158154
[Fact]
159155
public async Task TestRefreshTokenFailed()
160156
{
161-
var cbtcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
162-
bool? callbackArg = null;
163-
var credentialsProvider = new MockCredentialsProvider(_testOutputHelper, TimeSpan.FromSeconds(1));
164-
Task cb(bool arg)
157+
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
158+
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)))
165159
{
166-
callbackArg = arg;
167-
cbtcs.SetResult(true);
168-
return Task.CompletedTask;
169-
}
160+
using (CancellationTokenRegistration ctr = cts.Token.Register(() => tcs.TrySetCanceled()))
161+
{
162+
var credentialsProvider = new MockCredentialsProvider(_testOutputHelper, TimeSpan.FromSeconds(1));
170163

171-
var ex = new Exception();
172-
credentialsProvider.PasswordThrows(ex);
164+
Task cb(bool arg)
165+
{
166+
tcs.SetResult(arg);
167+
return Task.CompletedTask;
168+
}
173169

174-
try
175-
{
176-
_refresher.Register(credentialsProvider, cb);
177-
await cbtcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
178-
Assert.True(await cbtcs.Task);
170+
var ex = new Exception();
171+
credentialsProvider.PasswordThrows(ex);
179172

180-
Assert.True(credentialsProvider.RefreshCalled);
181-
Assert.False(callbackArg);
182-
}
183-
finally
184-
{
185-
Assert.True(_refresher.Unregister(credentialsProvider));
173+
_refresher.Register(credentialsProvider, cb);
174+
Assert.False(await tcs.Task);
175+
Assert.True(credentialsProvider.RefreshCalled);
176+
Assert.True(_refresher.Unregister(credentialsProvider));
177+
}
186178
}
187179
}
188180
}

0 commit comments

Comments
 (0)