AqHome: Added a test for sending/receiving asynchronously.

aqhome/libtest.c

@@ -11,6 +11,7 @@
 #include <gwenhywfar/buffer.h>
 
 #include <unistd.h>
+#include <time.h>
 
 
 
@@ -100,10 +101,82 @@ int testSend()
 
 
 
-int main(void)
+void _packetReceived(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len)
 {
-  //return testRecv();
+  GWEN_BUFFER *dbuf;
-  return testSend();
+  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);
+  fprintf(stdout, "- %s: Received:\n", GWEN_Buffer_GetStart(dbuf));
+  GWEN_Text_DumpString(ptr, len, 2);
+  GWEN_Time_free(ti);
+  GWEN_Buffer_free(dbuf);
+}
+
+
+
+int testLoop()
+{
+  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 };
+  time_t tLast;
+
+  fprintf(stdout, "Opening device...\n");
+  sr=AQH_Serial_new("/dev/ttyUSB0", 240);
+  rv=AQH_Serial_Open(sr);
+  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)>10) {
+      rv=AQH_Serial_StartWriting(sr, sendBuf, 5);
+      if (rv==0)
+        fprintf(stdout, "+ Sending data\n");
+      else if (rv==GWEN_ERROR_TRY_AGAIN)
+        fprintf(stdout, "W Outbuffer busy\n");
+      else {
+        DBG_ERROR(NULL, "here (%d)", rv);
+        AQH_Serial_Close(sr);
+        AQH_Serial_free(sr);
+        return rv;
+      }
+      tLast=t;
+    }
+  }
+
+  return 0;
+}
+
+
+
+
+
+int main(void)
+{
+  //return testRecv();
+  //return testSend();
+  return testLoop();
 }
 
 

aqhome/serial.c

@@ -18,12 +18,18 @@
 
 #include <string.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>
+
 
 
 int _attnLow(AQH_SERIAL *sr);
@@ -46,6 +52,8 @@ AQH_SERIAL *AQH_Serial_new(const char *deviceName, uint8_t addr)
   sr->address=addr;
   sr->fd=-1;
 
+  sr->bytesToRead=2;
+
   return sr;
 }
 
