aqhomecontrol/aqhome/ipc2/node_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 "./node_msgreader.h"
#include "./msgreader_p.h"
#include <aqhome/events2/fdobject.h>

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

#include <sys/socket.h>



/* ------------------------------------------------------------------------------------------------
 * definitions
 * ------------------------------------------------------------------------------------------------
 */

#define AQH_MSG_READER_HEADER_SIZE   2
#define AQH_MSG_READER_MAXMSGSIZE    32

#define AQH_MSG_READER_FLAG_SKIPPING 0x0001



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

static int _readMsg(AQH_OBJECT *o);
static int _readHeaderFromRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo);
static uint8_t _calcCrc8Checksum(const uint8_t *ptr, uint8_t len);



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

AQH_OBJECT *AQH_NodeMsgReader_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->bytesReceived<AQH_MSG_READER_HEADER_SIZE) {
      rv=_readHeaderFromRingbuffer(o, xo);
      if (rv<0) {
	DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
	return rv;
      }
    }

    if (xo->bytesReceived>=AQH_MSG_READER_HEADER_SIZE) {
      /* 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;

        if (_calcCrc8Checksum(msgPtr, msgLen)==0) {
          rv=AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_MSGRECVD, msgLen, (void*) msgPtr);
          if (rv==0) {
            DBG_INFO(AQH_LOGDOMAIN, "Received message ignored");
          }
        }
        else {
          DBG_ERROR(AQH_LOGDOMAIN, "Bad CRC");
          AQH_MsgReader_StartSkipping(o);
        }
        free(msgPtr);
	return 1;
      }
    }
    return 0;
  }
  return GWEN_ERROR_GENERIC;
}



int _readHeaderFromRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo)
{
  uint32_t remaining;
  int rv;
  uint32_t xferSize;
  uint32_t bytesInBuffer;

  bytesInBuffer=GWEN_RingBuffer_GetUsedBytes(xo->ringBuffer);
 
  /* still reading header */
  remaining=AQH_MSG_READER_HEADER_SIZE-xo->bytesReceived;
  if (bytesInBuffer<remaining)
    remaining=bytesInBuffer;
  xferSize=remaining;
  rv=GWEN_RingBuffer_ReadBytes(xo->ringBuffer, (char*) (xo->headerBuffer+xo->bytesReceived), &xferSize);
  if (rv<0) {
    DBG_INFO(AQH_LOGDOMAIN, "Ringbuffer empty");
    return 0;
  }
  if (xferSize<remaining) {
    DBG_ERROR(AQH_LOGDOMAIN, "Read fewer bytes than available?");
    return GWEN_ERROR_GENERIC;
  }
  xo->bytesReceived+=xferSize;
  if (xo->bytesReceived==AQH_MSG_READER_HEADER_SIZE) {
    uint32_t msgLen;

    /* full size received, parse msg size, allocate buffer */
    msgLen=xo->headerBuffer[1];
    if (msgLen>AQH_MSG_READER_MAXMSGSIZE) {
      DBG_ERROR(AQH_LOGDOMAIN, "Bad message size(%lu)", (unsigned long int) msgLen);
      AQH_MsgReader_StartSkipping(o);
      return GWEN_ERROR_GENERIC;
    }
    xo->currentMsgBuf=(uint8_t*) malloc(msgLen+3);   /* +3 (dest addr, msg len, crc) */
    memmove(xo->currentMsgBuf, xo->headerBuffer, xo->bytesReceived);
    xo->bytesLeft=(msgLen+3)-xo->bytesReceived;
  }

  return 0;
}



uint8_t _calcCrc8Checksum(const uint8_t *ptr, uint8_t len)
{
  int i;
  uint8_t x=0xff;

  for (i=0; i<len; i++, ptr++) {
    int j;

    x^=*ptr;
    for (j=0; j<8; j++) {
      if (x & 0x80)
	x=(uint8_t) (x<<1)^0x97;
      else
	x<<=1;
    }
  }

  return x;
}