More work on IPC code, added aqhomed daemon.

aqhome/0BUILD

@@ -46,22 +46,26 @@
     <headers dist="true" install="$(pkgincludedir)" >
       api.h
       aqhome.h
+      msgmanager.h
     </headers>
   
   
     <sources>
       $(local/typefiles)
       aqhome.c
+      msgmanager.c
     </sources>
 
     <subdirs>
       msg
+      ipc
       nodes
     </subdirs>
 
 
     <useTargets>
       aqhmsg
+      aqhipc
       aqhnodes
     </useTargets>
 

aqhome/ipc/0BUILD

@@ -0,0 +1,84 @@
+<?xml?>
+
+<gwbuild>
+
+  <target type="ConvenienceLibrary" name="aqhipc" >
+
+    <includes type="c" >
+      $(gwenhywfar_cflags)
+      -I$(topsrcdir)
+      -I$(topbuilddir)
+    </includes>
+  
+    <includes type="tm2" >
+      --include=$(builddir)
+      --include=$(srcdir)
+    </includes>
+
+
+    <define name="BUILDING_AQHOME" />
+
+    <setVar name="local/cflags">$(visibility_cflags)</setVar>
+
+  
+    <setVar name="tm2flags" >
+      --api=AQHOME_API
+    </setVar>
+
+    <setVar name="local/typefiles" >
+    </setVar>
+  
+    <setVar name="local/built_sources" >
+    </setVar>
+  
+    <setVar name="local/built_headers_pub">
+    </setVar>
+  
+  
+    <setVar name="local/built_headers_priv" >
+    </setVar>
+
+
+    <headers dist="false" install="$(pkgincludedir)/ipc" >
+      $(local/built_headers_pub)
+    </headers>
+  
+  
+    <headers dist="true" install="$(pkgincludedir)/ipc" >
+      endpoint_node_ipc.h
+      endpoint_node_ipc_tcp.h
+      msg_ipc.h
+      msg_forward.h
+    </headers>
+
+
+    <headers dist="true" >
+    </headers>
+  
+  
+    <sources>
+      $(local/typefiles)
+
+      endpoint_node_ipc.c
+      endpoint_node_ipc_tcp.c
+      msg_ipc.c
+      msg_forward.c
+    </sources>
+
+
+    <extradist>
+    </extradist>
+
+
+    <useTargets>
+    </useTargets>
+
+    <subdirs>
+    </subdirs>
+
+  
+  
+
+  </target>
+  
+</gwbuild>

aqhome/ipc/endpoint_node_ipc.c

@@ -0,0 +1,64 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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 "aqhome/ipc/endpoint_node_ipc.h"
+#include "aqhome/msg/endpoint_node.h"
+#include "aqhome/msg/msg_node.h"
+#include "aqhome/ipc/msg_forward.h"
+
+#include <gwenhywfar/list.h>
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/gwentime.h>
+#include <gwenhywfar/endpoint_ipc.h>
+
+
+#define AQH_MSG_ENDPOINT_NODEIPC_NAME            "nodeipc"
+
+
+
+static void _processOutMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *m);
+
+
+
+
+
+GWEN_MSG_ENDPOINT *AQH_IpcNodeEndpoint_new(const char *name, int groupId)
+{
+  int fd;
+  GWEN_MSG_ENDPOINT *ep;
+
+  ep=GWEN_IpcEndpoint_new(name?name:AQH_MSG_ENDPOINT_NODEIPC_NAME, groupId);
+  AQH_NodeEndpoint_Extend(ep);
+  GWEN_MsgEndpoint_SetProcessOutMsgFn(ep, _processOutMessage);
+
+  AQH_NodeEndpoint_SetAcceptedMsgGroups(ep, AQH_MSG_TYPEGROUP_ALL);
+
+  return ep;
+}
+
+
+
+void _processOutMessage(GWEN_MSG_ENDPOINT *ep, GWEN_MSG *nodeMsg)
+{
+  GWEN_MSG *ipcMsg;
+
+  ipcMsg=AQH_ForwardMsg_new(AQH_MSGTYPE_IPC_FORWARD, GWEN_Msg_GetConstBuffer(nodeMsg), GWEN_Msg_GetBytesInBuffer(nodeMsg));
+  GWEN_MsgEndpoint_AddSendMessage(ep, ipcMsg);
+  GWEN_Msg_free(nodeMsg);
+}
+
+
+
+
+
+

