AMQP Consumir Mensajes

Los consumidores consumen de las colas. Para consumir mensajes, debe existir una cola. Cuando se añade un nuevo consumidor, si ya hay mensajes listos en la cola, las entregas comienzan de inmediato.

La cola de destino puede estar vacía en el momento del registro del consumidor. En ese caso, las primeras entregas se producirán cuando se encolen nuevos mensajes.

 

El consumo de mensajes es una tarea asíncrona, lo que significa que cada vez que un nuevo mensaje puede entregarse a la cola del consumidor, el servidor lo envía automáticamente al cliente. Puede consultar un método alternativo para Recibir mensajes de forma síncrona.

 

Consumir

El método Consume crea un nuevo consumidor en la cola y, cada vez que llega un nuevo mensaje, este se entrega automáticamente al cliente consumidor.

 

El método tiene los siguientes argumentos:

 

• ChannelName: es el nombre del canal (debe estar abierto antes de llamar a este método).
• QueueName: es el nombre de la cola; no debe superar los 255 caracteres y no debe comenzar por "amq." (excepto si el parámetro passive es true).
• ConsumerTag: es el nombre del consumidor y debe ser único. Si no se establece, el servidor crea uno nuevo.
• NoLocal: si es true, significa que el consumidor nunca consume mensajes publicados en el mismo canal.
• NoAck: si es true, significa que el servidor no espera un acuse de recibo por cada mensaje entregado.
• Exclusive: si es true, evita que otros consumidores consuman mensajes de esta cola.
• NoWait: si es true, el servidor no enviará un acuse de recibo al cliente.
• Arguments: cadena que contiene argumentos personalizados; los valores deben pasarse como una cadena JSON, por ejemplo: {"x-dead-letter-exchange":"my-dlx"}.

 

Los mensajes se entregan en el evento OnAMQPBasicDeliver .

 

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;

Una llamada síncrona puede realizarse simplemente invocando el método ConsumeEx; este método devuelve true si el Consumer se ha creado correctamente y false si no se ha recibido confirmación del servidor.

 

Cancelar consumo

Este método se utiliza para cancelar una cola de consumidor existente.

 

El método tiene los siguientes argumentos:

 

• ChannelName: es el nombre del canal (debe estar abierto antes de llamar a este método).
• ConsumerTag: es el nombre del consumidor.
• NoWait: si es true, el servidor no enviará un acuse de recibo al cliente.

 

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

Se puede realizar una llamada síncrona invocando el método CancelConsumeEx; este método devuelve true si el Consumer ha sido cancelado y false si no ha llegado confirmación del servidor.