AMQP Exchange e code (1 / 3)

· Componentei

From sgcWebSockets 2022.1 il protocollo AMQP 0.9.1 e' supportato. Il Advanced Message Queuing Protocol (AMQP) e' un protocollo aperto standard di livello applicazione per middleware orientato ai messaggi. Le caratteristiche distintive di AMQP sono l'orientamento ai messaggi, l'accodamento, l'instradamento (incluso point-to-point e publish-and-subscribe), l'affidabilita' e la sicurezza.

AMQP e' un protocollo binario di livello applicazione, progettato per supportare in modo efficiente un'ampia gamma di applicazioni di messaging e pattern di comunicazione. Fornisce una comunicazione message-oriented con controllo di flusso e garanzie di consegna come at-most-once (in cui ogni messaggio e' consegnato al massimo una volta), at-least-once (in cui ogni messaggio e' garantito essere consegnato, ma puo' avvenire piu' volte) ed exactly-once (in cui il messaggio arriva una e una sola volta), nonche' autenticazione e/o crittografia basata su SASL e/o TLS. Presuppone un protocollo di trasporto affidabile sottostante come il Transmission Control Protocol (TCP).

Channels 

AMQP e' un protocollo multi-canale. I canali offrono un modo per multiplexare una pesante connessione TCP/IP in piu' connessioni leggere. Questo rende il protocollo piu' "firewall friendly", dato che l'uso delle porte e' prevedibile. Significa inoltre che traffic shaping e altre funzionalita' QoS di rete possono essere facilmente applicate.

Ogni canale viene eseguito in un proprio thread, quindi ogni volta che viene ricevuto un nuovo messaggio il client identifica prima il canale e accoda il messaggio in una coda elaborata dal thread del canale.

Il ciclo di vita del canale e' il seguente:

1. Il client apre un nuovo canale (Open).

2. Il server conferma che il nuovo canale e' pronto (Open-Ok).

3. Client e server usano il canale come necessario.

4. Uno dei peer (client o server) chiude il canale (Close).

5. L'altro peer effettua l'handshake di chiusura del canale (Close-Ok).


Per creare un nuovo canale chiama il metodo OpenChannel passando il nome del canale come argomento. L'evento OnAMQPChannelOpen viene sollevato come conferma inviata dal server dell'apertura del canale.

AMQP.OpenChannel('channel_name');
procedure OnAMQPChannelOpen(Sender: TObject; const aChannel: string);
begin
  DoLog('#AMQP_channel_open: ' + aChannel);
end;

Exchanges 

La classe exchange permette a un'applicazione di gestire gli exchange sul server. Permette all'applicazione di programmare il proprio wiring (invece di affidarsi a un'interfaccia di configurazione). Nota: la maggior parte delle applicazioni non ha bisogno di questo livello di sofisticazione, e i middleware legacy difficilmente supportano questa semantica.

Il ciclo di vita dell'exchange e':

1. Il client chiede al server di assicurarsi che l'exchange esista (Declare). Il client puo' specificare ulteriormente "crea l'exchange se non esiste", oppure "avvisami ma non crearlo, se non esiste".

2. Il client pubblica messaggi sull'exchange.

3. Il client puo' scegliere di eliminare l'exchange (Delete).

Il metodo DeclareExchange crea un nuovo exchange o verifica che ne esista gia' uno. Il metodo accetta i seguenti argomenti:


Per dichiarare un nuovo exchange chiama il metodo DeclareExchange e passa nome del canale, nome dell'exchange e tipo di exchange come argomenti. L'evento OnAMQPExchangeDeclare viene sollevato come conferma inviata dal server della dichiarazione dell'exchange.

AMQP.DeclareExchange('channel_name', 'exchange_name', 'direct');
procedure OnAMQPExchangeDeclare(Sender: TObject; const aChannel, aExchange: string);
begin
  DoLog('#AMQP_exchange_declare: [' + aChannel + '] ' + aExchange);
end;

Queues 

La classe queue permette a un'applicazione di gestire code di messaggi sul server. E' un passo basilare in quasi tutte le applicazioni che consumano messaggi, almeno per verificare che una coda attesa sia effettivamente presente.

Il ciclo di vita di una coda di messaggi durable e' abbastanza semplice:

1. Il client verifica che la coda di messaggi esista (Declare, con l'argomento "passive").

2. Il server conferma che la coda di messaggi esiste (Declare-Ok).

3. Il client legge i messaggi dalla coda.

Il ciclo di vita di una coda di messaggi temporanea e' piu' interessante:

1. Il client crea la coda di messaggi (Declare, spesso senza nome di coda in modo che il server ne assegni uno). Il server conferma (Declare-Ok).

2. Il client avvia un consumer sulla coda di messaggi. La funzionalita' precisa di un consumer e' definita dalla classe Basic.

3. Il client annulla il consumer, esplicitamente o chiudendo il canale e/o la connessione.

4. Quando l'ultimo consumer scompare dalla coda di messaggi e dopo un time-out di cortesia, il server elimina la coda.

AMQP implementa il meccanismo di delivery per le sottoscrizioni su topic come code di messaggi. Questo abilita strutture interessanti in cui una sottoscrizione puo' essere bilanciata fra un pool di subscriber cooperanti

applicazioni.

Il ciclo di vita di una sottoscrizione prevede una fase di bind aggiuntiva:

1. Il client crea la coda di messaggi (Declare) e il server conferma (Declare-Ok).

2. Il client effettua il bind della coda di messaggi a un topic exchange (Bind) e il server conferma (Bind-Ok).

3. Il client utilizza la coda di messaggi come negli esempi precedenti.

Il metodo DeclareQueue crea una nuova coda o verifica che esista gia'. Il metodo accetta i seguenti argomenti:


Per dichiarare una nuova coda chiama il metodo DeclareQueue passando il nome del canale e il nome della coda come argomenti. L'evento OnAMQPQueueDeclare viene sollevato come conferma inviata dal server della dichiarazione della coda.

AMQP.DeclareQueue('channel_name', 'queue_name');
procedure OnAMQPQueueDeclare(Sender: TObject; const aChannel, aQueue: string; aMessageCount, aConsumerCount: Integer);
begin
  DoLog('#AMQP_queue_declare: [' + aChannel + '] ' + aQueue);
end;