aqhome/ipc/endpoint_node_ipc.h

@@ -0,0 +1,24 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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_ENDPOINT_NODE_IPC_H
+#define AQH_ENDPOINT_NODE_IPC_H
+
+
+#include <aqhome/api.h>
+
+#include <gwenhywfar/endpoint.h>
+
+
+
+AQHOME_API GWEN_MSG_ENDPOINT *AQH_IpcNodeEndpoint_new(const char *name, int groupId);
+
+AQHOME_API int AQH_IpcNodeEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr);
+
+
+#endif

aqhome/ipc/endpoint_node_ipc_tcp.c

@@ -0,0 +1,63 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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 "aqhome/ipc/endpoint_node_ipc_tcp.h"
+#include "aqhome/ipc/endpoint_node_ipc.h"
+#include "aqhome/msg/endpoint_node.h"
+#include "aqhome/msg/msg_node.h"
+#include "aqhome/ipc/msg_forward.h"
+
+#include <gwenhywfar/list.h>
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/gwentime.h>
+#include <gwenhywfar/endpoint_ipc_tcp.h>
+
+
+#define AQH_MSG_ENDPOINT_NODEIPCTCP_NAME            "nodeipctcp"
+
+
+
+static GWEN_MSG_ENDPOINT *_createChild(GWEN_MSG_ENDPOINT *ep);
+
+
+
+
+
+GWEN_MSG_ENDPOINT *AQH_TcpIpcNodeEndpoint_new(const char *name, const char *host, int port, int groupId)
+{
+  int fd;
+  GWEN_MSG_ENDPOINT *ep;
+
+  ep=GWEN_TcpIpcEndpoint_new(name?name:AQH_MSG_ENDPOINT_NODEIPCTCP_NAME, host, port, groupId);
+  GWEN_MsgEndpoint_AddFlags(ep, AQH_MSGEP_NODE_FLAGS_NOMESSAGES);
+  GWEN_MsgEndpoint_SetCreateChildFn(ep, _createChild);
+
+  return ep;
+}
+
+
+
+GWEN_MSG_ENDPOINT *_createChild(GWEN_MSG_ENDPOINT *ep)
+{
+  DBG_INFO(AQH_LOGDOMAIN, "Creating child endpoint for %s", GWEN_MsgEndpoint_GetName(ep));
+  return AQH_IpcNodeEndpoint_new(NULL, GWEN_MsgEndpoint_GetGroupId(ep));
+}
+
+
+
+
+
+
+
+
+

aqhome/ipc/endpoint_node_ipc_tcp.h

@@ -0,0 +1,22 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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_ENDPOINT_NODE_IPC_TCP_H
+#define AQH_ENDPOINT_NODE_IPC_TCP_H
+
+
+#include <aqhome/api.h>
+
+#include <gwenhywfar/endpoint_ipc_tcp.h>
+
+
+
+AQHOME_API GWEN_MSG_ENDPOINT *AQH_TcpIpcNodeEndpoint_new(const char *name, const char *host, int port, int groupId);
+
+
+#endif

aqhome/ipc/msg_forward.c

