AMQP Consume Messages

Consumers consume from queues. In order to consume messages there has to be a queue. When a new consumer is added, assuming there are already messages ready in the queue, deliveries will start immediately.

The target queue can be empty at the time of consumer registration. In that case first deliveries will happen when new messages are enqueued.

 

Consuming messages is an asynchronous task, which means that every time a new message can be delivered to the consumer queue, it's pushed by the server to the client automatically. You can read an alternative method to Receive Message Synchronously.

 

Consume

The method Consume creates a new consumer in the queue, and every time there is a new message this will be delivered automatically to the consumer client.

 

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).
• ConsumerTag: it's the name of the consumer and must be unique. If it's not set, then the server creates a new one.
• NoLocal: if true means the consumer never consumes messages published on the same channel.
• NoAck: if true means the server doesn't expect an acknowledgment for every message delivered.
• Exclusive: if true prevents that other consumers consume messages from this queue.
• NoWait: if true, the server won't send 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"}.

 

The messages are delivered OnAMQPBasicDeliver event.

 

AMQP.Consume('channel_name', 'queue_name', 'consumer_tag');
 
procedure OnAMQPBasicDeliver(Sender: TObject;
  const aChannel: string;
  const aDeliver: TsgcAMQPFramePayload_Method_BasicDeliver;
  const aContent: TsgcAMQPMessageContent);
begin
  DoLog('#AMQP_basic_deliver: ' + aChannel + ' ' + aDeliver.ConsumerTag + ' ' +
    ' ' + aContent.Body.AsString);
end;

A Synchronous call can be done just calling the method ConsumeEx, this method returns true if the Consumer has been created and false if no confirmation from server has arrived.

 

Cancel Consume

This method is used to Cancel an existing consumer queue.

 

The method has the following arguments:

 

• ChannelName: it's the name of the channel (must be open before call this method).
• ConsumerTag: it's the name of the consumer.
• NoWait: if true, the server won't send an acknowledgment to the client.

 

AMQP.CancelConsume('channel_name', 'consumer_tag');
 
procedure OnAMQPBasicCancelConsume(Sender: TObject; const aChannel: string; const aConsumerTag);
begin
  DoLog('#AMQP_basic_cancel_consume: ' + aChannel + ' ' + aConsumerTag);
end;

A Synchronous call can be done just calling the method CancelConsumeEx, this method returns true if the Consumer has been cancelled and false if no confirmation from server has arrived.