MQTTClient, client for a MQTT server

• In questo articolo informazioni sul protocollo MQTT.
• In questo articolo come inviare dati alla piattaforma cloud Beebotte.

• Se l’oggetto aggiornato non è nell’ultima versione del package, vedi capitolo “Aggiornamento librerie” in questo articolo.
• Gli oggetti obsoleti sono inseriti nella libreria eLLabObsoleteLib fare riferimento al relativo manuale ed al manuale programmazione in formato pdf.

MQTTClient_v1

Per poter gestire la connessione in TLS è stata eliminata la gestione della connessione TCP, ora deve essere gestita esternamente ed occorre passare al FB lo stream di comunicazione nel parametro File.

MQTTClient_v2

Gestita allocazione automatica memoria in base a dimensione messaggio, ora il limite della dimensione è dato solo dalla memoria rilocabile a disposizione. Ritornati in uscita anzichè passati in ingresso, indirizzo di allocazione e dimensione del nome e valore del topic, i valori devono essere acquisiti su attivazione di Received.

MQTTClient_v3

Eliminati i parametri Publish, Subscribe, TxFlags, TxQoS, TxTopic, TxValue, TxVLength, Published, Subscribed, ora per la pubblicazione dei topics si utilizza il metodo Publish che si appoggia ad un registro FIFO. Parametro KeepAlive ora è di tipo TIME, inserito parametro Delay per impostazione tempo attesa invio dati al broker.

Con la connessione al broker (Comando CONNECT) è inviato anche il byte CFlags. I bits 7, 6 e 0 sono gestiti automaticamente dal FB.

+--------+--------+-----------+----+----+---------+-------------+--------+
|Username|Password|Will retain|Will QoS |Will flag|Clean session|Reserved|
+--------+--------+-----------+----+----+---------+-------------+--------+

• [7] Username: Gestita utomaticamente.
• [6] Password: Gestita utomaticamente.
• [5] Will retain:
• [4, 3] Will Qos:
• [2] Will flag:
• [1] Clean session: Se non attivo il broker riprende le comunicazioni con il client in base alla sessione identificata dal ClientID. Se non esiste alcuna sessione ne crea una nuova. Se attivo il broker elimina la sessione precedente e ne crea una nuova.
• [0] Reserved: Sempre 0.

I client si connettono ai broker per pubblicare o sottoscrivere degli argomenti (Topics), un client può essere sia publisher che subscriber. I client non comunicano direttamente tra di loro, i messaggi sono pubblicati e consegnati ad un broker MQTT. Quando un client si connette ad un broker, la connessione può essere persistente o non persistente (sessione pulita).

• Sessione non persistente: Può essere utilizzata dai i client che pubblicano solo messaggi e non si sottoscrivono ad alcun argomento. Il broker non immagazzina alcuna informazione dei messaggi non consegnati ai client. Alla disconnessione si perdono le sottoscrizioni e un client deve sottoscriversi nuovamente quando si riconnette al broker. I client informano il broker delle loro intenzioni utilizzando il flag di sessione pulita, che indica se è richiesta una sessione persistente.
• Sessione persistente: Una sessione persistente viene identificata da un client ID e permette ai client che sono offline di ricevere i messaggi che vengono inviati mentre sono disconnessi.

Si possono sottoscrivere uno o più topic, nella sottoscrizione si può specificare il nome esatto del topic al quale ci si vuole sottoscrivere o in alternativa è possibile utilizzare le wildcards, ovvero caratteri speciali per sottoscrivere un insieme di topics. Le wildcards non sono accettate nella pubblicazione.