@@ -0,0 +1,82 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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 <aqhome/ipc/msg_forward.h>
+
+#include <gwenhywfar/msg.h>
+#include <gwenhywfar/buffer.h>
+
+#include <gwenhywfar/misc.h>
+#include <gwenhywfar/list.h>
+#include <gwenhywfar/error.h>
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/text.h>
+#include <gwenhywfar/msg_ipc.h>
+
+
+
+
+GWEN_MSG *AQH_ForwardMsg_new(uint16_t code, const uint8_t *ptr, uint32_t len)
+{
+  return GWEN_IpcMsg_new(AQH_IPC_PROTOCOL_ID, AQH_IPC_PROTOCOL_VERSION, code, len, ptr);
+}
+
+
+
+const uint8_t *AQH_ForwardMsg_GetMsgPtr(const GWEN_MSG *msg)
+{
+  if (GWEN_Msg_GetBytesInBuffer(msg)>=AQH_MSG_FORWARD_MINSIZE)
+    return GWEN_Msg_GetConstBuffer(msg)+AQH_MSG_OFFS_FORWARD_MSG;
+  return NULL;
+}
+
+
+
+uint32_t AQH_ForwardMsg_GetMsgLen(const GWEN_MSG *msg)
+{
+  uint32_t len;
+
+  len=GWEN_Msg_GetBytesInBuffer(msg);
+  if (len>AQH_MSG_FORWARD_MINSIZE) {
+    return len-AQH_MSG_OFFS_FORWARD_MSG;
+  }
+  return 0;
+}
+
+
+
+void AQH_ForwardMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf, const char *sText)
+{
+  if (GWEN_Msg_GetBytesInBuffer(msg)>=AQH_MSG_FORWARD_MINSIZE) {
+    const uint8_t *ptr;
+    uint32_t len;
+
+    ptr=AQH_ForwardMsg_GetMsgPtr(msg);
+    len=AQH_ForwardMsg_GetMsgLen(msg);
+
+    GWEN_Buffer_AppendArgs(dbuf, "FORWARD (code=%d, protocol=%d, protocol version=%d)\n",
+                           GWEN_IpcMsg_GetCode(msg),
+                           GWEN_IpcMsg_GetProtoId(msg),
+                           GWEN_IpcMsg_GetProtoVersion(msg));
+    if (ptr && len) {
+      GWEN_Text_DumpString2Buffer(ptr, len, dbuf, 2);
+      GWEN_Buffer_AppendByte(dbuf, '\n');
+    }
+  }
+}
+
+
+
+
+
+
+

aqhome/ipc/msg_forward.h

@@ -0,0 +1,37 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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_MSG_FORWARD_H
+#define AQH_MSG_FORWARD_H
+
+
+#include <aqhome/api.h>
+#include <aqhome/ipc/msg_ipc.h>
+#include <aqhome/msg/msg_node.h>
+
+#include <gwenhywfar/msg.h>
+#include <gwenhywfar/buffer.h>
+
+
+
+#define AQH_MSG_OFFS_FORWARD_MSG        (GWEN_MSGIPC_OFFS_PAYLOAD+0)
+
+#define AQH_MSG_FORWARD_MINSIZE (AQH_MSG_OFFS_FORWARD_MSG+AQH_MSG_OFFS_ALL_DATA_BEGIN)
+
+
+
+AQHOME_API GWEN_MSG *AQH_ForwardMsg_new(uint16_t code, const uint8_t *ptr, uint32_t len);
+AQHOME_API const uint8_t *AQH_ForwardMsg_GetMsgPtr(const GWEN_MSG *msg);
+AQHOME_API uint32_t AQH_ForwardMsg_GetMsgLen(const GWEN_MSG *msg);
+AQHOME_API void AQH_ForwardMsg_DumpToBuffer(const GWEN_MSG *msg, GWEN_BUFFER *dbuf, const char *sText);
+
+
+#endif
+
+
+

aqhome/ipc/msg_ipc.c

@@ -0,0 +1,16 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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 <aqhome/ipc/msg_ipc.h>
+
+

aqhome/ipc/msg_ipc.h

@@ -0,0 +1,30 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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_MSG_IPC_H
+#define AQH_MSG_IPC_H
+
+
+#include <aqhome/api.h>
+
+#include <gwenhywfar/msg_ipc.h>
+#include <gwenhywfar/buffer.h>
+
+
+#define AQH_IPC_PROTOCOL_ID      1
+#define AQH_IPC_PROTOCOL_VERSION 1
+
+
+#define AQH_MSGTYPE_IPC_FORWARD   0x100
+
+
+
+#endif
+
+
+

aqhome/libtest.c

@@ -8,6 +8,8 @@
 #include "aqhome/msg/endpoint_node.h"
 #include "aqhome/msg/endpoint_log.h"
 #include "aqhome/msg/endpoint_tty.h"
+#include "aqhome/ipc/endpoint_node_ipc_tcp.h"
+#include "aqhome/msgmanager.h"
 
 #include "aqhome/aqhome.h"
 
