AMQP umożliwia odbieranie wiadomości w 2 trybach:
- Żądanie od klienta: za pomocą metody GetMessage. Jeśli w kolejce nie ma wiadomości, zostanie wywołane zdarzenie OnAMQPBasicGetEmpty.
- Wysłane przez serwer: za pomocą metody Consume.
Consume
Konsumenci odbierają wiadomości z kolejek. Aby móc konsumować wiadomości, musi istnieć kolejka. Gdy dodawany jest nowy konsument, a w kolejce są już gotowe wiadomości, dostarczanie zaczyna się natychmiast.
Kolejka docelowa może być pusta w chwili rejestracji konsumenta. W takim przypadku pierwsze dostarczenia nastąpią, gdy w kolejce pojawią się nowe wiadomości.
Konsumowanie wiadomości jest zadaniem asynchronicznym, co oznacza, że za każdym razem, gdy nowa wiadomość może zostać dostarczona do kolejki konsumenta, serwer automatycznie wypycha ją do klienta. Możesz zapoznać się z alternatywną metodą: odbieranie wiadomości synchronicznie.
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 dłuższa niż 255 znaków i nie zaczynająca się od "amq." (wyjątek: jeśli parametr passive jest true).
- ConsumerTag: nazwa konsumenta, musi być unikalna. Jeśli nie jest ustawiona, serwer tworzy nową.
- NoLocal: jeśli true, konsument nigdy nie odbiera wiadomości opublikowanych na tym samym kanale.
- NoAck: jeśli true, serwer nie oczekuje potwierdzenia dla każdej dostarczonej wiadomości.
- Exclusive: jeśli true, zapobiega konsumowaniu wiadomości z tej kolejki przez innych konsumentów.
- NoWait: jeśli true, serwer nie wysyła potwierdzenia do klienta.
Wiadomości są dostarczane za pomocą zdarzenia OnAMQPBasigGetOk.
AMQP.Consume('channel_name', 'exchange_name', 'consumer_tag');
procedure OnAMQPBasicGetOk(Sender: TObject; const aChannel: string; const aGetOk: TsgcAMQPFramePayload_Method_BasicGetOk; const aContent: TsgcAMQPMessageContent);
begin
DoLog('#AMQP_basic_GetOk: ' + aChannel + ' ' + IntToStr(aGetOk.MessageCount) + ' ' + aContent.Body.AsString);
end;
Pobieranie wiadomości
Pobieranie wiadomości to zadanie synchroniczne, co oznacza, że to klient pyta serwer, czy w kolejce są wiadomości. Możesz zapoznać się z alternatywną metodą: odbieranie wiadomości asynchronicznie.
Metoda GetMessage wysyła żądanie do serwera AMQP z pytaniem, czy w kolejce są dostępne wiadomości. Jeśli tak, zostaną dostarczone przez zdarzenie OnAMQPBasicGetOk, a jeśli kolejka jest pusta, zostanie wywołane zdarzenie OnAMQPBasicGetEmpty.
Metoda przyjmuje następujące argumenty:
- ChannelName: nazwa kanału (musi być otwarty przed wywołaniem tej metody).
- QueueName: nazwa kolejki, nie dłuższa niż 255 znaków i nie zaczynająca się od "amq." (wyjątek: jeśli parametr passive jest true).
- NoWait: jeśli true, serwer nie wysyła potwierdzenia do klienta.
AMQP.GetMessage('channel_name', 'exchange_name');
procedure OnAMQPBasicGetOk(Sender: TObject; const aChannel: string; const aGetOk: TsgcAMQPFramePayload_Method_BasicGetOk; const aContent: TsgcAMQPMessageContent);
begin
DoLog('#AMQP_basic_GetOk: ' + aChannel + ' ' + IntToStr(aGetOk.MessageCount) + ' ' + aContent.Body.AsString);
end;
procedure OnAMQPBasicGetEmpty(Sender: TObject; const aChannel: string);
begin
DoLog('#AMQP_basic_GetEmpty: ' + aChannel);
end;