aqhome: adapted server aqhome-mqttlog to events2 api.

aqhome/ipc2/mqttmsgreader.c

@@ -0,0 +1,182 @@
+/****************************************************************************
+ * 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 "./mqttmsgreader.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_ERROR(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_ERROR(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;
+}
+
+
+