Pico server "MQTT"

Detto terra terra un server MQTT (Message Queuing Telemetry Transport) permette il dispacciamento di messaggi suddivisi per tema (o topic, argomento, canale) tra unità connesse in rete. Le unità mittenti (publisher) inviano messaggi al server riguardanti uno specifico tema, e il server invia una copia dei messaggi (notifiche) a tutte le unità destinatarie (subscriber) che hanno sottoscritto la ricezione dei messaggi per quel tema.

Qual è il vantaggio dell’architettura publish/subscribe? Innanzitutto quello di non dover conoscere a priori il destinatario, né quanti destinatari ci siano, né se nel corso del tempo questi variano di numero. Non vi è un contatto diretto tra le unità, ogni unità trasmette al server un messaggio riguardante uno specifico tema, e riceve dal server solo i messaggi relativi ai temi che ha sottoscritto. Ogni unità subscriber può sottoscrivere più temi. Ogni unità può essere contemporaneamente publisher per certi temi, e subscriber per altri.

Allo stesso tempo al server non interessa nulla del significato dei topic o del contenuto dei messaggi, si limita a tenere in memoria una tabella di indirizzi sottoscrittori e a dispacciare i messaggi in arrivo. Quindi il suo compito è estremamente generale e indipendente da quello che fanno le unità.

Lo schemino seguente rappresenta quattro unità (con i relativi indirizzi di rete) che scambiano informazioni tramite un server intermedio. I canali tematici sono rappresentati tramite colori. Come si vede un messaggio può raggiungere più destinatari, e diversi mittenti possono scrivere sullo stesso canale tematico. Le unità possono sottoscrivere più temi e possono pubblicare messaggi riguardanti diversi temi.

Esempio uno: sensore di movimento

Un sensore di movimento può inviare un messaggio con topic “movimento in soggiorno”. Il server rigirerà il messaggio a tutte le unità che hanno sottoscritto quel topic, cioè quelle che per un motivo o per l’altro sono interessate a sapere se c’è del movimento in soggiorno.

L’informazione potrebbe sicuramente servire ad una centralina di allarme. Ma anche a una centralina di controllo luci, che potrebbe usare quell’informazione per spegnerle dopo un certo tempo di assenza (l’autore è famoso per addormentarsi con le luci accese, o, peggio, dimenticando la finestra del bagno aperta in inverno…)

Grazie all’archittettura publish/subscribe il sensore di movimento in questione non deve preoccuparsi di contattare una per una tutte le unità destinatarie (che potrebbero cambiare di numero o di indirizzo). Allo stesso modo alle unità destinatarie non interessa sapere da dove proviene fisicamente quel messaggio, ma solo che è relativo al topic di loro interesse.

Esempio due: sensori di temperatura

Uno o più sensori di temperatura possono pubblicare le loro misure nei momenti a loro più comodi (alcuni sensori richiedono discreti tempi per effettuare le misure).

Una o più centraline possono ricevere questi valori per decidere se avviare sistemi di riscaldamento o raffrescamento, o ancora generare degli allarmi.

Un’ulteriore centralina potrebbe invece sottoscrivere contemporaneamente i messaggi di tutti i sensori per registrare uno storico dei dati in un database.

Come si vede di nuovo, ai sensori non interessa cosa ci sia “a valle” o cosa viene fatto con le loro informazioni. Grazie al server intermedio si limitano semplicemente a trasmettere sullo specifico canale tematico.

Il pico server con Arduino

Il protocollo MQTT è diventato uno standard ISO ed è utilizzato in numerose applicazioni. In questo articolo però non si parla del “vero” protocollo MQTT standard, ma dell’applicazione del principio publish/subscribe in una forma molto più semplice che richiede pochissime risorse, per consentire la condivisione di informazioni tra semplici apparati come sensori e centraline in una piccola rete LAN domestica.

Un Arduino (UNO o Mega), una ethernet shield con chip W5100, e le librerie standard già presenti nell’IDE, sono tutto quello che serve per mettere in funzione un pico server, dalle funzionalità limitate, ma (credo) didatticamente interessante.