@@ -97,29 +99,37 @@ int testEndpoints()
   GWEN_MSG_ENDPOINT_MGR *emgr;
   GWEN_MSG_ENDPOINT *epTty;
   GWEN_MSG_ENDPOINT *epLog;
+  GWEN_MSG_ENDPOINT *epTcp;
 
   rv=AQH_Init();
   if (rv<0) {
   }
 
-  emgr=AQH_MsgEndpointMgr_new(0xc0);
-  epTty=AQH_TtyNodeEndpoint_new("/dev/ttyUSB0", AQH_MSG_ENDPOINTGROUP_NODE);
+  emgr=AQH_MsgManager_new(0xc0);
+  epTty=AQH_TtyNodeEndpoint_new("/dev/ttyUSB0", AQH_MSGMGR_ENDPOINTGROUP_NODE);
   if (epTty==NULL) {
     DBG_ERROR(NULL, "Error creating endpoint TTY");
     return 2;
   }
   GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTty);
 
-  epLog=AQH_LogEndpoint_new("endpoints.log", AQH_MSG_ENDPOINTGROUP_NODE);
+  epLog=AQH_LogEndpoint_new("endpoints.log", AQH_MSGMGR_ENDPOINTGROUP_NODE);
   if (epLog==NULL) {
     DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint LOG");
     return 2;
   }
   GWEN_MsgEndpointMgr_AddEndpoint(emgr, epLog);
 
+  epTcp=AQH_TcpIpcNodeEndpoint_new(NULL, "127.0.0.1", 45454, AQH_MSGMGR_ENDPOINTGROUP_IPC);
+  if (epTcp==NULL) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP");
+    return 2;
+  }
+  GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTcp);
+
   for (;;) {
     DBG_DEBUG(AQH_LOGDOMAIN, "Next loop");
-    AQH_MsgEndpointMgr_LoopOnce(emgr);
+    AQH_MsgManager_LoopOnce(emgr);
   }
   return 0;
 }

aqhome/msg/0BUILD

@@ -39,12 +39,12 @@
     </setVar>
 
 
-    <headers dist="false" install="$(pkgincludedir)/nodes" >
+    <headers dist="false" install="$(pkgincludedir)/msg" >
       $(local/built_headers_pub)
     </headers>
   
   
-    <headers dist="true" install="$(pkgincludedir)/nodes" >
+    <headers dist="true" install="$(pkgincludedir)/msg" >
       endpointmgr.h
       endpoint_node.h
       endpoint_log.h

aqhome/msg/endpoint_node.c

@@ -39,18 +39,25 @@ static void GWENHYWFAR_CB _freeData(void *bp, void *p);
 GWEN_MSG_ENDPOINT *AQH_NodeEndpoint_new(const char *name, int groupId)
 {
   GWEN_MSG_ENDPOINT *ep;
-  AQH_MSG_ENDPOINT_NODE *xep;
-  int fd;
 
   ep=GWEN_MsgEndpoint_new(name?name:AQH_MSG_ENDPOINT_NODE_NAME, groupId);
-  GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT_NODE, xep);
-  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep, xep, _freeData);
+  AQH_NodeEndpoint_Extend(ep);
 
   return ep;
 }
 
 
 
+void AQH_NodeEndpoint_Extend(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_NODE *xep;
+
+  GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT_NODE, xep);
+  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep, xep, _freeData);
+}
+
+
+
 void _freeData(void *bp, void *p)
 {
   AQH_MSG_ENDPOINT_NODE *xep;

aqhome/msg/endpoint_node.h

@@ -20,6 +20,7 @@
 
 
 AQHOME_API GWEN_MSG_ENDPOINT *AQH_NodeEndpoint_new(const char *name, int groupId);
+AQHOME_API void AQH_NodeEndpoint_Extend(GWEN_MSG_ENDPOINT *ep);
 
 AQHOME_API uint32_t AQH_NodeEndpoint_GetAcceptedMsgGroups(const GWEN_MSG_ENDPOINT *ep);
 AQHOME_API void AQH_NodeEndpoint_SetAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f);

aqhome/msg/endpointmgr.c

@@ -27,10 +27,8 @@ GWEN_INHERIT(GWEN_MSG_ENDPOINT_MGR, AQH_MSG_ENDPOINT_MGR);
 
 static void GWENHYWFAR_CB _freeData(void *bp, void *p);
 
