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

propagate socket closure up to endpoint.

Browse Source
This commit is contained in:
Martin Preuss
2026-06-30 23:30:48 +02:00
parent cdb9da2f9e
commit f23806d884
8 changed files with 96 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
aqhome/events2/fdobject.c
View File

@@ -20,6 +20,8 @@
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <sys/select.h>
GWEN_INHERIT(AQH_OBJECT, AQH_FDOBJECT)
@@ -202,6 +204,7 @@ int AQH_FdObject_Read(AQH_OBJECT *o, uint8_t *ptrBuffer, uint32_t lenBuffer)
DBG_INFO(AQH_LOGDOMAIN, "Error on read(%d): %s (%d)", xo->fd, strerror(errno), errno);
close(xo->fd);
xo->fd=-1;
AQH_Object_EmitSignal(o, AQH_FDOBJECT_SIGNAL_CLOSED, GWEN_ERROR_IO, NULL);
return GWEN_ERROR_IO;
}
}
@@ -233,6 +236,7 @@ int AQH_FdObject_FlushInput(AQH_OBJECT *o)
DBG_ERROR(AQH_LOGDOMAIN, "Error on flush, closing.");
close(xo->fd);
xo->fd=-1;
AQH_Object_EmitSignal(o, AQH_FDOBJECT_SIGNAL_CLOSED, GWEN_ERROR_IO, NULL);
}
}
return rv;
@@ -263,6 +267,7 @@ int AQH_FdObject_Write(AQH_OBJECT *o, const uint8_t *ptrBuffer, uint32_t lenBuff
DBG_ERROR(AQH_LOGDOMAIN, "Error on write: %s (%d)", strerror(errno), errno);
close(xo->fd);
xo->fd=-1;
AQH_Object_EmitSignal(o, AQH_FDOBJECT_SIGNAL_CLOSED, GWEN_ERROR_IO, NULL);
return GWEN_ERROR_IO;
}
}
@@ -310,6 +315,42 @@ void AQH_FdObject_EndMsg(AQH_OBJECT *o)
int AQH_FdObject_WaitForActivity(AQH_OBJECT *o, int timeoutInMillisecs)
{
if (o) {
AQH_FDOBJECT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_FDOBJECT, o);
if (xo) {
int fd;
fd=xo->fd;
if (fd>=0) {
int rv;
fd_set rset;
struct timeval tv;
tv.tv_sec=timeoutInMillisecs/1000;
tv.tv_usec=(timeoutInMillisecs%1000)*1000;
FD_SET(fd, &rset);
if (xo->fdMode==AQH_FDOBJECT_FDMODE_READ)
rv=select(fd+1, &rset, NULL, NULL, &tv);
else
rv=select(fd+1, NULL, &rset, NULL, &tv);
if (rv<0 && errno!=EINTR) {
DBG_ERROR(AQH_LOGDOMAIN, "Error on SELECT: %d (%s)", errno, strerror(errno));
return GWEN_ERROR_IO;
}
return rv;
}
}
}
return 0;
}
void _cbEnable(AQH_OBJECT *o)
{
if (o && !(AQH_Object_GetFlags(o) & AQH_OBJECT_FLAGS_ENABLED)) {
@@ -381,6 +422,20 @@ int _cbFlush(AQH_OBJECT *o)
void AQH_FdObject_Abort(AQH_OBJECT *o)
{
if (o) {
AQH_FDOBJECT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_FDOBJECT, o);
if (xo) {
close(xo->fd);
xo->fd=-1;
AQH_Object_EmitSignal(o, AQH_FDOBJECT_SIGNAL_CLOSED, GWEN_ERROR_ABORTED, NULL);
}
}
}

7
aqhome/events2/fdobject.h
View File

@@ -20,7 +20,9 @@ enum {
enum {
AQH_FDOBJECT_SIGNAL_ISREADY=AQH_OBJECT_SIGNAL_LAST
AQH_FDOBJECT_SIGNAL_ISREADY=AQH_OBJECT_SIGNAL_LAST,
AQH_FDOBJECT_SIGNAL_CLOSED,
AQH_FDOBJECT_SIGNAL_LAST
};
@@ -51,5 +53,8 @@ AQHOME_API AQH_FDOBJECT_STARTMSG_FN AQH_FdObject_SetStartMsgFn(AQH_OBJECT *o, AQ
AQHOME_API AQH_FDOBJECT_ENDMSG_FN AQH_FdObject_SetEndMsgFn(AQH_OBJECT *o, AQH_FDOBJECT_ENDMSG_FN f);
AQHOME_API AQH_FDOBJECT_FLUSH_FN AQH_FdObject_SetFlushFn(AQH_OBJECT *o, AQH_FDOBJECT_FLUSH_FN f);
AQHOME_API int AQH_FdObject_WaitForActivity(AQH_OBJECT *o, int timeoutInMillisecs);
AQHOME_API void AQH_FdObject_Abort(AQH_OBJECT *o);
#endif

1
aqhome/ipc2/endpoint.c
View File

@@ -92,6 +92,7 @@ AQH_OBJECT *AQH_Endpoint_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *msgReader, A
if (msgWriter) {
xo->msgWriter=msgWriter;
AQH_Object_AddLink(msgWriter, AQH_MSG_WRITER_SIGNAL_MSGSENT, AQH_ENDPOINT_SLOT_MSG_SENT, o);
AQH_Object_AddLink(msgWriter, AQH_MSG_WRITER_SIGNAL_CLOSED, AQH_ENDPOINT_SLOT_CLOSED, o);
}
return o;

20
aqhome/ipc2/msgreader.c
View File

@@ -39,6 +39,7 @@ 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, AQH_OBJECT *fdObject);
static int _handleSocketClosed(AQH_OBJECT *o, int param1);
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);
@@ -70,7 +71,8 @@ AQH_OBJECT *AQH_MsgReader_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *fdObject)
if (fdObject) {
xo->fdObject=fdObject;
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_ISREADY, AQH_MSGREADER_SLOT_SOCKETREADY, o);
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_ISREADY, AQH_MSGREADER_SLOT_SOCKETREADY, o);
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_CLOSED, AQH_MSGREADER_SLOT_SOCKETCLOSED, o);
}
return o;
@@ -158,11 +160,12 @@ AQH_MSG_READER *AQH_MsgReader_GetData(const AQH_OBJECT *o)
int _handleSignal(AQH_OBJECT *o,
uint32_t slotId,
AQH_OBJECT *senderObject,
GWEN_UNUSED int param1,
int param1,
GWEN_UNUSED void *param2)
{
switch(slotId) {
case AQH_MSGREADER_SLOT_SOCKETREADY: return _handleSocketReady(o, senderObject);
case AQH_MSGREADER_SLOT_SOCKETREADY: return _handleSocketReady(o, senderObject);
case AQH_MSGREADER_SLOT_SOCKETCLOSED: return _handleSocketClosed(o, param1);
default:
break;
}
@@ -242,6 +245,15 @@ int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
int _handleSocketClosed(AQH_OBJECT *o, int param1)
{
DBG_DEBUG(AQH_LOGDOMAIN, "Socket closed");
AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_CLOSED, param1, NULL);
return 1;
}
void AQH_MsgReader_StartSkipping(AQH_OBJECT *o)
{
AQH_MSG_READER *xo;
@@ -250,7 +262,7 @@ void AQH_MsgReader_StartSkipping(AQH_OBJECT *o)
if (xo) {
int rv;
DBG_ERROR(AQH_LOGDOMAIN, "Enter skip mode");
DBG_ERROR(AQH_LOGDOMAIN, "Skipping data in input buffers");
GWEN_RingBuffer_Reset(xo->ringBuffer);
_resetBuffers(xo);

3
aqhome/ipc2/msgreader.h
View File

@@ -23,7 +23,8 @@ enum {
enum {
AQH_MSGREADER_SLOT_SOCKETREADY=1
AQH_MSGREADER_SLOT_SOCKETREADY=1,
AQH_MSGREADER_SLOT_SOCKETCLOSED
};

15
aqhome/ipc2/msgwriter.c
View File

@@ -36,6 +36,7 @@ GWEN_INHERIT(AQH_OBJECT, AQH_MSG_WRITER)
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, AQH_OBJECT *fdObject);
static int _handleSocketClosed(AQH_OBJECT *o, int param1);
static int _startMsg(AQH_MSG_WRITER *xo, AQH_OBJECT *fdObject);
static void _endMsg(AQH_MSG_WRITER *xo, AQH_OBJECT *fdObject);
static void _resetBuffer(AQH_OBJECT *o);
@@ -111,11 +112,12 @@ void AQH_MsgWriter_SendMsg(AQH_OBJECT *o, const uint8_t *ptr, int len)
int _handleSignal(AQH_OBJECT *o,
uint32_t slotId,
GWEN_UNUSED AQH_OBJECT *senderObject,
GWEN_UNUSED int param1,
int param1,
GWEN_UNUSED void *param2)
{
switch(slotId) {
case AQH_MSGWRITER_SLOT_SOCKETREADY: return _handleSocketReady(o, senderObject);
case AQH_MSGWRITER_SLOT_SOCKETREADY: return _handleSocketReady(o, senderObject);
case AQH_MSGWRITER_SLOT_SOCKETCLOSED: return _handleSocketClosed(o, param1);
default:
break;
}
@@ -185,6 +187,15 @@ int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
int _handleSocketClosed(AQH_OBJECT *o, int param1)
{
DBG_DEBUG(AQH_LOGDOMAIN, "Socket closed");
AQH_Object_EmitSignal(o, AQH_MSG_WRITER_SIGNAL_CLOSED, param1, NULL);
return 1;
}
int _startMsg(AQH_MSG_WRITER *xo, AQH_OBJECT *fdObject)
{
int rv;

3
aqhome/ipc2/msgwriter.h
View File

@@ -22,7 +22,8 @@ enum {
enum {
AQH_MSGWRITER_SLOT_SOCKETREADY=1
AQH_MSGWRITER_SLOT_SOCKETREADY=1,
AQH_MSGWRITER_SLOT_SOCKETCLOSED
};

1
aqhome/ipc2/node_msgreader.c
View File

@@ -107,6 +107,7 @@ int _readMsg(AQH_OBJECT *o)
AQH_MsgReader_StartSkipping(o);
}
free(msgPtr);
/* handled a message */
return 1;
}
}