Se poi ci aggiungiamo anche la scatolina su misura stampata 3D… warping a parte sembra quasi un prodotto professionale…

Ho voluto sperimentare il dispacciamento di piccoli pacchetti UDP, che all’interno di una LAN domestica non dovrebbero soffrire di frammentazione o dropping.

Nel caso di unità connesse in WiFi tramite ESP8266 Esp-01, ogni tanto qualche pacchetto non viene ricevuto. Questo è insito in un tipo di protocollo di trasporto “non sicuro” come l’UDP, che per contro invece ha il vantaggio di una grande velocità.

Questo server elementare non effettua alcuna verifica sull’avvenuta ricezione. Se davvero occorre, va implementata a livello di applicazione nelle unità remote, che possono rispondersi direttamente, o attraverso appositi canali tematici “di ritorno”.

Protocollo server

La prima cosa da fare è stabilire un protocollo di livello applicativo per la comunicazione tra unità e server.

Per semplicità ho deciso che ogni unità possa trasmettere al server solo due tipi di messaggio: publish e subscribe. E che possa ricevere dal server solo due tipi di messaggio: notifica e ack. Questi tipi vengono identificati dal primo byte dei pacchetti contenente rispettivamente i caratteri P S N A.

Sempre per semplicità, il topic è rappresentato da un singolo byte di valore 0..255, questo significa poter gestire messaggi relativi a 256 diversi “argomenti”.

Segue un “pid”, un identificatore di pacchetto, o numero di sequenza, utile per riconoscere e gestire gli ack o le eventuali ritrasmissioni. In caso contrario può essere ignorato.

E infine il messaggio vero e proprio da recapitare ai destinatari è formato da 0 a 21 byte, per un totale di 24 byte massimi per ogni pacchetto UDP trasmesso.

Tipi messaggi ricevuti dal server:

  .-----.-------.-----.     .-----.
  | 'P' | topic | PID | ... |     |   publish
  '-----'-------'-----'     '-----'
  .-----.-------.-----.
  | 'S' | topic | PID |               subscribe
  '-----'-------'-----'

Tipi di messaggi trasmessi dal server:

  .-----.-------.-----.     .-----.
  | 'N' | topic | PID | ... |     |   notifica
  '-----'-------'-----'     '-----'
  .-----.-----.
  | 'A' | PID |                       acknowledge
  '-----'-----'

Il pacchetto acknowledge, trasmesso al mittente ad ogni sottoscrizione o ad ogni pubblicazione, permette alle unità di conoscere sempre lo stato di raggiungibilità del server, ed eventualmente tentare delle ritrasmissioni in caso di mancato ack. Va ricordato che questo ack riguarda solo la tratta mittente-server, e la sua ricezione garantisce solo al mittente che il server ha ricevuto correttamente il pacchetto.

Effettuare periodicamente le sottoscrizioni, ad esempio una volta al minuto, permette all’intera rete di tornare in piedi automaticamente anche se il server viene resettato.

Il bare minimum ethernet + UDP

Ovvero il codice minimo con le librerie da includere, per avere una ethernet funzionante e un canale di comunicazione UDP in grado di ricevere e trasmettere su una determinata porta, in questo caso la 9000.

//----------------------------------------------------------
//  PICOSERVER MQTT
//  Hardware ArduinoUNO + ethernet shield W5100
//----------------------------------------------------------
#include <Ethernet.h>
#include <EthernetUdp.h>

//----------------------------------------------------------
//  VARIABILI DI LAVORO GLOBALI
//----------------------------------------------------------
EthernetUDP    Udp;

//----------------------------------------------------------
//  FUNZIONE   : setup
//  DESCRIZIONE: Inizializza il sistema all'accensione
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void setup(void)
{
    byte MAC[] = { 0xFA, 0xCC, 0x10, 0xCA, 0xFF, 0xEE };
    IPAddress ipLocale(192, 168, 1, 30);
    Ethernet.init(10);                  // pin CS
    Ethernet.begin(MAC, ipLocale);
    Ethernet.setRetransmissionCount(10);
    Ethernet.setRetransmissionTimeout(30);
    Udp.begin(9000);
}

