aqhomecontrol/aqhome/ipc2/mqtt_msgreader.c

/****************************************************************************
 * This file is part of the project AqHome.
 * AqHome (c) by 2025 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_msgreader.h"
#include "./msgreader_p.h"
#include <aqhome/events2/fdobject.h>

#include <gwenhywfar/inherit.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endianfns.h>
#include <gwenhywfar/text.h>

#include <sys/socket.h>


#define AQH_MSG_READER_HEADER_SIZE 2
#define AQH_MSG_READER_MINMSGSIZE  12
#define AQH_MSG_READER_MAXMSGSIZE 10240

#define AQH_MSG_READER_FLAGS_READBODY 0x0001


/* ------------------------------------------------------------------------------------------------
 * forward declarations
 * ------------------------------------------------------------------------------------------------
 */

static int _readMsg(AQH_OBJECT *o);
static int _readHeaderFromRingbuffer(AQH_MSG_READER *xo);



/* ------------------------------------------------------------------------------------------------
 * implementation
 * ------------------------------------------------------------------------------------------------
 */

AQH_OBJECT *AQH_MqttMsgReader_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *fdObject)
{
  AQH_OBJECT *o;

  o=AQH_MsgReader_new(eventLoop, fdObject);
  AQH_MsgReader_SetReadMsgFn(o, _readMsg);

  return o;
}



int _readMsg(AQH_OBJECT *o)
{
  AQH_MSG_READER *xo;

  xo=AQH_MsgReader_GetData(o);
  if (xo) {
    int rv;

    if (!(xo->flags & AQH_MSG_READER_FLAGS_READBODY)) {
      DBG_INFO(AQH_LOGDOMAIN, "Reading header");
      rv=_readHeaderFromRingbuffer(xo);
      if (rv<0) {
	DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
	return rv;
      }
    }

    if (xo->flags & AQH_MSG_READER_FLAGS_READBODY) {
      DBG_INFO(AQH_LOGDOMAIN, "Reading body");
      /* reading remainder of msg directly into allocated buffer */
      rv=AQH_MsgReader_ReadRemainderFromRingbuffer(o);
      if (rv<0) {
	DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
	return rv;
      }
      else if (rv==1) {
	int msgLen;
	uint8_t *msgPtr;

	msgLen=xo->bytesReceived;
	msgPtr=xo->currentMsgBuf;

	xo->bytesReceived=0;
	xo->bytesLeft=0;
	xo->currentMsgBuf=NULL;
        xo->flags&=~AQH_MSG_READER_FLAGS_READBODY;

        DBG_INFO(NULL, "Received message:");
        //GWEN_Text_LogString((const char*) msgPtr, msgLen, NULL, GWEN_LoggerLevel_Error);
	rv=AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_MSGRECVD, msgLen, (void*) msgPtr);
	if (rv==0) {
	  DBG_ERROR(AQH_LOGDOMAIN, "Received message ignored");
	}
        free(msgPtr);
	return 1;
      }
    }
    return 0;
  }
  return GWEN_ERROR_GENERIC;
}



int _readHeaderFromRingbuffer(AQH_MSG_READER *xo)
{
  int remainingBytesInBuffer;
  uint32_t mqttPayloadLen=0;
  int idx=0;
  const uint8_t *ptr;
  uint8_t lenByte;
  int shift=0;
  int rv;

  ptr=xo->headerBuffer;

  remainingBytesInBuffer=xo->bytesReceived;
  if (remainingBytesInBuffer>AQH_MSG_READER_HEADERBUFFER_SIZE) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error in message (msg size not determined within %d bytes)", remainingBytesInBuffer);
    return GWEN_ERROR_BAD_DATA;
  }

  /* read type and flags (first byte) */
  if (xo->bytesReceived<=idx) {
    rv=GWEN_RingBuffer_ReadByte(xo->ringBuffer);
    if (rv!=-1) {
      xo->headerBuffer[idx]=rv;
      xo->bytesReceived++;
      remainingBytesInBuffer++;
    }
    else
      return 0;
  }
  idx++;

  /* read address bytes */
  while(idx<6) { /* max 4 bytes size plus type/flags byte */
    if (xo->bytesReceived<=idx) {
      rv=GWEN_RingBuffer_ReadByte(xo->ringBuffer);
      if (rv!=-1) {
        xo->headerBuffer[idx]=rv;
        xo->bytesReceived++;
        remainingBytesInBuffer++;
      }
      else
        return 0;
    }
    lenByte=ptr[idx];
    mqttPayloadLen+=(lenByte & 0x7f)<<shift;
    if (!(lenByte & 0x80)) {
      uint32_t fullMsgSize;
  
      /* last byte of size, finish */
      fullMsgSize=mqttPayloadLen+idx+1;
      xo->bytesLeft=(fullMsgSize-xo->bytesReceived);
  
      xo->currentMsgBuf=(uint8_t*) malloc(fullMsgSize);
      memmove(xo->currentMsgBuf, xo->headerBuffer, xo->bytesReceived);
      xo->bytesLeft=fullMsgSize-xo->bytesReceived;
      xo->flags|=AQH_MSG_READER_FLAGS_READBODY;
      DBG_VERBOUS(AQH_LOGDOMAIN,
                  "Got size: full size=%d, payload pos=%d, payload size=%d (%04x)",
                  fullMsgSize, idx+1, mqttPayloadLen, mqttPayloadLen);
      return 1; /* size successfully determined */
    }
    shift+=7;
    idx++;
  }
  DBG_ERROR(AQH_LOGDOMAIN, "Bad MQTT message (could not determine message length)");
  return GWEN_ERROR_BAD_DATA;
}