Skip to content
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

RabbitMQ connector consumer attributes : consumer-tag consumer-exclusive #2826

Merged
merged 1 commit into from
Nov 20, 2024
Merged

RabbitMQ connector consumer attributes : consumer-tag consumer-exclusive #2826

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions ...ing-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/RabbitMQConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,8 @@
@ConnectorAttribute(name = "routing-keys", direction = INCOMING, description = "A comma-separated list of routing keys to bind the queue to the exchange. Relevant only if 'exchange.type' is topic or direct", type = "string", defaultValue = "#")
@ConnectorAttribute(name = "arguments", direction = INCOMING, description = "A comma-separated list of arguments [key1:value1,key2:value2,...] to bind the queue to the exchange. Relevant only if 'exchange.type' is headers", type = "string")
@ConnectorAttribute(name = "consumer-arguments", direction = INCOMING, description = "A comma-separated list of arguments [key1:value1,key2:value2,...] for created consumer", type = "string")
@ConnectorAttribute(name = "consumer-tag", direction = INCOMING, description = "The consumer-tag option for created consumer, if not provided the consumer gets assigned a tag generated by the broker", type = "string")
@ConnectorAttribute(name = "consumer-exclusive", direction = INCOMING, description = "The exclusive flag for created consumer", type = "boolean")
@ConnectorAttribute(name = "content-type-override", direction = INCOMING, description = "Override the content_type attribute of the incoming message, should be a valid MINE type", type = "string")
@ConnectorAttribute(name = "max-outstanding-messages", direction = INCOMING, description = "The maximum number of outstanding/unacknowledged messages being processed by the connector at a time; must be a positive number", type = "int")

Expand Down
2 changes: 2 additions & 0 deletions .../main/java/io/smallrye/reactive/messaging/rabbitmq/internals/IncomingRabbitMQChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,8 @@ private Uni<String> declareQueue(
private Uni<RabbitMQConsumer> createConsumer(RabbitMQConnectorIncomingConfiguration ic, RabbitMQClient client) {
QueueOptions queueOptions = new QueueOptions();
queueOptions.setConsumerArguments(parseArguments(ic.getConsumerArguments()));
ic.getConsumerTag().ifPresent(queueOptions::setConsumerTag);
ic.getConsumerExclusive().ifPresent(queueOptions::setConsumerExclusive);
return client.basicConsumer(serverQueueName(RabbitMQClientHelper.getQueueName(ic)), queueOptions
.setAutoAck(ic.getAutoAcknowledgement())
.setMaxInternalQueueSize(ic.getMaxIncomingInternalQueueSize())
Expand Down