Le funzioni ‘setRetransmissionCount‘ e ‘setRetransmissionTimeout‘ sono importantissime. Il chip W5100 e la libreria per Arduino non gestiscono una tabella di ARP (associazione tra IP e MAC address), per cui ad ogni trasmissione il W5100 genera dei pacchetti broadcast per richiedere il MAC address all’ unità a cui deve inviare i pacchetti. Per default il transmissionCount è impostato a 8 e il timeout a 200ms. Questo vuol dire che in caso di destinatario off line, ogni trasmissione richiederebbe 1.6 secondi di tentativi bloccando tutto il resto.

Con le impostazioni riportate sopra invece vengono effettuati dei tentativi più veloci e per un tempo minore (300ms in totale). Questo è il vero limite alla velocità imposto da questa architettura, e bisogna anche considerare la possibilità di un blocco della comunicazione se molte unità che si erano sottoscritte vanno off line contemporaneamente. Per far fronte a questa eventualità si può prevedere la loro cancellazione dalla tabella sottoscrizioni, in modo che il blocco sia momentaneo.

Il buffer

Poi ci serve un buffer (un’area di memoria) per ricevere il contenuto dei pacchetti, o per contenere i dati da trasmettere. Oltre al buffer ci serve conoscere l’indirizzo del mittente (o specificare quello del destinatario quando vogliamo trasmettere), e la lunghezza dei dati da 0 a 21 byte.

Dato che si sperimenta solo all’interno della LAN locale, per semplicità l’indirizzo delle unità è solo l’ultimo byte dell’ indirizzo IP.

Quindi alle variabili globali aggiungiamo le dichiarazioni:

// buffer di ricezione e di
// trasmissione (mai contemporanee)
const uint8_t  MAXBUF = 24;
uint8_t        udpBuffer[MAXBUF];
uint8_t        udpADDR;
uint16_t       udpLen;

Ricezione UDP

Per ricevere un pacchetto si controlla che l’hardware ethernet sia funzionante, se il test è positivo si chiama la funzione parsePacket che, nel caso sia arrivato un pacchetto, restituisce la sua dimensione. Se la dimensione supera i 24 byte scartiamo il pacchetto, altrimenti lo leggiamo trasferendolo nel buffer.

Il chip W5100 è in grado di accodare più pacchetti in arrivo in un proprio buffer interno da qualche kbyte, da cui possono essere letti in momenti successivi con la funzione read.

//----------------------------------------------------------
//  FUNZIONE   : dropIncomingPacket
//  DESCRIZIONE: Scarta pacchetto ricevuto fuori specifiche
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void dropIncomingPacket(void)
{
    while (Udp.available()) { Udp.read(udpBuffer, MAXBUF); }
}

//----------------------------------------------------------
//  FUNZIONE   : loop
//  DESCRIZIONE: Ciclo principale del
//               programma ripetuto continuamente
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void loop(void)
{
    if (Ethernet.hardwareStatus() != EthernetNoHardware)
    {
        udpLen  = Udp.parsePacket();
        udpADDR = Udp.remoteIP()[3];
        if (udpLen > MAXBUF){ 

            dropIncomingPacket(); 
        
        }else if (udpLen > 0){
            
            Udp.read(udpBuffer, MAXBUF);
            processPacket();

        }else{
            // nothing
        }
    }
}

Processare i pacchetti

Una volta ricevuto un pacchetto (scartando eventuali più grandi di 24 byte che non c’entrano con il protocollo elementare stabilito prima), se ne può determinare il tipo (P S) per compiere l’azione adeguata in risposta.

