aqhome: fixed endpoint code. Now works again.

aqhome/0BUILD

@@ -46,50 +46,12 @@
     <headers dist="true" install="$(pkgincludedir)" >
       api.h
       aqhome.h
-      serial.h
-      serial_p.h
-      msg.h
-      msg_p.h
-      msg_value.h
-      msg_sendstats.h
-      msg_ping.h
-      msg_pong.h
-      msg_needaddr.h
-      msg_haveaddr.h
-      msg_claimaddr.h
-      msg_denyaddr.h
-      msg_setaccmsggrps.h
-      msgendpoint.h
-      msgendpoint_p.h
-      msgendpointtcp.h
-      msgendpointtty.h
-      msgendpointtty_p.h
-      msgendpointlog.h
-      msgendpointlog_p.h
-      msgendpointmanager.h
-      msgendpointmanager_p.h
     </headers>
   
   
     <sources>
       $(local/typefiles)
       aqhome.c
-      serial.c
-      msg.c
-      msg_value.c
-      msg_sendstats.c
-      msg_ping.c
-      msg_pong.c
-      msg_needaddr.c
-      msg_haveaddr.c
-      msg_claimaddr.c
-      msg_denyaddr.c
-      msg_setaccmsggrps.c
-      msgendpoint.c
-      msgendpointtcp.c
-      msgendpointtty.c
-      msgendpointlog.c
-      msgendpointmanager.c
     </sources>
 
     <subdirs>

aqhome/libtest.c

@@ -3,258 +3,86 @@
 # include <config.h>
 #endif
 
-#include "aqhome/serial.h"
+#include "aqhome/msg/msg_ping.h"
-#include "aqhome/msg_value.h"
+#include "aqhome/msg/endpointmgr.h"
-#include "aqhome/msg_sendstats.h"
+#include "aqhome/msg/endpoint_node.h"
-#include "aqhome/msg_ping.h"
+#include "aqhome/msg/endpoint_log.h"
-#include "aqhome/msg_pong.h"
+#include "aqhome/msg/endpoint_tty.h"
-#include "aqhome/msg_needaddr.h"
-#include "aqhome/msg_claimaddr.h"
-#include "aqhome/msg_haveaddr.h"
-#include "aqhome/msg_denyaddr.h"
 
 #include "aqhome/aqhome.h"
-#include "aqhome/msgendpointmanager.h"
-#include "aqhome/msgendpointtty.h"
-#include "aqhome/msgendpointlog.h"
 
 #include <gwenhywfar/debug.h>
 #include <gwenhywfar/text.h>
 #include <gwenhywfar/gwentime.h>
 #include <gwenhywfar/buffer.h>
+#include <gwenhywfar/endpoint_ipc_tcp.h>
 
 #include <unistd.h>
 #include <time.h>
 
 
 
