

API Docs - v1.0.17

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>", @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

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.