//----------------------------------------------------------
//  FUNZIONE   : sendPacket
//  DESCRIZIONE: Trasmette pacchetto UDP, se fallisce
//               lo scarta verso router
//  PARAMETRI  : outService se true abilita la
//               cancellazione indirizzo
//               da lista sottoscrizioni, usato per
//               pacchetti notifica se unita`
//               destinatarie non raggiungibili
//  RETURN CODE: ignorato
//----------------------------------------------------------
void sendPacket(bool outService)
{
    IPAddress ipRouter(192, 168, 1, 1);
    IPAddress ipDest(192, 168, 1, udpADDR);
    Udp.beginPacket(ipDest, 9000);
    Udp.write(udpBuffer, udpLen);
    uint8_t result = Udp.endPacket();

    if (0 == result)  // scarico pacchetto non trasmesso
    {
        Udp.beginPacket(ipRouter, 9000);
        Udp.print('\0');
        Udp.print('\0');
        Udp.endPacket();
        if (outService) { deleteSubscription(udpADDR); }
    }
}
//----------------------------------------------------------
//  FUNZIONE   : sendACK
//  DESCRIZIONE: Invia al mittente un pacchetto ACK + PID
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void sendACK(void)
{
    uint8_t tempTopic = udpBuffer[1];
    uint8_t tempLen = udpLen;
    udpBuffer[0] = 'A';
    udpBuffer[1] = udpBuffer[2];
    udpLen = 2;
    sendPacket(false);
    udpBuffer[1] = tempTopic;        
    udpLen = tempLen;
}
//----------------------------------------------------------
//  FUNZIONE   : processPacket
//  DESCRIZIONE: Elabora il pacchetto appena ricevuto
//               in udpBuffer lungo udpLen
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void processPacket(void)
{
    uint8_t type = udpBuffer[0];

    if (('P' == type) and (udpLen > 1)){ 

        sendACK();
        publish(); 
    
    }else if (('S' == type) and (3 == udpLen)){ 

        uint8_t topic = udpBuffer[1];
        sendACK();
        subscribe(topic, udpADDR);
    
    }else{
        // nothing
    }
}

La funzione ‘sendPacket’ è più complessa di quanto servirebbe perché il W5100 tiene in memoria i byte di un pacchetto che non è riuscito a trasmettere, e li aggiunge in testa alla prossima trasmissione creando pacchetti anomali.

Per questo motivo è necessario effettuare una trasmissione fittizia di “scarico buffer” verso un’unità sicuramente accesa e collegata via cavo, come il router.

Sulla mia LAN in fase notturna riscontro che per un’unità connessa in WiFi va a scarto circa lo 0.02% dei pacchetti ACK.

Dati per le sottoscrizioni

Adesso ci serve un posto per memorizzare i dati delle sottoscrizioni, cioè le coppie indirizzo unità e topic. Si possono quindi aggiungere alle variabili globali le seguenti dichiarazioni (per semplicità usiamo normali array):

// tabella sottoscrizioni topic/indirizzo
const uint16_t MAXSUBS = 200;
uint16_t       subs = 0;
uint8_t        mqttTopic[MAXSUBS] = { 0 };
uint8_t        mqttADDR[MAXSUBS]  = { 0 };

In totale possono esserci ‘MAXSUBS’ sottoscrizioni, il valore massimo possibile per ‘MAXSUBS’ è dato dalla memoria RAM occupata, due byte per ogni possibile sottoscrizione, quindi con il valore 200 allochiamo staticamente 400 byte di spazio.

Sottoscrivere

Quando arriva un pacchetto di tipo ‘S’ per richiedere una sottoscrizione bisogna prima vedere se è già esistente, poi se c’è ancora posto, ed infine (se non è già stata fatta e c’è ancora posto) aggiungere i nuovi dati agli array.

//----------------------------------------------------------
//  FUNZIONE   : subscribe
//  DESCRIZIONE: Aggiunge indirizzo/topic a tabella
//               sottoscrizioni. Se gia` esistente
//               o tabella piena annulla operazione
//  PARAMETRI  : topic 0..255 e indirizzo 1..254
//  RETURN CODE: ignorato
//----------------------------------------------------------
void subscribe(uint8_t topic, uint8_t addr)
{
    int16_t pos = -1;      // indice per posizione libera
    bool    giaPresente = false;  // stato sottoscrizione

    for (uint16_t i=0;  i < MAXSUBS;  i++)
    {
        uint8_t addrI = mqttADDR[i];
        if (0 == addrI){ 

            pos = i;              // posizione libera

        }else if ((addr == addrI) 
              and (topic == mqttTopic[i])){

            giaPresente = true;
            break;

        }else{
            // nothing
        }
    }

    if (!giaPresente  and  (pos >= 0))
    {
        mqttADDR[pos]  = addr;
        mqttTopic[pos] = topic;
    }
}

