Skip to content

add missing queues configurations #110

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 17, 2025
Merged

add missing queues configurations #110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
03bbe15
add missing queues configurations
Gsantomaggio Mar 14, 2025
43323c2
add missing queues configurations
Gsantomaggio Mar 14, 2025
6a38094
format
Gsantomaggio Mar 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion RabbitMQ.AMQP.Client/IEntities.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,18 @@ public enum OverFlowStrategy
{
DropHead,
RejectPublish,

RejectPublishDlx
// DROP_HEAD("drop-head"),
// REJECT_PUBLISH("reject-publish"),
// REJECT_PUBLISH_DLX("reject-publish-dlx");
}

public enum LeaderLocatorStrategy
{
ClientLocal,
Balanced
}

public interface IQueueSpecification : IEntityInfoSpecification<IQueueInfo>
{
public string QueueName { get; }
Expand All @@ -67,6 +72,8 @@ public interface IQueueSpecification : IEntityInfoSpecification<IQueueInfo>

IQueueSpecification MaxLengthBytes(ByteCapacity maxLengthBytes);

IQueueSpecification LeaderLocator(LeaderLocatorStrategy strategy);

// TODO: Add more tests for SingleActiveConsumer
IQueueSpecification SingleActiveConsumer(bool singleActiveConsumer);

Expand All @@ -93,6 +100,8 @@ public interface IStreamSpecification

public IStreamSpecification InitialClusterSize(int initialClusterSize);

public IStreamSpecification FileSizePerChunk(ByteCapacity fileSizePerChunk);

public IQueueSpecification Queue();
}

Expand All @@ -112,6 +121,8 @@ public interface IQuorumQueueSpecification

IQuorumQueueSpecification QuorumInitialGroupSize(int size);

IQuorumQueueSpecification QuorumTargetGroupSize(int size);

IQueueSpecification Queue();
}

Expand Down
25 changes: 25 additions & 0 deletions RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,17 @@ public IQueueSpecification MaxLengthBytes(ByteCapacity maxLengthBytes)
return this;
}

public IQueueSpecification LeaderLocator(LeaderLocatorStrategy leaderLocatorStrategy)
{
_queueArguments["x-queue-leader-locator"] = leaderLocatorStrategy switch
{
LeaderLocatorStrategy.ClientLocal => "client-local",
LeaderLocatorStrategy.Balanced => "balanced",
_ => throw new ArgumentOutOfRangeException(nameof(leaderLocatorStrategy), leaderLocatorStrategy, null)
};
return this;
}

public IQueueSpecification SingleActiveConsumer(bool singleActiveConsumer)
{
_queueArguments["x-single-active-consumer"] = singleActiveConsumer;
Expand Down Expand Up @@ -397,6 +408,13 @@ public IStreamSpecification InitialClusterSize(int initialClusterSize)
return this;
}

public IStreamSpecification FileSizePerChunk(ByteCapacity fileSizePerChunk)
{
Utils.ValidatePositive("x-stream-file-size-per-chunk", fileSizePerChunk);
_parent._queueArguments["x-stream-file-size-per-chunk"] = (long)fileSizePerChunk;
return this;
}

public IQueueSpecification Queue()
{
return _parent;
Expand Down Expand Up @@ -437,6 +455,13 @@ public IQuorumQueueSpecification QuorumInitialGroupSize(int size)
return this;
}

public IQuorumQueueSpecification QuorumTargetGroupSize(int size)
{
Utils.ValidatePositive("x-quorum-target-group-size", size);
_parent._queueArguments["x-quorum-target-group-size"] = size;
return this;
}