-static void _msgLoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr);
 static void _handleEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep);
 static void _handleNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep);
-static void _handleIpcEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep);
 static void _distributeMsgFromNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *srcEp, const GWEN_MSG *msg);
 
 
@@ -64,19 +62,19 @@ void _freeData(void *bp, void *p)
 
 
 
-int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
+int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr, int groupId)
 {
   int rv;
 
   rv=GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
-  _msgLoopOnce(emgr);
+  AQH_MsgEndpointMgr_LoopOnceOverNodeEndpoints(emgr, groupId);
   GWEN_MsgEndpointMgr_RunAllEndpoints(emgr);
   return rv;
 }
 
 
 
-void _msgLoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
+void AQH_MsgEndpointMgr_LoopOnceOverNodeEndpoints(GWEN_MSG_ENDPOINT_MGR *emgr, int groupId)
 {
   GWEN_MSG_ENDPOINT_LIST *endpointList;
 
@@ -88,7 +86,8 @@ void _msgLoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
     ep=GWEN_MsgEndpoint_List_First(endpointList);
     while(ep) {
       DBG_DEBUG(AQH_LOGDOMAIN, "- endpoint(%s)", GWEN_MsgEndpoint_GetName(ep));
-      _handleEndpoint(emgr, ep);
+      if (GWEN_MsgEndpoint_GetGroupId(ep)==groupId)
+        _handleNodeEndpoint(emgr, ep);
       ep=GWEN_MsgEndpoint_List_Next(ep);
     }
   }
@@ -96,17 +95,6 @@ void _msgLoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
 
 
 
-void _handleEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep)
-{
-  switch(GWEN_MsgEndpoint_GetGroupId(ep)) {
-  case AQH_MSG_ENDPOINTGROUP_NODE: _handleNodeEndpoint(emgr, ep); break;
-  case AQH_MSG_ENDPOINTGROUP_IPC:  _handleIpcEndpoint(emgr, ep); break;
-  default: break;
-  }
-}
-
-
-
 void _handleNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep)
 {
   GWEN_MSG *msg;
@@ -122,13 +110,6 @@ void _handleNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep)
 
 
 
-void _handleIpcEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep)
-{
-  /* TODO: handle IPC messages */
-}
-
-
-
 void _distributeMsgFromNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *srcEp, const GWEN_MSG *msg)
 {
   GWEN_MSG_ENDPOINT_LIST *endpointList;
@@ -159,7 +140,7 @@ void _distributeMsgFromNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOI
            ) {
           /* endpoint accepts this message */
           DBG_DEBUG(AQH_LOGDOMAIN, "  - endpoint %s accepts message", GWEN_MsgEndpoint_GetName(ep));
-          GWEN_MsgEndpoint_AddSendMessage(ep, GWEN_Msg_dup(msg));
+          GWEN_MsgEndpoint_ProcessOutMessage(ep, GWEN_Msg_dup(msg));
         }
         else {
           DBG_DEBUG(AQH_LOGDOMAIN, "  - endpoint %s does not accept message", GWEN_MsgEndpoint_GetName(ep));

aqhome/msg/endpointmgr.h

@@ -16,13 +16,10 @@
 
 
 
-#define AQH_MSG_ENDPOINTGROUP_NODE 1
-#define AQH_MSG_ENDPOINTGROUP_IPC  2
-
-
-
 AQHOME_API GWEN_MSG_ENDPOINT_MGR *AQH_MsgEndpointMgr_new(uint8_t busAddr);
-AQHOME_API int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr);
+AQHOME_API int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr, int groupId);
+
+AQHOME_API void AQH_MsgEndpointMgr_LoopOnceOverNodeEndpoints(GWEN_MSG_ENDPOINT_MGR *emgr, int groupId);
 
 
 

aqhome/msgmanager.c

