aqhome: started http endpoint.

aqhome/http/endpoint_http.c

@@ -1,5 +1,3 @@
-#if 0
-
 /****************************************************************************
  * This file is part of the project AqHome.
  * AqHome (c) by 2023 Martin Preuss, all rights reserved.
@@ -36,42 +34,45 @@
 GWEN_INHERIT(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP)
 
 
+/* ------------------------------------------------------------------------------------------------
+ * forward declarations
+ * ------------------------------------------------------------------------------------------------
+ */
 
 static void GWENHYWFAR_CB _freeData(void *bp, void *p);
-static int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
+static void _addSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_UNUSED GWEN_SOCKETSET *xSet);
-static int _readCommand(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen);
+static void _checkSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
-static int _readHeader(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen);
+static int _writeCurrentMessage(GWEN_MSG_ENDPOINT *ep);
-static int _readBody(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen);
+static int _readCurrentMessage(GWEN_MSG_ENDPOINT *ep);
-static int _readLineCheckEmpty(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen);
+static int _distributeBufferContent(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
-static int _calcCurrentLineLength(const GWEN_MSG_ENDPOINT *ep);
+static int _distributeBufferInCommandMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
-static int _parseCommandLineCheckIfHeaderNeeded(char *buffer, GWEN_DB_NODE *dbCurrentReadCommand);
+static int _distributeBufferInStatusMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
+static int _distributeBufferInHeaderMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
+static int _distributeBufferInBodyMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
+static int _distributeBufferAsLine(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen);
+static void _finishMessageAndStartNext(GWEN_MSG_ENDPOINT *ep);
+static GWEN_DB_NODE *_parseHeader(char *bufferPtr);
+static GWEN_DB_NODE *_parseCommand(const char *buffer);
 
 
 
+/* ------------------------------------------------------------------------------------------------
+ * implementations
+ * ------------------------------------------------------------------------------------------------
+ */
 
-GWEN_MSG_ENDPOINT *AQH_HttpEndpoint_new(const char *host, int port, const char *name, int groupId)
+void AQH_HttpEndpoint_Extend(GWEN_MSG_ENDPOINT *ep, uint32_t flags)
 {
-  GWEN_MSG_ENDPOINT *ep;
+  if (ep) {
-  AQH_ENDPOINT_HTTP *xep;
+    AQH_ENDPOINT_HTTP *xep;
 
-  ep=GWEN_TcpcEndpoint_new(host, port, name?name:AQH_ENDPOINT_HTTP_NAME, groupId);
+    GWEN_NEW_OBJECT(AQH_ENDPOINT_HTTP, xep);
-  if (ep==NULL) {
+    GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep, xep, _freeData);
-    DBG_INFO(AQH_LOGDOMAIN, "here");
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Command;
-    return NULL;
+
+    xep->addSocketsFn=GWEN_MsgEndpoint_SetAddSocketsFn(ep, _addSockets);
+    xep->checkSocketsFn=GWEN_MsgEndpoint_SetCheckSocketsFn(ep, _checkSockets);
   }
-
-  GWEN_NEW_OBJECT(AQH_ENDPOINT_HTTP, xep);
-  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep, xep, _freeData);
-
-  xep->currentReadBuffer=GWEN_Buffer_new(0, 256, 0, 1);
-  xep->currentReadHeader=GWEN_DB_Group_new("header");
-  xep->currentReadCommand=GWEN_DB_Group_new("cmd");
-
-  GWEN_MsgEndpoint_SetDefaultBufferSize(ep, AQH_ENDPOINT_HTTP_BUFFERSIZE);
-//  GWEN_MsgEndpoint_SetIsMsgCompleteFn(ep, _isMsgComplete);
-  GWEN_MsgEndpoint_SetHandleReadableFn(ep, _handleReadable);
-
-  return ep;
 }
 
 
@@ -86,217 +87,517 @@ void _freeData(void *bp, void *p)
   GWEN_DB_Group_free(xep->currentReadCommand);
   GWEN_Buffer_free(xep->currentReadBuffer);
 
-
   GWEN_FREE_OBJECT(xep);
 }
 
 
 
