efcasado commented on code in PR #90:
URL: https://github.com/apache/pulsar-connectors/pull/90#discussion_r3576391781
##########
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:
Thanks @david-streamlio!
Just pushed a commit that should address both issues 🤞
--
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]
