martin/aqhomecontrol
Public Access
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

aqhome: started rewriting message code, start using new event2 lib.

Browse Source
This commit is contained in:
Martin Preuss
2025-02-25 01:13:07 +01:00
parent f1f24168e5
commit cf8edbbd5f
58 changed files with 5393 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

400
aqhome/ipc2/msgreader.c
View File

@@ -18,14 +18,12 @@
#include <gwenhywfar/endianfns.h>
#include <sys/socket.h>
#include <time.h>
#define AQH_MSG_READER_MINMSGSIZE 12
#define AQH_MSG_READER_MAXMSGSIZE 10240
#define AQH_MSGREADER_SKIPTIME_IN_MS 20
enum {
AQH_MSGREADER_SLOT_SOCKETREADY=1
};
#define AQH_MSGREADER_FLAGS_SKIP 0x0001
@@ -39,13 +37,15 @@ GWEN_INHERIT(AQH_OBJECT, AQH_MSG_READER)
*/
static void GWENHYWFAR_CB _freeData(void *bp, void *p);
static int _handleSignal(AQH_OBJECT *o, uint32_t slotId, AQH_OBJECT *senderObject, int param1, void *param2);
static int _handleSocketReady(AQH_OBJECT *o);
static int _fillRingbuffer(AQH_OBJECT *o);
static int _readMsgFromRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo);
static int _readHeaderFromRingbuffer(AQH_MSG_READER *xo);
static int _readRemainderFromRingbuffer(AQH_MSG_READER *xo);
static int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject);
static int _fillRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo, AQH_OBJECT *fdObject);
static void _handleSkipping(AQH_OBJECT *o, AQH_OBJECT *fdObject);
static int _isStillSkipTime(AQH_MSG_READER *xo);
static uint64_t _getTimeInMilliSeconds(void);
static void _resetBuffers(AQH_MSG_READER *xo);
static void _cbEnable(AQH_OBJECT *o);
static void _cbDisable(AQH_OBJECT *o);
@@ -54,7 +54,7 @@ static int _readRemainderFromRingbuffer(AQH_MSG_READER *xo);
* ------------------------------------------------------------------------------------------------
*/
AQH_OBJECT *AQH_MsgReader_new(AQH_EVENT_LOOP *eventLoop, int fd)
AQH_OBJECT *AQH_MsgReader_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *fdObject)
{
AQH_OBJECT *o;
AQH_MSG_READER *xo;
@@ -62,15 +62,16 @@ AQH_OBJECT *AQH_MsgReader_new(AQH_EVENT_LOOP *eventLoop, int fd)
o=AQH_Object_new(eventLoop);
GWEN_NEW_OBJECT(AQH_MSG_READER, xo);
GWEN_INHERIT_SETDATA(AQH_OBJECT, AQH_MSG_READER, o, xo, _freeData);
xo->fdSocket=fd;
xo->ringBuffer=GWEN_RingBuffer_new(AQH_MSG_READER_RINGBUFFER_SIZE);
AQH_Object_SetSignalHandlerFn(o, _handleSignal);
AQH_Object_SetEnableFn(o, _cbEnable);
AQH_Object_SetDisableFn(o, _cbDisable);
/* create object for readable socket, connect to THIS, enable */
xo->fdObject=AQH_FdObject_new(AQH_Object_GetEventLoop(o), fd, AQH_FDOBJECT_FDMODE_READ);
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_ISREADY, AQH_MSGREADER_SLOT_SOCKETREADY, o);
AQH_Object_Enable(xo->fdObject);
if (fdObject) {
xo->fdObject=fdObject;
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_ISREADY, AQH_MSGREADER_SLOT_SOCKETREADY, o);
}
return o;
}
@@ -82,23 +83,77 @@ void GWENHYWFAR_CB _freeData(GWEN_UNUSED void *bp, void *p)
AQH_MSG_READER *xo;
xo=(AQH_MSG_READER*) p;
if (xo->fdObject) {
AQH_Object_Disable(xo->fdObject);
AQH_Object_free(xo->fdObject);
}
free(xo->currentMsgBuf);
GWEN_FREE_OBJECT(xo);
}
uint32_t AQH_MsgReader_GetFlags(const AQH_OBJECT *o)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
return xo?xo->flags:0;
}
void AQH_MsgReader_SetFlags(AQH_OBJECT *o, uint32_t f)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo)
xo->flags=f;
}
void AQH_MsgReader_AddFlags(AQH_OBJECT *o, uint32_t f)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo)
xo->flags|=f;
}
void AQH_MsgReader_SubFlags(AQH_OBJECT *o, uint32_t f)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo)
xo->flags&=~f;
}
AQH_MSG_READER *AQH_MsgReader_GetData(const AQH_OBJECT *o)
{
if (o) {
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
return xo;
}
return NULL;
}
int _handleSignal(AQH_OBJECT *o,
uint32_t slotId,
GWEN_UNUSED AQH_OBJECT *senderObject,
AQH_OBJECT *senderObject,
GWEN_UNUSED int param1,
GWEN_UNUSED void *param2)
{
switch(slotId) {
case AQH_MSGREADER_SLOT_SOCKETREADY: return _handleSocketReady(o);
case AQH_MSGREADER_SLOT_SOCKETREADY: return _handleSocketReady(o, senderObject);
default:
break;
}
@@ -108,7 +163,36 @@ int _handleSignal(AQH_OBJECT *o,
int _handleSocketReady(AQH_OBJECT *o)
int AQH_MsgReader_ReadMsg(AQH_OBJECT *o)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo)
return xo->readMsgFn(o);
return GWEN_ERROR_NOT_IMPLEMENTED;
}
AQH_MSG_READER_READMSG_FN AQH_MsgReader_SetReadMsgFn(AQH_OBJECT *o, AQH_MSG_READER_READMSG_FN f)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo) {
AQH_MSG_READER_READMSG_FN oldFn;
oldFn=xo->readMsgFn;
xo->readMsgFn=f;
return oldFn;
}
return NULL;
}
int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
{
AQH_MSG_READER *xo;
@@ -116,22 +200,28 @@ int _handleSocketReady(AQH_OBJECT *o)
if (xo) {
int rv;
if (xo->flags & AQH_MSGREADER_FLAGS_SKIP) {
if (_isStillSkipTime(xo)) {
_handleSkipping(o, fdObject);
return 1;
}
}
/* read available data into ringbuffer */
rv=_fillRingbuffer(o);
rv=_fillRingbuffer(o, xo, fdObject);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_ERROR, rv, NULL);
xo->fdSocket=-1;
return 1;
if (rv!=GWEN_ERROR_TRY_AGAIN) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_ERROR, rv, NULL);
}
}
/* read messages from ring buffer until buffer empty */
do {
rv=_readMsgFromRingbuffer(o, xo);
rv=AQH_MsgReader_ReadMsg(o);
} while (rv==1);
if (rv<0) {
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_ERROR, rv, NULL);
xo->fdSocket=-1;
}
return 1;
@@ -142,12 +232,66 @@ int _handleSocketReady(AQH_OBJECT *o)
int _fillRingbuffer(AQH_OBJECT *o)
void AQH_MsgReader_StartSkipping(AQH_OBJECT *o)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo && xo->fdSocket!=-1) {
if (xo) {
DBG_ERROR(AQH_LOGDOMAIN, "Enter skip mode");
GWEN_RingBuffer_Reset(xo->ringBuffer);
_resetBuffers(xo);
xo->flags|=AQH_MSGREADER_FLAGS_SKIP;
xo->timestamp=_getTimeInMilliSeconds();
}
}
void _handleSkipping(AQH_OBJECT *o, AQH_OBJECT *fdObject)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo) {
int rv;
rv=AQH_FdObject_FlushInput(fdObject);
if (rv<0) {
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_ERROR, rv, NULL);
}
else if (rv>0)
xo->timestamp=_getTimeInMilliSeconds();
}
}
int _isStillSkipTime(AQH_MSG_READER *xo)
{
uint64_t currentTime;
uint64_t diffTime;
currentTime=_getTimeInMilliSeconds();
diffTime=currentTime-xo->timestamp;
if (diffTime>=AQH_MSGREADER_SKIPTIME_IN_MS) {
xo->flags&=~AQH_MSGREADER_FLAGS_SKIP;
GWEN_RingBuffer_Reset(xo->ringBuffer);
_resetBuffers(xo);
DBG_ERROR(AQH_LOGDOMAIN, "Leaving skip mode");
return 0;
}
else {
xo->timestamp=currentTime;
return 1;
}
}
int _fillRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo, AQH_OBJECT *fdObject)
{
while (1) {
uint32_t len;
/* read data into ringbuffer */
@@ -155,160 +299,124 @@ int _fillRingbuffer(AQH_OBJECT *o)
if (len>0) {
int rv;
rv=AQH_FdObject_Read(xo->fdObject, (uint8_t*) GWEN_RingBuffer_GetWritePointer(xo->ringBuffer), len);
rv=AQH_FdObject_Read(fdObject, (uint8_t*) GWEN_RingBuffer_GetWritePointer(xo->ringBuffer), len);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
xo->fdSocket=-1;
if (rv!=GWEN_ERROR_TRY_AGAIN) {
DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
}
return rv;
}
else if (rv==0) {
DBG_INFO(AQH_LOGDOMAIN, "EOF met");
xo->fdSocket=-1;
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_CLOSED, 0, NULL);
return 0;
}
else {
/* bytes received */
GWEN_RingBuffer_SkipBytesWrite(xo->ringBuffer, rv);
return rv;
//return rv;
}
}
else {
DBG_INFO(AQH_LOGDOMAIN, "Ringbuffer full");
return GWEN_ERROR_BUFFER_OVERFLOW;
/*return GWEN_ERROR_BUFFER_OVERFLOW;*/
return 0;
}
}
}
int AQH_MsgReader_ReadRemainderFromRingbuffer(AQH_OBJECT *o)
{
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo) {
uint32_t bytesInRingBuffer;
uint32_t bytesToRead;
int rv;
bytesInRingBuffer=GWEN_RingBuffer_GetUsedBytes(xo->ringBuffer);
/* still reading header */
bytesToRead=xo->bytesLeft;
if (bytesInRingBuffer<bytesToRead)
bytesToRead=bytesInRingBuffer;
if (bytesToRead) {
uint32_t xferSize;
xferSize=bytesToRead;
rv=GWEN_RingBuffer_ReadBytes(xo->ringBuffer, (char*) (xo->currentMsgBuf+xo->bytesReceived), &xferSize);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "Ringbuffer empty");
return 0;
}
if (xferSize<bytesToRead) {
DBG_ERROR(AQH_LOGDOMAIN, "Read fewer bytes than available?");
return GWEN_ERROR_GENERIC;
}
xo->bytesReceived+=xferSize;
xo->bytesLeft-=xferSize;
if (xo->bytesLeft==0) {
/* msg finished */
DBG_INFO(AQH_LOGDOMAIN, "Message complete");
return 1;
}
}
return 0;
}
else {
DBG_INFO(AQH_LOGDOMAIN, "fd inactive (previous error or EOF?)");
DBG_ERROR(AQH_LOGDOMAIN, "Not a MSGREADER object");
return GWEN_ERROR_INVALID;
}
}
int _readMsgFromRingbuffer(AQH_OBJECT *o, AQH_MSG_READER *xo)
uint64_t _getTimeInMilliSeconds(void)
{
int rv;
struct timespec t ;
if (xo->bytesReceived<AQH_MSG_READER_HEADERBUFFER_SIZE) {
rv=_readHeaderFromRingbuffer(xo);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
return rv;
}
}
if (xo->bytesReceived>=AQH_MSG_READER_HEADERBUFFER_SIZE) {
/* reading remainder of msg directly into allocated buffer */
rv=_readRemainderFromRingbuffer(xo);
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;
rv=AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_MSGRECVD, msgLen, (void*) msgPtr);
if (rv==0) {
DBG_INFO(AQH_LOGDOMAIN, "Received message ignored");
free(msgPtr);
}
return 1;
}
}
return 0;
clock_gettime(CLOCK_REALTIME, &t);
return t.tv_sec*1000+(t.tv_nsec+500000)/1000000 ;
}
int _readHeaderFromRingbuffer(AQH_MSG_READER *xo)
void _resetBuffers(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_HEADERBUFFER_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, "here (%d)", rv);
return rv;
}
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_HEADERBUFFER_SIZE) {
uint32_t msgLen;
/* full size received, parse msg size, allocate buffer */
msgLen=GWEN_ENDIAN_LE32TOH(*((const uint32_t*)(xo->headerBuffer)));
if (msgLen<AQH_MSG_READER_MINMSGSIZE || msgLen>AQH_MSG_READER_MAXMSGSIZE) {
DBG_ERROR(AQH_LOGDOMAIN, "Bad message size(%lu)", (unsigned long int) msgLen);
return GWEN_ERROR_GENERIC;
}
xo->currentMsgBuf=(uint8_t*) malloc(msgLen+4);
memmove(xo->currentMsgBuf, xo->headerBuffer, xo->bytesReceived);
xo->bytesLeft=(msgLen+4)-xo->bytesReceived;
}
return 0;
free(xo->currentMsgBuf);
xo->currentMsgBuf=NULL;
xo->bytesReceived=0;
xo->bytesLeft=0;
}
int _readRemainderFromRingbuffer(AQH_MSG_READER *xo)
void _cbEnable(AQH_OBJECT *o)
{
uint32_t bytesInBuffer;
uint32_t bytesToRead;
int rv;
if (o && !(AQH_Object_GetFlags(o) & AQH_OBJECT_FLAGS_ENABLED)) {
AQH_MSG_READER *xo;
bytesInBuffer=GWEN_RingBuffer_GetUsedBytes(xo->ringBuffer);
/* still reading header */
bytesToRead=xo->bytesLeft;
if (bytesInBuffer<bytesToRead)
bytesToRead=bytesInBuffer;
if (bytesToRead) {
uint32_t xferSize;
xferSize=bytesToRead;
rv=GWEN_RingBuffer_ReadBytes(xo->ringBuffer, (char*) (xo->currentMsgBuf+xo->bytesReceived), &xferSize);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
return rv;
}
if (xferSize<bytesToRead) {
DBG_ERROR(AQH_LOGDOMAIN, "Read fewer bytes than available?");
return GWEN_ERROR_GENERIC;
}
xo->bytesReceived+=xferSize;
xo->bytesLeft-=xferSize;
if (xo->bytesLeft==0) {
/* msg finished */
DBG_INFO(AQH_LOGDOMAIN, "Message complete");
return 1;
}
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo && xo->fdObject)
AQH_Object_Enable(xo->fdObject);
}
return 0;
}
void _cbDisable(AQH_OBJECT *o)
{
if (o && (AQH_Object_GetFlags(o) & AQH_OBJECT_FLAGS_ENABLED)) {
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo && xo->fdObject)
AQH_Object_Disable(xo->fdObject);
}
}