AMQP Odbieranie wiadomości

Konsumenci odbierają wiadomości z kolejek. Aby pobierać wiadomości, musi istnieć kolejka. Po dodaniu nowego konsumenta, przy założeniu że w kolejce są już gotowe wiadomości, dostarczanie rozpoczyna się natychmiast.

Docelowa kolejka może być pusta w momencie rejestracji konsumenta. W takim przypadku pierwsze dostarczenia nastąpią po dodaniu nowych wiadomości do kolejki.

Odbieranie wiadomości jest zadaniem asynchronicznym, co oznacza, że za każdym razem, gdy nowa wiadomość może zostać dostarczona do kolejki konsumenta, jest ona automatycznie wypychana przez serwer do klienta. Alternatywną metodę synchronicznego odbierania wiadomości można znaleźć w odpowiedniej sekcji.

Odbieranie

Metoda Consume tworzy nowego konsumenta w kolejce, a każda nowa wiadomość jest automatycznie dostarczana do klienta-konsumenta.

Metoda przyjmuje następujące argumenty:

• ChannelName: nazwa kanału (musi być otwarty przed wywołaniem tej metody).
• QueueName: nazwa kolejki, nie może być dłuższa niż 255 znaków i nie może zaczynać się od "amq." (chyba że parametr passive ma wartość true).
• ConsumerTag: nazwa konsumenta, która musi być unikalna. Jeśli nie jest ustawiona, serwer tworzy nową.
• NoLocal: jeśli ma wartość true, konsument nigdy nie odbiera wiadomości opublikowanych na tym samym kanale.
• NoAck: jeśli true, oznacza, że serwer nie oczekuje potwierdzenia dla każdej dostarczonej wiadomości.
• Exclusive: jeśli wartość wynosi true, uniemożliwia innym konsumentom odbieranie wiadomości z tej kolejki.
• NoWait: jeśli true, serwer nie wyśle potwierdzenia do klienta.
• Arguments: ciąg znaków zawierający niestandardowe argumenty; wartości muszą być przekazane jako ciąg JSON, przykład: {"x-dead-letter-exchange":"my-dlx"}.

Wiadomości są dostarczane przez zdarzenie 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;

Możliwe jest również wywołanie synchroniczne poprzez metodę ConsumeEx; metoda ta zwraca true, jeśli konsument został utworzony, i false, jeśli nie otrzymano potwierdzenia od serwera.

Anuluj konsumowanie

Metoda służy do anulowania istniejącej kolejki konsumenta.

Metoda przyjmuje następujące argumenty:

• ChannelName: nazwa kanału (musi być otwarty przed wywołaniem tej metody).
• ConsumerTag: nazwa konsumenta.
• NoWait: jeśli true, serwer nie wyśle potwierdzenia do klienta.

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

Wywołanie synchroniczne można wykonać, wywołując metodę CancelConsumeEx. Metoda ta zwraca wartość true, jeśli konsument został anulowany, lub false, jeśli nie nadeszło potwierdzenie od serwera.