aqhomed, aqhome: added MQTT endpoint, publish value changes via MQTT.

aqhome/mqtt/endpoint_mqttc.c

@@ -13,6 +13,12 @@
 
 #include "aqhome/mqtt/endpoint_mqttc_p.h"
 #include "aqhome/mqtt/msg_mqtt.h"
+#include "aqhome/mqtt/msg_mqtt_connect.h"
+#include "aqhome/mqtt/msg_mqtt_publish.h"
+#include "aqhome/msg/endpoint_node.h"
+#include "aqhome/msg/msg_node.h"
+#include "aqhome/msg/msg_value2.h"
+#include "aqhome/msg/msg_sendstats.h"
 
 #include <gwenhywfar/endpoint_tcpc.h>
 #include <gwenhywfar/debug.h>
@@ -23,10 +29,21 @@
 #define GWEN_ENDPOINT_MQTTC_BUFFERSIZE  1024
 
 
+
 GWEN_INHERIT(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC)
 
 
 static void GWENHYWFAR_CB _freeData(void *bp, void *p);
+static void _run(GWEN_MSG_ENDPOINT *ep);
+static void _sendConnectMsg(GWEN_MSG_ENDPOINT *ep);
+static void _checkForConnAckMsg(GWEN_MSG_ENDPOINT *ep);
+static void _processOutMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg);
+static void _processValue2Message(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg);
+static void _processSendStatsMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg);
+static void _publishDouble(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, double v);
+static void _publishInt(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, int v);
+static void _publishString(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, const char *v);
+static const char *_valueTypeToString(int t);
 static int _isMsgComplete(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *msg);
 static int _calcAndSetPayloadSizeAndOffset(GWEN_MSG *msg);
 
@@ -43,11 +60,16 @@ GWEN_MSG_ENDPOINT *AQH_MqttClientEndpoint_new(const char *host, int port, const
     DBG_INFO(AQH_LOGDOMAIN, "here");
     return NULL;
   }
+  AQH_NodeEndpoint_Extend(ep);
+  AQH_NodeEndpoint_SetAcceptedMsgGroups(ep, AQH_MSG_TYPEGROUP_ALL);
+
   GWEN_NEW_OBJECT(AQH_ENDPOINT_MQTTC, xep);
   GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep, xep, _freeData);
 
   GWEN_MsgEndpoint_SetDefaultBufferSize(ep, GWEN_ENDPOINT_MQTTC_BUFFERSIZE);
   GWEN_MsgEndpoint_SetIsMsgCompleteFn(ep, _isMsgComplete);
+  xep->previousRunFn=GWEN_MsgEndpoint_SetRunFn(ep, _run);
+  GWEN_MsgEndpoint_SetProcessOutMsgFn(ep, _processOutMessage);
 
   return ep;
 }
@@ -64,6 +86,305 @@ void _freeData(void *bp, void *p)
 
 
 