@@ -252,6 +260,13 @@ int AQH_Serial_SendPacket(AQH_SERIAL *sr, uint8_t destAddr, const uint8_t *ptr,
 
 
 
+void AQH_Serial_SetPacketReceivedFn(AQH_SERIAL *sr, AQH_SERIAL_PACKETRECEIVED_FN fn)
+{
+  sr->packetReceivedFn=fn;
+}
+
+
+
 int _attnLow(AQH_SERIAL *sr)
 {
   int status;
@@ -347,6 +362,14 @@ int _writeForced(AQH_SERIAL *sr, const uint8_t *buf, uint8_t len)
 
 
 
+void AQH_Serial_PacketReceived(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len)
+{
+  if (sr->packetReceivedFn)
+    sr->packetReceivedFn(sr, ptr, len);
+}
+
+
+
 int _check(const uint8_t *ptr, uint8_t len)
 {
   int i;
@@ -379,6 +402,155 @@ uint8_t _calcChecksum(const uint8_t *ptr, uint8_t len)
 
 
 
+int _readFromFd(AQH_SERIAL *sr)
+{
+  int rv;
+  uint8_t pos;
+  uint8_t *buf;
+  int len;
+
+  pos=sr->readPos;
+  buf=sr->readBuffer+pos;
+  len=sr->bytesToRead;
+  rv=(int) read(sr->fd, buf, len);
+  if (rv<0) {
+    if (errno!=EINTR) {
+      DBG_ERROR(NULL, "Error on readFromFd(%s): %s (%d)", sr->deviceName, strerror(errno), errno);
+      return GWEN_ERROR_IO;
+    }
+    return 0;
+  }
+  else if (rv==0) {
+    DBG_ERROR(NULL, "EOF met on read(%s)", sr->deviceName);
+    return GWEN_ERROR_IO;
+  }
+  else {
+    sr->readPos+=rv;
+    sr->bytesToRead-=rv;
+
+    if (sr->readPos==2) {
+      uint8_t i;
+
+      /* we have 2 bytes now, so we now the total msg len */
+      i=sr->readBuffer[1];
+      i++;              /* remainder of message plus XOR byte */
+      sr->bytesToRead=i;
+    }
+    else {
+      if (sr->bytesToRead==0) {
+        /* msg received, handle */
+        AQH_Serial_PacketReceived(sr, sr->readBuffer, sr->readPos);
+        sr->readPos=0;
+        sr->bytesToRead=2;
+      }
+    }
+    return 0;
+  }
+}
+
+
+
+int _writeToFd(AQH_SERIAL *sr)
+{
+  if (sr->bytesToWrite) {
+    int rv;
+    uint8_t pos;
+    uint8_t *buf;
+    int len;
+  
+    pos=sr->writePos;
+    buf=sr->writeBuffer+pos;
+    len=sr->bytesToWrite;
+    rv=(int) write(sr->fd, buf, len);
+    if (rv<0) {
+      if (errno!=EINTR) {
+        DBG_ERROR(NULL, "Error on writeToFd(%s): %s (%d)", sr->deviceName, strerror(errno), errno);
+        return GWEN_ERROR_IO;
+      }
+      return 0;
+    }
+    else {
+      sr->writePos+=rv;
+      sr->bytesToWrite-=rv;
+      if (sr->bytesToWrite==0)
+        _attnHigh(sr);
+      return 0;
+    }
+  }
+
+  return 0;
+}
+
+
+
+int AQH_Serial_StartWriting(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len)
+{
+  int rv;
+
+  if (sr->bytesToWrite) {
+    DBG_ERROR(NULL, "Write buffer in use");
+    return GWEN_ERROR_TRY_AGAIN;
+  }
+
+  memmove(sr->writeBuffer, ptr, len);
+  sr->bytesToWrite=len;
+  sr->writePos=0;
+
+  rv=_attnLow(sr);
+  if (rv<0) {
+    DBG_INFO(NULL, "here (%d)", rv);
+    return rv;
+  }
+
+  usleep(10);
+  return 0;
+}
+
+
+
+
+int AQH_Serial_Loop(AQH_SERIAL *sr)
+{
+  fd_set readSet;
+  fd_set writeSet;
+  struct timeval tv;
+  int rv;
+
+  tv.tv_sec=5;
+  tv.tv_usec=0;
+
+  if (sr->bytesToWrite) {
+    FD_ZERO(&writeSet);
+    FD_SET(sr->fd, &writeSet);
+  }
+  FD_ZERO(&readSet);
+  FD_SET(sr->fd, &readSet);
+
+  rv=select(sr->fd+1, &readSet, (sr->bytesToWrite)?(&writeSet):NULL, NULL, &tv);
+  if (rv<0) {
+  }
+  else if (rv) {
+    if (FD_ISSET(sr->fd, &readSet)) {
+      rv=_readFromFd(sr);
+      if (rv<0) {
+        DBG_INFO(NULL, "here (%d)", rv);
+        return rv;
+      }
+    }
+    if (FD_ISSET(sr->fd, &writeSet)) {
+      rv=_writeToFd(sr);
+      if (rv<0) {
+        DBG_INFO(NULL, "here (%d)", rv);
+        return rv;
+      }
+    }
+  }
+
+  return 0;
+}
+
+
+
 
 
 

aqhome/serial.h

@@ -24,6 +24,10 @@ extern "C" {
 typedef struct AQH_SERIAL AQH_SERIAL;
 
 
+typedef void (*AQH_SERIAL_PACKETRECEIVED_FN)(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len);
+
+
+
 AQHOME_API AQH_SERIAL *AQH_Serial_new(const char *deviceName, uint8_t addr);
 AQHOME_API void AQH_Serial_free(AQH_SERIAL *sr);
 
@@ -39,6 +43,13 @@ AQHOME_API int AQH_Serial_Send(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len);
 AQHOME_API int AQH_Serial_SendPacket(AQH_SERIAL *sr, uint8_t destAddr, const uint8_t *ptr, uint8_t len);
 
 
+AQHOME_API int AQH_Serial_Loop(AQH_SERIAL *sr);
+AQHOME_API int AQH_Serial_StartWriting(AQH_SERIAL *sr, const uint8_t *ptr, uint8_t len);
+
+
+AQHOME_API void AQH_Serial_SetPacketReceivedFn(AQH_SERIAL *sr, AQH_SERIAL_PACKETRECEIVED_FN fn);
+
+
 #ifdef __cplusplus
 }
 #endif

aqhome/serial_p.h

@@ -14,10 +14,24 @@
 #include <inttypes.h>
 
 
+#define AQH_SERIAL_BUFFERSIZE 32
+
+
+
 struct AQH_SERIAL {
   char *deviceName;
   int fd;
   uint8_t address;
+
+  uint8_t readBuffer[AQH_SERIAL_BUFFERSIZE];
+  uint8_t bytesToRead;
+  uint8_t readPos;
+
+  uint8_t writeBuffer[AQH_SERIAL_BUFFERSIZE];
+  uint8_t bytesToWrite;
+  uint8_t writePos;
+
+  AQH_SERIAL_PACKETRECEIVED_FN packetReceivedFn;
 };