Conduit Connector for RabbitMQ

The RabbitMQ connector is one of Conduit standalone plugins. It provides both a source and a destination connector for RabbitMQ.

It uses the AMQP 0-9-1 Model to connect to RabbitMQ.

What data does the OpenCDC record consist of?

Field	Description
record.Position	json object with the delivery tag and the queue name from where the record was read from.
record.Operation	currently fixed as "create".
record.Metadata	a string to string map, with keys prefixed as rabbitmq.{DELIVERY_PROPERTY}.
record.Key	the message id from the read message.
record.Payload.Before
record.Payload.After	the message body

How to Build?

Run make build to compile the connector.

Testing

Execute make test to perform all non-tls tests. Execute make test-tls for the TLS tests. Both commands use docker files located at test/docker-compose.yml and test/docker-compose-tls.yml respectively. Tests require docker-compose v2.

Source Configuration Parameters

Name	Description	Required	Default Value
url	The RabbitMQ server's URL.	Yes
tls.enabled	Flag to enable or disable TLS.	false	false
tls.clientCert	Path to the client certificate for TLS.	No
tls.clientKey	Path to the client's key for TLS.	No
tls.caCert	Path to the CA (Certificate Authority) certificate for TLS.	No
queue.name	The name of the RabbitMQ queue to consume messages from.	Yes
queue.durable	Specifies whether the queue is durable.	No	true
queue.autoDelete	If the queue will auto-delete.	No	false
queue.exclusive	If the queue is exclusive.	No	false
queue.noWait	If the queue is declared without waiting for server reply.	No	false
consumer.name	The name of the consumer.	No
consumer.autoAck	If the server should consider messages acknowledged once delivered.	No	false
consumer.exclusive	If the consumer should be exclusive.	No	false
consumer.noLocal	If the server should not deliver messages published by the same connection.	No	false
consumer.noWait	If the consumer should be declared without waiting for server confirmation.	No	false

Destination Configuration Parameters

Name	Description	Required	Default Value
url	The RabbitMQ server's URL.	Yes
tls.enabled	Flag to enable or disable TLS.	false	false
tls.clientCert	Path to the client certificate for TLS.	No
tls.clientKey	Path to the client's key for TLS.	No
tls.caCert	Path to the CA (Certificate Authority) certificate for TLS.	No
queue.name	The name of the RabbitMQ queue where messages will be published to.	Yes
queue.durable	Specifies whether the queue is durable.	No	true
queue.autoDelete	If the queue will auto-delete.	No	false
queue.exclusive	If the queue is exclusive.	No	false
queue.noWait	If the queue is declared without waiting for server reply.	No	false
contentType	The MIME content type of the messages written to RabbitMQ.	No	text/plain
delivery.contentEncoding	The content encoding for the message.	No
delivery.deliveryMode	Delivery mode of the message. Non-persistent (1) or persistent (2).	No	2
delivery.priority	The priority of the message.	No	0
delivery.correlationID	The correlation id associated with the message.	No
delivery.replyTo	Address to reply to.	No
delivery.messageTypeName	The type name of the message.	No
delivery.userID	The user id associated with the message.	No
delivery.appID	The application id associated with the message.	No
delivery.mandatory	Indicates if this message is mandatory.	No	false
delivery.immediate	Indicates if this message should be treated as immediate.	No	false
delivery.expiration	Indicates the message expiration time, if any.	No
exchange.name	The name of the exchange to publish to.	No
exchange.type	The type of the exchange to publish to.	No	direct
exchange.durable	Specifies whether the exchange is durable.	No	true
exchange.autoDelete	If the exchange will auto-delete.	No	false
exchange.internal	If the exchange is internal.	No	false
exchange.noWait	If the exchange is declared without waiting for server reply.	No	false
routingKey	The routing key to use when publishing to an exchange.	No

Example pipeline.yml file

Here's an example of a pipeline.yml file using file to RabbitMQ and RabbitMQ to file pipelines:

version: 2.0
pipelines:
  - id: file-to-rabbitmq
    status: running
    connectors:
      - id: file.in
        type: source
        plugin: builtin:file
        name: file-destination
        settings:
          path: ./file.in
      - id: rabbitmq.out
        type: destination
        plugin: standalone:rabbitmq
        name: rabbitmq-source
        settings:
          url: amqp://guest:guest@localhost:5672/
          queue.name: demo-queue
          sdk.record.format: template
          sdk.record.format.options: '{{ printf "%s" .Payload.After }}'

  - id: rabbitmq-to-file
    status: running
    connectors:
      - id: rabbitmq.in
        type: source
        plugin: standalone:rabbitmq
        name: rabbitmq-source
        settings:
          url: amqp://guest:guest@localhost:5672/
          queue.name: demo-queue

      - id: file.out
        type: destination
        plugin: builtin:file
        name: file-destination
        settings:
          path: ./file.out
          sdk.record.format: template
          sdk.record.format.options: '{{ printf "%s" .Payload.After }}'