More work on node/pc interface.

- added AQH_MSG_ENDPOINT - added AQH_MsgEndpointLog - added AQH_MsgEndpointTcp - added AQH_MsgEndpointTty - added AQH_MsgEndpointMgr
2023-02-20 23:45:10 +01:00
parent e1bf53e93f
commit caa85edfc6
20 changed files with 2104 additions and 6 deletions
--- a/aqhome/msgendpoint.c
+++ b/aqhome/msgendpoint.c
@@ -0,0 +1,555 @@
+/****************************************************************************
+ * 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/msgendpoint_p.h"
+
+
+#include <gwenhywfar/misc.h>
+#include <gwenhywfar/list.h>
+#include <gwenhywfar/error.h>
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/text.h>
+
+#include <stdio.h>
+#include <fcntl.h>
+#include <termios.h>
+#include <unistd.h>
+#include <errno.h>
+#include <sys/ioctl.h>
+
+#include <stdlib.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#ifdef HAVE_ARPA_INET_H
+# include <arpa/inet.h>
+#endif
+
+
+#define AQH_MSG_ENDPOINT_BUFFERSIZE 32
+
+
+GWEN_LIST_FUNCTIONS(AQH_MSG_ENDPOINT, AQH_MsgEndpoint)
+GWEN_INHERIT_FUNCTIONS(AQH_MSG_ENDPOINT)
+
+
+static int _internalHandleReadable(AQH_MSG_ENDPOINT *ep);
+static int _internalHandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr);
+static int _setSocketNonBlocking(int fd);
+
+
+
+
+AQH_MSG_ENDPOINT *AQH_MsgEndpoint_new(int fd, int groupId)
+{
+  AQH_MSG_ENDPOINT *ep;
+
+  GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT, ep);
+  GWEN_INHERIT_INIT(AQH_MSG_ENDPOINT, ep);
+  GWEN_LIST_INIT(AQH_MSG_ENDPOINT, ep);
+  ep->fd=fd;
+  ep->groupId=groupId;
+  ep->receivedMessageList=AQH_Msg_List_new();
+  ep->sendMessageList=AQH_Msg_List_new();
+
+  return ep;
+}
+
+
+
+void AQH_MsgEndpoint_free(AQH_MSG_ENDPOINT *ep)
+{
+  if (ep) {
+    GWEN_LIST_FINI(AQH_MSG_ENDPOINT, ep);
+    GWEN_INHERIT_FINI(AQH_MSG_ENDPOINT, ep);
+    if (ep->fd>=0)
+      close(ep->fd);
+    AQH_Msg_free(ep->currentlyReceivedMsg);
+    AQH_Msg_List_free(ep->receivedMessageList);
+    AQH_Msg_List_free(ep->sendMessageList);
+    GWEN_FREE_OBJECT(ep);
+  }
+}
+
+
+
+int AQH_MsgEndpoint_GetFd(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->fd;
+}
+
+
+
+uint32_t AQH_MsgEndpoint_GetGroupId(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->groupId;
+}
+
+
+
+uint32_t AQH_MsgEndpoint_GetAcceptedMsgGroups(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->acceptedMsgGroups;
+}
+
+
+
+void AQH_MsgEndpoint_SetAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedMsgGroups=f;
+}
+
+
+
+void AQH_MsgEndpoint_AddAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedMsgGroups|=f;
+}
+
+
+
+void AQH_MsgEndpoint_DelAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedMsgGroups&=~f;
+}
+
+
+
+uint32_t AQH_MsgEndpoint_GetAcceptedEndpointGroups(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->acceptedEndpointGroups;
+}
+
+
+
+void AQH_MsgEndpoint_SetAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedEndpointGroups=f;
+}
+
+
+
+void AQH_MsgEndpoint_AddAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedEndpointGroups|=f;
+}
+
+
+
+void AQH_MsgEndpoint_DelAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->acceptedEndpointGroups&=~f;
+}
+
+
+
+uint32_t AQH_MsgEndpoint_GetFlags(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->flags;
+}
+
+
+
+void AQH_MsgEndpoint_SetFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->flags=f;
+}
+
+
+
+void AQH_MsgEndpoint_AddFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->flags|=f;
+}
+
+
+
+void AQH_MsgEndpoint_SubFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
+{
+  ep->flags&=~f;
+}
+
+
+
+
+AQH_MSG_LIST *AQH_MsgEndpoint_GetReceivedMessageList(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->receivedMessageList;
+}
+
+
+
+AQH_MSG *AQH_MsgEndpoint_TakeFirstReceivedMessage(AQH_MSG_ENDPOINT *ep)
+{
+  AQH_MSG *msg;
+
+  msg=AQH_Msg_List_First(ep->receivedMessageList);
+  if (msg)
+    AQH_Msg_List_Del(msg);
+  return msg;
+}
+
+
+
+AQH_MSG_LIST *AQH_MsgEndpoint_GetSendMessageList(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->sendMessageList;
+}
+
+
+
+void AQH_MsgEndpoint_AddReceivedMessage(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
+{
+  AQH_Msg_List_Add(m, ep->receivedMessageList);
+}
+
+
+
+void AQH_MsgEndpoint_AddSendMessage(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
+{
+  AQH_Msg_List_Add(m, ep->sendMessageList);
+}
+
+
+
+AQH_MSG *AQH_MsgEndpoint_GetCurrentlyReceivedMsg(const AQH_MSG_ENDPOINT *ep)
+{
+  return ep->currentlyReceivedMsg;
+}
+
+
+
+void AQH_MsgEndpoint_SetCurrentlyReceivedMsg(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
+{
+  AQH_Msg_free(ep->currentlyReceivedMsg);
+  ep->currentlyReceivedMsg=m;
+}
+
+
+
+int AQH_MsgEndpoint_HandleReadable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
+{
+  return (ep->handleReadableFn)?(ep->handleReadableFn(ep, emgr)):_internalHandleReadable(ep);
+}
+
+
+
+int AQH_MsgEndpoint_HandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
+{
+  return (ep->handleWritableFn)?(ep->handleWritableFn(ep, emgr)):_internalHandleWritable(ep, emgr);
+}
+
+
+
+void AQH_MsgEndpoint_Run(AQH_MSG_ENDPOINT *ep)
+{
+  if (ep->runFn)
+    ep->runFn(ep);
+}
+
+
+
+int AQH_MsgEndpoint_StartMsg(AQH_MSG_ENDPOINT *ep)
+{
+  int rv;
+
+  rv=(ep->startMsgFn)?(ep->startMsgFn(ep)):0;
+  if (rv==0)
+    ep->sendingMessage=1;
+  return rv;
+}
+
+
+
+int AQH_MsgEndpoint_EndMsg(AQH_MSG_ENDPOINT *ep)
+{
+  int rv;
+
+  rv=(ep->endMsgFn)?(ep->endMsgFn(ep)):0;
+  if (rv==0)
+    ep->sendingMessage=0;
+  return rv;
+}
+
+
+
+int AQH_MsgEndpoint_CheckMsg(AQH_MSG_ENDPOINT *ep)
+{
+  return (ep->checkMsgFn)?(ep->checkMsgFn(ep)):1;
+}
+
+
+
+AQH_MSG_ENDPOINT_HANDLEREADABLE_FN AQH_MsgEndpoint_SetHandleReadableFn(AQH_MSG_ENDPOINT *ep,
+                                                                       AQH_MSG_ENDPOINT_HANDLEREADABLE_FN f)
+{
+  AQH_MSG_ENDPOINT_HANDLEREADABLE_FN oldFn;
+
+  oldFn=ep->handleReadableFn;
+  ep->handleReadableFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN AQH_MsgEndpoint_SetHandleWritableFn(AQH_MSG_ENDPOINT *ep,
+                                                                       AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN f)
+{
+  AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN oldFn;
+
+  oldFn=ep->handleWritableFn;
+  ep->handleWritableFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_GET_READFD_FN AQH_MsgEndpoint_SetGetReadFdFn(AQH_MSG_ENDPOINT *ep,
+                                                              AQH_MSG_ENDPOINT_GET_READFD_FN f)
+{
+  AQH_MSG_ENDPOINT_GET_READFD_FN oldFn;
+
+  oldFn=ep->getReadFdFn;
+  ep->getReadFdFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_GET_WRITEFD_FN AQH_MsgEndpoint_SetGetWriteFdFn(AQH_MSG_ENDPOINT *ep,
+                                                                AQH_MSG_ENDPOINT_GET_WRITEFD_FN f)
+{
+  AQH_MSG_ENDPOINT_GET_WRITEFD_FN oldFn;
+
+  oldFn=ep->getWriteFdFn;
+  ep->getWriteFdFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_RUN_FN AQH_MsgEndpoint_SetRunFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_RUN_FN f)
+{
+  AQH_MSG_ENDPOINT_RUN_FN oldFn;
+
+  oldFn=ep->runFn;
+  ep->runFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_STARTMSG_FN AQH_MsgEndpoint_SetStartMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_STARTMSG_FN f)
+{
+  AQH_MSG_ENDPOINT_STARTMSG_FN oldFn;
+
+  oldFn=ep->startMsgFn;
+  ep->startMsgFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_ENDMSG_FN AQH_MsgEndpoint_SetEndMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_ENDMSG_FN f)
+{
+  AQH_MSG_ENDPOINT_ENDMSG_FN oldFn;
+
+  oldFn=ep->endMsgFn;
+  ep->endMsgFn=f;
+  return oldFn;
+}
+
+
+
+AQH_MSG_ENDPOINT_CHECKMSG_FN AQH_MsgEndpoint_SetCheckMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_CHECKMSG_FN f)
+{
+  AQH_MSG_ENDPOINT_CHECKMSG_FN oldFn;
+
+  oldFn=ep->checkMsgFn;
+  ep->checkMsgFn=f;
+  return oldFn;
+}
+
+
+
+int AQH_MsgEndpoint_GetReadFd(AQH_MSG_ENDPOINT *ep)
+{
+  return (ep->getReadFdFn)?(ep->getReadFdFn(ep)):(ep->fd);
+}
+
+
+
+int AQH_MsgEndpoint_GetWriteFd(AQH_MSG_ENDPOINT *ep)
+{
+  if (ep->getWriteFdFn)
+    return ep->getWriteFdFn(ep);
+  else {
+    int somethingToWrite;
+
+    somethingToWrite=(AQH_Msg_List_First(ep->sendMessageList)!=NULL)?1:0;
+    if (somethingToWrite)
+      return ep->fd;
+  }
+  return GWEN_ERROR_NO_DATA;
+}
+
+
+
+
+int AQH_MsgEndpoint_DiscardInput(AQH_MSG_ENDPOINT *ep)
+{
+  int rv;
+  uint8_t buffer[AQH_MSG_ENDPOINT_BUFFERSIZE];
+
+  do {
+    rv=read(ep->fd, buffer, sizeof(buffer));
+  } while( (rv>0 || (rv<0) && errno==EINTR));
+  if (rv<0 && errno!=EAGAIN && errno!=EWOULDBLOCK) {
+    DBG_ERROR(NULL, "Error on read(): %s (%d)", strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  else if (rv==0) {
+    DBG_ERROR(NULL, "EOF met on read()");
+    return GWEN_ERROR_IO;
+  }
+  return 0;
+}
+
+
+
+int _internalHandleReadable(AQH_MSG_ENDPOINT *ep)
+{
+  int rv;
+  uint8_t buffer[AQH_MSG_ENDPOINT_BUFFERSIZE*2];
+  int len;
+  int i;
+
+  do {
+    rv=read(ep->fd, buffer, sizeof(buffer));
+  } while( (rv<0) && errno==EINTR);
+  if (rv<0) {
+    if (errno==EAGAIN || errno==EWOULDBLOCK)
+      return GWEN_ERROR_TRY_AGAIN;
+    DBG_ERROR(NULL, "Error on read(): %s (%d)", strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  else if (rv==0) {
+    DBG_ERROR(NULL, "EOF met on read()");
+    return GWEN_ERROR_IO;
+  }
+  len=rv;
+
+  for (i=0; i<len; i++) {
+    if (ep->currentlyReceivedMsg==NULL)
+      ep->currentlyReceivedMsg=AQH_Msg_new();
+    rv=AQH_Msg_AddByte(ep->currentlyReceivedMsg, buffer[i]);
+    if (rv<0) {
+      DBG_ERROR(NULL, "here (%d)", rv);
+      return rv;
+    }
+    rv=AQH_Msg_IsMsgComplete(ep->currentlyReceivedMsg);
+    if (rv<0) {
+      /* invalid message */
+      DBG_ERROR(NULL, "Invalid message, discarding");
+      AQH_Msg_free(ep->currentlyReceivedMsg);
+      ep->currentlyReceivedMsg=NULL;
+      rv=AQH_MsgEndpoint_DiscardInput(ep);
+      if (rv<0) {
+	DBG_ERROR(NULL, "here (%d)", rv);
+	return rv;
+      }
+    }
+    else if (rv>0) {
+      if (!AQH_Msg_IsChecksumValid(ep->currentlyReceivedMsg)) {
+	DBG_ERROR(NULL, "Invalid checksum, discarding message");
+        GWEN_Text_DumpString(AQH_Msg_GetBuffer(ep->currentlyReceivedMsg), AQH_Msg_GetBytesInBuffer(ep->currentlyReceivedMsg), 6);
+	AQH_Msg_free(ep->currentlyReceivedMsg);
+	ep->currentlyReceivedMsg=NULL;
+	rv=AQH_MsgEndpoint_DiscardInput(ep);
+	if (rv<0) {
+	  DBG_ERROR(NULL, "here (%d)", rv);
+	  return rv;
+	}
+      }
+      else {
+	/* valid msg received, add to list */
+	AQH_Msg_List_Add(ep->currentlyReceivedMsg, ep->receivedMessageList);
+	ep->currentlyReceivedMsg=NULL;
+      }
+    }
+  }
+
+  return 0;
+}
+
+
+
+int _internalHandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
+{
+  AQH_MSG *msg;
+
+  msg=AQH_Msg_List_First(ep->sendMessageList);
+  if (msg) {
+    uint8_t pos;
+    int len;
+    int remaining;
+    int rv;
+
+    rv=AQH_MsgEndpoint_CheckMsg(ep);
+    if (rv<0 || rv==1) {
+      DBG_ERROR(NULL, "Line busy, not sending");
+      usleep(100);
+      return 0;
+    }
+    pos=AQH_Msg_GetCurrentPos(msg);
+    remaining=AQH_Msg_GetRemainingBytes(msg);
+    if (remaining>0) {
+      const uint8_t *buf;
+
+      rv=AQH_MsgEndpoint_StartMsg(ep);
+      if (rv<0) {
+        DBG_ERROR(NULL, "here (%d)", rv);
+        return rv;
+      }
+
+      buf=AQH_Msg_GetBuffer(msg)+pos;
+      do {
+	rv=write(ep->fd, buf, remaining);
+      } while(rv<0 && errno==EINTR);
+      if (rv<0) {
+        if (errno==EAGAIN || errno==EWOULDBLOCK)
+          return GWEN_ERROR_TRY_AGAIN;
+        DBG_ERROR(NULL, "Error on write(): %s (%d)", strerror(errno), errno);
+        return GWEN_ERROR_IO;
+      }
+      AQH_Msg_IncCurrentPos(msg, rv);
+      if (rv==remaining) {
+        rv=AQH_MsgEndpoint_EndMsg(ep);
+        // TODO: callback msg sent
+	AQH_Msg_List_Del(msg);
+	AQH_Msg_free(msg);
+	if (rv<0) {
+	  DBG_ERROR(NULL, "here (%d)", rv);
+	  return rv;
+	}
+      }
+    }
+  }
+  return 0;
+}
+
+
+