Skip to main content

rabbitmq (Sink)

The rabbitmq sink pushes the events into a rabbitmq broker using the AMQP protocol

Syntax

CREATE SINK <NAME> WITH (type="rabbitmq", map.type="<STRING>", uri="<STRING>", heartbeat="<INT>", exchange.name="<STRING>", exchange.type="<STRING>", exchange.durable.enabled="<BOOL>", exchange.autodelete.enabled="<BOOL>", delivery.mode="<INT>", content.type="<STRING>", content.encoding="<STRING>", priority="<INT>", correlation.id="<STRING>", reply.to="<STRING>", expiration="<STRING>", message.id="<STRING>", timestamp="<STRING>", type="<STRING>", user.id="<STRING>", app.id="<STRING>", routing.key="<STRING>", headers="<STRING>", tls.enabled="<BOOL>", tls.truststore.path="<STRING>", tls.truststore.password="<STRING>", tls.truststore.type="<STRING>", tls.version="<STRING>")

Query Parameters

NameDescriptionDefault ValuePossible Data TypesOptionalDynamic
uriThe URI that used to connect to an AMQP server. If no URI is specified, an error is logged in the CLI.e.g., amqp://guest:guest, amqp://guest:guest@localhost:5672STRINGNoNo
heartbeatThe period of time (in seconds) after which the peer TCP connection should be considered unreachable (down) by RabbitMQ and client libraries.60INTYesNo
exchange.nameThe name of the exchange that decides what to do with a message it sends.If the exchange.name already exists in the RabbitMQ server, then the system uses that exchange.name instead of redeclaring.STRINGNoYes
exchange.typeThe type of the exchange.name. The exchange types available are direct, fanout, topic and headers. For a detailed description of each type, see [RabbitMQ - AMQP Concepts](https://www.rabbitmq.com/tutorials/amqp-concepts.html)directSTRINGYesYes
exchange.durable.enabledIf this is set to true, the exchange remains declared even if the broker restarts.falseBOOLYesYes
exchange.autodelete.enabledIf this is set to true, the exchange is automatically deleted when it is not used anymore.falseBOOLYesYes
delivery.modeThis determines whether the connection should be persistent or not. The value must be either 1 or 2.If the delivery.mode = 1, then the connection is not persistent. If the delivery.mode = 2, then the connection is persistent.1INTYesNo
content.typeThe message content type. This should be the MIME content type.nullSTRINGYesNo
content.encodingThe message content encoding. The value should be MIME content encoding.nullSTRINGYesNo
prioritySpecify a value within the range 0 to 9 in this parameter to indicate the message priority.0INTYesYes
correlation.idThe message correlated to the current message. e.g., The request to which this message is a reply. When a request arrives, a message describing the task is pushed to the queue by the front end server. After that the frontend server blocks to wait for a response message with the same correlation ID. A pool of worker machines listen on queue, and one of them picks up the task, performs it, and returns the result as message. Once a message with right correlation ID arrives, thefront end server continues to return the response to the caller.nullSTRINGYesYes
reply.toThis is an anonymous exclusive callback queue. When the RabbitMQ receives a message with the reply.to property, it sends the response to the mentioned queue. This is commonly used to name a reply queue (or any other identifier that helps a consumer application to direct its response).nullSTRINGYesNo
expirationThe expiration time after which the message is deleted. The value of the expiration field describes the TTL (Time To Live) period in milliseconds.nullSTRINGYesNo
message.idThe message identifier. If applications need to identify messages, it is recommended that they use this attribute instead of putting it into the message payload.nullSTRINGYesYes
timestampTimestamp of the moment when the message was sent. If you do not specify a value for this parameter, the system automatically generates the current date and time as the timestamp value. The format of the timestamp value is dd/mm/yyyy.current timestampSTRINGYesNo
typeThe type of the message. e.g., The type of the event or the command represented by the message.nullSTRINGYesNo
user.idThe user ID specified here is verified by RabbitMQ against theuser name of the actual connection. This is an optional parameter.nullSTRINGYesNo
app.idThe identifier of the application that produced the message.nullSTRINGYesNo
routing.keyThe key based on which the excahnge determines how to route the message to the queue. The routing key is similar to an address for the message.emptySTRINGYesYes
headersThe headers of the message. The attributes used for routing are taken from the this paremeter. A message is considered matching if the value of the header equals the value specified upon binding.nullSTRINGYesYes
tls.enabledThis parameter specifies whether an encrypted communication channel should be established or not. When this parameter is set to true, the tls.truststore.path and tls.truststore.password parameters are initialized.falseBOOLYesNo
tls.truststore.pathThe file path to the location of the truststore of the client that sends the RabbitMQ events via the AMQP protocol. A custom client-truststore can be specified if required. If a custom truststore is not specified, then the system uses the default client-trustore in the ${carbon.home}/resources/security directory.\${carbon.home}/resources/security/client-truststore.jksSTRINGYesNo
tls.truststore.passwordThe password for the client-truststore. A custom password can be specified if required. If no custom password is specified, then the system uses gdncarbon as the default password.gdncarbonSTRINGYesNo
tls.truststore.typeThe type of the truststore.JKSSTRINGYesNo
tls.versionThe version of the tls/ssl.SSLSTRINGYesNo

Example 1

@App:name('TestExecutionPlan')
CREATE STREAM FooStream (symbol string, price float, volume long);

@info(name = 'query1')
CREATE SINK BarStream WITH (type ='rabbitmq', uri = 'amqp://guest:guest@localhost:5672', exchange.name = 'direct', routing.key= 'direct', map.type='xml') (symbol string, price float, volume long);

insert into BarStream
from FooStream select symbol, price, volume ;

This query publishes events to the direct exchange with the direct exchange type and the directTest routing key.