public IQueueSpecification Queue()
{
return _parent;
Expand Down
9 changes: 9 additions & 0 deletions RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,7 @@ RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.Exclusive(bool isExclusive) ->
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.Expires(System.TimeSpan expiration) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.IsAutoDelete.get -> bool
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.IsExclusive.get -> bool
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.LeaderLocator(RabbitMQ.AMQP.Client.LeaderLocatorStrategy leaderLocatorStrategy) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.MaxLength(long maxLength) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.MaxLengthBytes(RabbitMQ.AMQP.Client.ByteCapacity! maxLengthBytes) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification.MessageTtl(System.TimeSpan ttl) -> RabbitMQ.AMQP.Client.IQueueSpecification!
Expand All @@ -474,6 +475,7 @@ RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.DeadLetterStrategy(RabbitMQ.AM
RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.DeliveryLimit(int limit) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.Queue() -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.QuorumInitialGroupSize(int size) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.QuorumTargetGroupSize(int size) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpRpcClient
RabbitMQ.AMQP.Client.Impl.AmqpRpcClient.AmqpRpcClient(RabbitMQ.AMQP.Client.Impl.RpcClientConfiguration! configuration) -> void
RabbitMQ.AMQP.Client.Impl.AmqpRpcClient.PublishAsync(RabbitMQ.AMQP.Client.IMessage! message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IMessage!>!
Expand All @@ -499,6 +501,7 @@ RabbitMQ.AMQP.Client.Impl.AmqpRpcServerBuilder.RequestQueue(RabbitMQ.AMQP.Client
RabbitMQ.AMQP.Client.Impl.AmqpRpcServerBuilder.RequestQueue(string! requestQueue) -> RabbitMQ.AMQP.Client.IRpcServerBuilder!
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification.AmqpStreamSpecification(RabbitMQ.AMQP.Client.Impl.AmqpQueueSpecification! parent) -> void
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification.FileSizePerChunk(RabbitMQ.AMQP.Client.ByteCapacity! fileSizePerChunk) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification.InitialClusterSize(int initialClusterSize) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification.MaxAge(System.TimeSpan maxAge) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.Impl.AmqpStreamSpecification.MaxSegmentSizeBytes(RabbitMQ.AMQP.Client.ByteCapacity! maxSegmentSize) -> RabbitMQ.AMQP.Client.IStreamSpecification!
Expand Down Expand Up @@ -631,6 +634,7 @@ RabbitMQ.AMQP.Client.IQueueSpecification.Exclusive(bool isExclusive) -> RabbitMQ
RabbitMQ.AMQP.Client.IQueueSpecification.Expires(System.TimeSpan expiration) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IQueueSpecification.IsAutoDelete.get -> bool
RabbitMQ.AMQP.Client.IQueueSpecification.IsExclusive.get -> bool
RabbitMQ.AMQP.Client.IQueueSpecification.LeaderLocator(RabbitMQ.AMQP.Client.LeaderLocatorStrategy strategy) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IQueueSpecification.MaxLength(long maxLength) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IQueueSpecification.MaxLengthBytes(RabbitMQ.AMQP.Client.ByteCapacity! maxLengthBytes) -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IQueueSpecification.MessageTtl(System.TimeSpan ttl) -> RabbitMQ.AMQP.Client.IQueueSpecification!
Expand All @@ -649,6 +653,7 @@ RabbitMQ.AMQP.Client.IQuorumQueueSpecification.DeadLetterStrategy(RabbitMQ.AMQP.
RabbitMQ.AMQP.Client.IQuorumQueueSpecification.DeliveryLimit(int limit) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.IQuorumQueueSpecification.Queue() -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IQuorumQueueSpecification.QuorumInitialGroupSize(int size) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.IQuorumQueueSpecification.QuorumTargetGroupSize(int size) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
RabbitMQ.AMQP.Client.IRecoveryConfiguration
RabbitMQ.AMQP.Client.IRecoveryConfiguration.Activated(bool activated) -> RabbitMQ.AMQP.Client.IRecoveryConfiguration!
RabbitMQ.AMQP.Client.IRecoveryConfiguration.BackOffDelayPolicy(RabbitMQ.AMQP.Client.IBackOffDelayPolicy! backOffDelayPolicy) -> RabbitMQ.AMQP.Client.IRecoveryConfiguration!
Expand Down Expand Up @@ -680,12 +685,16 @@ RabbitMQ.AMQP.Client.IRpcServerBuilder.ReplyPostProcessor(System.Func<RabbitMQ.A
RabbitMQ.AMQP.Client.IRpcServerBuilder.RequestQueue(RabbitMQ.AMQP.Client.IQueueSpecification! requestQueue) -> RabbitMQ.AMQP.Client.IRpcServerBuilder!
RabbitMQ.AMQP.Client.IRpcServerBuilder.RequestQueue(string! requestQueue) -> RabbitMQ.AMQP.Client.IRpcServerBuilder!
RabbitMQ.AMQP.Client.IStreamSpecification
RabbitMQ.AMQP.Client.IStreamSpecification.FileSizePerChunk(RabbitMQ.AMQP.Client.ByteCapacity! fileSizePerChunk) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.IStreamSpecification.InitialClusterSize(int initialClusterSize) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.IStreamSpecification.MaxAge(System.TimeSpan maxAge) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.IStreamSpecification.MaxSegmentSizeBytes(RabbitMQ.AMQP.Client.ByteCapacity! maxSegmentSize) -> RabbitMQ.AMQP.Client.IStreamSpecification!
RabbitMQ.AMQP.Client.IStreamSpecification.Queue() -> RabbitMQ.AMQP.Client.IQueueSpecification!
RabbitMQ.AMQP.Client.IUriSelector
RabbitMQ.AMQP.Client.IUriSelector.Select(System.Collections.Generic.ICollection<System.Uri!>! uris) -> System.Uri!
RabbitMQ.AMQP.Client.LeaderLocatorStrategy
RabbitMQ.AMQP.Client.LeaderLocatorStrategy.Balanced = 1 -> RabbitMQ.AMQP.Client.LeaderLocatorStrategy
RabbitMQ.AMQP.Client.LeaderLocatorStrategy.ClientLocal = 0 -> RabbitMQ.AMQP.Client.LeaderLocatorStrategy
RabbitMQ.AMQP.Client.LifeCycleCallBack
RabbitMQ.AMQP.Client.MessageHandler
RabbitMQ.AMQP.Client.MetricsReporter
Expand Down
10 changes: 10 additions & 0 deletions Tests/Management/ManagementTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,17 +151,23 @@ public async Task DeclareStreamQueueWithArguments()

IQueueInfo queueInfo = await _management.Queue()
.Name(_queueName)
.MaxLengthBytes(ByteCapacity.Kb(1024))
.LeaderLocator(LeaderLocatorStrategy.Balanced)
.Stream()
.MaxAge(TimeSpan.FromSeconds(10))
.MaxSegmentSizeBytes(ByteCapacity.Kb(1024))
.InitialClusterSize(1)
.FileSizePerChunk(ByteCapacity.Kb(1024))
.Queue()
.DeclareAsync();

Assert.Equal(_queueName, queueInfo.Name());
Assert.Equal("10s", queueInfo.Arguments()["x-max-age"]);
Assert.Equal(1024000L, queueInfo.Arguments()["x-stream-max-segment-size-bytes"]);
Assert.Equal(1, queueInfo.Arguments()["x-initial-cluster-size"]);
Assert.Equal("balanced", queueInfo.Arguments()["x-queue-leader-locator"]);
Assert.Equal(1024000L, queueInfo.Arguments()["x-stream-file-size-per-chunk"]);
Assert.Equal(1024000L, queueInfo.Arguments()["x-max-length-bytes"]);
// NB: DisposeAsync will delete the queue with name _queueName
}

Expand All @@ -173,17 +179,21 @@ public async Task DeclareQuorumQueueWithArguments()

IQueueInfo queueInfo = await _management.Queue()
.Name(_queueName)
.LeaderLocator(LeaderLocatorStrategy.ClientLocal)
.Quorum()
.DeliveryLimit(12)
.DeadLetterStrategy(QuorumQueueDeadLetterStrategy.AtLeastOnce)
.QuorumInitialGroupSize(3)
.QuorumTargetGroupSize(5)
.Queue()
.DeclareAsync();

Assert.Equal(_queueName, queueInfo.Name());
Assert.Equal(12, queueInfo.Arguments()["x-max-delivery-limit"]);
Assert.Equal("at-least-once", queueInfo.Arguments()["x-dead-letter-strategy"]);
Assert.Equal(3, queueInfo.Arguments()["x-quorum-initial-group-size"]);
Assert.Equal(5, queueInfo.Arguments()["x-quorum-target-group-size"]);
Assert.Equal("client-local", queueInfo.Arguments()["x-queue-leader-locator"]);
// NB: DisposeAsync will delete the queue with name _queueName
}

Expand Down