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

aqhome: More work on endpoint system. Basically works.

Browse Source
This commit is contained in:
Martin Preuss
2023-02-22 17:54:17 +01:00
parent 3803afbdea
commit b2138af652
12 changed files with 242 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
aqhome/msgendpointmanager.c
View File

@@ -110,42 +110,57 @@ int _ioLoopOnce(AQH_MSG_ENDPOINT_MGR *emgr)
tv.tv_sec=2;
tv.tv_usec=0;
DBG_INFO(AQH_LOGDOMAIN, "Sampling sockets");
ep=AQH_MsgEndpoint_List_First(emgr->endpointList);
if (ep==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "No endpoints.");
return GWEN_ERROR_GENERIC;
}
while(ep) {
DBG_INFO(AQH_LOGDOMAIN, "- checking endpoint %s", AQH_MsgEndpoint_GetName(ep));
if (!(AQH_MsgEndpoint_GetFlags(ep) & AQH_MSG_ENDPOINT_FLAGS_NOIO)) {
int fd;
fd=AQH_MsgEndpoint_GetReadFd(ep);
if (fd!=-1) {
if (fd>=0) {
DBG_INFO(AQH_LOGDOMAIN, " - adding socket %d for read", fd);
FD_SET(fd, &readSet);
highestRdFd=(fd>highestRdFd)?fd:highestRdFd;
}
fd=AQH_MsgEndpoint_GetWriteFd(ep);
if (fd!=-1) {
if (fd>=0) {
DBG_INFO(AQH_LOGDOMAIN, " - adding socket %d for write", fd);
FD_SET(fd, &writeSet);
highestWrFd=(fd>highestWrFd)?fd:highestWrFd;
}
}
else {
DBG_INFO(AQH_LOGDOMAIN, " - endpoint %s does not support IO", AQH_MsgEndpoint_GetName(ep));
}
ep=AQH_MsgEndpoint_List_Next(ep);
}
DBG_INFO(AQH_LOGDOMAIN, "Calling select (highest read socket: %d, highest write socket: %d)", highestRdFd, highestWrFd);
rv=select(((highestRdFd>highestWrFd)?highestRdFd:highestWrFd)+1,
(highestRdFd<0)?NULL:&readSet,
(highestWrFd<0)?NULL:&writeSet,
NULL,
&tv);
DBG_INFO(AQH_LOGDOMAIN, "Return from select (%d, %d=%s)", rv, (rv<0)?errno:0, (rv<0)?strerror(errno):"no error");
if (rv<0) {
if (errno!=EINTR) {
DBG_ERROR(NULL, "Error on select");
DBG_ERROR(AQH_LOGDOMAIN, "Error on select");
return GWEN_ERROR_IO;
}
}
else if (rv==0) {
/* timeout */
DBG_INFO(AQH_LOGDOMAIN, "timeout");
return GWEN_ERROR_TRY_AGAIN;
}
else if (rv) {
DBG_INFO(AQH_LOGDOMAIN, "Letting all endpoints handle IO");
ep=AQH_MsgEndpoint_List_First(emgr->endpointList);
while(ep) {
AQH_MSG_ENDPOINT *epNext;
@@ -155,15 +170,19 @@ int _ioLoopOnce(AQH_MSG_ENDPOINT_MGR *emgr)
epNext=AQH_MsgEndpoint_List_Next(ep);
fd=AQH_MsgEndpoint_GetFd(ep);
if (fd!=-1 && FD_ISSET(fd, &readSet)) {
DBG_INFO(AQH_LOGDOMAIN, "- endpoint(%s): read", AQH_MsgEndpoint_GetName(ep));
rv=AQH_MsgEndpoint_HandleReadable(ep, emgr);
if (rv<0 && rv!=GWEN_ERROR_TRY_AGAIN) {
DBG_INFO(AQH_LOGDOMAIN, "error, removing endpoint %s", AQH_MsgEndpoint_GetName(ep));
fd=-1;
AQH_MsgEndpointMgr_DelEndpoint(emgr, ep);
}
}
if (fd!=-1 && FD_ISSET(fd, &writeSet)) {
DBG_INFO(AQH_LOGDOMAIN, "- endpoint(%s): write", AQH_MsgEndpoint_GetName(ep));
rv=AQH_MsgEndpoint_HandleWritable(ep, emgr);
if (rv<0 && rv!=GWEN_ERROR_TRY_AGAIN) {
DBG_INFO(AQH_LOGDOMAIN, "error, removing endpoint %s", AQH_MsgEndpoint_GetName(ep));
fd=-1;
AQH_MsgEndpointMgr_DelEndpoint(emgr, ep);
}
@@ -181,21 +200,30 @@ void _msgLoopOnce(AQH_MSG_ENDPOINT_MGR *emgr)
{
AQH_MSG_ENDPOINT *ep;
DBG_INFO(AQH_LOGDOMAIN, "Handle endpoint messages");
ep=AQH_MsgEndpoint_List_First(emgr->endpointList);
while(ep) {
AQH_MSG *msg;
DBG_INFO(AQH_LOGDOMAIN, "- endpoint(%s)", AQH_MsgEndpoint_GetName(ep));
while( (msg=AQH_MsgEndpoint_TakeFirstReceivedMessage(ep)) ) {
uint32_t msgGroup;
DBG_INFO(AQH_LOGDOMAIN,
" - msg %d from %d to %d",
AQH_Msg_GetMsgType(msg), AQH_Msg_GetSourceAddress(msg), AQH_Msg_GetDestAddress(msg));
msgGroup=_getMsgGroup(AQH_Msg_GetMsgType(msg));
if (msgGroup & AQH_MSG_ENDPOINT_MSGGROUP_ADMIN) {
if (!(AQH_MsgEndpoint_GetGroupId(ep) & AQH_MSG_ENDPOINT_ENDPOINTGROUP_BUS))
if (!(AQH_MsgEndpoint_GetGroupId(ep) & AQH_MSG_ENDPOINT_ENDPOINTGROUP_BUS)) {
/* only handle admin messages not from nodes */
DBG_INFO(AQH_LOGDOMAIN, " - handling admin message");
_handleAdminMsg(emgr, ep, msg);
}
}
else
else {
DBG_INFO(AQH_LOGDOMAIN, " - distributing message");
_distributeMsg(emgr, ep, msg);
}
AQH_Msg_free(msg);
}
@@ -209,11 +237,13 @@ void _runAllEndpoints(AQH_MSG_ENDPOINT_MGR *emgr)
{
AQH_MSG_ENDPOINT *ep;
DBG_INFO(AQH_LOGDOMAIN, "Running all endpoints");
ep=AQH_MsgEndpoint_List_First(emgr->endpointList);
while(ep) {
AQH_MSG_ENDPOINT *next;
next=AQH_MsgEndpoint_List_Next(ep);
DBG_INFO(AQH_LOGDOMAIN, "- running endpoint %s", AQH_MsgEndpoint_GetName(ep));
AQH_MsgEndpoint_Run(ep);
ep=next;
}
@@ -236,6 +266,7 @@ void _distributeMsg(AQH_MSG_ENDPOINT_MGR *emgr, AQH_MSG_ENDPOINT *srcEp, const A
uint32_t acceptedGroupIds;
uint32_t acceptedMsgGroups;
DBG_INFO(AQH_LOGDOMAIN, "- checking endpoint %s", AQH_MsgEndpoint_GetName(ep));
acceptedGroupIds=AQH_MsgEndpoint_GetAcceptedMsgGroups(ep);
acceptedMsgGroups=AQH_MsgEndpoint_GetAcceptedMsgGroups(ep);
@@ -245,8 +276,12 @@ void _distributeMsg(AQH_MSG_ENDPOINT_MGR *emgr, AQH_MSG_ENDPOINT *srcEp, const A
(acceptedGroupIds & srcGroupId)
) {
/* endpoint accepts this message */
DBG_INFO(AQH_LOGDOMAIN, " - endpoint %s accepts message", AQH_MsgEndpoint_GetName(ep));
AQH_MsgEndpoint_AddSendMessage(ep, AQH_Msg_dup(msg));
}
else {
DBG_INFO(AQH_LOGDOMAIN, " - endpoint %s does not accept message", AQH_MsgEndpoint_GetName(ep));
}
}
ep=AQH_MsgEndpoint_List_Next(ep);
}