Implemented RabbitMQ source connector. #31
Conversation
| String connectionName = config.getString(CONNECTION_NAME); | ||
| String queueName = config.getString(QUEUE_NAME); | ||
| if (consumer == null) { | ||
| throw new IllegalArgumentException("consumeFunction cannot be null."); |
There was a problem hiding this comment.
Can this really happen this late in the initialization?
There was a problem hiding this comment.
Yes, this can happen when a user of the connectors framework creates RabbitMQSourceConnector object, doesn't invoke setConsumer method and invokes open method. Probably, it'd make sense to create only one constructor for RabbitMQSourceConnector such as:
public RabbitMQSourceConnector(Consumer<Collection<byte[]>> consumeFunction) {
//...
}There was a problem hiding this comment.
I see - I think an example usecase would be helpful to see it in action.
| rabbitMQChannel.queueDeclare(queueName, false, false, false, null); | ||
| com.rabbitmq.client.Consumer consumer = new RabbitMQConsumer(this.consumer, rabbitMQChannel); | ||
| rabbitMQChannel.basicConsume(queueName, consumer); | ||
|
|
There was a problem hiding this comment.
Log line that connection has been opened with some details might be helpful for debugging any issues that come up.
Well, there seems to be nothing to test with unit tests. As for integration test, I think they should be added along with final distributed implementation. Correct me if I'm mistaken. |
|
This looks to be fine for me any concerns before we merge it , tests can be discussed later |
|
am merging this for now |
No description provided.