-int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
+void _addSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_UNUSED GWEN_SOCKETSET *xSet)
 {
-  AQH_ENDPOINT_HTTP *xep;
+  if (ep) {
-  int rv;
+    AQH_ENDPOINT_HTTP *xep;
-  uint8_t buffer[AQH_ENDPOINT_HTTP_BUFFERSIZE];
-  const uint8_t *ptr;
-  int len;
 
-  DBG_DEBUG(GWEN_LOGDOMAIN, "Reading from endpoint %s", GWEN_MsgEndpoint_GetName(ep));
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+    if (xep) {
-  do {
+      if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED) {
-    rv=read(GWEN_MsgEndpoint_GetFd(ep), buffer, sizeof(buffer));
+        GWEN_SOCKET *sk;
-  } while( (rv<0) && errno==EINTR);
-  if (rv<0) {
-    if (errno==EAGAIN || errno==EWOULDBLOCK)
-      return GWEN_ERROR_TRY_AGAIN;
-    DBG_INFO(GWEN_LOGDOMAIN, "Error on read(): %s (%d)", strerror(errno), errno);
-    return GWEN_ERROR_IO;
-  }
-  else if (rv==0) {
-    DBG_INFO(GWEN_LOGDOMAIN, "EOF met on read()");
-    return GWEN_ERROR_IO;
-  }
-  len=rv;
-  ptr=buffer;
 
-  /* len=number of bytes read into buffer */
+        sk=GWEN_MsgEndpoint_GetSocket(ep);
-  while(len) {
+        if (sk) {
-    switch(xep->readMode) {
+          DBG_DEBUG(AQH_LOGDOMAIN, "Endpoint %s: Adding socket %d to read set",
-    case AQH_EndpointHttpd_ReadMode_Command:
+                    GWEN_MsgEndpoint_GetName(ep),
-      rv=_readCommand(ep, &ptr, &len);
+                    GWEN_Socket_GetSocketInt(sk));
-      break;
+          GWEN_SocketSet_AddSocket(readSet, sk);
-    case AQH_EndpointHttpd_ReadMode_Headers:
+          if (GWEN_MsgEndpoint_HaveMessageToSend(ep)) {
-      rv=_readHeader(ep, &ptr, &len);
+            DBG_DEBUG(AQH_LOGDOMAIN, "Endpoint %s: Adding socket %d to write set",
-      break;
+                      GWEN_MsgEndpoint_GetName(ep),
-    case AQH_EndpointHttpd_ReadMode_Body:
+                      GWEN_Socket_GetSocketInt(sk));
-      rv=_readBody(ep, &ptr, &len);
+            GWEN_SocketSet_AddSocket(writeSet, sk);
-      break;
+          }
-    default:
+        } /* if socket */
-      DBG_ERROR(AQH_LOGDOMAIN, "Unexpected read mode %d", xep->readMode);
+      }
-      return GWEN_ERROR_INTERNAL;
+      else if (xep->addSocketsFn) {
+        DBG_INFO(AQH_LOGDOMAIN, "Endpoint %s: Not connected, calling base function", GWEN_MsgEndpoint_GetName(ep));
+        xep->addSocketsFn(ep, readSet, writeSet, xSet);
+      }
+    } /* if (xep) */
+  } /* if (ep) */
+}
+
+
+
+void _checkSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
+{
+  if (ep) {
+    AQH_ENDPOINT_HTTP *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+    if (xep) {
+      int rv;
+
+      if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED) {
+        GWEN_SOCKET *sk;
+      
+        sk=GWEN_MsgEndpoint_GetSocket(ep);
+        if (sk) {
+          if (GWEN_SocketSet_HasSocket(writeSet, sk)) {
+            DBG_DEBUG(AQH_LOGDOMAIN, "Endpoint %s: Has socket in write set", GWEN_MsgEndpoint_GetName(ep));
+            rv=_writeCurrentMessage(ep);
+            if (rv<0 && rv!=GWEN_ERROR_TIMEOUT) {
+              DBG_INFO(AQH_LOGDOMAIN,
+                       "Endpoint %s: Error writing current message (%d), disconnecting",
+                       GWEN_MsgEndpoint_GetName(ep),
+                       rv);
+              GWEN_MsgEndpoint_Disconnect(ep);
+              return;
+            }
+          }
+
+          if (GWEN_SocketSet_HasSocket(readSet, sk)) {
+            DBG_DEBUG(AQH_LOGDOMAIN, "Endpoint %s: Has socket in read set", GWEN_MsgEndpoint_GetName(ep));
+            rv=_readCurrentMessage(ep);
+            if (rv<0 && rv!=GWEN_ERROR_TIMEOUT) {
+              DBG_INFO(AQH_LOGDOMAIN,
+                       "Endpoint %s: Error reading current message (%d), disconnecting",
+                       GWEN_MsgEndpoint_GetName(ep),
+                       rv);
+              GWEN_MsgEndpoint_Disconnect(ep);
+              return;
+            }
+          }
+        }
+      } /* if connected */
+      else if (xep->checkSocketsFn) {
+        DBG_INFO(AQH_LOGDOMAIN, "Endpoint %s: Not connected, calling base function", GWEN_MsgEndpoint_GetName(ep));
+        xep->checkSocketsFn(ep, readSet, writeSet, xSet);
+      }
     }
   }
-
-  return 0;
 }
 
 
 
