AMQP — Consommer des messages

Les consommateurs consomment depuis des files d'attente. Pour consommer des messages, il doit y avoir une file d'attente. Lorsqu'un nouveau consommateur est ajouté, en supposant qu'il y ait déjà des messages prêts dans la file d'attente, les livraisons commenceront immédiatement.

La file cible peut être vide au moment de l'enregistrement du consommateur. Dans ce cas, les premières livraisons auront lieu lorsque de nouveaux messages seront mis en file d'attente.

 

La consommation de messages est une tâche asynchrone, ce qui signifie que chaque fois qu'un nouveau message peut être livré à la file d'attente du consommateur, il est transmis automatiquement par le serveur au client. Vous pouvez lire une méthode alternative pour recevoir un message de manière synchrone.

 

Consommer

La méthode Consume crée un nouveau consommateur dans la file d'attente, et chaque fois qu'un nouveau message arrive, il sera automatiquement livré au client consommateur.

 

La méthode possède les arguments suivants :

 

• ChannelName : c'est le nom du canal (doit être ouvert avant d'appeler cette méthode).
• QueueName : c'est le nom de la file d'attente, ne doit pas dépasser 255 caractères et ne doit pas commencer par « amq. » (sauf si le paramètre passive est true).
• ConsumerTag : c'est le nom du consommateur et doit être unique. S'il n'est pas défini, le serveur en crée un nouveau.
• NoLocal : si True, le consommateur ne consomme jamais les messages publiés sur le même canal.
• NoAck : si true, signifie que le serveur n'attend pas d'accusé de réception pour chaque message distribué.
• Exclusif : si true, empêche les autres consommateurs de consommer des messages depuis cette file d'attente.
• NoWait : si true, le serveur n'enverra pas d'acquittement au client.
• Arguments : chaîne contenant des arguments personnalisés, les valeurs doivent être passées sous forme de chaîne JSON, exemple : {"x-dead-letter-exchange":"my-dlx"}.

 

Les messages sont livrés dans l'événement 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;

Un appel synchrone peut être effectué en appelant simplement la méthode ConsumeEx ; cette méthode retourne true si le Consommateur a été créé et false si aucune confirmation du serveur n'est arrivée.

 

Annuler la consommation

Cette méthode est utilisée pour annuler une file de consommateur existante.

 

La méthode possède les arguments suivants :

 

• ChannelName : c'est le nom du canal (doit être ouvert avant d'appeler cette méthode).
• ConsumerTag : c'est le nom du consommateur.
• NoWait : si true, le serveur n'enverra pas d'acquittement au 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;

Un appel synchrone peut être effectué en appelant simplement la méthode CancelConsumeEx ; cette méthode renvoie true si le consommateur a été annulé et false si aucune confirmation du serveur n'est arrivée.