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

aqhome: more work on transformation to event2/ipc2.

Browse Source
This commit is contained in:
Martin Preuss
2025-02-27 14:08:44 +01:00
parent bebc4c1b0d
commit d887747b3c
45 changed files with 2446 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
aqhome/ipc2/0BUILD
View File

@@ -51,10 +51,12 @@
nodemsgreader.h
ttyobject.h
tcpd_object.h
tcp_object.h
endpoint.h
message.h
msgrequest.h
ipc_server.h
ipc_client.h
tty_endpoint.h
</headers>
@@ -79,10 +81,12 @@
nodemsgreader.c
ttyobject.c
tcpd_object.c
tcp_object.c
endpoint.c
message.c
msgrequest.c
ipc_server.c
ipc_client.c
tty_endpoint.c
</sources>

264
aqhome/ipc2/endpoint.c
View File

@@ -30,9 +30,14 @@
* ------------------------------------------------------------------------------------------------
*/
/*#define LOG_MESSAGES*/
enum {
AQH_ENDPOINT_SLOT_MSG_RECVD=1,
AQH_ENDPOINT_SLOT_MSG_SENT
AQH_ENDPOINT_SLOT_MSG_SENT,
AQH_ENDPOINT_SLOT_CLOSED
};
@@ -54,8 +59,8 @@ GWEN_INHERIT(AQH_OBJECT, AQH_ENDPOINT)
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 _handleMsgRecvd(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr);
static int _handleMsgSent(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr);
static void _dumpMsg(const AQH_MESSAGE *msg, const char *sText);
static int _handleMsgSent(AQH_OBJECT *o);
static int _handleClosed(AQH_OBJECT *o);
@@ -81,6 +86,7 @@ AQH_OBJECT *AQH_Endpoint_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *msgReader, A
if (msgReader) {
xo->msgReader=msgReader;
AQH_Object_AddLink(msgReader, AQH_MSG_READER_SIGNAL_MSGRECVD, AQH_ENDPOINT_SLOT_MSG_RECVD, o);
AQH_Object_AddLink(msgReader, AQH_MSG_READER_SIGNAL_CLOSED, AQH_ENDPOINT_SLOT_CLOSED, o);
}
if (msgWriter) {
@@ -98,6 +104,8 @@ void GWENHYWFAR_CB _freeData(GWEN_UNUSED void *bp, void *p)
AQH_ENDPOINT *xo;
xo=(AQH_ENDPOINT*) p;
free(xo->serviceName);
free(xo->userName);
AQH_Message_List_free(xo->msgOutList);
AQH_Message_List_free(xo->msgInList);
AQH_Object_free(xo->msgWriter);
@@ -107,6 +115,197 @@ void GWENHYWFAR_CB _freeData(GWEN_UNUSED void *bp, void *p)
const char *AQH_Endpoint_GetServiceName(const AQH_OBJECT *o)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return xo->serviceName;
}
return NULL;
}
void AQH_Endpoint_SetServiceName(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo) {
free(xo->serviceName);
xo->serviceName=s?strdup(s):NULL;
}
}
}
const char *AQH_Endpoint_GetUserName(const AQH_OBJECT *o)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return xo->userName;
}
return NULL;
}
void AQH_Endpoint_SetUserName(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo) {
free(xo->userName);
xo->userName=s?strdup(s):NULL;
}
}
}
uint32_t AQH_Endpoint_GetPermissions(const AQH_OBJECT *o)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return xo->permissions;
}
return 0;
}
void AQH_Endpoint_SetPermissions(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->permissions=i;
}
}
void AQH_Endpoint_AddPermissions(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->permissions|=i;
}
}
void AQH_Endpoint_SubPermissions(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->permissions&=~i;
}
}
uint32_t AQH_Endpoint_GetFlags(const AQH_OBJECT *o)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return xo->flags;
}
return 0;
}
void AQH_Endpoint_SetFlags(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->flags=i;
}
}
void AQH_Endpoint_AddFlags(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->flags|=i;
}
}
void AQH_Endpoint_SubFlags(AQH_OBJECT *o, uint32_t i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->flags&=~i;
}
}
int AQH_Endpoint_GetState(const AQH_OBJECT *o)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return xo->state;
}
return 0;
}
void AQH_Endpoint_SetState(AQH_OBJECT *o, int i)
{
if (o) {
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
xo->state=i;
}
}
AQH_MESSAGE_LIST *AQH_Endpoint_GetMsgOutList(const AQH_OBJECT *o)
{
if (o) {
@@ -144,6 +343,7 @@ void AQH_Endpoint_AddMsgOut(AQH_OBJECT *o, AQH_MESSAGE *msg)
if (xo) {
AQH_Message_List_Add(msg, xo->msgOutList);
if (xo->msgWriter && AQH_Message_List_GetCount(xo->msgOutList)==1) {
DBG_INFO(AQH_LOGDOMAIN, "Enabling msgWriter, sending message");
AQH_MsgWriter_SendMsg(xo->msgWriter, AQH_Message_GetMsgPointer(msg), AQH_Message_GetUsedSize(msg));
AQH_Object_Enable(xo->msgWriter);
}
@@ -173,8 +373,15 @@ AQH_MESSAGE *AQH_Endpoint_GetNextMsgIn(AQH_OBJECT *o)
AQH_ENDPOINT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo)
return AQH_Message_List_First(xo->msgInList);
if (xo) {
AQH_MESSAGE *msg;
msg=AQH_Message_List_First(xo->msgInList);
if (msg) {
AQH_Message_List_Del(msg);
return msg;
}
}
}
return NULL;
}
@@ -189,7 +396,7 @@ void AQH_Endpoint_AddMsgIn(AQH_OBJECT *o, AQH_MESSAGE *msg)
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo) {
AQH_Message_List_Add(msg, xo->msgInList);
DBG_ERROR(AQH_LOGDOMAIN, "now %d msgs in list", AQH_Message_List_GetCount(xo->msgInList));
DBG_INFO(AQH_LOGDOMAIN, "now %d msgs in list", AQH_Message_List_GetCount(xo->msgInList));
}
}
}
@@ -228,7 +435,8 @@ int _handleSignal(AQH_OBJECT *o, uint32_t slotId, GWEN_UNUSED AQH_OBJECT *sender
{
switch(slotId) {
case AQH_ENDPOINT_SLOT_MSG_RECVD: return _handleMsgRecvd(o, param1, param2);
case AQH_ENDPOINT_SLOT_MSG_SENT: return _handleMsgSent(o, param1, param2);
case AQH_ENDPOINT_SLOT_MSG_SENT: return _handleMsgSent(o);
case AQH_ENDPOINT_SLOT_CLOSED: return _handleClosed(o);
default:
break;
}
@@ -242,9 +450,9 @@ int _handleMsgRecvd(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr)
{
AQH_MESSAGE *msg;
DBG_ERROR(AQH_LOGDOMAIN, "Msg received:");
DBG_INFO(AQH_LOGDOMAIN, "Msg received:");
/*GWEN_Text_LogString((const char*) msgPtr, msgLen, AQH_LOGDOMAIN, GWEN_LoggerLevel_Error);*/
msg=AQH_NodeMessage_fromBuffer(msgPtr, msgLen);
_dumpMsg(msg, "Received");
AQH_Endpoint_AddMsgIn(o, msg);
return 1;
@@ -252,11 +460,9 @@ int _handleMsgRecvd(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr)
int _handleMsgSent(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr)
int _handleMsgSent(AQH_OBJECT *o)
{
DBG_ERROR(AQH_LOGDOMAIN, "Msg sent:");
GWEN_Text_LogString((const char*) msgPtr, msgLen, AQH_LOGDOMAIN, GWEN_LoggerLevel_Error);
DBG_INFO(AQH_LOGDOMAIN, "Msg sent");
if (o) {
AQH_ENDPOINT *xo;
@@ -264,6 +470,7 @@ int _handleMsgSent(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr)
if (xo) {
AQH_MESSAGE *msg;
DBG_INFO(AQH_LOGDOMAIN, "Messages in outlist: %d", AQH_Message_List_GetCount(xo->msgOutList));
msg=AQH_Message_List_First(xo->msgOutList);
if (msg) {
/* remove sent message from list */
@@ -273,33 +480,38 @@ int _handleMsgSent(AQH_OBJECT *o, int msgLen, const uint8_t *msgPtr)
/* get next message in list */
msg=AQH_Message_List_First(xo->msgOutList);
if (msg) {
DBG_ERROR(AQH_LOGDOMAIN, "Sending next message");
DBG_INFO(AQH_LOGDOMAIN, "Sending next message");
AQH_MsgWriter_SendMsg(xo->msgWriter, AQH_Message_GetMsgPointer(msg), AQH_Message_GetUsedSize(msg));
}
}
else {
DBG_ERROR(AQH_LOGDOMAIN, "Last message sent, disabling writer");
AQH_Object_Disable(xo->msgWriter);
else {
DBG_INFO(AQH_LOGDOMAIN, "Last message sent, disabling writer");
AQH_Object_Disable(xo->msgWriter);
}
}
}
}
return 1;
}
void _dumpMsg(const AQH_MESSAGE *msg, const char *sText)
int _handleClosed(AQH_OBJECT *o)
{
if (msg) {
GWEN_BUFFER *mbuf;
AQH_ENDPOINT *xo;
mbuf=GWEN_Buffer_new(0, 256, 0, 1);
AQH_NodeMessage_DumpSpecificToBuffer(msg, mbuf, sText);
DBG_ERROR(AQH_LOGDOMAIN, "%s", GWEN_Buffer_GetStart(mbuf));
GWEN_Buffer_free(mbuf);
DBG_ERROR(AQH_LOGDOMAIN, "Connection closed.");
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_ENDPOINT, o);
if (xo) {
AQH_Object_Disable(xo->msgWriter);
AQH_Object_Disable(xo->msgReader);
if (0==AQH_Object_EmitSignal(o, AQH_ENDPOINT_SIGNAL_CLOSED, 0, NULL)) {
DBG_ERROR(AQH_LOGDOMAIN, "Signal CLOSED not handled");
}
return 1;
}
return 0;
}

25
aqhome/ipc2/endpoint.h
View File

@@ -14,6 +14,12 @@
enum {
AQH_ENDPOINT_SIGNAL_CLOSED=AQH_OBJECT_SIGNAL_LAST,
};
/**
* Constructor.
*
@@ -26,6 +32,25 @@
*/
AQHOME_API AQH_OBJECT *AQH_Endpoint_new(AQH_EVENT_LOOP *eventLoop, AQH_OBJECT *msgReader, AQH_OBJECT *msgWriter);
AQHOME_API const char *AQH_Endpoint_GetServiceName(const AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_SetServiceName(AQH_OBJECT *o, const char *s);
AQHOME_API const char *AQH_Endpoint_GetUserName(const AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_SetUserName(AQH_OBJECT *o, const char *s);
AQHOME_API uint32_t AQH_Endpoint_GetPermissions(const AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_SetPermissions(AQH_OBJECT *o, uint32_t i);
AQHOME_API void AQH_Endpoint_AddPermissions(AQH_OBJECT *o, uint32_t i);
AQHOME_API void AQH_Endpoint_SubPermissions(AQH_OBJECT *o, uint32_t i);
AQHOME_API uint32_t AQH_Endpoint_GetFlags(const AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_SetFlags(AQH_OBJECT *o, uint32_t i);
AQHOME_API void AQH_Endpoint_AddFlags(AQH_OBJECT *o, uint32_t i);
AQHOME_API void AQH_Endpoint_SubFlags(AQH_OBJECT *o, uint32_t i);
AQHOME_API int AQH_Endpoint_GetState(const AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_SetState(AQH_OBJECT *o, int i);
AQHOME_API AQH_MESSAGE_LIST *AQH_Endpoint_GetMsgOutList(const AQH_OBJECT *o);
AQHOME_API AQH_MESSAGE *AQH_Endpoint_GetNextMsgOut(AQH_OBJECT *o);
AQHOME_API void AQH_Endpoint_AddMsgOut(AQH_OBJECT *o, AQH_MESSAGE *msg);

5
aqhome/ipc2/endpoint_p.h
View File

@@ -21,6 +21,11 @@ struct AQH_ENDPOINT {
AQH_OBJECT *msgReader;
uint32_t lastMsgId;
int state;
uint32_t permissions;
uint32_t flags;
char *serviceName;
char *userName;
};

53
aqhome/ipc2/ipc_client.c Normal file
View File

@@ -0,0 +1,53 @@
/****************************************************************************
* 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 "./ipc_client.h"
#include <aqhome/ipc2/ipcmsgreader.h>
#include <aqhome/ipc2/msgwriter.h>
#include <aqhome/events2/fdobject.h>
#include <gwenhywfar/debug.h>
#include <unistd.h>
/* ------------------------------------------------------------------------------------------------
* code
* ------------------------------------------------------------------------------------------------
*/
AQH_OBJECT *AQH_IpcClientObject_new(AQH_EVENT_LOOP *eventLoop, int fd)
{
int fdCopy;
AQH_OBJECT *fdReader;
AQH_OBJECT *fdWriter;
AQH_OBJECT *msgReader;
AQH_OBJECT *msgWriter;
AQH_OBJECT *endpoint;
fdCopy=dup(fd);
fdReader=AQH_FdObject_new(eventLoop, fd, AQH_FDOBJECT_FDMODE_READ);
msgReader=AQH_IpcMsgReader_new(eventLoop, fdReader);
AQH_Object_Enable(msgReader);
fdWriter=AQH_FdObject_new(eventLoop, fdCopy, AQH_FDOBJECT_FDMODE_WRITE);
msgWriter=AQH_MsgWriter_new(eventLoop, fdWriter);
endpoint=AQH_Endpoint_new(eventLoop, msgReader, msgWriter);
return endpoint;
}

28
aqhome/ipc2/ipc_client.h Normal file
View File

@@ -0,0 +1,28 @@
/****************************************************************************
* 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.
****************************************************************************/
#ifndef AQH_IPC_CLIENT_H
#define AQH_IPC_CLIENT_H
#include <aqhome/ipc2/endpoint.h>
/**
* Create an endpoint object (see @ref AQH_Endpoint_new) which works over a given socket.
*
* @param eventLoop pointer to eventLoop
* @param fd connected non-blocking socket to work with (see @ref AQH_TcpObject_CreateConnectedSocket).
*/
AQHOME_API AQH_OBJECT *AQH_IpcClientObject_new(AQH_EVENT_LOOP *eventLoop, int fd);
#endif

3
aqhome/ipc2/ipc_server.c
View File

@@ -101,7 +101,7 @@ int _handleSignal(AQH_OBJECT *o,
GWEN_UNUSED void *param2)
{
switch(slotId) {
case AQH_TCPD_OBJECT_SIGNAL_NEWCONN: return _handleNewConn(o, param1);
case AQH_IPCD_OBJECT_SLOT_NEWCONN: return _handleNewConn(o, param1);
default:
break;
}
@@ -127,6 +127,7 @@ int _handleNewConn(AQH_OBJECT *o, int newFd)
fdReader=AQH_FdObject_new(eventLoop, newFd, AQH_FDOBJECT_FDMODE_READ);
msgReader=AQH_IpcMsgReader_new(eventLoop, fdReader);
AQH_Object_Enable(msgReader);
fdWriter=AQH_FdObject_new(eventLoop, fdCopy, AQH_FDOBJECT_FDMODE_WRITE);
msgWriter=AQH_MsgWriter_new(eventLoop, fdWriter);

9
aqhome/ipc2/ipcmsgreader.c
View File

@@ -17,6 +17,7 @@
#include <gwenhywfar/inherit.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endianfns.h>
#include <gwenhywfar/text.h>
#include <sys/socket.h>
@@ -63,6 +64,7 @@ int _readMsg(AQH_OBJECT *o)
int rv;
if (xo->bytesReceived<AQH_MSG_READER_HEADER_SIZE) {
DBG_INFO(AQH_LOGDOMAIN, "Reading header");
rv=_readHeaderFromRingbuffer(xo);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
@@ -71,6 +73,7 @@ int _readMsg(AQH_OBJECT *o)
}
if (xo->bytesReceived>=AQH_MSG_READER_HEADER_SIZE) {
DBG_INFO(AQH_LOGDOMAIN, "Reading body");
/* reading remainder of msg directly into allocated buffer */
rv=AQH_MsgReader_ReadRemainderFromRingbuffer(o);
if (rv<0) {
@@ -90,7 +93,7 @@ int _readMsg(AQH_OBJECT *o)
rv=AQH_Object_EmitSignal(o, AQH_MSG_READER_SIGNAL_MSGRECVD, msgLen, (void*) msgPtr);
if (rv==0) {
DBG_INFO(AQH_LOGDOMAIN, "Received message ignored");
DBG_ERROR(AQH_LOGDOMAIN, "Received message ignored");
}
free(msgPtr);
return 1;
@@ -136,9 +139,9 @@ int _readHeaderFromRingbuffer(AQH_MSG_READER *xo)
DBG_ERROR(AQH_LOGDOMAIN, "Bad message size(%lu)", (unsigned long int) msgLen);
return GWEN_ERROR_GENERIC;
}
xo->currentMsgBuf=(uint8_t*) malloc(msgLen+4); /* +4 because of msg len (4 bytes) */
xo->currentMsgBuf=(uint8_t*) malloc(msgLen);
memmove(xo->currentMsgBuf, xo->headerBuffer, xo->bytesReceived);
xo->bytesLeft=(msgLen+4)-xo->bytesReceived;
xo->bytesLeft=msgLen-xo->bytesReceived;
}
return 0;

5
aqhome/ipc2/msgreader.c
View File

@@ -201,6 +201,7 @@ int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
{
AQH_MSG_READER *xo;
DBG_INFO(AQH_LOGDOMAIN, "Socket ready");
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
if (xo) {
int rv;
@@ -401,7 +402,7 @@ void _resetBuffers(AQH_MSG_READER *xo)
void _cbEnable(AQH_OBJECT *o)
{
if (o && !(AQH_Object_GetFlags(o) & AQH_OBJECT_FLAGS_ENABLED)) {
if (o) {
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);
@@ -414,7 +415,7 @@ void _cbEnable(AQH_OBJECT *o)
void _cbDisable(AQH_OBJECT *o)
{
if (o && (AQH_Object_GetFlags(o) & AQH_OBJECT_FLAGS_ENABLED)) {
if (o) {
AQH_MSG_READER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_READER, o);

18
aqhome/ipc2/msgwriter.c
View File

@@ -129,11 +129,12 @@ int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
{
AQH_MSG_WRITER *xo;
DBG_INFO(AQH_LOGDOMAIN, "Socket ready");
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_MSG_WRITER, o);
if (xo) {
if (xo->bytesLeft) {
int rv;
int rv;
if (xo->bytesLeft) {
if (!(xo->flags & AQH_MSGWRITER_FLAGS_MSGSTARTED)) {
rv=_startMsg(xo, fdObject);
if (rv<0) {
@@ -161,12 +162,17 @@ int _handleSocketReady(AQH_OBJECT *o, AQH_OBJECT *fdObject)
}
else {
if (xo->bytesLeft==0) {
int msgLen;
const uint8_t *msgPtr;
_endMsg(xo, fdObject);
rv=AQH_Object_EmitSignal(o, AQH_MSG_WRITER_SIGNAL_MSGSENT, xo->msgBufLen, (void*) (xo->msgBufPtr));
if (rv==0) {
DBG_INFO(AQH_LOGDOMAIN, "Sent message ignored");
}
msgPtr=xo->msgBufPtr;
msgLen=xo->msgBufLen;
_resetBuffer(o);
rv=AQH_Object_EmitSignal(o, AQH_MSG_WRITER_SIGNAL_MSGSENT, msgLen, (void*) msgPtr);
if (rv==0) {
DBG_ERROR(AQH_LOGDOMAIN, "Sent message ignored");
}
}
}
}

178
aqhome/ipc2/tcp_object.c Normal file
View File

@@ -0,0 +1,178 @@
/****************************************************************************
* 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 "./tcp_object.h"
#include <gwenhywfar/debug.h>
#include <unistd.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <string.h>
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static int _socketSetBlocking(int sk, int fl);
static int _translateHError(int herr);
static int _setHostAddr(struct in_addr *inetAddr, const char *sAddr);
static int _setHostName(struct in_addr *inetAddr, const char *sAddr);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
int AQH_TcpObject_CreateConnectedSocket(const char *addr, int port)
{
int sk;
struct sockaddr_in inetAddr;
int rv;
memset(&inetAddr, 0, sizeof(inetAddr));
inetAddr.sin_family=AF_INET;
rv=_setHostAddr(&inetAddr.sin_addr, addr); /* try tuple */
if (rv<0) {
rv=_setHostName(&inetAddr.sin_addr, addr); /* lookup name */
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
return rv;
}
}
inetAddr.sin_port=htons(port);
sk=socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sk<0) {
DBG_INFO(AQH_LOGDOMAIN, "socket(): %s", strerror(errno));
return GWEN_ERROR_IO;
}
rv=connect(sk, (struct sockaddr*) &inetAddr, sizeof(inetAddr));
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
close(sk);
return rv;
}
rv=_socketSetBlocking(sk, 0);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
close(sk);
return rv;
}
return sk;
}
int _setHostAddr(struct in_addr *inetAddr, const char *sAddr)
{
inetAddr->s_addr=0;
if (!inet_aton(sAddr, inetAddr)) {
DBG_ERROR(AQH_LOGDOMAIN, "Invalid address \"%s\"", sAddr);
return GWEN_ERROR_INVALID;
}
return 0;
}
int _setHostName(struct in_addr *inetAddr, const char *sAddr)
{
struct hostent *he;
he=gethostbyname(sAddr);
if (!he) {
DBG_ERROR(AQH_LOGDOMAIN, "gethostbyname(\"%s\"): %s", sAddr, hstrerror(h_errno));
return _translateHError(h_errno);
}
/* name resolved, store address */
memcpy(inetAddr, he->h_addr_list[0], sizeof(struct in_addr));
return 0;
}
int _socketSetBlocking(int sk, int fl)
{
int prevFlags;
int newFlags;
/* get current socket flags */
prevFlags=fcntl(sk, F_GETFL);
if (prevFlags==-1) {
DBG_INFO(AQH_LOGDOMAIN, "fcntl(): %s", strerror(errno));
return GWEN_ERROR_IO;
}
/* set nonblocking/blocking */
if (fl)
newFlags=prevFlags&(~O_NONBLOCK);
else
newFlags=prevFlags|O_NONBLOCK;
if (-1==fcntl(sk, F_SETFL, newFlags)) {
DBG_INFO(AQH_LOGDOMAIN, "fcntl(): %s", strerror(errno));
return GWEN_ERROR_IO;
}
prevFlags=fcntl(sk, F_GETFL);
if (prevFlags!=newFlags) {
DBG_ERROR(AQH_LOGDOMAIN, "fcntl() did not set flags correctly (%08x!=%08x)", prevFlags, newFlags);
return GWEN_ERROR_IO;
}
return 0;
}
int _translateHError(int herr)
{
int rv;
switch (herr) {
case HOST_NOT_FOUND:
rv=GWEN_ERROR_HOST_NOT_FOUND;
break;
#ifdef NO_ADDRESS
case NO_ADDRESS:
rv=GWEN_ERROR_NO_ADDRESS;
break;
#endif
case NO_RECOVERY:
rv=GWEN_ERROR_NO_RECOVERY;
break;
case TRY_AGAIN:
rv=GWEN_ERROR_TRY_AGAIN;
break;
default:
rv=GWEN_ERROR_UNKNOWN_DNS_ERROR;
break;
} /* switch */
return rv;
}

24
aqhome/ipc2/tcp_object.h Normal file
View File

@@ -0,0 +1,24 @@
/****************************************************************************
* 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.
****************************************************************************/
#ifndef AQH_TCP_OBJECT_H
#define AQH_TCP_OBJECT_H
#include <aqhome/events2/object.h>
/**
* Helper function to create and connect a TCP socket.
*/
AQHOME_API int AQH_TcpObject_CreateConnectedSocket(const char *addr, int port);
#endif

42
aqhome/ipc2/tcpd_object.c
View File

@@ -47,6 +47,8 @@ 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 void _cbEnable(AQH_OBJECT *o);
static void _cbDisable(AQH_OBJECT *o);
static int _socketSetReuseAddress(int sk, int fl);
static int _socketSetBlocking(int sk, int fl);
@@ -71,16 +73,12 @@ AQH_OBJECT *AQH_TcpdObject_new(AQH_EVENT_LOOP *eventLoop, int fd)
GWEN_NEW_OBJECT(AQH_TCPD_OBJECT, xo);
GWEN_INHERIT_SETDATA(AQH_OBJECT, AQH_TCPD_OBJECT, o, xo, _freeData);
AQH_Object_SetSignalHandlerFn(o, _handleSignal);
AQH_Object_SetEnableFn(o, _cbEnable);
AQH_Object_SetDisableFn(o, _cbDisable);
xo->fdSocket=fd;
xo->fdObject=AQH_FdObject_new(eventLoop, fd, AQH_FDOBJECT_FDMODE_READ);
#if 0
/* create object for readable socket, connect to THIS, enable */
xo->fdObject=AQH_FdObject_new(AQH_Object_GetEventLoop(o), rv, AQH_FDOBJECT_FDMODE_READ);
AQH_Object_AddLink(xo->fdObject, AQH_FDOBJECT_SIGNAL_ISREADY, AQH_TCPD_OBJECT_SLOT_SOCKETREADY, o);
AQH_Object_Enable(xo->fdObject);
#endif
return o;
}
@@ -178,17 +176,19 @@ int _handleSocketReady(AQH_OBJECT *o)
{
AQH_TCPD_OBJECT *xo;
DBG_ERROR(NULL, "Socket ready");
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_TCPD_OBJECT, o);
if (xo) {
int clientSk;
clientSk=_acceptConnection(xo->fdSocket);
if (clientSk<0) {
DBG_INFO(AQH_LOGDOMAIN, "here (%d)", clientSk);
DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", clientSk);
}
else {
DBG_ERROR(AQH_LOGDOMAIN, "New connection");
if (0==AQH_Object_EmitSignal(o, AQH_TCPD_OBJECT_SIGNAL_NEWCONN, clientSk, NULL)) {
DBG_INFO(AQH_LOGDOMAIN, "New connection not handled");
DBG_ERROR(AQH_LOGDOMAIN, "New connection not handled");
close(clientSk);
}
}
@@ -199,6 +199,32 @@ int _handleSocketReady(AQH_OBJECT *o)
void _cbEnable(AQH_OBJECT *o)
{
if (o) {
AQH_TCPD_OBJECT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_TCPD_OBJECT, o);
if (xo && xo->fdObject)
AQH_Object_Enable(xo->fdObject);
}
}
void _cbDisable(AQH_OBJECT *o)
{
if (o) {
AQH_TCPD_OBJECT *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_TCPD_OBJECT, o);
if (xo && xo->fdObject)
AQH_Object_Disable(xo->fdObject);
}
}
int _setHostAddr(struct in_addr *inetAddr, const char *sAddr)
{
inetAddr->s_addr=0;