aqhome: add and implement a flush function for fd/tty objects.

aqhome/events2/fdobject.c

@@ -34,6 +34,7 @@ GWEN_INHERIT(AQH_OBJECT, AQH_FDOBJECT)
 static void GWENHYWFAR_CB _freeData(void *bp, void *p);
 static void _cbEnable(AQH_OBJECT *o);
 static void _cbDisable(AQH_OBJECT *o);
+static int _cbFlush(AQH_OBJECT *o);
 
 
 
@@ -53,6 +54,7 @@ AQH_OBJECT *AQH_FdObject_new(AQH_EVENT_LOOP *eventLoop, int fd, int mode)
   GWEN_INHERIT_SETDATA(AQH_OBJECT, AQH_FDOBJECT, o, xo, _freeData);
   xo->fd=fd;
   xo->fdMode=mode;
+  xo->flushFn=_cbFlush;
 
   AQH_Object_SetEnableFn(o, _cbEnable);
   AQH_Object_SetDisableFn(o, _cbDisable);
@@ -110,6 +112,23 @@ AQH_FDOBJECT_ENDMSG_FN AQH_FdObject_SetEndMsgFn(AQH_OBJECT *o, AQH_FDOBJECT_ENDM
 
 
 
+AQH_FDOBJECT_FLUSH_FN AQH_FdObject_SetFlushFn(AQH_OBJECT *o, AQH_FDOBJECT_FLUSH_FN f)
+{
+  AQH_FDOBJECT *xo;
+
+  xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_FDOBJECT, o);
+  if (xo) {
+    AQH_FDOBJECT_FLUSH_FN oldFn;
+
+    oldFn=xo->flushFn;
+    xo->flushFn=f;
+    return oldFn;
+  }
+  return NULL;
+}
+
+
+
 int AQH_FdObject_GetFdMode(const AQH_OBJECT *o)
 {
   if (o) {
@@ -205,40 +224,21 @@ int AQH_FdObject_FlushInput(AQH_OBJECT *o)
     AQH_FDOBJECT *xo;
 
     xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_FDOBJECT, o);
-    if (xo) {
+    if (xo && xo->flushFn) {
-      if (xo->fd!=-1) {
+      int rv;
-        ssize_t rv;
-        ssize_t bytesRead=0;
-        uint8_t tbuf[16];
 
-        do {
+      rv=xo->flushFn(o);
-          rv=read(xo->fd, tbuf, sizeof(tbuf));
+      if (rv<0) {
-          if (rv>0)
+        if (xo->fd!=-1) {
-            bytesRead+=rv;
+          DBG_ERROR(AQH_LOGDOMAIN, "Error on flush, closing.");
-        } while(rv>=0);
+          close(xo->fd);
-
+          xo->fd=-1;
-        if (rv==0) {
-          DBG_INFO(AQH_LOGDOMAIN, "EOF met");
-          return GWEN_ERROR_EOF;
         }
-        else if (rv<0) {
-          if (errno!=EINTR && errno!=EWOULDBLOCK && errno!=EAGAIN) {
-            DBG_ERROR(AQH_LOGDOMAIN, "Error on read: %s (%d)", strerror(errno), errno);
-            close(xo->fd);
-            xo->fd=-1;
-            return GWEN_ERROR_IO;
-          }
-        }
-        return (bytesRead>0)?1:0;
-      }
-      else {
-        DBG_ERROR(AQH_LOGDOMAIN, "Previous error, not reading.");
-        return GWEN_ERROR_IO;
       }
+      return rv;
     }
   }
-
+  return 0;
-  return GWEN_ERROR_INVALID;
 }
 
 
@@ -340,6 +340,48 @@ void _cbDisable(AQH_OBJECT *o)
 
 
 
+int _cbFlush(AQH_OBJECT *o)
+{
+  if (o) {
+    AQH_FDOBJECT *xo;
+
+    xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_FDOBJECT, o);
+    if (xo) {
+      if (xo->fd!=-1) {
+        ssize_t rv;
+        ssize_t bytesRead=0;
+        uint8_t tbuf[16];
+
+        do {
+          rv=read(xo->fd, tbuf, sizeof(tbuf));
+          if (rv>0)
+            bytesRead+=rv;
+        } while(rv>=0);
+
+        if (rv==0) {
+          DBG_INFO(AQH_LOGDOMAIN, "EOF met");
+          return GWEN_ERROR_EOF;
+        }
+        else if (rv<0) {
+          if (errno!=EINTR && errno!=EWOULDBLOCK && errno!=EAGAIN) {
+            DBG_ERROR(AQH_LOGDOMAIN, "Error on read: %s (%d)", strerror(errno), errno);
+            return GWEN_ERROR_IO;
+          }
+        }
+        return (bytesRead>0)?1:0;
+      }
+      else {
+        DBG_ERROR(AQH_LOGDOMAIN, "Previous error, not reading.");
+        return GWEN_ERROR_IO;
+      }
+    }
+  }
+  return GWEN_ERROR_INVALID;
+}
+
+
+
+
 
 
 

aqhome/events2/fdobject.h

@@ -26,6 +26,7 @@ enum {
 
 typedef int (*AQH_FDOBJECT_STARTMSG_FN)(AQH_OBJECT *o);
 typedef void (*AQH_FDOBJECT_ENDMSG_FN)(AQH_OBJECT *o);
+typedef int (*AQH_FDOBJECT_FLUSH_FN)(AQH_OBJECT *o);
 
 
 /**
@@ -48,6 +49,7 @@ AQHOME_API void AQH_FdObject_EndMsg(AQH_OBJECT *o);
 
 AQHOME_API AQH_FDOBJECT_STARTMSG_FN AQH_FdObject_SetStartMsgFn(AQH_OBJECT *o, AQH_FDOBJECT_STARTMSG_FN f);
 AQHOME_API AQH_FDOBJECT_ENDMSG_FN AQH_FdObject_SetEndMsgFn(AQH_OBJECT *o, AQH_FDOBJECT_ENDMSG_FN f);
+AQHOME_API AQH_FDOBJECT_FLUSH_FN AQH_FdObject_SetFlushFn(AQH_OBJECT *o, AQH_FDOBJECT_FLUSH_FN f);
 
 
 #endif

aqhome/events2/fdobject_p.h

@@ -20,6 +20,7 @@ struct AQH_FDOBJECT {
 
   AQH_FDOBJECT_STARTMSG_FN startMsgFn;
   AQH_FDOBJECT_ENDMSG_FN endMsgFn;
+  AQH_FDOBJECT_FLUSH_FN flushFn;
 };
 
 

aqhome/ipc2/tty_object.c

@@ -32,8 +32,9 @@
  * ------------------------------------------------------------------------------------------------
  */
 
-static int _startMsg(AQH_OBJECT *o);
+static int _cbStartMsg(AQH_OBJECT *o);
-static void _endMsg(AQH_OBJECT *o);
+static void _cbEndMsg(AQH_OBJECT *o);
+static int _cbFlush(AQH_OBJECT *o);
 static int _getAttn(int fd);
 static int _setAttn(int fd, int val);
 static int _fdSetBlocking(int sk, int fl);
@@ -51,14 +52,15 @@ AQH_OBJECT *AQH_TtyObject_new(AQH_EVENT_LOOP *eventLoop, int fd, int fdMode)
   AQH_OBJECT *o;
 
   o=AQH_FdObject_new(eventLoop, fd, fdMode);
-  AQH_FdObject_SetStartMsgFn(o, _startMsg);
+  AQH_FdObject_SetStartMsgFn(o, _cbStartMsg);
-  AQH_FdObject_SetEndMsgFn(o, _endMsg);
+  AQH_FdObject_SetEndMsgFn(o, _cbEndMsg);
+  AQH_FdObject_SetFlushFn(o, _cbFlush);
   return o;
 }
 
 
 
-int _startMsg(AQH_OBJECT *o)
+int _cbStartMsg(AQH_OBJECT *o)
 {
   if (o) {
     if (AQH_Object_GetFlags(o) & AQH_TTYOBJECT_FLAGS_NOATTN)
@@ -91,7 +93,7 @@ int _startMsg(AQH_OBJECT *o)
 
 
 
-void _endMsg(AQH_OBJECT *o)
+void _cbEndMsg(AQH_OBJECT *o)
 {
   if (o) {
     if (!(AQH_Object_GetFlags(o) & AQH_TTYOBJECT_FLAGS_NOATTN)) {
@@ -107,6 +109,32 @@ void _endMsg(AQH_OBJECT *o)
 
 
 
+int _cbFlush(AQH_OBJECT *o)
+{
+  int fd;
+
+  fd=AQH_FdObject_GetFd(o);
+  if (fd>=0) {
+    int rv;
+
+    rv=tcflush(fd, TCIFLUSH);
+    if (rv<0) {
+      if (errno!=EINTR && errno!=EWOULDBLOCK && errno!=EAGAIN) {
+        DBG_ERROR(AQH_LOGDOMAIN, "Error on tcflush: %s (%d)", strerror(errno), errno);
+        return GWEN_ERROR_IO;
+      }
+    }
+  }
+  else {
+    DBG_ERROR(AQH_LOGDOMAIN, "Previous error");
+    return GWEN_ERROR_IO;
+  }
+
+  return 0;
+}
+
+
+
 int _getAttn(int fd)
 {
   int rv;
@@ -243,6 +271,7 @@ int _fdSetBlocking(int fd, int fl)
 
 
 
+
 #if 0
 int _msleep(long int msec)
 {