Skip to content

API Docs - v3.0.4

Tested Siddhi Core version: 5.1.5

It could also support other Siddhi Core minor versions.

Sink

rabbitmq (Sink)

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

Syntax

@sink(type="rabbitmq", 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>", @map(...)))

QUERY PARAMETERS

Name Description Default Value Possible Data Types Optional Dynamic
uri

The 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:5672

STRING No No
heartbeat

The period of time (in seconds) after which the peer TCP connection should be considered unreachable (down) by RabbitMQ and client libraries.

60 INT Yes No
exchange.name

The 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.

STRING No Yes
exchange.type

The 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

direct STRING Yes Yes
exchange.durable.enabled

If this is set to true, the exchange remains declared even if the broker restarts.

false BOOL Yes Yes
exchange.autodelete.enabled

If this is set to true, the exchange is automatically deleted when it is not used anymore.

false BOOL Yes Yes
delivery.mode

This 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.

1 INT Yes No
content.type

The message content type. This should be the MIME content type.

null STRING Yes No
content.encoding

The message content encoding. The value should be MIME content encoding.

null STRING Yes No
priority

Specify a value within the range 0 to 9 in this parameter to indicate the message priority.

0 INT Yes Yes
correlation.id

The 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.

null STRING Yes Yes
reply.to

This 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).

null STRING Yes No
expiration

The expiration time after which the message is deleted. The value of the expiration field describes the TTL (Time To Live) period in milliseconds.

null STRING Yes No
message.id

The message identifier. If applications need to identify messages, it is recommended that they use this attribute instead of putting it into the message payload.

null STRING Yes Yes
timestamp

Timestamp 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 timestamp STRING Yes No
type

The type of the message. e.g., The type of the event or the command represented by the message.

null STRING Yes No
user.id

The user ID specified here is verified by RabbitMQ against theuser name of the actual connection. This is an optional parameter.

null STRING Yes No
app.id

The identifier of the application that produced the message.

null STRING Yes No
routing.key

The 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.

empty STRING Yes Yes
headers

The 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.

null STRING Yes Yes
tls.enabled

This 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.

false BOOL Yes No
tls.truststore.path

The 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.jks STRING Yes No
tls.truststore.password

The password for the client-truststore. A custom password can be specified if required. If no custom password is specified, then the system uses wso2carbon as the default password.

wso2carbon STRING Yes No
tls.truststore.type

The type of the truststore.

JKS STRING Yes No
tls.version

The version of the tls/ssl.

SSL STRING Yes No

Examples EXAMPLE 1

@App:name('TestExecutionPlan') 
define stream FooStream (symbol string, price float, volume long); 
@info(name = 'query1')
@sink(type ='rabbitmq',
uri = 'amqp://guest:guest@localhost:5672',
exchange.name = 'direct',
routing.key= 'direct',
@map(type='xml'))
Define stream BarStream (symbol string, price float, volume long);
from FooStream select symbol, price, volume insert into BarStream;

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

Source

rabbitmq (Source)

The rabbitmq source receives the events from the rabbitmq broker via the AMQP protocol.

Syntax

@source(type="rabbitmq", uri="<STRING>", heartbeat="<INT>", exchange.name="<STRING>", exchange.type="<STRING>", exchange.durable.enabled="<BOOL>", exchange.autodelete.enabled="<BOOL>", routing.key="<STRING>", headers="<STRING>", queue.name="<STRING>", queue.durable.enabled="<BOOL>", queue.exclusive.enabled="<BOOL>", queue.autodelete.enabled="<BOOL>", tls.enabled="<BOOL>", tls.truststore.path="<STRING>", tls.truststore.password="<STRING>", tls.truststore.type="<STRING>", tls.version="<STRING>", auto.ack="<BOOL>", @map(...)))

QUERY PARAMETERS

Name Description Default Value Possible Data Types Optional Dynamic
uri

The URI that is 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:5672

STRING No No
heartbeat

The period of time (in seconds) after which the peer TCP connection should be considered unreachable (down) by RabbitMQ and client libraries.

60 INT Yes No
exchange.name

The name of the exchange that decides what to do with a message it receives.If the exchange.name already exists in the RabbitMQ server, then the system uses that exchange.name instead of redeclaring.

STRING No No
exchange.type

The 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.

direct STRING Yes No
exchange.durable.enabled

If this is set to true, the exchange remains declared even if the broker restarts.

false BOOL Yes No
exchange.autodelete.enabled

If this is set to true, the exchange is automatically deleted when it is not used anymore.

false BOOL Yes No
routing.key

The key based on which the exchange determines how to route the message to queues. The routing key is like an address for the message. The routing.key must be initialized when the value for the exchange.type parameter is direct or topic.

empty STRING Yes No
headers

The 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.

null STRING Yes No
queue.name

A queue is a buffer that stores messages. If the queue name already exists in the RabbitMQ server, then the system usees that queue name instead of redeclaring it. If no value is specified for this parameter, the system uses the unique queue name that is automatically generated by the RabbitMQ server.

system generated queue name STRING Yes No
queue.durable.enabled

If this parameter is set to true, the queue remains declared even if the broker restarts

false BOOL Yes No
queue.exclusive.enabled

If this parameter is set to true, the queue is exclusive for the current connection. If it is set to false, it is also consumable by other connections.

false BOOL Yes No
queue.autodelete.enabled

If this parameter is set to true, the queue is automatically deleted when it is not used anymore.

false BOOL Yes No
tls.enabled

This 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.

false BOOL Yes No
tls.truststore.path

The file path to the location of the truststore of the client that receives 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.jks STRING Yes No
tls.truststore.password

The password for the client-truststore. A custom password can be specified if required. If no custom password is specified, then the system uses wso2carbon as the default password.

wso2carbon STRING Yes No
tls.truststore.type

The type of the truststore.

JKS STRING Yes No
tls.version

The version of the tls/ssl.

SSL STRING Yes No
auto.ack

If this parameter is set to false, the server should expect explicit messages acknowledgements once delivered

true BOOL Yes No

Examples EXAMPLE 1

@App:name('TestExecutionPlan') 
define stream FooStream (symbol string, price float, volume long); 
@info(name = 'query1') 
@source(type ='rabbitmq',
uri = 'amqp://guest:guest@localhost:5672',
exchange.name = 'direct',
routing.key= 'direct',
@map(type='xml'))
Define stream BarStream (symbol string, price float, volume long);
from FooStream select symbol, price, volume insert into BarStream;

This query receives events from the direct exchange with the directexchange type, and the directTest routing key.