Pubblicare

Quando arriva un pacchetto di tipo ‘P’ per pubblicare un messaggio relativo a un topic, si scorre la lista delle sottoscrizioni alla ricerca di possibili destinatari, e si invia a ciascuno una copia del messaggio ricevuto sotto forma di pacchetto tipo ‘N’ (notifica).

//----------------------------------------------------------
//  FUNZIONE   : publish
//  DESCRIZIONE: Notifica un messaggio a tutti
//               i sottoscrittori
//  PARAMETRI  : nessuno
//  RETURN CODE: ignorato
//----------------------------------------------------------
void publish(void)
{
    uint8_t topic = udpBuffer[1];
    udpBuffer[0] = 'N';
    for (uint16_t i=0;  i < MAXSUBS;  i++)
    {
        udpADDR = mqttADDR[i];
        if ((udpADDR != 0) and (topic == mqttTopic[i]))
        { 
            sendPacket(true); 
        }
    }
}

Cancellare

Cancellazione delle sottoscrizioni di un’unità destinataria che non risponde alla richiesta ARP del W5100

//----------------------------------------------------------
//  FUNZIONE   : deleteSubscription
//  DESCRIZIONE: Cancella un indirizzo dalla tabella
//               sottoscrizioni
//  PARAMETRI  : indirizzo 1..254
//  RETURN CODE: ignorato
//----------------------------------------------------------
void deleteSubscription(uint8_t addr)
{
    for (uint16_t i=0;  i < MAXSUBS;  i++)
    {
        if (addr == mqttADDR[i]) { mqttADDR[i] = 0; }
    }
}

Una blanda sicurezza

In una LAN locale composta sempre dalle stesse unità con IP statici, è possibile aggiungere un minimo di sicurezza (davvero poca, ma comunque meglio di niente) facendo accettare alle unità solo pacchetti provenienti dall’indirizzo del server, e allo stesso modo facendo accettare al server solo pacchetti provenienti da un insieme di indirizzi autorizzati.

Si può quindi aggiungere alle variabili globali un array di indirizzi autorizzati (anche qui specificando solo l’ultimo byte).

const uint8_t   IPAUTHORIZED[] = { 18, 20, 22, 35, 122 };

E infine una funzione da richiamare per sapere se l’indirizzo attualmente presente nella variabile ‘udpADDR’ è autorizzato.

//----------------------------------------------------------
//  FUNZIONE   : authorized
//  DESCRIZIONE: Verifica se un IP e` autorizzato a
//               trasmettere al server
//  PARAMETRI  : L'ultimo byte dell'IP
//  RETURN CODE: true se autorizzato
//----------------------------------------------------------
bool authorized(uint8_t n)
{
    uint8_t nAuthorized = sizeof(IPAUTHORIZED);
    for (uint8_t i=0;  i < nAuthorized;  i++)
    {
        if (n == IPAUTHORIZED[i]) { return true; }
    }
    return false;
}

Basta quindi modificare la sola sezione di ricezione dei pacchetti:

        }else if (udpLen > 0){

            if (not authorized(udpADDR))
            {
                dropIncomingPacket();
            }
            else
            {
                Udp.read(udpBuffer, MAXBUF);
                processPacket();
            }

Se si volesse maggiore sicurezza, nel senso che le unità sappiano di star parlando al vero server e viceversa, bisognerebbe implementare uno scambio (magari periodico) di chiavi random, da “trattare” in modo non banale per ottenere un codice di verifica da aggiungere ai messaggi.

Questo però implicherebbe un’ulteriore tabella indirizzi/chiavi nel server, una funzione per ottenere e verificare il codice sia sulle unità che sul server, e magari una ulteriore blacklist lato server per le unità che sforino le cinque autenticazioni errate (tentativi brute force).

Pico server “MQTT”