+uint16_t AQH_MqttClientEndpoint_GetNextPacketId(GWEN_MSG_ENDPOINT *ep)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep)
+      return ++(xep->lastPacketId);
+  }
+  return 0;
+}
+
+
+
+uint16_t AQH_MqttClientEndpoint_GetKeepAliveTime(const GWEN_MSG_ENDPOINT *ep)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep)
+      return xep->keepAliveTime;
+  }
+  return 0;
+}
+
+
+
+void AQH_MqttClientEndpoint_SetKeepAliveTime(GWEN_MSG_ENDPOINT *ep, uint16_t i)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep)
+      xep->keepAliveTime=i;
+  }
+}
+
+
+
+const char *AQH_MqttClientEndpoint_GetClientId(const GWEN_MSG_ENDPOINT *ep)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep)
+      return xep->clientId;
+  }
+  return NULL;
+}
+
+
+
+void AQH_MqttClientEndpoint_SetClientId(GWEN_MSG_ENDPOINT *ep, const char *s)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep) {
+      free(xep->clientId);
+      xep->clientId=s?strdup(s):NULL;
+    }
+  }
+}
+
+
+
+const char *AQH_MqttClientEndpoint_GetTopicPrefix(const GWEN_MSG_ENDPOINT *ep)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep)
+      return xep->topicPrefix;
+  }
+  return NULL;
+}
+
+
+
+void AQH_MqttClientEndpoint_SetTopicPrefix(GWEN_MSG_ENDPOINT *ep, const char *s)
+{
+  if (ep) {
+    AQH_ENDPOINT_MQTTC *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+    if (xep) {
+      free(xep->topicPrefix);
+      xep->topicPrefix=s?strdup(s):NULL;
+    }
+  }
+}
+
+
+
+void _run(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_ENDPOINT_MQTTC *xep;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+  if (xep) {
+    int state;
+
+    if (xep->previousRunFn)
+      xep->previousRunFn(ep);
+
+    state=GWEN_TcpcEndpoint_GetState(ep);
+    if (state==GWEN_MSG_ENDPOINT_TCPC_STATE_UNCONNECTED) {
+      int rv;
+
+      rv=GWEN_TcpcEndpoint_StartConnect(ep);
+      if (rv<0) {
+        DBG_INFO(AQH_LOGDOMAIN, "Error starting to connect (%d)", rv);
+      }
+    }
+    else if (state==GWEN_MSG_ENDPOINT_TCPC_STATE_CONNECTING) {
+      DBG_DEBUG(AQH_LOGDOMAIN, "Still connecting");
+    }
+    else if (state==GWEN_MSG_ENDPOINT_TCPC_STATE_CONNECTED)
+      _sendConnectMsg(ep);
+    else if (state==GWEN_ENDPOINT_MQTTC_STATE_WAITFORCONNACK)
+      _checkForConnAckMsg(ep);
+    else if (state==GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED){
+      /* nothing to do */
+    }
+    else {
+      DBG_ERROR(AQH_LOGDOMAIN, "Unhandled connection status %d", state);
+    }
+  }
+}
+
+
+
+void _sendConnectMsg(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_ENDPOINT_MQTTC *xep;
+  GWEN_MSG *msg;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+  /* send CONNECT */
+  msg=GWEN_ConnectMqttMsg_new("MQTT", 4, 0, xep->keepAliveTime, xep->clientId, NULL, NULL);
+  if (msg) {
+    DBG_INFO(AQH_LOGDOMAIN, "Sending MQTT CONNECT request.");
+    GWEN_MsgEndpoint_AddSendMessage(ep, msg);
+    GWEN_TcpcEndpoint_SetState(ep, GWEN_ENDPOINT_MQTTC_STATE_WAITFORCONNACK);
+  }
+}
+
+
+
+void _checkForConnAckMsg(GWEN_MSG_ENDPOINT *ep)
+{
+  GWEN_MSG *msg;
+
+  msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(ep);
+  if (msg) {
+    uint8_t msgType;
+
+    msgType=AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0;
+    if (msgType==AQH_MQTTMSG_MSGTYPE_CONNACK) {
+      DBG_INFO(AQH_LOGDOMAIN, "MQTT CONNACK received, logical connection established.");
+      GWEN_TcpcEndpoint_SetState(ep, GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED);
+    }
+    else {
+      DBG_ERROR(AQH_LOGDOMAIN, "Unexpected message received (%s)", AQH_MqttMsg_MsgTypeToString(msgType));
+    }
+    GWEN_Msg_free(msg);
+  }
+}
+
+
+
+void _processOutMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg)
+{
+  if (GWEN_TcpcEndpoint_GetState(ep)==GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED) {
+    DBG_DEBUG(AQH_LOGDOMAIN, "Processing output message");
+    switch(AQH_NodeMsg_GetMsgType(nodeMsg)) {
+    case AQH_MSG_TYPE_VALUE2:
+      _processValue2Message(ep, nodeMsg);
+      break;
+    case AQH_MSG_TYPE_COMSENDSTATS:
+      _processSendStatsMessage(ep, nodeMsg);
+      break;
+    default:
+      break;
+    }
+    GWEN_Msg_free(nodeMsg);
+  }
+}
+
+
+
+void _processValue2Message(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg)
+{
+  _publishDouble(ep,
+                 AQH_Value2Msg_GetUid(nodeMsg),
+                 AQH_Value2Msg_GetValueId(nodeMsg),
+                 _valueTypeToString(AQH_Value2Msg_GetValueId(nodeMsg)),
+                 AQH_Value2Msg_GetValue(nodeMsg));
+}
+
+
+
+void _processSendStatsMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg)
+{
+  uint16_t packetsOutInt;
+
+  packetsOutInt=AQH_SendStatsMsg_GetPacketsOut(nodeMsg);
+  if (packetsOutInt) {
+    double packetsOut;
+    double collisions;
+    double aborted;
+    double collisionsPercentage=0.0;
+    double abortedPercentage=0.0;
+
+    packetsOut=(double) packetsOutInt;
+    collisions=(double)AQH_SendStatsMsg_GetCollisions(nodeMsg);
+    aborted=(double)AQH_SendStatsMsg_GetAborted(nodeMsg);
+
+    collisionsPercentage=collisions*100.0/packetsOut;
+    abortedPercentage=aborted*100.0/packetsOut;
+
+    _publishInt(ep, AQH_SendStatsMsg_GetUid(nodeMsg), 0, "net/packetsOut", packetsOutInt);
+    _publishInt(ep, AQH_SendStatsMsg_GetUid(nodeMsg), 0, "net/collisions", (int) AQH_SendStatsMsg_GetCollisions(nodeMsg));
+    _publishDouble(ep, AQH_SendStatsMsg_GetUid(nodeMsg), 0, "net/collisionsPercent", collisionsPercentage);
+    _publishDouble(ep, AQH_SendStatsMsg_GetUid(nodeMsg), 0, "net/abortedPercent", abortedPercentage);
+  }
+}
+
+
+
+void _publishDouble(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, double v)
+{
+  char numBuf[16];
+
+  snprintf(numBuf, sizeof(numBuf)-1, "%f", v);
+  numBuf[sizeof(numBuf)-1]=0;
+  _publishString(ep, uid, valueId, valuePath, numBuf);
+}
+
+
+
+void _publishInt(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, int v)
+{
+  char numBuf[16];
+
+  snprintf(numBuf, sizeof(numBuf)-1, "%d", v);
+  numBuf[sizeof(numBuf)-1]=0;
+  _publishString(ep, uid, valueId, valuePath, numBuf);
+}
+
+
+
+void _publishString(GWEN_MSG_ENDPOINT *ep, uint32_t uid, int valueId, const char *valuePath, const char *v)
+{
+  AQH_ENDPOINT_MQTTC *xep;
+  GWEN_BUFFER *bufTopic;
+  GWEN_MSG *pubMsg;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_MQTTC, ep);
+
+  bufTopic=GWEN_Buffer_new(0, 64, 0, 1);
+  if (valueId>0)
+    GWEN_Buffer_AppendArgs(bufTopic, "%s/%08x/%d/%s",
+                           xep->topicPrefix,
+                           uid,
+                           valueId,
+                           valuePath);
+  else
+    GWEN_Buffer_AppendArgs(bufTopic, "%s/%08x/%s",
+                           xep->topicPrefix,
+                           uid,
+                           valuePath);
+
+  pubMsg=GWEN_PublishMqttMsg_new(0, 0, GWEN_Buffer_GetStart(bufTopic), (const uint8_t*) v, strlen(v));
+  if (pubMsg) {
+    DBG_INFO(AQH_LOGDOMAIN, "MQTT PUBLISH %s: %s", GWEN_Buffer_GetStart(bufTopic), v);
+    GWEN_MsgEndpoint_AddSendMessage(ep, pubMsg);
+  }
+  GWEN_Buffer_free(bufTopic);
+}
+
+
+
+const char *_valueTypeToString(int t)
+{
+  switch(t) {
+  case AQH_MSG_VALUE2_TYPE_TEMP:     return "temperature";
+  case AQH_MSG_VALUE2_TYPE_HUMIDITY: return "humidity";
+  default:                           return "unknown";
+  }
+}
+
+
+
 int _isMsgComplete(GWEN_UNUSED GWEN_MSG_ENDPOINT *ep, GWEN_MSG *msg)
 {
   int rv;

aqhome/mqtt/endpoint_mqttc.h

@@ -14,10 +14,21 @@
 
 #include <gwenhywfar/endpoint.h>
 
+#define GWEN_ENDPOINT_MQTTC_STATE_WAITFORCONNACK (GWEN_MSG_ENDPOINT_TCPC_STATE_NEXTFREE+0)
+#define GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED    (GWEN_MSG_ENDPOINT_TCPC_STATE_NEXTFREE+1)
 
 
 AQHOME_API GWEN_MSG_ENDPOINT *AQH_MqttClientEndpoint_new(const char *host, int port, const char *name, int groupId);
 
+AQHOME_API uint16_t AQH_MqttClientEndpoint_GetNextPacketId(GWEN_MSG_ENDPOINT *ep);
+AQHOME_API uint16_t AQH_MqttClientEndpoint_GetKeepAliveTime(const GWEN_MSG_ENDPOINT *ep);
+AQHOME_API void AQH_MqttClientEndpoint_SetKeepAliveTime(GWEN_MSG_ENDPOINT *ep, uint16_t i);
+
+AQHOME_API const char *AQH_MqttClientEndpoint_GetClientId(const GWEN_MSG_ENDPOINT *ep);
+AQHOME_API void AQH_MqttClientEndpoint_SetClientId(GWEN_MSG_ENDPOINT *ep, const char *s);
+
+AQHOME_API const char *AQH_MqttClientEndpoint_GetTopicPrefix(const GWEN_MSG_ENDPOINT *ep);
+AQHOME_API void AQH_MqttClientEndpoint_SetTopicPrefix(GWEN_MSG_ENDPOINT *ep, const char *s);
 
 
 

aqhome/mqtt/endpoint_mqttc_p.h

@@ -15,7 +15,12 @@
 
 typedef struct AQH_ENDPOINT_MQTTC AQH_ENDPOINT_MQTTC;
 struct AQH_ENDPOINT_MQTTC {
-  int dummy;
+  char *clientId;
+  char *topicPrefix;
+  uint16_t lastPacketId;
+  uint16_t keepAliveTime;
+
+  GWEN_MSG_ENDPOINT_RUN_FN previousRunFn;
 };
 
 

aqhome/mqtt/msg_mqtt_pubresponse.c

@@ -35,7 +35,7 @@ GWEN_MSG *GWEN_PubResponseMqttMsg_new(uint8_t typeAndFlags, uint16_t pkgId)
 
 
 
-void AQH_PubResponseMqttMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf, const char *sText)
+uint16_t AQH_PubResponseMqttMsg_GetPacketId(const GWEN_MSG *msg)
 {
   const uint8_t *msgPtr;
   uint32_t msgLen;
@@ -47,12 +47,24 @@ void AQH_PubResponseMqttMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf,
     const uint8_t *payloadPtr;
     uint32_t payloadLen;
 
-    GWEN_Buffer_AppendArgs(dbuf, "%s %s", AQH_MqttMsg_MsgTypeToString(msgPtr[0] & 0xf0), sText);
     payloadLen=GWEN_Msg_GetParsedPayloadSize(msg);
     payloadPtr=msgPtr+GWEN_Msg_GetParsedPayloadOffset(msg);
-    if (payloadLen>=2)
-      GWEN_Buffer_AppendArgs(dbuf, "(packet id=%04x)", (payloadPtr[0]<<8)+payloadPtr[1]);
+    if (payloadPtr && payloadLen>=2)
+      return (payloadPtr[0]<<8)+payloadPtr[1];
   }
+  return 0;
+}
+
+
+
+void AQH_PubResponseMqttMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf, const char *sText)
+{
+  uint8_t msgType;
+  uint16_t packetId;
+
+  msgType=AQH_MqttMsg_GetMsgTypeAndFlags(msg);
+  packetId=AQH_PubResponseMqttMsg_GetPacketId(msg);
+  GWEN_Buffer_AppendArgs(dbuf, "%s %s (packet id: %04x)", AQH_MqttMsg_MsgTypeToString(msgType & 0xf0), sText, packetId);
 }
 
 

aqhome/mqtt/msg_mqtt_pubresponse.h

@@ -22,6 +22,7 @@
  * Use for PUBACK, PUBREC, PUBREL and PUBCOMP.
  */
 AQHOME_API GWEN_MSG *GWEN_PubResponseMqttMsg_new(uint8_t typeAndFlags, uint16_t pkgId);
+AQHOME_API uint16_t AQH_PubResponseMqttMsg_GetPacketId(const GWEN_MSG *msg);
 AQHOME_API void AQH_PubResponseMqttMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf, const char *sText);