AMQP Consume Messages

Consumenten verbruiken uit wachtrijen. Om berichten te kunnen verbruiken moet er een wachtrij zijn. Wanneer een nieuwe consument wordt toegevoegd en er al berichten klaarstaan in de wachtrij, beginnen leveringen onmiddellijk.

De doelwachtrij kan leeg zijn op het moment van consumentregistratie. In dat geval vinden de eerste afleveringen plaats wanneer nieuwe berichten in de wachtrij worden geplaatst.

 

Het verwerken van berichten is een asynchrone taak, wat betekent dat elke keer dat een nieuw bericht aan de consumentenwachtrij kan worden bezorgd, het automatisch door de server naar de client wordt gepusht. U kunt een alternatieve methode lezen om Berichten synchroon te ontvangen.

 

Verbruiken

De methode Consume maakt een nieuwe consument in de wachtrij aan, en elke keer dat er een nieuw bericht is, wordt dit automatisch afgeleverd aan de consumerende client.

 

De methode has het volgende arguments:

 

• ChannelName: de naam van het kanaal (moet geopend zijn voordat deze methode wordt aangeroepen).
• QueueName: dit is de naam van de queue, mag niet langer zijn dan 255 tekens en mag niet beginnen met "amq." (tenzij de passieve parameter true is).
• ConsumerTag: dit is de naam van de consument en moet uniek zijn. Als het niet is ingesteld, maakt de server een nieuwe aan.
• NoLocal: if true means the consumer nooit consumes messages published on the dezelfde kanaal.
• NoAck: als true betekent dit dat de server geen bevestiging verwacht voor elk afgeleverd bericht.
• Exclusief: als true, voorkomt dit dat andere consumenten berichten van deze wachtrij verbruiken.
• NoWait: als true, stuurt de server geen bevestiging naar de client.
• Arguments: tekenreeks met aangepaste argumenten; de waarden moeten worden doorgegeven als een JSON-tekenreeks, voorbeeld: {"x-dead-letter-exchange":"my-dlx"}.

 

De berichten worden afgeleverd via de gebeurtenis 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;

Er kan ook een synchrone aanroep worden gedaan door de methode ConsumeEx aan te roepen; deze methode retourneert true als de Consumer is aangemaakt en false als er geen bevestiging van de server is ontvangen.

 

Verbruik annuleren

Deze methode wordt gebruikt om een bestaande consumerwachtrij te annuleren.

 

De methode has het volgende arguments:

 

• ChannelName: de naam van het kanaal (moet geopend zijn voordat deze methode wordt aangeroepen).
• ConsumerTag: dit is de naam van de consumer.
• NoWait: als true, stuurt de server geen bevestiging naar de 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;

Een synchrone aanroep kan worden gedaan door simpelweg de methode CancelConsumeEx aan te roepen; deze methode retourneert true als de consument is geannuleerd en false als er geen bevestiging van de server is ontvangen.