-int testRecv()
+GWEN_MSG *createPingMsg(uint8_t destAddr, uint8_t srcAddr)
 {
-  AQH_SERIAL *sr;
+  GWEN_MSG *msg;
-  int rv;
-  int i;
-  GWEN_BUFFER *dbuf;
-
-  fprintf(stdout, "Opening device...\n");
-  sr=AQH_Serial_new("/dev/ttyUSB0", 240);
-  rv=AQH_Serial_Open(sr, AQH_SerialReadWriteMode_ReadOnly);
-  if (rv<0) {
-    DBG_ERROR(NULL, "ERROR opening device (%d)", rv);
-    AQH_Serial_free(sr);
-    return 2;
-  }
-  fprintf(stdout, "Device open, waiting for packets\n");
-
-  dbuf=GWEN_Buffer_new(0, 256, 0, 1);
-  for (i=0; ; i++) {
-    uint8_t buffer[1024];
-    GWEN_TIME *ti;
-
-    fprintf(stdout, "Waiting for packet...\n");
-    rv=AQH_Serial_Recv(sr, buffer, sizeof(buffer)-1);
-    if (rv<0)
-      break;
-    ti=GWEN_CurrentTime();
-    GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss", dbuf);
-    fprintf(stdout, "%s: Received:\n", GWEN_Buffer_GetStart(dbuf));
-    GWEN_Text_DumpString(buffer, rv, 2);
-    GWEN_Time_free(ti);
-    GWEN_Buffer_Reset(dbuf);
-    fprintf(stdout, "\n");
-  }
-  GWEN_Buffer_free(dbuf);
-
-  AQH_Serial_Close(sr);
-  AQH_Serial_free(sr);
-  return 0;
-}
-
-
-
-int testSend()
-{
-  AQH_SERIAL *sr;
-  int rv;
-  int i;
-  GWEN_BUFFER *dbuf;
-  uint8_t sendBuf[5]={0x01, 0x02, 0xdb, 0x00, 0xd8 };
-  uint8_t sendBuf2[2]={0x02, 0xdb };
-
-  fprintf(stdout, "Opening device...\n");
-  sr=AQH_Serial_new("/dev/ttyUSB0", 240);
-  rv=AQH_Serial_Open(sr, AQH_SerialReadWriteMode_ReadWrite);
-  if (rv<0) {
-    DBG_ERROR(NULL, "ERROR opening device (%d)", rv);
-    AQH_Serial_free(sr);
-    return 2;
-  }
-  fprintf(stdout, "Device open, waiting for packets\n");
-
-  dbuf=GWEN_Buffer_new(0, 256, 0, 1);
-  for (i=0; ; i++) {
-    GWEN_TIME *ti;
-
-    //rv=AQH_Serial_Send(sr, sendBuf, sizeof(sendBuf));
-    rv=AQH_Serial_SendPacket(sr, 1, sendBuf2, sizeof(sendBuf2));
-    if (rv<0)
-      break;
-    ti=GWEN_CurrentTime();
-    GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss", dbuf);
-    fprintf(stdout, "%s: Packet %d sent.\n", GWEN_Buffer_GetStart(dbuf), i);
-    GWEN_Time_free(ti);
-    GWEN_Buffer_Reset(dbuf);
-    sleep(3);
-  }
-  GWEN_Buffer_free(dbuf);
-
-  AQH_Serial_Close(sr);
-  AQH_Serial_free(sr);
-  return 0;
-}
-
-
-
-void _packetReceived(AQH_SERIAL *sr, AQH_MSG *msg)
-{
-  const uint8_t *ptr;
-  uint8_t len;
-  uint8_t msgType;
-  int msgIsValid;
-  GWEN_BUFFER *dbuf;
-  GWEN_TIME *ti;
-
-  dbuf=GWEN_Buffer_new(0, 256, 0, 1);
-  ti=GWEN_CurrentTime();
-  GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss  ", dbuf);
-  GWEN_Time_free(ti);
-  ti=NULL;
-
-  msgIsValid=(AQH_Msg_IsChecksumValid(msg) && AQH_Msg_IsMsgComplete(msg));
-  ptr=AQH_Msg_GetBuffer(msg);
-  len=AQH_Msg_GetBytesInBuffer(msg);
-  msgType=AQH_Msg_GetMsgType(msg);
-
-  if (msgType==AQH_MSG_TYPE_PING) {
-    AQH_MsgPing_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_PONG) {
-    AQH_MsgPong_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_COMSENDSTATS) { /* CPRO_CMD_COMSENDSTATS */
-    AQH_MsgSendStats_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_TWIBUSMEMBER) { /* CPRO_CMD_TWIBUSMEMBER */
-    int i2cAddr;
-    int availability;
-
-    i2cAddr=ptr[4];
-    availability=ptr[5];
-    fprintf(stdout, "-> TWI DEVICE %02x: %s\n", i2cAddr, (availability==0)?"not available":"FOUND");
-  }
-  else if (msgType==40) { /* CPRO_CMD_DEBUG */
-    uint8_t param1;
-    uint8_t param2;
-
-    param1=ptr[4];
-    param2=ptr[5];
-    fprintf(stdout, "-> Debug param1=%02x, param2=%02x\n", param1, param2);
-  }
-  else if (msgType==AQH_MSG_TYPE_VALUE) { /* CPRO_CMD_VALUE */
-    AQH_MsgValue_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-    //GWEN_Text_DumpString(ptr, len, 6);
-  }
-  else if (msgType==AQH_MSG_TYPE_NEED_ADDRESS) {
-    AQH_MsgNeedAddr_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_CLAIM_ADDRESS) {
-    AQH_MsgClaimAddr_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_HAVE_ADDRESS) {
-    AQH_MsgHaveAddr_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else if (msgType==AQH_MSG_TYPE_DENY_ADDRESS) {
-    AQH_MsgDenyAddr_DumpToBuffer(msg, dbuf, "received");
-    fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-  }
-  else {
-    fprintf(stdout, "    %s: Received (%d):\n", GWEN_Buffer_GetStart(dbuf), msgType);
-    GWEN_Text_DumpString(ptr, len, 6);
-  }
-  GWEN_Buffer_free(dbuf);
-
-  AQH_Msg_free(msg);
-}
-
-
-
-AQH_MSG *createPingMsg(AQH_SERIAL *sr, uint8_t destAddr)
-{
-  AQH_MSG *msg;
   int rv;
 
-  msg=AQH_Msg_new();
+  msg=GWEN_Msg_new(AQH_MAXMSGSIZE);
-  rv=AQH_Msg_AddByte(msg, destAddr);
+  rv=GWEN_Msg_AddByte(msg, destAddr);
   if (rv<0) {
     fprintf(stderr, "ERROR1: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, 6); /* msglen */
+  rv=GWEN_Msg_AddByte(msg, 6); /* msglen */
   if (rv<0) {
     fprintf(stderr, "ERROR2: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, AQH_MSG_TYPE_PING); /* ping */
+  rv=GWEN_Msg_AddByte(msg, AQH_MSG_TYPE_PING); /* ping */
   if (rv<0) {
     fprintf(stderr, "ERROR3: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, AQH_Serial_GetAddress(sr)); /* src addr */
+  rv=GWEN_Msg_AddByte(msg, srcAddr); /* src addr */
   if (rv<0) {
     fprintf(stderr, "ERROR4: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
 
-  rv=AQH_Msg_AddByte(msg, 0); /* timestamp */
+  rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */
   if (rv<0) {
     fprintf(stderr, "ERROR5: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, 0); /* timestamp */
+  rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */
   if (rv<0) {
     fprintf(stderr, "ERROR6: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, 0); /* timestamp */
+  rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */
   if (rv<0) {
     fprintf(stderr, "ERROR7: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
-  rv=AQH_Msg_AddByte(msg, 0); /* timestamp */
+  rv=GWEN_Msg_AddByte(msg, 0); /* timestamp */
   if (rv<0) {
     fprintf(stderr, "ERROR8: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
 
 
-  rv=AQH_Msg_AddChecksum(msg);
+  rv=AQH_NodeMsg_AddChecksum(msg);
   if (rv<0) {
     fprintf(stderr, "ERROR9: %d\n", rv);
-    AQH_Msg_free(msg);
+    GWEN_Msg_free(msg);
     return NULL;
   }
 
@@ -263,151 +91,36 @@ AQH_MSG *createPingMsg(AQH_SERIAL *sr, uint8_t destAddr)
 
 
 
-int testLoop()
-{
-  AQH_SERIAL *sr;
-  int rv;
-  int i;
-  GWEN_BUFFER *dbuf;
-  time_t tLast;
-
-  fprintf(stdout, "Opening device...\n");
-  sr=AQH_Serial_new("/dev/ttyUSB0", 240);
-  rv=AQH_Serial_Open(sr, AQH_SerialReadWriteMode_ReadWrite);
-  if (rv<0) {
-    DBG_ERROR(NULL, "ERROR opening device (%d)", rv);
-    AQH_Serial_free(sr);
-    return 2;
-  }
-  fprintf(stdout, "Device open, waiting for packets\n");
-  AQH_Serial_SetPacketReceivedFn(sr, _packetReceived);
-
-  tLast=time(NULL);
-  for (;;) {
-    time_t t;
-
-    rv=AQH_Serial_Loop(sr);
-    if (rv<0) {
-      AQH_Serial_Close(sr);
-      AQH_Serial_free(sr);
-      break;
-    }
-    t=time(NULL);
-    if (difftime(t, tLast)>61) {
-      AQH_MSG *msg;
-
-      msg=createPingMsg(sr, 1);
-      if (msg) {
-        rv=AQH_Serial_AddMessageToSend(sr, msg);
-        if (rv<0) {
-          fprintf(stderr, "ERROR: %d\n", rv);
-          AQH_Msg_free(msg);
-          return rv;
-        }
-
-        if (rv==0) {
-          GWEN_BUFFER *dbuf;
-          GWEN_TIME *ti;
-        
-          dbuf=GWEN_Buffer_new(0, 256, 0, 1);
-          ti=GWEN_CurrentTime();
-          GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss  ", dbuf);
-          GWEN_Time_free(ti);
-          ti=NULL;
-          AQH_MsgPing_DumpToBuffer(msg, dbuf, "enqueued");
-          fprintf(stdout, "%s", GWEN_Buffer_GetStart(dbuf));
-          GWEN_Buffer_free(dbuf);
-        }
-        else {
-          DBG_ERROR(NULL, "here (%d)", rv);
-          AQH_Serial_Close(sr);
-          AQH_Serial_free(sr);
-          return rv;
-        }
-      }
-      else {
-        DBG_ERROR(NULL, "here (%d)", rv);
-        AQH_Serial_Close(sr);
-        AQH_Serial_free(sr);
-        return GWEN_ERROR_INTERNAL;
-      }
-      tLast=t;
-    }
-  }
-
-  return 0;
-}
-
-
-
-int testReadLoop()
-{
-  AQH_SERIAL *sr;
-  int rv;
-  int i;
-  GWEN_BUFFER *dbuf;
-  time_t tLast;
-
-  fprintf(stdout, "Opening device...\n");
-  sr=AQH_Serial_new("/dev/ttyUSB0", 240);
-  rv=AQH_Serial_Open(sr, AQH_SerialReadWriteMode_ReadOnly);
-  if (rv<0) {
-    DBG_ERROR(NULL, "ERROR opening device (%d)", rv);
-    AQH_Serial_free(sr);
-    return 2;
-  }
-  fprintf(stdout, "Device open, waiting for packets\n");
-  AQH_Serial_SetPacketReceivedFn(sr, _packetReceived);
-
-  tLast=time(NULL);
-  for (;;) {
-    time_t t;
-
-    rv=AQH_Serial_ReadOnlyLoop(sr);
-    if (rv<0) {
-      AQH_Serial_Close(sr);
-      AQH_Serial_free(sr);
-      break;
-    }
-    t=time(NULL);
-  }
-
-  return 0;
-}
-
-
-
 int testEndpoints()
 {
   int rv;
-  AQH_MSG_ENDPOINT_MGR *emgr;
+  GWEN_MSG_ENDPOINT_MGR *emgr;
-  AQH_MSG_ENDPOINT *epTty;
+  GWEN_MSG_ENDPOINT *epTty;
-  AQH_MSG_ENDPOINT *epLog;
+  GWEN_MSG_ENDPOINT *epLog;
 
   rv=AQH_Init();
   if (rv<0) {
   }
 
-  emgr=AQH_MsgEndpointManager_new(0xc0);
+  emgr=AQH_MsgEndpointMgr_new(0xc0);
-  epTty=AQH_MsgEndpointTty_new("/dev/ttyUSB0");
+  epTty=AQH_TtyNodeEndpoint_new("/dev/ttyUSB0", AQH_MSG_ENDPOINTGROUP_NODE);
   if (epTty==NULL) {
     DBG_ERROR(NULL, "Error creating endpoint TTY");
     return 2;
   }
-  AQH_MsgEndpointManager_AddEndpoint(emgr, epTty);
+  GWEN_MsgEndpointMgr_AddEndpoint(emgr, epTty);
 
-  epLog=AQH_MsgEndpointLog_new("endpoints.log");
+  epLog=AQH_LogEndpoint_new("endpoints.log", AQH_MSG_ENDPOINTGROUP_NODE);
   if (epLog==NULL) {
     DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint LOG");
     return 2;
   }
-  AQH_MsgEndpointManager_AddEndpoint(emgr, epLog);
+  GWEN_MsgEndpointMgr_AddEndpoint(emgr, epLog);
 
   for (;;) {
     DBG_DEBUG(AQH_LOGDOMAIN, "Next loop");
-    AQH_MsgEndpointManager_LoopOnce(emgr);
+    AQH_MsgEndpointMgr_LoopOnce(emgr);
   }
-
   return 0;
 }
 
@@ -417,25 +130,7 @@ int testEndpoints()
 
 int main(int argc, char **argv)
 {
-  const char *cmd;
+  return testEndpoints();
-
-  if (argc<2)
-    return testLoop();
-
-  cmd=argv[1];
-  if (strcasecmp(cmd, "monitor")==0 || strcasecmp(cmd, "mon")==0)
-    return testReadLoop();
-  else if (strcasecmp(cmd, "pingtest")==0) {
-  }
-  else if (strcasecmp(cmd, "rwtest")==0) {
-    return testLoop();
-  }
-  else if (strcasecmp(cmd, "endpoint")==0) {
-    return testEndpoints();
-  }
-  //return testRecv();
-  //return testSend();
-  return testLoop();
 }
 
 

aqhome/msg/0BUILD

@@ -48,6 +48,7 @@
       endpointmgr.h
       endpoint_node.h
       endpoint_log.h
+      endpoint_tty.h
       msg_node.h
       msg_ping.h
       msg_pong.h
@@ -64,6 +65,7 @@
       endpointmgr_p.h
       endpoint_node_p.h
       endpoint_log_p.h
+      endpoint_tty_p.h
       msg_node_p.h
     </headers>
   
@@ -74,6 +76,7 @@
       endpointmgr.c
       endpoint_node.c
       endpoint_log.c
+      endpoint_tty.c
       msg_node.c
       msg_ping.c
       msg_pong.c

aqhome/msg/endpoint_log.c

@@ -12,6 +12,7 @@
 
 
 #include "aqhome/msg/endpoint_log_p.h"
+#include "aqhome/msg/endpoint_node.h"
 
 #include "aqhome/msg/msg_value.h"
 #include "aqhome/msg/msg_sendstats.h"
@@ -50,7 +51,7 @@ GWEN_MSG_ENDPOINT *AQH_LogEndpoint_new(const char *filename, int groupId)
   GWEN_MSG_ENDPOINT *ep;
   AQH_MSG_ENDPOINT_LOG *xep;
 
-  ep=GWEN_MsgEndpoint_new(AQH_MSG_ENDPOINT_LOG_NAME, groupId);
+  ep=AQH_NodeEndpoint_new(AQH_MSG_ENDPOINT_LOG_NAME, groupId);
   GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT_LOG, xep);
   xep->filename=strdup(filename);
   GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_LOG, ep, xep, _freeData);
@@ -108,6 +109,7 @@ void _logMessage(GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
   GWEN_TIME *ti;
 
   xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_LOG, ep);
+  assert(xep);
   dbuf=GWEN_Buffer_new(0, 256, 0, 1);
   ti=GWEN_CurrentTime();
   GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss  ", dbuf);

aqhome/msg/endpoint_node.c

@@ -18,19 +18,10 @@
 #include <gwenhywfar/debug.h>
 #include <gwenhywfar/text.h>
 
-#include <fcntl.h>
-#include <termios.h>
-#include <errno.h>
-#include <sys/ioctl.h>
-#include <unistd.h>
 
 
 #define AQH_MSG_ENDPOINT_NODE_NAME            "node"
 
-#define AQH_MSG_ENDPOINT_NODE_BAUDRATE        B19200
-#define AQH_MSG_ENDPOINT_NODE_BYTE_MICROSECS  520
-
-#define AQH_MSG_ENDPOINT_NODE_BUFFERSIZE      128
 
 
 
@@ -39,52 +30,22 @@ GWEN_INHERIT(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE)
 
 
 
-static int _getReadFd(GWEN_MSG_ENDPOINT *ep);
-static int _getWriteFd(GWEN_MSG_ENDPOINT *ep);
-static int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
-static int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
-
 static void GWENHYWFAR_CB _freeData(void *bp, void *p);
 
-static int _startMsg(GWEN_MSG_ENDPOINT *ep);
-static int _endMsg(GWEN_MSG_ENDPOINT *ep);
-static int _isLineBusy(GWEN_MSG_ENDPOINT *ep);
-
-static int _openDevice(GWEN_MSG_ENDPOINT *ep);
-static int _attnLow(GWEN_MSG_ENDPOINT *ep);
-static int _attnHigh(GWEN_MSG_ENDPOINT *ep);
-static int _isAttnLow(GWEN_MSG_ENDPOINT *ep);
 
 
 
 
-
+GWEN_MSG_ENDPOINT *AQH_NodeEndpoint_new(const char *name, int groupId)
-GWEN_MSG_ENDPOINT *AQH_MsgEndpointNode_new(const char *devicePath, int groupId)
 {
   GWEN_MSG_ENDPOINT *ep;
   AQH_MSG_ENDPOINT_NODE *xep;
   int fd;
 
-  ep=GWEN_MsgEndpoint_new(AQH_MSG_ENDPOINT_NODE_NAME, groupId);
+  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);
 
-  GWEN_MsgEndpoint_SetHandleReadableFn(ep, _handleReadable);
-  GWEN_MsgEndpoint_SetHandleWritableFn(ep, _handleWritable);
-  GWEN_MsgEndpoint_SetGetReadFdFn(ep, _getReadFd);
-  GWEN_MsgEndpoint_SetGetWriteFdFn(ep, _getWriteFd);
-
-  xep->deviceName=strdup(devicePath);
-
-  fd=_openDevice(ep);
-  if (fd<0) {
-    DBG_INFO(NULL, "here (%d)", fd);
-    GWEN_MsgEndpoint_free(ep);
-    return NULL;
-  }
-  GWEN_MsgEndpoint_SetFd(ep, fd);
-  _attnHigh(ep);
-
   return ep;
 }
 
@@ -105,6 +66,7 @@ uint32_t AQH_NodeEndpoint_GetAcceptedMsgGroups(const GWEN_MSG_ENDPOINT *ep)
   const AQH_MSG_ENDPOINT_NODE *xep;
 
   xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
+  assert(xep);
   if (xep)
     return xep->acceptedMsgGroups;
   return 0;
@@ -117,6 +79,7 @@ void AQH_NodeEndpoint_SetAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
   AQH_MSG_ENDPOINT_NODE *xep;
 
   xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
+  assert(xep);
   if (xep)
     xep->acceptedMsgGroups=f;
 }
@@ -128,6 +91,7 @@ void AQH_NodeEndpoint_AddAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
   AQH_MSG_ENDPOINT_NODE *xep;
 
   xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
+  assert(xep);
   if (xep)
     xep->acceptedMsgGroups|=f;
 }
@@ -139,337 +103,10 @@ void AQH_NodeEndpoint_DelAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
   AQH_MSG_ENDPOINT_NODE *xep;
 
   xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
+  assert(xep);
   if (xep)
     xep->acceptedMsgGroups&=~f;
 }
 
 
 
-int _getReadFd(GWEN_MSG_ENDPOINT *ep)
-{
-  return GWEN_MsgEndpoint_GetFd(ep);
-}
-
-
-
-int _getWriteFd(GWEN_MSG_ENDPOINT *ep)
-{
-  return GWEN_MsgEndpoint_HaveMessageToSend(ep)?GWEN_MsgEndpoint_GetFd(ep):GWEN_ERROR_NO_DATA;
-}
-
-
-
-int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
-{
-  int rv;
-  uint8_t buffer[AQH_MSG_ENDPOINT_NODE_BUFFERSIZE];
-  int len;
-  int i;
-
-  DBG_DEBUG(GWEN_LOGDOMAIN, "Reading from endpoint %s", GWEN_MsgEndpoint_GetName(ep));
-  do {
-    rv=read(GWEN_MsgEndpoint_GetFd(ep), buffer, sizeof(buffer));
-  } while( (rv<0) && errno==EINTR);
-  if (rv<0) {
-    if (errno==EAGAIN || errno==EWOULDBLOCK)
-      return GWEN_ERROR_TRY_AGAIN;
-    DBG_ERROR(GWEN_LOGDOMAIN, "Error on read(): %s (%d)", strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  else if (rv==0) {
-    DBG_ERROR(GWEN_LOGDOMAIN, "EOF met on read()");
-    return GWEN_ERROR_IO;
-  }
-  len=rv;
-
-  for (i=0; i<len; i++) {
-    GWEN_MSG *msg;
-
-    msg=GWEN_MsgEndpoint_GetCurrentlyReceivedMsg(ep);
-    if (msg==NULL) {
-      msg=GWEN_Msg_new(AQH_MSG_ENDPOINT_NODE_BUFFERSIZE);
-      GWEN_Msg_SetGroupId(msg, GWEN_MsgEndpoint_GetGroupId(ep));
-      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, msg);
-    }
-    rv=GWEN_Msg_AddByte(msg, buffer[i]);
-    if (rv<0) {
-      DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
-      return rv;
-    }
-    rv=AQH_NodeMsg_IsMsgComplete(msg);
-    if (rv<0) {
-      /* invalid message */
-      DBG_ERROR(GWEN_LOGDOMAIN, "Invalid message, discarding");
-      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
-      rv=GWEN_MsgEndpoint_DiscardInput(ep);
-      if (rv<0) {
-        DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
-        return rv;
-      }
-    }
-    else if (rv>0) {
-      /* complete msg received, add to list */
-      if (!AQH_NodeMsg_IsChecksumValid(msg)) {
-        DBG_ERROR(AQH_LOGDOMAIN, "Invalid checksum, discarding message");
-        GWEN_Text_DumpString(GWEN_Msg_GetBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 6);
-        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
-        rv=GWEN_MsgEndpoint_DiscardInput(ep);
-        if (rv<0) {
-          DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
-          return rv;
-        }
-      }
-      else {
-        GWEN_Msg_Attach(msg);
-        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
-        GWEN_MsgEndpoint_AddReceivedMessage(ep, msg);
-      }
-    }
-  } /* for */
-
-  return 0;
-}
-
-
-
-int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
-{
-  GWEN_MSG *msg;
-
-  DBG_DEBUG(GWEN_LOGDOMAIN, "Writing to endpoint %s", GWEN_MsgEndpoint_GetName(ep));
-  msg=GWEN_MsgEndpoint_GetFirstSendMessage(ep);
-  if (msg) {
-    uint8_t pos;
-    int remaining;
-    int rv;
-
-    pos=GWEN_Msg_GetCurrentPos(msg);
-    remaining=GWEN_Msg_GetRemainingBytes(msg);
-    if (remaining>0) {
-      const uint8_t *buf;
-      int fd;
-
-      if (pos==0) {
-        /* start new message */
-        rv=_isLineBusy(ep);
-        if (rv<0 || rv==1) {
-          DBG_ERROR(AQH_LOGDOMAIN, "Line busy, not sending");
-          usleep(100);
-          return 0;
-        }
-        rv=_startMsg(ep);
-        if (rv<0) {
-          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
-          return rv;
-        }
-      }
-
-      fd=GWEN_MsgEndpoint_GetFd(ep);
-      /* start new message */
-      buf=GWEN_Msg_GetBuffer(msg)+pos;
-      do {
-        rv=write(fd, buf, remaining);
-      } while(rv<0 && errno==EINTR);
-      if (rv<0) {
-        if (errno==EAGAIN || errno==EWOULDBLOCK)
-          return GWEN_ERROR_TRY_AGAIN;
-        DBG_ERROR(GWEN_LOGDOMAIN, "Error on write(): %s (%d)", strerror(errno), errno);
-        return GWEN_ERROR_IO;
-      }
-      GWEN_Msg_IncCurrentPos(msg, rv);
-      if (rv==remaining) {
-        /* end current message */
-        rv=_endMsg(ep);
-        GWEN_Msg_List_Del(msg);
-	GWEN_Msg_free(msg);
-        if (rv<0) {
-          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
-          return rv;
-        }
-      }
-    }
-  }
-  return 0;
-}
-
-
-
-int _openDevice(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-  int fd;
-  int status;
-  int i;
-  struct termios options;
-  int rv;
-  int m;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  fd=open(xep->deviceName, O_NOCTTY | O_NDELAY | O_RDWR);
-  if (fd<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on open(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  rv=tcgetattr(fd, &(xep->previousOptions));
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcgetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  memset(&options, 0, sizeof(options)); /* preset */
-
-  options.c_cflag=CLOCAL | CREAD | CS8;
-  options.c_iflag=IGNPAR | IGNBRK;
-  options.c_oflag=0;
-  options.c_lflag=0;
-  cfmakeraw(&options);
-  options.c_cc[VTIME]=0;  /* read timeout in deciseconds */
-  options.c_cc[VMIN]=0;   /* no minimum number of receive bytes */
-
-  rv=cfsetispeed(&options, AQH_MSG_ENDPOINT_NODE_BAUDRATE);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetispeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  rv=cfsetospeed(&options, AQH_MSG_ENDPOINT_NODE_BAUDRATE);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetospeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-
-  rv=tcflush(fd, TCIOFLUSH);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcflush(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-
-  rv=tcsetattr(fd, TCSANOW, &options);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcsetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-
-  return fd;
-}
-
-
-
-int _startMsg(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  if (xep->intendedAttnState==1) {
-    int rv;
-
-    rv=_attnLow(ep);
-    if (rv<0) {
-      DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
-      return rv;
-    }
-    usleep(AQH_MSG_ENDPOINT_NODE_BYTE_MICROSECS/5);
-  }
-
-  return 0;
-}
-
-
-
-int _endMsg(GWEN_MSG_ENDPOINT *ep)
-{
-  /* TODO: flush before releasing ATTN */
-  _attnHigh(ep);
-}
-
-
-
-int _isLineBusy(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  if (xep->intendedAttnState==0) {
-    /* we pulled the line low ourselves, and because of the circuitry nobody can pull it high */
-    return 0;
-  }
-  return _isAttnLow(ep);
-}
-
-
-
-int _attnLow(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-  int status;
-  int rv;
-  int fd;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  fd=GWEN_MsgEndpoint_GetFd(ep);
-  rv=ioctl(fd, TIOCMGET, &status); /* GET the State of MODEM bits in Status */
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  status |= TIOCM_DTR | TIOCM_RTS;        /* clear the DTR pin (cave: signals inverted!) */
-  rv=ioctl(fd, TIOCMSET, &status);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  xep->intendedAttnState=0;
-  return 0;
-}
-
-
-
-int _attnHigh(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-  int status;
-  int rv;
-  int fd;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  fd=GWEN_MsgEndpoint_GetFd(ep);
-  rv=ioctl(fd, TIOCMGET, &status);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  status |= TIOCM_DTR;        /* Set the DTR pin */
-  status &= ~ (TIOCM_DTR | TIOCM_RTS);      /* clear the DTR pin (cave: signals inverted!) */
-  rv=ioctl(fd, TIOCMSET, &status);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-
-  xep->intendedAttnState=1;
-  return 0;
-}
-
-
-
-int _isAttnLow(GWEN_MSG_ENDPOINT *ep)
-{
-  AQH_MSG_ENDPOINT_NODE *xep;
-  int status;
-  int rv;
-  int fd;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
-  fd=GWEN_MsgEndpoint_GetFd(ep);
-  rv=ioctl(fd, TIOCMGET, &status);
-  if (rv<0) {
-    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  return (status & TIOCM_CTS)?1:0;
-}
-
-
-
-
-
-
-

aqhome/msg/endpoint_node.h

@@ -15,12 +15,11 @@
 #include <gwenhywfar/endpoint.h>
 
 
-#define AQH_MSGEP_NODE_FLAGS_NOMESSAGES 0x0001
+#define AQH_MSGEP_NODE_FLAGS_NOMESSAGES 0x0002
-#define AQH_MSGEP_NODE_FLAGS_NOIO       0x0002
 
 
 
-AQHOME_API GWEN_MSG_ENDPOINT *AQH_NodeEndpointNode_new(const char *devicePath, int groupId);
+AQHOME_API GWEN_MSG_ENDPOINT *AQH_NodeEndpoint_new(const char *name, int groupId);
 
 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/endpoint_node_p.h

@@ -22,10 +22,6 @@
 typedef struct AQH_MSG_ENDPOINT_NODE AQH_MSG_ENDPOINT_NODE;
 struct AQH_MSG_ENDPOINT_NODE {
   uint32_t acceptedMsgGroups;
-
-  char *deviceName;
-  struct termios previousOptions;
-  int intendedAttnState;
 };
 
 

aqhome/msg/endpoint_tty.c

@@ -0,0 +1,437 @@
+/****************************************************************************
+ * 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/msg/endpoint_tty_p.h"
+#include "aqhome/msg/endpoint_node.h"
+#include "aqhome/msg/msg_node.h"
+
+#include <gwenhywfar/inherit.h>
+#include <gwenhywfar/debug.h>
+#include <gwenhywfar/text.h>
+
+#include <fcntl.h>
+#include <termios.h>
+#include <errno.h>
+#include <sys/ioctl.h>
+#include <unistd.h>
+
+
+#define AQH_MSG_ENDPOINT_TTY_NAME            "tty"
+
+#define AQH_MSG_ENDPOINT_TTY_BAUDRATE        B19200
+#define AQH_MSG_ENDPOINT_TTY_BYTE_MICROSECS  520
+
+#define AQH_MSG_ENDPOINT_TTY_BUFFERSIZE      128
+
+
+
+GWEN_INHERIT(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY)
+
+
+
+
+static int _getReadFd(GWEN_MSG_ENDPOINT *ep);
+static int _getWriteFd(GWEN_MSG_ENDPOINT *ep);
+static int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
+static int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
+
+static void GWENHYWFAR_CB _freeData(void *bp, void *p);
+
+static int _startMsg(GWEN_MSG_ENDPOINT *ep);
+static int _endMsg(GWEN_MSG_ENDPOINT *ep);
+static int _isLineBusy(GWEN_MSG_ENDPOINT *ep);
+
+static int _openDevice(GWEN_MSG_ENDPOINT *ep);
+static int _attnLow(GWEN_MSG_ENDPOINT *ep);
+static int _attnHigh(GWEN_MSG_ENDPOINT *ep);
+static int _isAttnLow(GWEN_MSG_ENDPOINT *ep);
+
+
+
+
+
+GWEN_MSG_ENDPOINT *AQH_TtyNodeEndpoint_new(const char *devicePath, int groupId)
+{
+  GWEN_MSG_ENDPOINT *ep;
+  AQH_MSG_ENDPOINT_TTY *xep;
+  int fd;
+
+  ep=AQH_NodeEndpoint_new(AQH_MSG_ENDPOINT_TTY_NAME, groupId);
+  GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT_TTY, xep);
+  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep, xep, _freeData);
+
+  GWEN_MsgEndpoint_SetHandleReadableFn(ep, _handleReadable);
+  GWEN_MsgEndpoint_SetHandleWritableFn(ep, _handleWritable);
+  GWEN_MsgEndpoint_SetGetReadFdFn(ep, _getReadFd);
+  GWEN_MsgEndpoint_SetGetWriteFdFn(ep, _getWriteFd);
+
+  xep->deviceName=strdup(devicePath);
+
+  fd=_openDevice(ep);
+  if (fd<0) {
+    DBG_INFO(NULL, "here (%d)", fd);
+    GWEN_MsgEndpoint_free(ep);
+    return NULL;
+  }
+  GWEN_MsgEndpoint_SetFd(ep, fd);
+  _attnHigh(ep);
+
+  return ep;
+}
+
+
+
+void _freeData(void *bp, void *p)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+
+  xep=(AQH_MSG_ENDPOINT_TTY*) p;
+  GWEN_FREE_OBJECT(xep);
+}
+
+
+
+int _getReadFd(GWEN_MSG_ENDPOINT *ep)
+{
+  return GWEN_MsgEndpoint_GetFd(ep);
+}
+
+
+
+int _getWriteFd(GWEN_MSG_ENDPOINT *ep)
+{
+  return GWEN_MsgEndpoint_HaveMessageToSend(ep)?GWEN_MsgEndpoint_GetFd(ep):GWEN_ERROR_NO_DATA;
+}
+
+
+
+int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
+{
+  int rv;
+  uint8_t buffer[AQH_MSG_ENDPOINT_TTY_BUFFERSIZE];
+  int len;
+  int i;
+
+  DBG_DEBUG(GWEN_LOGDOMAIN, "Reading from endpoint %s", GWEN_MsgEndpoint_GetName(ep));
+  do {
+    rv=read(GWEN_MsgEndpoint_GetFd(ep), buffer, sizeof(buffer));
+  } while( (rv<0) && errno==EINTR);
+  if (rv<0) {
+    if (errno==EAGAIN || errno==EWOULDBLOCK)
+      return GWEN_ERROR_TRY_AGAIN;
+    DBG_ERROR(GWEN_LOGDOMAIN, "Error on read(): %s (%d)", strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  else if (rv==0) {
+    DBG_ERROR(GWEN_LOGDOMAIN, "EOF met on read()");
+    return GWEN_ERROR_IO;
+  }
+  len=rv;
+
+  for (i=0; i<len; i++) {
+    GWEN_MSG *msg;
+
+    msg=GWEN_MsgEndpoint_GetCurrentlyReceivedMsg(ep);
+    if (msg==NULL) {
+      msg=GWEN_Msg_new(AQH_MSG_ENDPOINT_TTY_BUFFERSIZE);
+      GWEN_Msg_SetGroupId(msg, GWEN_MsgEndpoint_GetGroupId(ep));
+      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, msg);
+    }
+    rv=GWEN_Msg_AddByte(msg, buffer[i]);
+    if (rv<0) {
+      DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
+      return rv;
+    }
+    rv=AQH_NodeMsg_IsMsgComplete(msg);
+    if (rv<0) {
+      /* invalid message */
+      DBG_ERROR(GWEN_LOGDOMAIN, "Invalid message, discarding");
+      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
+      rv=GWEN_MsgEndpoint_DiscardInput(ep);
+      if (rv<0) {
+        DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
+        return rv;
+      }
+    }
+    else if (rv>0) {
+      /* complete msg received, add to list */
+      if (!AQH_NodeMsg_IsChecksumValid(msg)) {
+        DBG_ERROR(AQH_LOGDOMAIN, "Invalid checksum, discarding message");
+        GWEN_Text_DumpString(GWEN_Msg_GetBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 6);
+        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
+        rv=GWEN_MsgEndpoint_DiscardInput(ep);
+        if (rv<0) {
+          DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
+          return rv;
+        }
+      }
+      else {
+        GWEN_Msg_Attach(msg);
+        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
+        GWEN_MsgEndpoint_AddReceivedMessage(ep, msg);
+      }
+    }
+  } /* for */
+
+  return 0;
+}
+
+
+
+int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
+{
+  GWEN_MSG *msg;
+
+  DBG_DEBUG(GWEN_LOGDOMAIN, "Writing to endpoint %s", GWEN_MsgEndpoint_GetName(ep));
+  msg=GWEN_MsgEndpoint_GetFirstSendMessage(ep);
+  if (msg) {
+    uint8_t pos;
+    int remaining;
+    int rv;
+
+    pos=GWEN_Msg_GetCurrentPos(msg);
+    remaining=GWEN_Msg_GetRemainingBytes(msg);
+    if (remaining>0) {
+      const uint8_t *buf;
+      int fd;
+
+      if (pos==0) {
+        /* start new message */
+        rv=_isLineBusy(ep);
+        if (rv<0 || rv==1) {
+          DBG_ERROR(AQH_LOGDOMAIN, "Line busy, not sending");
+          usleep(100);
+          return 0;
+        }
+        rv=_startMsg(ep);
+        if (rv<0) {
+          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
+          return rv;
+        }
+      }
+
+      fd=GWEN_MsgEndpoint_GetFd(ep);
+      /* start new message */
+      buf=GWEN_Msg_GetBuffer(msg)+pos;
+      do {
+        rv=write(fd, buf, remaining);
+      } while(rv<0 && errno==EINTR);
+      if (rv<0) {
+        if (errno==EAGAIN || errno==EWOULDBLOCK)
+          return GWEN_ERROR_TRY_AGAIN;
+        DBG_ERROR(GWEN_LOGDOMAIN, "Error on write(): %s (%d)", strerror(errno), errno);
+        return GWEN_ERROR_IO;
+      }
+      GWEN_Msg_IncCurrentPos(msg, rv);
+      if (rv==remaining) {
+        /* end current message */
+        rv=_endMsg(ep);
+        GWEN_Msg_List_Del(msg);
+	GWEN_Msg_free(msg);
+        if (rv<0) {
+          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
+          return rv;
+        }
+      }
+    }
+  }
+  return 0;
+}
+
+
+
+int _openDevice(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+  int fd;
+  int status;
+  int i;
+  struct termios options;
+  int rv;
+  int m;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  fd=open(xep->deviceName, O_NOCTTY | O_NDELAY | O_RDWR);
+  if (fd<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on open(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  rv=tcgetattr(fd, &(xep->previousOptions));
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcgetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  memset(&options, 0, sizeof(options)); /* preset */
+
+  options.c_cflag=CLOCAL | CREAD | CS8;
+  options.c_iflag=IGNPAR | IGNBRK;
+  options.c_oflag=0;
+  options.c_lflag=0;
+  cfmakeraw(&options);
+  options.c_cc[VTIME]=0;  /* read timeout in deciseconds */
+  options.c_cc[VMIN]=0;   /* no minimum number of receive bytes */
+
+  rv=cfsetispeed(&options, AQH_MSG_ENDPOINT_TTY_BAUDRATE);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetispeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  rv=cfsetospeed(&options, AQH_MSG_ENDPOINT_TTY_BAUDRATE);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetospeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+
+  rv=tcflush(fd, TCIOFLUSH);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcflush(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+
+  rv=tcsetattr(fd, TCSANOW, &options);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcsetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+
+  return fd;
+}
+
+
+
+int _startMsg(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  if (xep->intendedAttnState==1) {
+    int rv;
+
+    rv=_attnLow(ep);
+    if (rv<0) {
+      DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
+      return rv;
+    }
+    usleep(AQH_MSG_ENDPOINT_TTY_BYTE_MICROSECS/5);
+  }
+
+  return 0;
+}
+
+
+
+int _endMsg(GWEN_MSG_ENDPOINT *ep)
+{
+  /* TODO: flush before releasing ATTN */
+  _attnHigh(ep);
+}
+
+
+
+int _isLineBusy(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  if (xep->intendedAttnState==0) {
+    /* we pulled the line low ourselves, and because of the circuitry nobody can pull it high */
+    return 0;
+  }
+  return _isAttnLow(ep);
+}
+
+
+
+int _attnLow(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+  int status;
+  int rv;
+  int fd;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  fd=GWEN_MsgEndpoint_GetFd(ep);
+  rv=ioctl(fd, TIOCMGET, &status); /* GET the State of MODEM bits in Status */
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  status |= TIOCM_DTR | TIOCM_RTS;        /* clear the DTR pin (cave: signals inverted!) */
+  rv=ioctl(fd, TIOCMSET, &status);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  xep->intendedAttnState=0;
+  return 0;
+}
+
+
+
+int _attnHigh(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+  int status;
+  int rv;
+  int fd;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  fd=GWEN_MsgEndpoint_GetFd(ep);
+  rv=ioctl(fd, TIOCMGET, &status);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  status |= TIOCM_DTR;        /* Set the DTR pin */
+  status &= ~ (TIOCM_DTR | TIOCM_RTS);      /* clear the DTR pin (cave: signals inverted!) */
+  rv=ioctl(fd, TIOCMSET, &status);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+
+  xep->intendedAttnState=1;
+  return 0;
+}
+
+
+
+int _isAttnLow(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_MSG_ENDPOINT_TTY *xep;
+  int status;
+  int rv;
+  int fd;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_TTY, ep);
+  assert(xep);
+  fd=GWEN_MsgEndpoint_GetFd(ep);
+  rv=ioctl(fd, TIOCMGET, &status);
+  if (rv<0) {
+    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
+    return GWEN_ERROR_IO;
+  }
+  return (status & TIOCM_CTS)?1:0;
+}
+
+
+
+
+
+
+

aqhome/msg/endpoint_tty.h

@@ -0,0 +1,21 @@
+/****************************************************************************
+ * 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_MSGENDPOINT_TTY_H
+#define AQH_MSGENDPOINT_TTY_H
+
+
+#include <aqhome/api.h>
+
+#include <gwenhywfar/endpoint.h>
+
+
+AQHOME_API GWEN_MSG_ENDPOINT *AQH_TtyNodeEndpoint_new(const char *devicePath, int groupId);
+
+
+#endif

aqhome/msg/endpoint_tty_p.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_MSGENDPOINT_TTY_P_H
+#define AQH_MSGENDPOINT_TTY_P_H
+
+
+#include <aqhome/api.h>
+
+#include "aqhome/msg/endpoint_tty.h"
+
+#include <termios.h>
+
+
+
+
+typedef struct AQH_MSG_ENDPOINT_TTY AQH_MSG_ENDPOINT_TTY;
+struct AQH_MSG_ENDPOINT_TTY {
+  char *deviceName;
+  struct termios previousOptions;
+  int intendedAttnState;
+};
+
+
+
+
+#endif

aqhome/msg/endpointmgr.c

@@ -26,6 +26,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);
@@ -62,6 +64,18 @@ void _freeData(void *bp, void *p)
 
 
 
+int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
+{
+  int rv;
+
+  rv=GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
+  _msgLoopOnce(emgr);
+  GWEN_MsgEndpointMgr_RunAllEndpoints(emgr);
+  return rv;
+}
+
+
+
 void _msgLoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr)
 {
   GWEN_MSG_ENDPOINT_LIST *endpointList;
@@ -131,17 +145,17 @@ void _distributeMsgFromNodeEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOI
     ep=GWEN_MsgEndpoint_List_First(endpointList);
     while(ep) {
       if (ep!=srcEp) {
-        uint32_t acceptedGroupIds;
         uint32_t acceptedMsgGroups;
 
-        DBG_DEBUG(AQH_LOGDOMAIN, "- checking endpoint %s", GWEN_MsgEndpoint_GetName(ep));
-        acceptedGroupIds=AQH_NodeEndpoint_GetAcceptedMsgGroups(ep);
         acceptedMsgGroups=AQH_NodeEndpoint_GetAcceptedMsgGroups(ep);
+        DBG_DEBUG(AQH_LOGDOMAIN, "- checking endpoint %s (msgGroup=%08x, accept: %08x, flags: %08x)",
+                  GWEN_MsgEndpoint_GetName(ep),
+                  msgGroup, acceptedMsgGroups,
+                  GWEN_MsgEndpoint_GetFlags(ep));
 
         if (
             !(GWEN_MsgEndpoint_GetFlags(ep) & AQH_MSGEP_NODE_FLAGS_NOMESSAGES) &&
-            (acceptedMsgGroups & msgGroup) &&
+            (msgGroup & acceptedMsgGroups)
-            (acceptedGroupIds & srcGroupId)
            ) {
           /* endpoint accepts this message */
           DBG_DEBUG(AQH_LOGDOMAIN, "  - endpoint %s accepts message", GWEN_MsgEndpoint_GetName(ep));

aqhome/msg/endpointmgr.h

@@ -22,6 +22,7 @@
 
 
 AQHOME_API GWEN_MSG_ENDPOINT_MGR *AQH_MsgEndpointMgr_new(uint8_t busAddr);
+AQHOME_API int AQH_MsgEndpointMgr_LoopOnce(GWEN_MSG_ENDPOINT_MGR *emgr);