aqhome: completed adapting to msgio2 interface.

apps/aqhome-mqttlog/mqtt.c

@@ -0,0 +1,179 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 Martin Preuss, all rights reserved.
+ *
+ * The license for this file can be found in the file COPYING which you
+ * should have received along with this file.
+ ****************************************************************************/
+
+#ifdef HAVE_CONFIG_H
+# include <config.h>
+#endif
+
+#include "./mqtt.h"
+
+#include <aqhome/api.h>
+#include <aqhome/aqhome.h>
+
+#include "aqhome/mqtt/endpoint2_mqttc.h"
+#include <aqhome/mqtt/msg_mqtt_connect.h>
+#include <aqhome/mqtt/msg_mqtt_connack.h>
+#include <aqhome/mqtt/msg_mqtt_publish.h>
+#include <aqhome/mqtt/msg_mqtt_subscribe.h>
+#include <aqhome/mqtt/msg_mqtt_suback.h>
+
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/text.h>
+
+#include <time.h>
+#include <unistd.h>
+
+
+
+/* ------------------------------------------------------------------------------------------------
+ * defines
+ * ------------------------------------------------------------------------------------------------
+ */
+
+#define AQHOME_MQTTLOG_DEFAULT_CMDTIMEOUT 10000
+
+
+
+/* ------------------------------------------------------------------------------------------------
+ * forward declarations
+ * ------------------------------------------------------------------------------------------------
+ */
+
+static GWEN_MSG *_awaitPacket(GWEN_MSG_ENDPOINT2 *epTcp, uint8_t expectedPacketType, int timeoutInSeconds);
+
+
+
+/* ------------------------------------------------------------------------------------------------
+ * implementations
+ * ------------------------------------------------------------------------------------------------
+ */
+
+GWEN_MSG_ENDPOINT2 *AqHomeMqttLog_CreateMqttEndpoint(GWEN_DB_NODE *dbArgs)
+{
+  const char *mqttAddress;
+  int mqttPort;
+  const char *mqttClientId;
+  const char *mqttTopicPrefix;
+  int mqttKeepAlive;
+
+  mqttAddress=GWEN_DB_GetCharValue(dbArgs, "mqttAddress", 0, NULL);
+  mqttPort=GWEN_DB_GetIntValue(dbArgs, "mqttPort", 0, 1883);
+  mqttClientId=GWEN_DB_GetCharValue(dbArgs, "mqttClientId", 0, "aqhome-mqttlog");
+  mqttTopicPrefix=GWEN_DB_GetCharValue(dbArgs, "mqttTopicPrefix", 0, "aqhome/sensors");
+  mqttKeepAlive=GWEN_DB_GetIntValue(dbArgs, "mqttKeepAlive", 0, 600);
+
+  if (mqttAddress && *mqttAddress && mqttPort) {
+    GWEN_MSG_ENDPOINT2 *epMqtt;
+
+    DBG_INFO(AQH_LOGDOMAIN, "Connecting to %s (port %d)", mqttAddress, mqttPort);
+    epMqtt=AQH_MqttClientEndpoint2_new(mqttClientId, mqttAddress, mqttPort, NULL, 0);
+    if (epMqtt==NULL) {
+      DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP");
+      return NULL;
+    }
+    if (mqttTopicPrefix && *mqttTopicPrefix)
+      AQH_MqttClientEndpoint2_SetTopicPrefix(epMqtt, mqttTopicPrefix);
+    AQH_MqttClientEndpoint2_SetKeepAliveTime(epMqtt, mqttKeepAlive);
+
+    return epMqtt;
+  }
+
+  return NULL;
+}
+
+
+
+int AqHomeMqttLog_MqttConnect(GWEN_MSG_ENDPOINT2 *epTcp)
+{
+  if (GWEN_MsgEndpoint2_GetState(epTcp)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
+    int rv;
+
+    rv=AQH_MqttClientEndpoint2_StartConnect(epTcp);
+    if (rv<0) {
+      DBG_ERROR(NULL, "Error starting to connect (%d)", rv);
+      return rv;
+    }
+  }
+
+  while(GWEN_MsgEndpoint2_GetState(epTcp)!=GWEN_MSG_ENDPOINT_STATE_CONNECTED) {
+    DBG_DEBUG(NULL, "Next loop");
+    GWEN_MsgEndpoint2_IoLoop(epTcp, 2000); /* 2000 ms */
+  }
+  return 0;
+}
+
+
+
+int AqHomeMqttLog_Subscribe(GWEN_MSG_ENDPOINT2 *epTcp, const char *topicFilter)
+{
+  uint16_t pckId;
+  GWEN_MSG *msgOut;
+  GWEN_MSG *msgIn;
+
+  DBG_INFO(NULL, "Sending SUBSCRIBE %s", topicFilter);
+  pckId=AQH_MqttClientEndpoint2_GetNextPacketId(epTcp);
+  msgOut=GWEN_SubscribeMqttMsg_new(AQH_MQTTMSG_MSGTYPE_SUBSCRIBE, pckId, topicFilter, 0);
+  if (msgOut==NULL) {
+    DBG_ERROR(NULL, "Error creating message");
+    return GWEN_ERROR_INTERNAL;
+  }
+  DBG_ERROR(NULL, "Sending this message:");
+  GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msgOut), GWEN_Msg_GetBytesInBuffer(msgOut), 2);
+
+  GWEN_MsgEndpoint2_AddSendMessage(epTcp, msgOut);
+
+  DBG_INFO(NULL, "Waiting for response");
+  msgIn=_awaitPacket(epTcp, AQH_MQTTMSG_MSGTYPE_SUBACK, AQHOME_MQTTLOG_DEFAULT_CMDTIMEOUT);
+  if (msgIn) {
+    GWEN_BUFFER *buf;
+
+    buf=GWEN_Buffer_new(0, 256, 0, 1);
+    AQH_SubAckMqttMsg_DumpToBuffer(msgIn, buf, "received");
+    DBG_INFO(NULL, "%s", GWEN_Buffer_GetStart(buf));
+    GWEN_Buffer_free(buf);
+    GWEN_Msg_free(msgIn);
+  }
+
+  return 0;
+}
+
+
+
+GWEN_MSG *_awaitPacket(GWEN_MSG_ENDPOINT2 *epTcp, uint8_t expectedPacketType, int timeoutInSeconds)
+{
+  time_t startTime;
+
+  startTime=time(NULL);
+
+  for (;;) {
+    GWEN_MSG *msg;
+    time_t now;
+
+    GWEN_MsgEndpoint2_IoLoop(epTcp, 2000); /* 2000 ms */
+    msg=GWEN_MsgEndpoint2_TakeFirstReceivedMessage(epTcp);
+    if (msg) {
+      if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(expectedPacketType & 0xf0)) {
+        return msg;
+      }
+      else {
+        DBG_ERROR(NULL, "Received this message:");
+        GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
+      }
+      GWEN_Msg_free(msg);
+    }
+    now=time(NULL);
+    if (now-startTime>timeoutInSeconds) {
+      DBG_INFO(NULL, "Timeout");
+      break;
+    }
+  }
+
+  return NULL;
+}
+
+