@@ -0,0 +1,81 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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 "aqhome/msgmanager.h"
+#include "aqhome/msg/endpointmgr.h"
+
+#include <gwenhywfar/misc.h>
+#include <gwenhywfar/debug.h>
+
+
+
+void _loopOnceOverIpcEndpoints(GWEN_MSG_ENDPOINT_MGR *emgr);
+void _handleIpcEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep);
+
+
+
+
+
+GWEN_MSG_ENDPOINT_MGR *AQH_MsgManager_new(uint8_t busAddr)
+{
+  return AQH_MsgEndpointMgr_new(busAddr);
+}
+
+
+
+int AQH_MsgManager_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
+{
+  int rv;
+
+  rv=GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
+  AQH_MsgEndpointMgr_LoopOnceOverNodeEndpoints(emgr, AQH_MSGMGR_ENDPOINTGROUP_NODE);
+  _loopOnceOverIpcEndpoints(emgr);
+  GWEN_MsgEndpointMgr_RunAllEndpoints(emgr);
+  return rv;
+}
+
+
+
+void _loopOnceOverIpcEndpoints(GWEN_MSG_ENDPOINT_MGR *emgr)
+{
+  GWEN_MSG_ENDPOINT_LIST *endpointList;
+
+  DBG_DEBUG(AQH_LOGDOMAIN, "Handle endpoint messages");
+  endpointList=GWEN_MsgEndpointMgr_GetEndpointList(emgr);
+  if (endpointList) {
+    GWEN_MSG_ENDPOINT *ep;
+
+    ep=GWEN_MsgEndpoint_List_First(endpointList);
+    while(ep) {
+      DBG_DEBUG(AQH_LOGDOMAIN, "- endpoint(%s)", GWEN_MsgEndpoint_GetName(ep));
+      if (GWEN_MsgEndpoint_GetGroupId(ep)==AQH_MSGMGR_ENDPOINTGROUP_IPC)
+        _handleIpcEndpoint(emgr, ep);
+      ep=GWEN_MsgEndpoint_List_Next(ep);
+    }
+  }
+}
+
+
+
+void _handleIpcEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *ep)
+{
+  GWEN_MSG *msg;
+  
+  while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(ep)) ) {
+    /* exec IPC message */
+    GWEN_Msg_free(msg);
+  }
+}
+
+
+

aqhome/msgmanager.h

@@ -0,0 +1,32 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2023 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_MSG_MANAGER_H
+#define AQH_MSG_MANAGER_H
+
+
+#include <aqhome/api.h>
+
+#include <gwenhywfar/endpointmgr.h>
+
+
+
+#define AQH_MSGMGR_ENDPOINTGROUP_NODE 1
+#define AQH_MSGMGR_ENDPOINTGROUP_IPC  2
+
+
+
+AQHOME_API GWEN_MSG_ENDPOINT_MGR *AQH_MsgManager_new(uint8_t busAddr);
+AQHOME_API int AQH_MsgManager_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr);
+
+
+
+
+
+
+#endif

aqhome/nodes/nodedb.c

@@ -80,32 +80,14 @@ int AQH_NodeDb_AddNodeInfo(AQH_NODE_DB *ndb, AQH_NODE_INFO *ni)
 
 AQH_NODE_INFO *AQH_NodeDb_GetNodeInfoByBusAddr(AQH_NODE_DB *ndb, uint8_t busAddr)
 {
-  AQH_NODE_INFO *ni;
-
-  ni=AQH_NodeInfo_List_First(ndb->nodeList);
-  while(ni) {
-    if (busAddr==0 || busAddr==AQH_NodeInfo_GetBusAddress(ni))
-      return ni;
-    ni=AQH_NodeInfo_List_Next(ni);
-  }
-
-  return NULL;
+  return AQH_NodeInfo_List_GetByBusAddress(ndb->nodeList, busAddr);
 }
 
 
 
 AQH_NODE_INFO *AQH_NodeDb_GetNodeInfoByUid(AQH_NODE_DB *ndb, uint32_t uid)
 {
-  AQH_NODE_INFO *ni;
-
-  ni=AQH_NodeInfo_List_First(ndb->nodeList);
-  while(ni) {
-    if (uid==0 || uid==AQH_NodeInfo_GetUid(ni))
-      return ni;
-    ni=AQH_NodeInfo_List_Next(ni);
-  }
-
-  return NULL;
+  return AQH_NodeInfo_List_GetByUid(ndb->nodeList, uid);
 }