• Wildcard (+): Permette di sottoscrivere un singolo livello. Vediamo un esempio: sottoscrivendo il topic myhome/+/temperature si riceveranno tutti i topics che contengono una qualsiasi scritta al posto del wildcard, ad esempio myhome/bedroom/temperature, myhome/livingroom/temperature, myhome/kitchen/temperature, ecc.
• Wildcard (#): Permette di sottoscrivere un multilivello. Il carattere deve occupare sempre la posizione finale del topic e deve essere preceduto dal carattere /. Ad esempio sottoscrivendo il topic myhome/bedroom/# si riceveranno tutti itopics che iniziano con la definizione che precede il wildcard, ad esempio myhome/bedroom/temperature, myhome/bedroom/humidity, myhome/bedroom/airquality, ecc.

Questo metodo permette di compilare la struttura dati MQTT_TS_DATA di sottoscrizione topic sul broker. Quando il valore del topic sul broker cambia viene automaticamente aggiornata la variabile definita. E’ possibile gestire un array di strutture dati, una per ogni topic, nel campo IDx indicare l’indice della struttura da compilare. Nel caso si desideri eseguire la sola sottoscrizione delegando la ricezione dei dati dal broker al programma utente occorre definire DBuffer:=eNULL e DBSize:=0.

IDx (UDINT) Indice struttura dati.
Topic (@STRING) Definizione topic da sottoscrivere.
DBuffer (PVOID) Definizione indirizzo buffer valore topic da aggiornare su ricezione da broker.
DBSize (UDINT) Dimensione buffer valore.
QoS (USINT) Quality of service

Il metodo ritorna un (BOOL) FALSE se errore esecuzione, TRUE se esecuzione corretta.

Permette di pubblicare un topic sul broker, il valore presente nel buffer PValue viene inserito nel registro FIFO di appoggio. Questo permette di pubblicare più topic contemporaneamente, il FB provvederà ad eseguire lo spool dei dati dal FIFO ed a eseguirne la pubblicazione in base alla disponibilità del broker. Se la pubblicazione di un record presente nel FIFO non và a buon fine dopo alcuni tentativi viene scartato continuando la pubblicazione dei records successivi. Le variabili sono pubblicate come array di bytes come sono presenti in memoria mantenendo l’endianness del sistema. Per pubblicare valori numerici occorre utilizzare variabili stringa.

Topic (@STRING) Definizione topic da sottoscrivere.
PValue (PVOID) Definizione indirizzo buffer valore topic da pubblicare sul broker.
PVLength (UDINT) Dimensione valore da pubblicare, non definire se variabile stringa (Massimo 512-lunghezza nome topic).
TRetain (BOOL) Retain topic
QoS (USINT) Quality of service pubblicazione topic. Valori accettati 0 e 1.

Il metodo ritorna un (BOOL) FALSE se errore esecuzione, TRUE se esecuzione corretta.

• QoS:=0: Il topic viene pubblicato senza alcuna risposta da parte del broker.
• QoS:=1: Alla pubblicazione del topic il broker risponde con un messaggio PUBACK di conferma. Se non ricevuto il FB esegue una nuova pubblicazione.

PROGRAM ST_MQTTClient
VAR
    Publish : BOOL; (* Publish command *)
    BOOLValue : ARRAY[0..1] OF BOOL := [true, false]; (* BOOL value [Tx][Rx] *)
    REALValue : ARRAY[0..1] OF REAL := [10.0, 0.0]; (* REAL value [Tx][Rx] *)
    STRINGValue : ARRAY[0..1] OF STRING[ 16 ]; (* STRING value [Rx] *)
    SPData : ARRAY[0..2] OF MQTT_TS_DATA; (* MQTT topic subscribe data *)
    TCPClient : SysTCPClient; (* TCP client management *)
    FIFO : FIFOFile_v1; (* FIFO on file *)
    MQTT : MQTTClient_v3; (* MQTT client FB *)
END_VAR

// *****************************************************************************
// PROGRAM "ST_MQTTClient"
// *****************************************************************************
// This program connects to a free broker, it shows how to subscribe and
// publish topics on it.
// -----------------------------------------------------------------------------

    // -------------------------------------------------------------------------
    // INITIALIZATION
    // -------------------------------------------------------------------------
    // Initialize the FBs parameters.

    IF (SysFirstLoop) THEN

        // Set FIFO parameters.

        FIFO.FIFOFilename:=eNULL; //Path and name of FIFO file
        FIFO.FIFOSize:=2048; //FIFO file size
        FIFO.FIFOIDx:=eNULL; //FIFO indexes

        // Set TCPClient parameters.

        TCPClient.PeerAdd:=ADR('broker.mqttdashboard.com'); //Peer address
        TCPClient.PeerPort:=1883; //Peer port
        TCPClient.LocalAdd:=ADR('0.0.0.0'); //Local address
        TCPClient.LocalPort:=0; //Local port
        TCPClient.FlushTm:=0; //Flush time (mS)
        TCPClient.LifeTm:=90; //Life time (S)
        TCPClient.RxSize:=512; //Rx buffer size
        TCPClient.TxSize:=512; //Tx buffer size

        // Set MQTTClient parameters.

        MQTT.SpyOn:=TRUE; //Spy active
        MQTT.FIFOFile:=ADR(FIFO); //FIFO on file
        MQTT.CFlags:=16#02; //Clean session
        MQTT.Username:=eNULL; //Broker username
        MQTT.Password:=eNULL; //Broker password
        MQTT.ClientID:=eNULL; //Client identifier
        MQTT.KeepAlive:=T#90s; //Keep alive time
        MQTT.Delay:=T#2s; //Send delay time
        MQTT.Timeout:=T#5s; //Execution timeout

        // Topic subscribe definitions.

        MQTT.TSData:=ADR(SPData); //Topic subscribe data
        MQTT.TSNumber:=SIZEOF(SPData)/SIZEOF(SPData[0]); //Topic subscribe number

        eTO_JUNK(MQTT.Subscribe(0, ADR('ElsistTest/BOOLValue'), ADR(BOOLValue[1]), SIZEOF(BOOLValue[1]), 0));
        eTO_JUNK(MQTT.Subscribe(1, ADR('ElsistTest/REALValue'), ADR(REALValue[1]), SIZEOF(REALValue[1]), 0));
        eTO_JUNK(MQTT.Subscribe(2, ADR('ElsistTest/STRINGValue'), ADR(STRINGValue[1]), 16, 0));
    END_IF;

    // -------------------------------------------------------------------------
    // MQTT MANAGEMENT
    // -------------------------------------------------------------------------
    // FBs management.

    TCPClient(Connect:=MQTT.Connect); //TCPClient management
    MQTT(Enable:=TRUE, File:=TCPClient.File); //MQTTClient management

    // To publish topics set by debug the "Publish" variable.

    IF NOT(Publish) THEN RETURN; END_IF;
    Publish:=FALSE; //Publish command
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/BOOLValue'), ADR(BOOLValue[0]), SIZEOF(BOOLValue[0]), FALSE, 1));
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/REALValue'), ADR(REALValue[0]), SIZEOF(REALValue[0]), FALSE, 1));

    eTO_JUNK(SysVsnprintf(ADR(STRINGValue[0]), 16, ADR('%.3f'), REAL_TYPE, ADR(REALValue[0])));
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/STRINGValue'), ADR(STRINGValue[0]), LEN(STRINGValue[0]), FALSE, 1));

// [End of file]

PROGRAM ST_TLSMQTTClient
VAR
    Publish : BOOL; (* Publish command *)
    BOOLValue : ARRAY[0..1] OF BOOL := [true, false]; (* BOOL value [Tx][Rx] *)
    REALValue : ARRAY[0..1] OF REAL := [10.0, 0.0]; (* REAL value [Tx][Rx] *)
    STRINGValue : ARRAY[0..1] OF STRING[ 16 ]; (* STRING value [Rx] *)
    SPData : ARRAY[0..2] OF MQTT_TS_DATA; (* MQTT topic subscribe data *)
    TCPClient : SysTCPClient; (* TCP client management *)
    TLSClient : SysTLSClient; (* TLS client *)
    FIFO : FIFOFile_v1; (* FIFO on file *)
    MQTT : MQTTClient_v3; (* MQTT client FB *)
END_VAR

// *****************************************************************************
// PROGRAM "ST_MQTTSpool"
// *****************************************************************************
// This program connects to a free broker, it shows how to subscribe and
// publish topics on it.
// -----------------------------------------------------------------------------

    // -------------------------------------------------------------------------
    // INITIALIZATION
    // -------------------------------------------------------------------------
    // Initialize the FBs parameters.

    IF (SysFirstLoop) THEN

        // Set FIFO parameters.

        FIFO.FIFOFilename:=eNULL; //Path and name of FIFO file
        FIFO.FIFOSize:=2048; //FIFO file size
        FIFO.FIFOIDx:=eNULL; //FIFO indexes

        // Set TCPClient parameters.

        TCPClient.PeerAdd:=ADR('mqtt-dashboard.com'); //Peer address
        TCPClient.PeerPort:=8883; //Peer port
        TCPClient.LocalAdd:=ADR('0.0.0.0'); //Local address
        TCPClient.LocalPort:=0; //Local port
        TCPClient.FlushTm:=0; //Flush time (mS)
        TCPClient.LifeTm:=90; //Life time (S)
        TCPClient.RxSize:=512; //Rx buffer size
        TCPClient.TxSize:=512; //Tx buffer size

        // Set TLSClient parameters.

        TLSClient.SpyOn:=FALSE; //Spy On
        TLSClient.Mode:=TLS_MODE_TCP; //Connection mode
        TLSClient.Server:=TCPClient.PeerAdd; //Server name
        TLSClient.CipherSuites:=eNULL; //Cipher suites supported
        TLSClient.Psk:=eNULL; //Preshared key
        TLSClient.PskIdentity:=eNULL; //Preshared key identity

        // Without server certification.

        TLSClient.CAVerify:=TLS_CA_NONE; //Certificates verify type
        TLSClient.CAFile:=eNULL; //CA file

        // Set MQTTClient parameters.

        MQTT.SpyOn:=TRUE; //Spy active
        MQTT.FIFOFile:=ADR(FIFO); //FIFO on file
        MQTT.CFlags:=16#02; //Clean session
        MQTT.Username:=eNULL; //Broker username
        MQTT.Password:=eNULL; //Broker password
        MQTT.ClientID:=eNULL; //Client identifier
        MQTT.KeepAlive:=T#90s; //Keep alive time
        MQTT.Delay:=T#2s; //Send delay time
        MQTT.Timeout:=T#5s; //Execution timeout

        // Topic subscribe definitions.

        MQTT.TSData:=ADR(SPData); //Topic subscribe data
        MQTT.TSNumber:=SIZEOF(SPData)/SIZEOF(SPData[0]); //Topic subscribe number

        eTO_JUNK(MQTT.Subscribe(0, ADR('ElsistTest/BOOLValue'), ADR(BOOLValue[1]), SIZEOF(BOOLValue[1]), 0));
        eTO_JUNK(MQTT.Subscribe(1, ADR('ElsistTest/REALValue'), ADR(REALValue[1]), SIZEOF(REALValue[1]), 0));
        eTO_JUNK(MQTT.Subscribe(2, ADR('ElsistTest/STRINGValue'), ADR(STRINGValue[1]), 16, 0));
    END_IF;

    // -------------------------------------------------------------------------
    // MQTT SPOOLING MANAGEMENT
    // -------------------------------------------------------------------------
    // FBs management.

    TCPClient(Connect:=MQTT.Connect); //TCPClient management
    TLSClient(IFile:=TCPClient.File); //TLS client management
    MQTT(Enable:=TRUE, File:=TLSClient.OFile); //MQTTClient management

    // To publish topics set by debug the "Publish" variable.

    IF NOT(Publish) THEN RETURN; END_IF;
    Publish:=FALSE; //Publish command
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/BOOLValue'), ADR(BOOLValue[0]), SIZEOF(BOOLValue[0]), FALSE, 1));
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/REALValue'), ADR(REALValue[0]), SIZEOF(REALValue[0]), FALSE, 1));

    eTO_JUNK(SysVsnprintf(ADR(STRINGValue[0]), 16, ADR('%.3f'), REAL_TYPE, ADR(REALValue[0])));
    eTO_JUNK(MQTT.Publish(ADR('ElsistTest/STRINGValue'), ADR(STRINGValue[0]), LEN(STRINGValue[0]), FALSE, 1));

// [End of file]