Skip to content

Commit cffe0fd

Browse files
lukebakkenlukebakken
committed
* Make CreateChannelOptions an immutable class.
1 parent 4a65bd9 commit cffe0fd

18 files changed

+79
-57
lines changed

‎projects/Applications/GH-1647/Program.cs

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,11 +40,7 @@
4040
Password = "guest"
4141
};
4242

43-
var channelOptions = new CreateChannelOptions
44-
{
45-
PublisherConfirmationsEnabled = true,
46-
PublisherConfirmationTrackingEnabled = true
47-
};
43+
var channelOptions = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true);
4844

4945
var props = new BasicProperties();
5046
byte[] msg = Encoding.UTF8.GetBytes("test");

‎projects/Applications/MassPublish/Program.cs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,9 @@ await consumeChannel.BasicConsumeAsync(queue: QueueName, autoAck: true, consumer
137137

138138
publishTasks.Add(Task.Run(async () =>
139139
{
140-
using IChannel publishChannel = await publishConnection.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
140+
var createChannelOptions = new CreateChannelOptions(publisherConfirmationsEnabled: true,
141+
publisherConfirmationTrackingEnabled: true);
142+
using IChannel publishChannel = await publishConnection.CreateChannelAsync(createChannelOptions);
141143
publishChannel.ChannelShutdownAsync += Channel_ChannelShutdownAsync;
142144

143145
for (int i = 0; i < ItemsPerBatch; i++)

‎projects/Applications/PublisherConfirms/PublisherConfirms.cs

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,11 @@
4343
const int MESSAGE_COUNT = 50_000;
4444
bool debug = false;
4545

46-
var channelOpts = new CreateChannelOptions
47-
{
48-
PublisherConfirmationsEnabled = true,
49-
PublisherConfirmationTrackingEnabled = true,
50-
OutstandingPublisherConfirmationsRateLimiter = new ThrottlingRateLimiter(MAX_OUTSTANDING_CONFIRMS)
51-
};
46+
var channelOpts = new CreateChannelOptions(
47+
publisherConfirmationsEnabled: true,
48+
publisherConfirmationTrackingEnabled: true,
49+
outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(MAX_OUTSTANDING_CONFIRMS)
50+
);
5251

5352
var props = new BasicProperties
5453
{
@@ -177,7 +176,7 @@ async Task HandlePublishConfirmsAsynchronously()
177176

178177
await using IConnection connection = await CreateConnectionAsync();
179178

180-
channelOpts.PublisherConfirmationTrackingEnabled = false;
179+
channelOpts = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: false);
181180
await using IChannel channel = await connection.CreateChannelAsync(channelOpts);
182181

183182
// declare a server-named queue

‎projects/RabbitMQ.Client/CreateChannelOptions.cs

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public sealed class CreateChannelOptions
5353
/// <see cref="IChannel.GetNextPublishSequenceNumberAsync(System.Threading.CancellationToken)"/> to allow correlation
5454
/// of the response with the correct message.
5555
/// </summary>
56-
public bool PublisherConfirmationsEnabled { get; set; } = false;
56+
public readonly bool PublisherConfirmationsEnabled = false;
5757

5858
/// <summary>
5959
/// Should this library track publisher confirmations for you? Defaults to <c>false</c>
@@ -63,7 +63,7 @@ public sealed class CreateChannelOptions
6363
/// If the broker then sends a <c>basic.return</c> response for the message, this library can
6464
/// then correctly handle the message.
6565
/// </summary>
66-
public bool PublisherConfirmationTrackingEnabled { get; set; } = false;
66+
public readonly bool PublisherConfirmationTrackingEnabled = false;
6767

6868
/// <summary>
6969
/// If the publisher confirmation tracking is enabled, this represents the rate limiter used to
@@ -72,7 +72,7 @@ public sealed class CreateChannelOptions
7272
/// Defaults to a <see cref="ThrottlingRateLimiter"/> with a limit of 128 and a throttling percentage of 50% with a delay during throttling.
7373
/// </summary>
7474
/// <remarks>Setting the rate limiter to <c>null</c> disables the rate limiting entirely.</remarks>
75-
public RateLimiter? OutstandingPublisherConfirmationsRateLimiter { get; set; } = new ThrottlingRateLimiter(128);
75+
public readonly RateLimiter? OutstandingPublisherConfirmationsRateLimiter = new ThrottlingRateLimiter(128);
7676

7777
/// <summary>
7878
/// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one <see cref="IAsyncBasicConsumer"/>
@@ -84,10 +84,17 @@ public sealed class CreateChannelOptions
8484
/// For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
8585
/// In addition to that consumers need to be thread/concurrency safe.
8686
/// </summary>
87-
public ushort? ConsumerDispatchConcurrency { get; set; } = null;
87+
public readonly ushort? ConsumerDispatchConcurrency = null;
8888

89-
public CreateChannelOptions()
89+
public CreateChannelOptions(bool publisherConfirmationsEnabled,
90+
bool publisherConfirmationTrackingEnabled,
91+
RateLimiter? outstandingPublisherConfirmationsRateLimiter = null,
92+
ushort? consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency)
9093
{
94+
PublisherConfirmationsEnabled = publisherConfirmationsEnabled;
95+
PublisherConfirmationTrackingEnabled = publisherConfirmationTrackingEnabled;
96+
OutstandingPublisherConfirmationsRateLimiter = outstandingPublisherConfirmationsRateLimiter;
97+
ConsumerDispatchConcurrency = consumerDispatchConcurrency;
9198
}
9299

93100
internal ushort InternalConsumerDispatchConcurrency
@@ -116,22 +123,22 @@ internal CreateChannelOptions(ConnectionConfig connectionConfig)
116123
_connectionConfigContinuationTimeout = connectionConfig.ContinuationTimeout;
117124
}
118125

119-
private void WithConnectionConfig(ConnectionConfig connectionConfig)
126+
private CreateChannelOptions WithConnectionConfig(ConnectionConfig connectionConfig)
120127
{
121128
_connectionConfigConsumerDispatchConcurrency = connectionConfig.ConsumerDispatchConcurrency;
122129
_connectionConfigContinuationTimeout = connectionConfig.ContinuationTimeout;
130+
return this;
123131
}
124132

125133
internal static CreateChannelOptions CreateOrUpdate(CreateChannelOptions? createChannelOptions, ConnectionConfig config)
126134
{
127-
if (createChannelOptions is not null)
135+
if (createChannelOptions is null)
128136
{
129-
createChannelOptions.WithConnectionConfig(config);
130-
return createChannelOptions;
137+
return new CreateChannelOptions(config);
131138
}
132139
else
133140
{
134-
return new CreateChannelOptions(config);
141+
return createChannelOptions.WithConnectionConfig(config);
135142
}
136143
}
137144
}

‎projects/RabbitMQ.Client/PublicAPI.Shipped.txt

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -921,3 +921,8 @@ static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client
921921
static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync<T>(this RabbitMQ.Client.IChannel! channel, RabbitMQ.Client.PublicationAddress! addr, T basicProperties, System.ReadOnlyMemory<byte> body, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
922922
static readonly RabbitMQ.Client.Constants.DefaultContinuationTimeout -> System.TimeSpan
923923
static readonly RabbitMQ.Client.Constants.DefaultHandshakeContinuationTimeout -> System.TimeSpan
924+
RabbitMQ.Client.CreateChannelOptions.CreateChannelOptions(bool publisherConfirmationsEnabled, bool publisherConfirmationTrackingEnabled, System.Threading.RateLimiting.RateLimiter? outstandingPublisherConfirmationsRateLimiter = null, ushort? consumerDispatchConcurrency = 1) -> void
925+
readonly RabbitMQ.Client.CreateChannelOptions.ConsumerDispatchConcurrency -> ushort?
926+
readonly RabbitMQ.Client.CreateChannelOptions.OutstandingPublisherConfirmationsRateLimiter -> System.Threading.RateLimiting.RateLimiter?
927+
readonly RabbitMQ.Client.CreateChannelOptions.PublisherConfirmationsEnabled -> bool
928+
readonly RabbitMQ.Client.CreateChannelOptions.PublisherConfirmationTrackingEnabled -> bool

‎projects/Test/Common/IntegrationFixture.cs

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,9 @@ public abstract class IntegrationFixture : IAsyncLifetime
6666
protected IConnection _conn;
6767
protected IChannel _channel;
6868

69+
protected static readonly CreateChannelOptions _createChannelOptions = new(publisherConfirmationsEnabled: true,
70+
publisherConfirmationTrackingEnabled: true);
71+
6972
protected static readonly Encoding _encoding = new UTF8Encoding();
7073
protected static readonly int _processorCount = Environment.ProcessorCount;
7174

@@ -153,7 +156,7 @@ public virtual async Task InitializeAsync()
153156

154157
if (_openChannel)
155158
{
156-
_channel = await _conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
159+
_channel = await _conn.CreateChannelAsync(_createChannelOptions);
157160
}
158161

159162
if (IsVerbose)

‎projects/Test/Common/TestConnectionRecoveryBase.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -199,7 +199,7 @@ protected async Task PublishMessagesWhileClosingConnAsync(string queueName)
199199
{
200200
using (AutorecoveringConnection publishingConn = await CreateAutorecoveringConnectionAsync())
201201
{
202-
using (IChannel publishingChannel = await publishingConn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true }))
202+
using (IChannel publishingChannel = await publishingConn.CreateChannelAsync(_createChannelOptions))
203203
{
204204
for (ushort i = 0; i < TotalMessageCount; i++)
205205
{
@@ -342,7 +342,7 @@ public virtual Task PostHandleDeliveryAsync(ulong deliveryTag,
342342

343343
protected static async Task<bool> SendAndConsumeMessageAsync(IConnection conn, string queue, string exchange, string routingKey)
344344
{
345-
using (IChannel ch = await conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true }))
345+
using (IChannel ch = await conn.CreateChannelAsync(_createChannelOptions))
346346
{
347347
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
348348

‎projects/Test/Integration/ConnectionRecovery/TestConnectionRecovery.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ await _channel.ExchangeDeclareAsync(exchange: exchangeName,
9696
consumer.ReceivedAsync += MessageReceived;
9797
await _channel.BasicConsumeAsync(queueName, true, consumer);
9898

99-
await using (IChannel pubCh = await _conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true }))
99+
await using (IChannel pubCh = await _conn.CreateChannelAsync(_createChannelOptions))
100100
{
101101
await pubCh.BasicPublishAsync(exchange: exchangeName, routingKey: routingKey, body: body);
102102
await pubCh.CloseAsync();
@@ -106,7 +106,7 @@ await _channel.ExchangeDeclareAsync(exchange: exchangeName,
106106

107107
await CloseAndWaitForRecoveryAsync();
108108

109-
await using (IChannel pubCh = await _conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true }))
109+
await using (IChannel pubCh = await _conn.CreateChannelAsync(_createChannelOptions))
110110
{
111111
await pubCh.BasicPublishAsync(exchange: exchangeName, routingKey: "unused", body: body);
112112
await pubCh.CloseAsync();

‎projects/Test/Integration/TestAsyncConsumer.cs

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -213,7 +213,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
213213
});
214214
return Task.CompletedTask;
215215
};
216-
await using (IChannel publishChannel = await publishConn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true }))
216+
await using (IChannel publishChannel = await publishConn.CreateChannelAsync(_createChannelOptions))
217217
{
218218
AddCallbackExceptionHandlers(publishConn, publishChannel);
219219
publishChannel.DefaultConsumer = new DefaultAsyncConsumer(publishChannel,
@@ -646,7 +646,7 @@ public async Task TestCreateChannelWithinAsyncConsumerCallback_GH650()
646646
var consumer1 = new AsyncEventingBasicConsumer(_channel);
647647
consumer1.ReceivedAsync += async (sender, args) =>
648648
{
649-
await using IChannel innerChannel = await _conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
649+
await using IChannel innerChannel = await _conn.CreateChannelAsync(_createChannelOptions);
650650
await innerChannel.BasicPublishAsync(exchangeName, queue2Name,
651651
mandatory: true,
652652
body: Encoding.ASCII.GetBytes(nameof(TestCreateChannelWithinAsyncConsumerCallback_GH650)));
@@ -707,9 +707,15 @@ private async Task ValidateConsumerDispatchConcurrency()
707707
AutorecoveringChannel autorecoveringChannel = (AutorecoveringChannel)_channel;
708708
Assert.Equal(ConsumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
709709
Assert.Equal(_consumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
710-
await using IChannel ch = await _conn.CreateChannelAsync(
711-
new CreateChannelOptions { ConsumerDispatchConcurrency = expectedConsumerDispatchConcurrency });
710+
711+
var createChannelOptions = new CreateChannelOptions(publisherConfirmationsEnabled: false,
712+
publisherConfirmationTrackingEnabled: false,
713+
outstandingPublisherConfirmationsRateLimiter: null,
714+
consumerDispatchConcurrency: expectedConsumerDispatchConcurrency);
715+
716+
await using IChannel ch = await _conn.CreateChannelAsync(createChannelOptions);
712717
AutorecoveringChannel ach = (AutorecoveringChannel)ch;
718+
713719
Assert.Equal(expectedConsumerDispatchConcurrency, ach.ConsumerDispatcher.Concurrency);
714720
}
715721

‎projects/Test/Integration/TestAsyncEventingBasicConsumer.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ public async Task TestAsyncEventingBasicConsumer_GH1038()
105105
await _channel.BasicConsumeAsync(queueName, false, consumer);
106106

107107
//publisher
108-
await using IChannel publisherChannel = await _conn.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
108+
await using IChannel publisherChannel = await _conn.CreateChannelAsync(_createChannelOptions);
109109
byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
110110
var props = new BasicProperties();
111111
await publisherChannel.BasicPublishAsync(exchange: exchangeName, routingKey: string.Empty,

0 commit comments

Comments
 (0)