IotMqttv5Sample

Beispiel zur MQTTv5-Kommunikation mithilfe einer Message-Queue

In diesem Beispiel wird die Kommunikation zu einem MQTT Broker unter Verwendung von MQTTv5 dargestellt. Als Voraussetzung muss der verwendete Message Broker MQTTv5 unterstützen. Es werden Nachrichten verschickt („Publish“) und empfangen („Subscribe“). Dies erfolgt in zwei Schritten. Zuerst wird anhand des Topics entschieden, welche Nachrichten empfangen werden sollen. Anschließend werden die empfangenen Nachrichten in einer Message Queue eingesammelt, um von hier aus gelesen und ausgewertet werden zu können. Das Sample verschickt und empfängt zu Demonstrationszwecken Nachrichten auf demselben Topic.

Handhabung der MessageQueue Im Gegensatz zum FB_IotMqttClient Baustein für MQTTv3, müssen Sie im FB_IotMqtt5Client Baustein die Message Queue nicht mehr separat als Objekt instanziieren und mit dem Baustein verknüpfen. Anstelle dessen können Sie direkt auf den entsprechenden Ausgang des Bausteins zugreifen um mit der Message Queue zu arbeiten.

Projektstruktur

PROGRAM PrgMqttCom
VAR
    fbMqttClient    : FB_IotMqtt5Client;
    bSetParameter   : BOOL := TRUE;
    bConnect        : BOOL := TRUE;
END_VAR

    (* published message *)
    sTopicPub : STRING(255) := 'MyTopic';
    sPayloadPub : STRING(255);
    fbTimer : TON := (PT:=T#1S);
    rSensor1 : REAL;
    nSensor2 : DINT;
    bSensor3 : BOOL;

    (* received message *)
    bSubscribed    : BOOL;
    sTopicSub      : STRING(255) := 'MyTopic';
    {attribute 'TcEncoding':='UTF-8'}
    sTopicRcv      : STRING(255);
    {attribute 'TcEncoding':='UTF-8'}
    sPayloadRcv    : STRING(255);
    fbMessage      : FB_IotMqtt5Message;

IF bSetParameter THEN
    bSetParameter := FALSE;
    fbMqttClient.sHostName := 'localhost';
    fbMqttClient.nHostPort := 1883;
END_IF

fbMqttClient.Execute(bConnect);

IF fbMqttClient.bConnected THEN
    IF NOT bSubscribed THEN
        bSubscribed := fbMqttClient.Subscribe(sTopic:=sTopicSub, eQoS:=TcIotMqttQos.AtMostOnceDelivery);
    END_IF
    fbTimer(IN:=TRUE);
    IF fbTimer.Q THEN // publish new payload every second
        fbTimer(IN:=FALSE);
        rSensor1 := rSensor1 + 0.1;
        nSensor2 := nSensor2 + 1;
        bSensor3 := NOT bSensor3;
        sPayloadPub := F_CreateMessage(rSensor1, nSensor2, bSensor3);
        fbMqttClient.Publish(sTopic:= sTopicPub, pPayload:= ADR(sPayloadPub), nPayloadSize:= LEN2(ADR(sPayloadPub))+1, eQoS:= TcIotMqttQos.AtMostOnceDelivery, bRetain:= FALSE, bQueue:= FALSE );
    END_IF
END_IF

IF fbMqttClient.fbMessageQueue.nQueuedMessages > 0 THEN
    IF fbMqttClient.fbMessageQueue.Dequeue(fbMessage:=fbMessage) THEN
        fbMessage.GetTopic(pTopic:=ADR(sTopicRcv), nTopicSize:=SIZEOF(sTopicRcv));
        fbMessage.GetPayload(pPayload:=ADR(sPayloadRcv), nPayloadSize:=SIZEOF(sPayloadRcv), bSetNullTermination:=FALSE);
    END_IF
END_IF

FUNCTION F_CreateMessage : STRING(255)
VAR_INPUT
  Sensor1 : REAL;
  Sensor2 : DINT;
  Sensor3 : BOOL;
END_VAR
VAR
  dtTimestamp : DATE_AND_TIME;
  timeAsFileTime : T_FILETIME64;
  fbJson : FB_JsonSaxWriter;
END_VAR

timeAsFileTime := F_GetSystemTime();
dtTimestamp := FILETIME64_TO_DT( timeAsFileTime );
fbJson.StartObject();
fbJson.AddKey('Timestamp');
fbJson.AddDateTime(dtTimestamp);
fbJson.AddKey('Values');
fbJson.StartObject();
fbJson.AddKey('Sensor1');
fbJson.AddReal(Sensor1);
fbJson.AddKey('Sensor2');
fbJson.AddDint(Sensor2);
fbJson.AddKey('Sensor3');
fbJson.AddBool(Sensor3);
fbJson.EndObject();
fbJson.EndObject();
F_CreateMessage := fbJson.GetDocument();
fbJson.ResetDocument();

Weitere Schritte

Das Beispiel kann abgewandelt werden für Applikationen, in denen sich auf mehrere Topics angemeldet wird. Dann werden MQTT-Nachrichten mit verschiedenen Topics empfangen. Die Auswertung der Nachrichten könnten Sie wie folgt für diesen Anwendungsfall erweitern:

VAR
    (* received payload for each subscribed topic *)
    sPayloadRcv1 : STRING(255);
    sPayloadRcv2 : STRING(255);
END_VAR
VAR CONSTANT
    (* subscriptions *)
    sTopicSub1 : STRING(255) := 'my first topic';
    sTopicSub2 : STRING(255) := 'my second topic'; 
END_VAR
----------------------------------------------------------------

IF fbMqttClient.fbMessageQueue.nQueuedMessages > 0 THEN
    IF fbMqttClient.fbMessageQueue.Dequeue(fbMessage:=fbMessage) THEN
        IF fbMessage.CompareTopic(sTopic:=sTopicSub1) THEN
            fbMessage.GetPayload(pPayload:=ADR(sPayloadRcv1), nPayloadSize:=SIZEOF(sPayloadRcv1), bSetNullTermination:=FALSE);
        ELSIF fbMessage.CompareTopic(sTopic:=sTopicSub2) THEN
            fbMessage.GetPayload(pPayload:=ADR(sPayloadRcv2), nPayloadSize:=SIZEOF(sPayloadRcv2), bSetNullTermination:=FALSE);
        END_IF
    END_IF
END_IF