-/* rv: 1 if packet complete, <0 on error, 0 if packet not complete */
+int _writeCurrentMessage(GWEN_MSG_ENDPOINT *ep)
-int _readCommand(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen)
 {
-  AQH_ENDPOINT_HTTP *xep;
+  GWEN_MSG *msg;
-  int rv;
 
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+  DBG_DEBUG(AQH_LOGDOMAIN, "Writing to endpoint %s", GWEN_MsgEndpoint_GetName(ep));
-  rv=_readLineCheckEmpty(ep, pPtr, pLen);
+  msg=GWEN_MsgEndpoint_GetFirstSendMessage(ep);
-  if (rv>=0) {
+  if (msg) {
-    if (rv==1) {
+    uint8_t pos;
-      DBG_ERROR(AQH_LOGDOMAIN, "Empty command line received");
+    int remaining;
-      return GWEN_ERROR_BAD_DATA;
+    int rv;
+
+    pos=GWEN_Msg_GetCurrentPos(msg);
+    remaining=GWEN_Msg_GetRemainingBytes(msg);
+    if (remaining>0) {
+      const uint8_t *buf;
+
+      /* start new message */
+      buf=GWEN_Msg_GetBuffer(msg)+pos;
+      rv=GWEN_MsgEndpoint_WriteToSocket(ep, buf, remaining);
+      if (rv<0) {
+        if (rv==GWEN_ERROR_TIMEOUT)
+          return rv;
+        DBG_ERROR(AQH_LOGDOMAIN, "Error on write() (%d)", rv);
+        return rv;
+      }
+      GWEN_Msg_IncCurrentPos(msg, rv);
+      if (rv==remaining) {
+        DBG_INFO(AQH_LOGDOMAIN, "Message completely sent");
+        /* end current message */
+        GWEN_Msg_List_Del(msg);
+        GWEN_Msg_free(msg);
+      }
     }
-
-    /* line complete, parse command */
-    rv=_parseCommandLineCheckIfHeaderNeeded(GWEN_Buffer_GetStart(xep->currentReadBuffer), xep->currentReadCommand);
-    if (rv==0) {
-      /* no header follows, stay in readMode AQH_EndpointHttpd_ReadMode_Command for next command */
-      DBG_INFO(AQH_LOGDOMAIN, "No header expected, packet finished");
-      return 1;
-    }
-
-    /* line complete, set bookmark 0=pos of header begin, advance readMode */
-    GWEN_Buffer_SetBookmark(xep->currentReadBuffer, AQH_ENDPOINT_HTTP_BOOKMARK_HEADER, GWEN_Buffer_GetPos(xep->currentReadBuffer));
-    xep->readMode=AQH_EndpointHttpd_ReadMode_Headers;
-    DBG_INFO(AQH_LOGDOMAIN, "Start reading headers");
-  }
-
-  return 0;
-}
-
-
-
-int _readHeader(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen)
-{
-  AQH_ENDPOINT_HTTP *xep;
-  int rv;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
-  rv=_readLineCheckEmpty(ep, pPtr, pLen);
-  if (rv==1) {
-    /* line complete and empty */
-    // TODO: parse header
-    GWEN_Buffer_SetBookmark(xep->currentReadBuffer, AQH_ENDPOINT_HTTP_BOOKMARK_BODY, GWEN_Buffer_GetPos(xep->currentReadBuffer));
-    xep->readMode=AQH_EndpointHttpd_ReadMode_Body;
-  }
-  return 0;
-}
-
-
-
-int _readBody(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen)
-{
-}
-
-
-
-/* ret: -1 if line not complete,  1 if line complete and empty, 0 if line complete and not empty */
-int _readLineCheckEmpty(GWEN_MSG_ENDPOINT *ep, const uint8_t **pPtr, int *pLen)
-{
-  AQH_ENDPOINT_HTTP *xep;
-  const uint8_t *ptr=*pPtr;
-  int len=*pLen;
-  const uint8_t *p2;
-  int rv;
-
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
-
-  p2=memchr(ptr, 10, len);
-  if (p2) {
-    int l2;
-    uint32_t pos;
-    int currentLineLength;
-
-    l2=p2-ptr+1; /* also count the LF byte itself */
-    GWEN_Buffer_AppendBytes(xep->currentReadBuffer, (const char*) ptr, l2);
-    currentLineLength=_calcCurrentLineLength(ep);
-    pos=GWEN_Buffer_GetPos(xep->currentReadBuffer);
-    /* set bookmark 0: pos of header begin */
-    GWEN_Buffer_SetBookmark(xep->currentReadBuffer, 0, pos);
-    xep->lastLineStartPos=pos;
-    ptr+=l2;
-    len-=l2;
-    rv=(currentLineLength==0)?1:0;
   }
   else {
-    GWEN_Buffer_AppendBytes(xep->currentReadBuffer, (const char*) ptr, len);
+    DBG_INFO(AQH_LOGDOMAIN, "Nothing to send");
-    ptr+=len;
+  }
-    len=0;
+  return 0;
-    rv=-1;
+}
+
+
+
+
+int _readCurrentMessage(GWEN_MSG_ENDPOINT *ep)
+{
+  int rv;
+  uint8_t buffer[AQH_ENDPOINT_HTTP_BUFFERSIZE];
+
+  DBG_DEBUG(AQH_LOGDOMAIN, "Reading from endpoint %s", GWEN_MsgEndpoint_GetName(ep));
+  rv=GWEN_MsgEndpoint_ReadFromSocket(ep, buffer, sizeof(buffer));
+  if (rv<0 && rv!=GWEN_ERROR_TIMEOUT) {
+    DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
+    return rv;
+  }
+  else if (rv==0) {
+    DBG_INFO(AQH_LOGDOMAIN, "EOF met on read()");
+    return GWEN_ERROR_IO;
+  }
+
+  rv=_distributeBufferContent(ep, buffer, rv);
+  if (rv<0) {
+    DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
+    return rv;
+  }
+
+  return 0;
+}
+
+
+
+int _distributeBufferContent(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
+{
+  if (ep) {
+    AQH_ENDPOINT_HTTP *xep;
+
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+    if (xep) {
+      int rv;
+      if (xep->currentReadBuffer==NULL)
+	xep->currentReadBuffer=GWEN_Buffer_new(0, 256, 0, 1);
+
+      while(bufferLen) {
+	switch(xep->readMode) {
+	case AQH_EndpointHttpd_ReadMode_Command:
+	  rv=_distributeBufferInCommandMode(ep, bufferPtr, bufferLen);
+	  break;
+	case AQH_EndpointHttpd_ReadMode_Status:
+	  rv=_distributeBufferInStatusMode(ep, bufferPtr, bufferLen);
+	  break;
+	case AQH_EndpointHttpd_ReadMode_Headers:
+	  rv=_distributeBufferInHeaderMode(ep, bufferPtr, bufferLen);
+	  break;
+	case AQH_EndpointHttpd_ReadMode_Body:
+	  rv=_distributeBufferInBodyMode(ep, bufferPtr, bufferLen);
+	  break;
+	default:
+	  DBG_ERROR(AQH_LOGDOMAIN, "Invalid read mode %d", xep->readMode);
+	  return GWEN_ERROR_GENERIC;
+	}
+	if (rv<0) {
+	  DBG_INFO(AQH_LOGDOMAIN, "here (%d)", rv);
+	  return rv;
+	}
+	else if (rv==0) {
+	  DBG_INFO(AQH_LOGDOMAIN, "No bytes used? SNH!");
+	  return GWEN_ERROR_INTERNAL;
+	}
+	bufferPtr+=rv;
+	bufferLen-=rv;
+      } /* while */
+      return 0;
+    } /* if (xep) */
+  } /* if (ep) */
+  return GWEN_ERROR_GENERIC;
+}
+
+
+
+int _distributeBufferInCommandMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
+{
+  int rv;
+
+  rv=_distributeBufferAsLine(ep, bufferPtr, bufferLen);
+  if (rv<0) {
+    AQH_ENDPOINT_HTTP *xep;
+    GWEN_DB_NODE *db;
+    const char *s;
+
+    /* line complete, TODO: parse status/command line */
+    rv=-rv;
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+
+    db=_parseCommand(GWEN_Buffer_GetStart(xep->currentReadBuffer));
+    if (db==NULL) {
+      DBG_INFO(AQH_LOGDOMAIN, "Error parsing command line [%s]", GWEN_Buffer_GetStart(xep->currentReadBuffer));
+      return GWEN_ERROR_BAD_DATA;
+    }
+    s=GWEN_DB_GetCharValue(db, "protocol", 0, "HTTP/0.9");
+    if (s && *s && strcasecmp(s, "HTTP/0.9")==0) {
+      DBG_INFO(AQH_LOGDOMAIN, "HTTP 0.9, no header, message finished");
+      _finishMessageAndStartNext(ep);
+      return rv;
+    }
+    DBG_INFO(AQH_LOGDOMAIN, "Command line complete, advancing to header read mode");
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Headers;
+    xep->currentBodyPos=GWEN_Buffer_GetPos(xep->currentReadBuffer);
   }
 
-  *pPtr=ptr;
-  *pLen=len;
   return rv;
 }
 
 
 
-int _calcCurrentLineLength(const GWEN_MSG_ENDPOINT *ep)
+int _distributeBufferInStatusMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
 {
-  AQH_ENDPOINT_HTTP *xep;
+  int rv;
-  const uint8_t *ptr;
-  int len=0;
 
-  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+  rv=_distributeBufferAsLine(ep, bufferPtr, bufferLen);
-  ptr=(const uint8_t*) (GWEN_Buffer_GetStart(xep->currentReadBuffer)+xep->lastLineStartPos);
+  if (rv<0) {
-  while(*ptr && *ptr!=10 && *ptr!=13)
+    AQH_ENDPOINT_HTTP *xep;
-    len++;
+
-  return len;
+    /* line complete, TODO: parse status/command line */
+    rv=-rv;
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+    DBG_INFO(AQH_LOGDOMAIN, "Line complete, advancing to header read mode");
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Headers;
+    xep->currentBodyPos=GWEN_Buffer_GetPos(xep->currentReadBuffer);
+  }
+
+  return rv;
 }
 
 
 
-int _parseCommandLineCheckIfHeaderNeeded(char *buffer, GWEN_DB_NODE *dbCurrentReadCommand)
+int _distributeBufferInHeaderMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
 {
+  int rv;
+
+  rv=_distributeBufferAsLine(ep, bufferPtr, bufferLen);
+  if (rv<0) {
+    AQH_ENDPOINT_HTTP *xep;
+    int lineLength;
+
+    /* line complete, TODO: parse status/command line */
+    rv=-rv;
+    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+    DBG_INFO(AQH_LOGDOMAIN, "Line complete, advancing to header read mode");
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Headers;
+
+    lineLength=GWEN_Buffer_GetPos(xep->currentReadBuffer)-xep->lastLineStartPos;
+    xep->lastLineStartPos=GWEN_Buffer_GetPos(xep->currentReadBuffer);
+    if (lineLength==2) {
+      char *copyOfHeader;
+      GWEN_DB_NODE *db;
+      int contentLength;
+
+      /* Empty line received, TODO: parse header */
+      DBG_INFO(AQH_LOGDOMAIN, "Empty header line received, end of header reached.");
+      copyOfHeader=strdup(GWEN_Buffer_GetStart(xep->currentReadBuffer)+xep->currentHeaderPos);
+      db=_parseHeader(copyOfHeader);
+      if (db==NULL) {
+	DBG_INFO(AQH_LOGDOMAIN, "Error parsing HTTP header");
+	free(copyOfHeader);
+	return GWEN_ERROR_BAD_DATA;
+      }
+      free(copyOfHeader);
+      contentLength=GWEN_DB_GetIntValue(db, "Content-Length", 0, -1);
+      if (contentLength==0 || contentLength==-1) {
+	_finishMessageAndStartNext(ep);
+      }
+      else {
+	xep->currentBodyPos=GWEN_Buffer_GetPos(xep->currentReadBuffer);
+	xep->currentBodySize=contentLength;
+	xep->readMode=AQH_EndpointHttpd_ReadMode_Body;
+      }
+    }
+  }
+
+  return rv;
+}
+
+
+
+int _distributeBufferInBodyMode(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
+{
+  AQH_ENDPOINT_HTTP *xep;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+  if (xep->currentBodySize>0) {
+    int len;
+
+    len=bufferLen;
+    if (len>xep->currentBodySize)
+      len=xep->currentBodySize;
+    if (len) {
+      GWEN_Buffer_AppendBytes(xep->currentReadBuffer, (const char*) bufferPtr, len);
+      xep->currentBodySize-=len;
+      if (xep->currentBodySize==0) {
+	DBG_INFO(AQH_LOGDOMAIN, "Body completely received");
+	_finishMessageAndStartNext(ep);
+      }
+      return len;
+    }
+    else {
+      DBG_ERROR(AQH_LOGDOMAIN, "No bytes left to read, SNH!");
+      return GWEN_ERROR_INTERNAL;
+    }
+  }
+  else {
+    DBG_ERROR(AQH_LOGDOMAIN, "No body size, aborting");
+    return GWEN_ERROR_GENERIC;
+  }
+}
+
+
+
+/* return negative number of bytes handled when LF encountered, positive value otherwise */
+int _distributeBufferAsLine(GWEN_MSG_ENDPOINT *ep, const uint8_t *bufferPtr, int bufferLen)
+{
+  AQH_ENDPOINT_HTTP *xep;
+  const char *s;
+  int i;
+
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+
+  s=(const char*) bufferPtr;
+  i=0;
+  while(i<bufferLen && *s!=10) {   /* scan until LF is encountered */
+    s++;
+    i++;
+  }
+  GWEN_Buffer_AppendBytes(xep->currentReadBuffer, (const char*) bufferPtr, i+1);
+  if (*s==10) {
+    DBG_INFO(AQH_LOGDOMAIN, "Received full line");
+    i=-i;
+  }
+  return i;
+}
+
+
+
+void _finishMessageAndStartNext(GWEN_MSG_ENDPOINT *ep)
+{
+  AQH_ENDPOINT_HTTP *xep;
+  GWEN_MSG *msg;
+  
+  DBG_INFO(AQH_LOGDOMAIN, "Message completely received.");
+  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_ENDPOINT_HTTP, ep);
+  msg=GWEN_Msg_fromBytes((const uint8_t*)GWEN_Buffer_GetStart(xep->currentReadBuffer), GWEN_Buffer_GetUsedBytes(xep->currentReadBuffer));
+  GWEN_MsgEndpoint_AddReceivedMessage(ep, msg);
+
+  GWEN_Buffer_Reset(xep->currentReadBuffer);
+  xep->currentHeaderPos=0;
+  xep->currentBodyPos=0;
+  xep->currentBodySize=0;
+  xep->lastLineStartPos=0;
+
+  if (xep->flags & AQH_ENDPOINT_HTTP_FLAGS_PASSIVE)
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Command;
+  else
+    xep->readMode=AQH_EndpointHttpd_ReadMode_Status;
+}
+
+
+
+GWEN_DB_NODE *_parseHeader(char *bufferPtr)
+{
+  GWEN_DB_NODE *db;
+  char *p;
+
+  db=GWEN_DB_Group_new("header");
+
+  /* resolve line continuations */
+  p=bufferPtr;
+  while (*p) {
+    p=strchr(p, 10);
+    if (p) {
+      if (p[1]==32 || p[1]==9)
+        /* found a continuation */
+        *p=32;
+      p++;
+    }
+  }
+
+  /* parse every line */
+  p=bufferPtr;
+  while (p && *p) {
+    char *pNext;
+    char *pVarBegin;
+    char *pVarEnd;
+
+    /* skip blanks */
+    pNext=strchr(p, 10);
+    if (pNext) {
+      *pNext=0;
+      pNext++;
+    }
+    while (*p && (*p==32 || *p==9))
+      p++;
+    if (*p) {
+      pVarBegin=p;
+      while (*p && *p!=':' && *p>32 && *p<127)
+        p++;
+      pVarEnd=p;
+      if (*p!=':') {
+        DBG_INFO(AQH_LOGDOMAIN, "No separator after variable name in received header");
+	GWEN_DB_Group_free(db);
+	return NULL;
+      }
+      *pVarEnd=0;
+      p++;
+
+      while (*p && (*p==32 || *p==9))
+        p++;
+      if (*p)
+	GWEN_DB_SetCharValue(db, GWEN_PATH_FLAGS_CREATE_VAR, pVarBegin, p);
+    }
+    p=pNext;
+  }
+
+  return db;
+}
+
+
+
+GWEN_DB_NODE *_parseCommand(const char *buffer)
+{
+  GWEN_DB_NODE *db;
+  char *tmp;
   char *p;
   char *s;
 
-  s=buffer;
+  db=GWEN_DB_Group_new("command");
+
+  tmp=strdup(buffer);
+  s=tmp;
 
   /* read command */
   p=strchr(s, ' ');
   if (!p) {
     DBG_ERROR(AQH_LOGDOMAIN, "Bad format of HTTP request (%s)", buffer);
-    return GWEN_ERROR_INVALID;
+    free(tmp);
+    GWEN_DB_Group_free(db);
+    return NULL;
   }
   *p=0;
-  GWEN_DB_SetCharValue(dbCurrentReadCommand, GWEN_DB_FLAGS_OVERWRITE_VARS, "command", s);
-  *p=' '; /* restore buffer */
   p++;
+
+  GWEN_DB_SetCharValue(db, GWEN_DB_FLAGS_OVERWRITE_VARS, "command", s);
   s=p;
 
   /* read URL */
   p=strchr(s, ' ');
   if (!p) {
     DBG_ERROR(AQH_LOGDOMAIN, "Bad format of HTTP request (%s)", buffer);
-    return GWEN_ERROR_INVALID;
+    free(tmp);
+    GWEN_DB_Group_free(db);
+    return NULL;
   }
   *p=0;
-  GWEN_DB_SetCharValue(dbCurrentReadCommand, GWEN_DB_FLAGS_OVERWRITE_VARS, "url", s);
-  *p=' '; /* restore buffer */
   p++;
+
+  GWEN_DB_SetCharValue(db, GWEN_DB_FLAGS_OVERWRITE_VARS, "url", s);
   s=p;
 
   if (*s==0) {
     /* no protocol information follows, so we assume HTTP/0.9 */
-    return 0;
+    DBG_ERROR(AQH_LOGDOMAIN, "Bad request (not in HTTP>=1.0)");
+    free(tmp);
+    GWEN_DB_Group_free(db);
+    return NULL;
   }
   else {
-    GWEN_DB_SetCharValue(dbCurrentReadCommand, GWEN_DB_FLAGS_OVERWRITE_VARS, "protocol", s);
+    GWEN_DB_SetCharValue(db, GWEN_DB_FLAGS_OVERWRITE_VARS, "protocol", s);
   }
 
-  return 1;
+  free(tmp);
+  return db;
 }
 
 
 
-#endif
+
+

aqhome/http/endpoint_http.h

@@ -15,6 +15,12 @@
 #include <gwenhywfar/endpoint.h>
 
 
+#define AQH_ENDPOINT_HTTP_FLAGS_PASSIVE 0x0001
+
+
+
+AQHOME_API void AQH_HttpEndpoint_Extend(GWEN_MSG_ENDPOINT *ep, uint32_t flags);
+
 
 
 

aqhome/http/endpoint_http_p.h

@@ -17,8 +17,10 @@
 #include <gwenhywfar/buffer.h>
 
 
+
 enum {
   AQH_EndpointHttpd_ReadMode_Command=0,
+  AQH_EndpointHttpd_ReadMode_Status,
   AQH_EndpointHttpd_ReadMode_Headers,
   AQH_EndpointHttpd_ReadMode_Body
 };
@@ -28,9 +30,16 @@ enum {
 typedef struct AQH_ENDPOINT_HTTP AQH_ENDPOINT_HTTP;
 struct AQH_ENDPOINT_HTTP {
   int readMode;
+
+  uint32_t flags;
+
+  GWEN_MSG_ENDPOINT_ADDSOCKETS_FN addSocketsFn;
+  GWEN_MSG_ENDPOINT_CHECKSOCKETS_FN checkSocketsFn;
+
   GWEN_BUFFER *currentReadBuffer;
   GWEN_DB_NODE *currentReadCommand;
   GWEN_DB_NODE *currentReadHeader;
+  int currentHeaderPos;
   int currentBodyPos;
   int currentBodySize;
   int lastLineStartPos;