aqhomecontrol/apps/aqhome-mqttlog/mqtt.c

/****************************************************************************
 * 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 && rv!=GWEN_ERROR_IN_PROGRESS) {
      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;
  }
  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;
}



int AqHomeMqttLog_Ping(GWEN_MSG_ENDPOINT2 *epTcp)
{
  GWEN_MSG *msgOut;

  DBG_INFO(AQH_LOGDOMAIN, "Sending PING");
  msgOut=GWEN_MqttMsg_new(AQH_MQTTMSG_MSGTYPE_PINGREQ, 0, NULL);
  if (msgOut==NULL) {
    DBG_ERROR(NULL, "Error creating message");
    return GWEN_ERROR_INTERNAL;
  }
  GWEN_MsgEndpoint2_AddSendMessage(epTcp, msgOut);

  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;
}