Commands | AMQP Queues

The queue class lets an application manage message queues on the server. This is a basic step in almost all applications that consume messages, at least to verify that an expected message queue is actually present.

 

The life-cycle for a durable message queue is fairly simple:

 

1. The client asserts that the message queue exists (Declare, with the "passive" argument).

2. The server confirms that the message queue exists (Declare-Ok).

3. The client reads messages off the message queue.

 

The life-cycle for a temporary message queue is more interesting:

 

1. The client creates the message queue (Declare, often with no message queue name so the server will assign a name). The server confirms (Declare-Ok).

2. The client starts a consumer on the message queue. The precise functionality of a consumer is defined by the Basic class.

3. The client cancels the consumer, either explicitly or by closing the channel and/or connection.

4. When the last consumer disappears from the message queue, and after a polite time-out, the server deletes the message queue.

 

AMQP implements the delivery mechanism for topic subscriptions as message queues. This enables interesting structures where a subscription can be load balanced among a pool of co-operating subscriber

applications.

 

The life-cycle for a subscription involves an extra bind stage:

 

1. The client creates the message queue (Declare), and the server confirms (Declare-Ok).

2. The client binds the message queue to a topic exchange (Bind) and the server confirms (Bind-Ok).

3. The client uses the message queue as in the previous examples.

 

Declare Queue

This method creates a new queue or verifies that a Queue already exists. The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• QueueName: it's the name of the queue, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• Passive: if passive is true, the server only verifies that the queue is already declared. If passive is false, and the queue not exists, the server will create a new one.
• Durable: if true, the queue will be re-created when the server starts. If false, the queue will be deleted when the server stops.
• Exclusive: if true means the queue is only accessed by the current connection.
• AutoDelete: if true, the queue will be deleted when all consumers no longer use the queue.
• NoWait: if true, the server doesn't sends an acknowledgment to the client.
• Arguments: string which contains custom arguments, the values must be passed as a json string, example: {"x-dead-letter-exchange":"my-dlx"}.

 

To Declare a new Queue just call the method DeclareQueue and pass the channel name and  queue name as arguments. The event OnAMQPQueueDeclare is raised as a confirmation sent by the server that the exchange has been declared.

AMQP.DeclareQueue('channel_name', 'queue_name');
 
procedure OnAMQPQueueDeclare(Sender: TObject; const aChannel, aQueue: string; 
  aMessageCount, aConsumerCount: Integer);
begin
  DoLog('#AMQP_queue_declare: [' + aChannel + '] ' + aQueue);
end;

A Synchronous call can be done too calling the method DeclareQueueEx, this method returns true if the Queue has been Declared and false if no confirmation from server has arrived.

 

if AMQP.DeclareQueueEx('channel_name', 'queue_name') then
  DoLog('#AMQP_queue_declare: [' + aChannel + '] ' + aQueue);
else
  DoLog('#AMQP_queue_declare_error');

Delete Queue

This method is used to delete an existing Queue. The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• QueueName: it's the name of the queue, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• IfUnused: the server only deletes the queue if there aren't any consumers attachedt to it.
• IfEmpty: the server only deletes the queue if there are no messages.
• NoWait: if true, the server doesn't sends an acknowledgment to the client.

 

To Delete an existing Queue call the method DeleteQueue and pass the channel name and queue name as arguments. The event OnAMQPQueueDelete is raised as a confirmation sent by the server that the queue has been deleted.

 

A Synchronous call can be done too calling the method DeleteQueueEx, this method returns true if the Queue has been Deleted and false if no confirmation from server has arrived.

 

 

Bind Queue

This method is used to bind a Queue to a Exchange. The Exchanges use the bindings to know which queues will be used to route the messages.

 

All AMQP Servers bind automatically all the queues to the default exchange (it's a "direct" exchange without name) using the Queue Name as the binding routing key. This allows to send a message to a specific queue without declare a binding. Just call the method PublishMessage, pass an empty value as Exchange Name and set the RoutingKey with the value of the Queue Name.

 

The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• QueueName: it's the name of the queue, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• ExchangeName: it's the name of the exchange, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• RoutingKey: it's the binding's routing key.
• NoWait: if true, the server doesn't sends an acknowledgment to the client.
• Arguments: string which contains custom arguments, the values must be passed as a json string, example: {"x-dead-letter-exchange":"my-dlx"}.

 

To Bind a Queue to a Exchange call the method BindQueue and pass the channel name,  queue name, exchange and routing key as arguments. The event OnAMQPQueueBind is raised as a confirmation sent by the server that the queue has been bind.

AMQP.BindQueue('channel_name', 'queue_name', 'exchange_name', 'routing_key');
 
procedure OnAMQPQueueBind(Sender: TObject; const aChannel, aQueue, aExchange: string);
begin
  DoLog('#AMQP_queue_bind: [' + aChannel + '] ' + aQueue + ' -->-- ' + aExchange)
end;

 

A Synchronous call can be done too calling the method BindQueueEx, this method returns true if the Queue has been Bind and false if no confirmation from server has arrived.

if AMQP.BindQueueEx('channel_name', 'queue_name', 'exchange_name', 'routing_key') then
  DoLog('#AMQP_queue_bind: [' + aChannel + '] ' + aQueue + ' --><-- ' + aExchange)
else
  DoLog('#AMQP_queue_bind_error');

UnBind Queue

This method deletes an existing queue binding.

 

The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• QueueName: it's the name of the queue, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• ExchangeName: it's the name of the exchange, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• RoutingKey: it's the binding's routing key.

 

To UnBind a Queue just call the method UnBindQueue and pass the channel name,  queue name, exchange and routing key as arguments. The event OnAMQPQueueUnBind is raised as a confirmation sent by the server that the queue has been unbind.

 

A Synchronous call can be done too calling the method UnBindQueueEx, this method returns true if the Queue has been UnBind and false if no confirmation from server has arrived.

 

Purge Queue

This method purges all messages of a queue. All the messages that have been sent but are awaiting acknowledgment are not affected.

 

The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• QueueName: it's the name of the queue, must be no longer of 255 characters and not begin with "amq." (except if passive parameter is true).
• NoWait: if true, the server doesn't sends an acknowledgment to the client.

 

To Purge a Queue just call the method PurgeQueue and pass the channel name and  queue name as arguments. The event OnAMQPQueuePurge is raised as a confirmation sent by the server that the queue has been Purged.

 

A Synchronous call can be done too calling the method PurgeQueueEx, this method returns true if the Queue has been Purged and false if no confirmation from server has arrived.