efcasado commented on code in PR #90:
URL: https://github.com/apache/pulsar-connectors/pull/90#discussion_r3563528010
##########
rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java:
##########
@@ -64,19 +66,33 @@ public void open(Map<String, Object> config, SourceContext
sourceContext) throws
rabbitMQConnection.getPort()
);
rabbitMQChannel = rabbitMQConnection.createChannel();
+ AMQP.Queue.DeclareOk queueDeclaration;
if (rabbitMQSourceConfig.isPassive()) {
-
rabbitMQChannel.queueDeclarePassive(rabbitMQSourceConfig.getQueueName());
+ queueDeclaration =
rabbitMQChannel.queueDeclarePassive(rabbitMQSourceConfig.getQueueName());
} else {
- rabbitMQChannel.queueDeclare(rabbitMQSourceConfig.getQueueName(),
false, false, false, null);
+ queueDeclaration =
rabbitMQChannel.queueDeclare(rabbitMQSourceConfig.getQueueName(),
+
rabbitMQSourceConfig.isDurable(),
+
rabbitMQSourceConfig.isExclusive(),
+
rabbitMQSourceConfig.isAutoDelete(),
+ null
+ );
}
+ queueName = queueDeclaration.getQueue();
logger.info("Setting channel.basicQos({}, {}).",
rabbitMQSourceConfig.getPrefetchCount(),
rabbitMQSourceConfig.isPrefetchGlobal()
);
rabbitMQChannel.basicQos(rabbitMQSourceConfig.getPrefetchCount(),
rabbitMQSourceConfig.isPrefetchGlobal());
+ String exchange = rabbitMQSourceConfig.getExchangeName();
+ String routingKey = rabbitMQSourceConfig.getRoutingKey();
+ if (exchange != null && !exchange.isEmpty()) {
+ // note: the exchange is expected to already exist; it is not
declared here
+ rabbitMQChannel.queueBind(queueName, exchange, routingKey);
+ logger.info("Bound queue {} to exchange {} with routing key {}.",
queueName, exchange, routingKey);
+ }
com.rabbitmq.client.Consumer consumer = new RabbitMQConsumer(this,
rabbitMQChannel);
- rabbitMQChannel.basicConsume(rabbitMQSourceConfig.getQueueName(),
consumer);
- logger.info("A consumer for queue {} has been successfully started.",
rabbitMQSourceConfig.getQueueName());
+ rabbitMQChannel.basicConsume(queueName, consumer);
+ logger.info("A consumer for queue {} has been successfully started.",
queueName);
}
Review Comment:
This issue wasn't introduced by our changes. I guess Copilot raised it
because we changed the line to be able to log the right queue name when using
server-generated queues. That is, when the configured queue name is `""` and we
let the server generate the queue name.
I am leaning towards ignoring this for now, since so far the goal of this
pull request was to be as backwards-compatible as possible, but I am happy to
give this a go if you deem it necessary 😄
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
