AMQP Consuma Messaggi

I consumer consumano dalle code. Per consumare messaggi deve esistere una coda. Quando viene aggiunto un nuovo consumer, supponendo che siano già presenti messaggi pronti nella coda, le consegne inizieranno immediatamente.

La coda di destinazione può essere vuota al momento della registrazione del consumer. In tal caso le prime consegne avverranno quando verranno accodati nuovi messaggi.

 

Il consumo di messaggi è un'attività asincrona, il che significa che ogni volta che un nuovo messaggio può essere consegnato alla coda del consumatore, viene inviato automaticamente dal server al client. È possibile consultare un metodo alternativo per ricevere messaggi in modo sincrono.

 

Consuma

Il metodo Consume crea un nuovo consumer nella coda e ogni volta che è presente un nuovo messaggio questo verrà recapitato automaticamente al client consumer.

 

Il metodo ha i seguenti argomenti:

 

• ChannelName: è il nome del canale (deve essere aperto prima di chiamare questo metodo).
• QueueName: è il nome della coda, non deve superare i 255 caratteri e non deve iniziare con "amq." (tranne se il parametro passive è true).
• ConsumerTag: è il nome del consumer e deve essere univoco. Se non è impostato, il server ne crea uno nuovo.
• NoLocal: se true, il consumer non consuma mai i messaggi pubblicati sullo stesso canale.
• NoAck: se true significa che il server non si aspetta una conferma per ogni messaggio recapitato.
• Exclusive: se true impedisce ad altri consumer di consumare messaggi da questa coda.
• NoWait: se true, il server non invierà un acknowledgement al client.
• Arguments: stringa contenente argomenti personalizzati; i valori devono essere passati come stringa json, esempio: {"x-dead-letter-exchange":"my-dlx"}.

 

I messaggi vengono consegnati nell'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 chiamata sincrona può essere effettuata chiamando il metodo ConsumeEx; questo metodo restituisce true se il Consumer è stato creato e false se non è arrivata alcuna conferma dal server.

 

Annulla Consume

Questo metodo viene utilizzato per annullare una coda consumer esistente.

 

Il metodo ha i seguenti argomenti:

 

• ChannelName: è il nome del canale (deve essere aperto prima di chiamare questo metodo).
• ConsumerTag: è il nome del consumer.
• NoWait: se true, il server non invierà un acknowledgement al 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;

È possibile effettuare una chiamata sincrona chiamando il metodo CancelConsumeEx; questo metodo restituisce true se il Consumer è stato annullato e false se non è arrivata alcuna conferma dal server.