add ehttp_*

src/common/inc/elog.h

+#ifndef _elog_h_8897be44_dda8_45b6_9d37_8d8691cb05fb_
+#define _elog_h_8897be44_dda8_45b6_9d37_8d8691cb05fb_
+
+#include <stdio.h>
+
+typedef enum {
+  ELOG_DEBUG,
+  ELOG_INFO,
+  ELOG_WARN,
+  ELOG_ERROR,
+  ELOG_CRITICAL,
+  ELOG_VERBOSE,
+  ELOG_ABORT,
+} ELOG_LEVEL;
+
+void elog_set_level(ELOG_LEVEL base); // only log those not less than base
+void elog_set_thread_name(const char *name);
+
+void elog(ELOG_LEVEL level, int fd, const char *file, int line, const char *func, const char *fmt, ...)
+#ifdef __GNUC__
+ __attribute__((format(printf, 6, 7)))
+#endif
+;
+
+#define DLOG(fd, fmt, ...)     elog(ELOG_DEBUG,    fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define ILOG(fd, fmt, ...)     elog(ELOG_INFO,     fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define WLOG(fd, fmt, ...)     elog(ELOG_WARN,     fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define ELOG(fd, fmt, ...)     elog(ELOG_ERROR,    fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define CLOG(fd, fmt, ...)     elog(ELOG_CRITICAL, fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define VLOG(fd, fmt, ...)     elog(ELOG_VERBOSE,  fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define ALOG(fd, fmt, ...)     elog(ELOG_ABORT,    fd, __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+
+#define D(fmt, ...)     elog(ELOG_DEBUG,    fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define I(fmt, ...)     elog(ELOG_INFO,     fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define W(fmt, ...)     elog(ELOG_WARN,     fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define E(fmt, ...)     elog(ELOG_ERROR,    fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define C(fmt, ...)     elog(ELOG_CRITICAL, fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define V(fmt, ...)     elog(ELOG_VERBOSE,  fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+#define A(fmt, ...)     elog(ELOG_ABORT,    fileno(stdout), __FILE__, __LINE__, __FUNCTION__, fmt, ##__VA_ARGS__)
+
+
+
+// NOTE: https://en.wikipedia.org/wiki/Fail-fast
+// for the sake of simplicity, both implementation and usage,
+// we'll follow `fail-fast` or `let-it-crash` philosophy.
+
+// assertion in both debug/release build
+#define EQ_ABORT(fmt, ...) A("Assertion failure: "fmt, ##__VA_ARGS__)
+
+#define EQ_ASSERT(statement) do {                                            \
+  if (statement) break;                                                      \
+  A("Assertion failure: %s", #statement);                                    \
+} while (0)
+
+#define EQ_ASSERT_EXT(statement, fmt, ...) do {                              \
+  if (statement) break;                                                      \
+  A("Assertion failure: %s: "fmt, #statement, ##__VA_ARGS__);                \
+} while (0)
+
+#define EQ_ASSERT_API0(statement) do {                                             \
+  if (statement) break;                                                            \
+  A("Assertion failure: %s failed: [%d]%s", #statement, errno, strerror(errno));   \
+} while (0)
+
+#define EQ_ASSERT_API(api) do {                                              \
+  A("Assertion failure: %s failed: [%d]%s", #api, errno, strerror(errno));   \
+} while (0)
+
+
+#endif // _elog_h_8897be44_dda8_45b6_9d37_8d8691cb05fb_
+

src/common/src/elog.c

+#include "elog.h"
+
+#include <libgen.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <sys/syscall.h>
+#include <time.h>
+#include <unistd.h>
+
+#define gettid() syscall(__NR_gettid)
+
+static ELOG_LEVEL         elog_level_base = ELOG_DEBUG;
+
+static __thread long      elog_thread_id;
+static __thread char      elog_thread_name[24] = {0};
+
+void elog_set_level(ELOG_LEVEL base) {
+  elog_level_base = base;
+}
+
+void elog_set_thread_name(const char *name) {
+  elog_thread_id = gettid();
+  snprintf(elog_thread_name, sizeof(elog_thread_name), "%s", name);
+}
+
+void elog(ELOG_LEVEL level, int fd, const char *file, int line, const char *func, const char *fmt, ...)
+{
+  if (level < elog_level_base) return;
+  if (fd == -1) return;
+
+  if (elog_thread_name[0]=='\0') {
+    elog_set_thread_name("unknown");
+  }
+
+  char     *p;
+  int       n;
+  size_t    bytes;
+
+  char buf[4096];
+  snprintf(buf, sizeof(buf), "%s", file);
+
+  char fn[1024];
+  snprintf(fn, sizeof(fn), "%s", basename(buf));
+
+  char      C;
+  switch (level) {
+    case ELOG_DEBUG:        C = 'D'; break;
+    case ELOG_INFO:         C = 'I'; break;
+    case ELOG_WARN:         C = 'W'; break;
+    case ELOG_ERROR:        C = 'E'; break;
+    case ELOG_CRITICAL:     C = 'C'; break;
+    case ELOG_VERBOSE:      C = 'V'; break;
+    case ELOG_ABORT:        C = 'A'; break;
+    default:                return;
+  }
+
+  struct tm      t;
+  struct timeval tv;
+
+  if (gettimeofday(&tv, NULL)) return;
+  if (!localtime_r(&tv.tv_sec, &t)) return;
+
+  p     = buf;
+  bytes = sizeof(buf);
+
+  n = snprintf(p, bytes, "%c[%02d/%02d %02d:%02d:%02d.%06ld][%06ld]: ==",
+               C,
+               t.tm_mon + 1, t.tm_mday,
+               t.tm_hour, t.tm_min, t.tm_sec,
+               tv.tv_usec,
+               elog_thread_id);
+  p += n; bytes -= n;
+
+  va_list arg;
+  va_start(arg, fmt);
+  if (bytes>0) {
+    n = vsnprintf(p, bytes, fmt, arg);
+    p += n; bytes -= n;
+  }
+  va_end(arg);
+
+  if (bytes>0) {
+    n = snprintf(p, bytes, "== t:%s#%s[%d]#%s()",
+                 elog_thread_name, fn, line, func);
+  }
+
+  dprintf(fd, "%s\n", buf);
+
+  if (level == ELOG_ABORT) {
+    abort();
+  }
+}
+

src/plugins/http/inc/ehttp_gzip.h

+#ifndef _ehttp_gzip_h_9196791b_ac2a_4d73_9979_f4b41abbc4c0_
+#define _ehttp_gzip_h_9196791b_ac2a_4d73_9979_f4b41abbc4c0_
+
+#include <stddef.h>
+
+#define EHTTP_GZIP_CHUNK_SIZE_DEFAULT      (1024*16)
+
+typedef struct ehttp_gzip_s              ehttp_gzip_t;
+
+typedef struct ehttp_gzip_callbacks_s    ehttp_gzip_callbacks_t;
+typedef struct ehttp_gzip_conf_s         ehttp_gzip_conf_t;
+
+struct ehttp_gzip_callbacks_s {
+  void (*on_data)(ehttp_gzip_t *gzip, void *arg, const char *buf, size_t len);
+};
+
+struct ehttp_gzip_conf_s {
+  int        get_header:2; // 0: not fetching header info
+  size_t     chunk_size;   // 0: fallback to default: EHTTP_GZIP_CHUNK_SIZE_DEFAULT
+};
+
+ehttp_gzip_t* ehttp_gzip_create_decompressor(ehttp_gzip_conf_t conf, ehttp_gzip_callbacks_t callbacks, void *arg);
+ehttp_gzip_t* ehttp_gzip_create_compressor(ehttp_gzip_conf_t conf, ehttp_gzip_callbacks_t callbacks, void *arg);
+void          ehttp_gzip_destroy(ehttp_gzip_t *gzip);
+
+int           ehttp_gzip_write(ehttp_gzip_t *gzip, const char *buf, size_t len);
+int           ehttp_gzip_finish(ehttp_gzip_t *gzip);
+
+#endif // _ehttp_gzip_h_9196791b_ac2a_4d73_9979_f4b41abbc4c0_
+

src/plugins/http/inc/ehttp_parser.h

+#ifndef _ehttp_parser_fc7f9ac9_52da_4ee3_b556_deb2e1c3866e
+#define _ehttp_parser_fc7f9ac9_52da_4ee3_b556_deb2e1c3866e
+
+#include <stddef.h>
+
+typedef struct ehttp_parser_s               ehttp_parser_t;
+typedef struct ehttp_parser_callbacks_s     ehttp_parser_callbacks_t;
+typedef struct ehttp_parser_conf_s          ehttp_parser_conf_t;
+typedef struct ehttp_status_code_s          ehttp_status_code_t;
+
+struct ehttp_parser_callbacks_s {
+  void (*on_request_line)(void *arg, const char *method, const char *target, const char *version, const char *target_raw);
+  void (*on_status_line)(void *arg, const char *version, int status_code, const char *reason_phrase);
+  void (*on_header_field)(void *arg, const char *key, const char *val);
+  void (*on_body)(void *arg, const char *chunk, size_t len);
+  void (*on_end)(void *arg);
+  void (*on_error)(void *arg, int status_code);
+};
+
+struct ehttp_parser_conf_s {
+  size_t             flush_block_size;       // <=0: immediately
+};
+
+ehttp_parser_t* ehttp_parser_create(ehttp_parser_callbacks_t callbacks, ehttp_parser_conf_t conf, void *arg);
+void            ehttp_parser_destroy(ehttp_parser_t *parser);
+int             ehttp_parser_parse(ehttp_parser_t *parser, const char *buf, size_t len);
+int             ehttp_parser_parse_string(ehttp_parser_t *parser, const char *str);
+int             ehttp_parser_parse_char(ehttp_parser_t *parser, const char c);
+int             ehttp_parser_parse_end(ehttp_parser_t *parser);
+
+char* ehttp_parser_urldecode(const char *enc);
+
+const char* ehttp_status_code_get_desc(const int status_code);
+
+#endif // _ehttp_parser_fc7f9ac9_52da_4ee3_b556_deb2e1c3866e
+

src/plugins/http/inc/ehttp_util_string.h

+#ifndef _ehttp_util_string_h_99dacde5_2e7d_4662_97d6_04611fde683b_
+#define _ehttp_util_string_h_99dacde5_2e7d_4662_97d6_04611fde683b_
+
+#include <stddef.h>
+
+typedef struct ehttp_util_string_s   ehttp_util_string_t;
+
+struct ehttp_util_string_s {
+  char                        *str;
+  size_t                       len;
+};
+
+void ehttp_util_string_cleanup(ehttp_util_string_t *str);
+int  ehttp_util_string_append(ehttp_util_string_t *str, const char *s, size_t len);
+void ehttp_util_string_clear(ehttp_util_string_t *str);
+
+#endif // _ehttp_util_string_h_99dacde5_2e7d_4662_97d6_04611fde683b_
+

src/plugins/http/inc/httpInt.h

@@ -28,6 +28,8 @@
 #include "httpLog.h"
 #include "httpJson.h"
 
+#include "ehttp_parser.h"
+
 #define HTTP_MAX_CMD_SIZE           1024
 #define HTTP_MAX_BUFFER_SIZE        1024*1024
 
@@ -162,6 +164,11 @@ typedef struct {
   int32_t len;
 } HttpBuf;
 
+typedef enum {
+  EHTTP_CONTEXT_PROCESS_FAILED = 0x01,
+  EHTTP_CONTEXT_PARSER_FAILED  = 0x02
+} EHTTP_CONTEXT_FAILED_CAUSE;
+
 typedef struct {
   char              buffer[HTTP_BUFFER_SIZE];
   int               bufsize;
@@ -172,6 +179,10 @@ typedef struct {
   HttpBuf           data;                // body content
   HttpBuf           token;               // auth token
   HttpDecodeMethod *pMethod;
+
+  ehttp_parser_t        *parser;
+  int                    inited:2;
+  int                    failed:4;
 } HttpParser;
 
 typedef struct HttpContext {

src/plugins/http/src/ehttp_gzip.c

+#include "ehttp_gzip.h"
+
+#include "os.h"
+#include "zlib.h"
+
+#include <stdlib.h>
+
+typedef enum {
+  EHTTP_GZIP_INITING,
+  EHTTP_GZIP_READY,
+  EHTTP_GZIP_CLOSED,
+} EHTTP_GZIP_STATE;
+
+struct ehttp_gzip_s {
+  ehttp_gzip_conf_t       conf;
+  ehttp_gzip_callbacks_t  callbacks;
+  void                   *arg;
+  z_stream               *gzip;
+  gz_header              *header;
+  char                   *chunk;
+
+  int                     state;
+};
+
+static void dummy_on_data(ehttp_gzip_t *gzip, void *arg, const char *buf, size_t len) {
+}
+
+static void ehttp_gzip_cleanup(ehttp_gzip_t *gzip) {
+  switch(gzip->state) {
+    case EHTTP_GZIP_READY: {
+      inflateEnd(gzip->gzip);
+    } break;
+    default: break;
+  }
+  if (gzip->gzip) {
+    free(gzip->gzip);
+    gzip->gzip = NULL;
+  }
+  if (gzip->header) {
+    free(gzip->header);
+    gzip->header = NULL;
+  }
+  if (gzip->chunk) {
+    free(gzip->chunk);
+    gzip->chunk = NULL;
+  }
+  gzip->state = EHTTP_GZIP_CLOSED;
+}
+
+ehttp_gzip_t* ehttp_gzip_create_decompressor(ehttp_gzip_conf_t conf, ehttp_gzip_callbacks_t callbacks, void *arg) {
+  ehttp_gzip_t *gzip = (ehttp_gzip_t*)calloc(1, sizeof(*gzip));
+  if (!gzip) return NULL;
+
+  do {
+    gzip->conf                = conf;
+    gzip->callbacks           = callbacks;
+    gzip->arg                 = arg;
+    if (gzip->callbacks.on_data == NULL) gzip->callbacks.on_data = dummy_on_data;
+    gzip->gzip                = (z_stream*)calloc(1, sizeof(*gzip->gzip));
+    if (gzip->conf.get_header) {
+      gzip->header            = (gz_header*)calloc(1, sizeof(*gzip->header));
+    }
+    if (gzip->conf.chunk_size<=0) gzip->conf.chunk_size = EHTTP_GZIP_CHUNK_SIZE_DEFAULT;
+    gzip->chunk                 = (char*)malloc(gzip->conf.chunk_size);
+    if (!gzip->gzip || (gzip->conf.get_header && !gzip->header) || !gzip->chunk) break;
+    gzip->gzip->zalloc      = Z_NULL;
+    gzip->gzip->zfree       = Z_NULL;
+    gzip->gzip->opaque      = Z_NULL;
+
+  // 863      windowBits can also be greater than 15 for optional gzip decoding.  Add
+  // 864    32 to windowBits to enable zlib and gzip decoding with automatic header
+  // 865    detection, or add 16 to decode only the gzip format (the zlib format will
+  // 866    return a Z_DATA_ERROR).  If a gzip stream is being decoded, strm->adler is a
+  // 867    CRC-32 instead of an Adler-32.  Unlike the gunzip utility and gzread() (see
+  // 868    below), inflate() will not automatically decode concatenated gzip streams.
+  // 869    inflate() will return Z_STREAM_END at the end of the gzip stream.  The state
+  // 870    would need to be reset to continue decoding a subsequent gzip stream.
+    int ret = inflateInit2(gzip->gzip, 32); // 32/16? 32/16 + MAX_WBITS
+    if (ret != Z_OK) break;
+    if (gzip->header) {
+      ret = inflateGetHeader(gzip->gzip, gzip->header);
+    }
+    if (ret != Z_OK) break;
+
+    gzip->gzip->next_out           = (z_const Bytef*)gzip->chunk;
+    gzip->gzip->avail_out          = gzip->conf.chunk_size;
+    gzip->state = EHTTP_GZIP_READY;
+    return gzip;
+  } while (0);
+
+  ehttp_gzip_destroy(gzip);
+  return NULL;
+}
+
+ehttp_gzip_t* ehttp_gzip_create_compressor(ehttp_gzip_conf_t conf, ehttp_gzip_callbacks_t callbacks, void *arg);
+
+void ehttp_gzip_destroy(ehttp_gzip_t *gzip) {
+  ehttp_gzip_cleanup(gzip);
+
+  free(gzip);
+}
+
+int ehttp_gzip_write(ehttp_gzip_t *gzip, const char *buf, size_t len) {
+  if (gzip->state != EHTTP_GZIP_READY) return -1;
+  if (len <= 0) return 0;
+
+  gzip->gzip->next_in        = (z_const Bytef*)buf;
+  gzip->gzip->avail_in       = len;
+
+  while (gzip->gzip->avail_in) {
+    int ret;
+    if (gzip->header) {
+      ret = inflate(gzip->gzip, Z_BLOCK);
+    } else {
+      ret = inflate(gzip->gzip, Z_SYNC_FLUSH);
+    }
+    if (ret != Z_OK && ret != Z_STREAM_END) return -1;
+
+    if (gzip->gzip->avail_out>0) {
+      if (ret!=Z_STREAM_END) continue;
+    }
+
+    size_t len = gzip->gzip->next_out - (z_const Bytef*)gzip->chunk;
+
+    gzip->gzip->next_out[0] = '\0';
+    gzip->callbacks.on_data(gzip, gzip->arg, gzip->chunk, len);
+    gzip->gzip->next_out       = (z_const Bytef*)gzip->chunk;
+    gzip->gzip->avail_out      = gzip->conf.chunk_size;
+  }
+
+  return 0;
+}
+
+int ehttp_gzip_finish(ehttp_gzip_t *gzip) {
+  if (gzip->state != EHTTP_GZIP_READY) return -1;
+
+  gzip->gzip->next_in        = NULL;
+  gzip->gzip->avail_in       = 0;
+
+  int ret;
+  ret = inflate(gzip->gzip, Z_FINISH);
+
+  if (ret != Z_STREAM_END) return -1;
+
+  size_t len = gzip->gzip->next_out - (z_const Bytef*)gzip->chunk;
+
+  gzip->gzip->next_out[0] = '\0';
+  gzip->callbacks.on_data(gzip, gzip->arg, gzip->chunk, len);
+  gzip->gzip->next_out       = NULL;
+  gzip->gzip->avail_out      = 0;
+
+  return 0;
+}
+

src/plugins/http/src/ehttp_util_string.c

+#include "ehttp_util_string.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+void ehttp_util_string_cleanup(ehttp_util_string_t *str) {
+  free(str->str);
+  str->str = NULL;
+  str->len = 0;
+}
+
+int ehttp_util_string_append(ehttp_util_string_t *str, const char *s, size_t len) {
+  // int   n     = str->str?strlen(str->str):0;
+  int   n     = str->len;
+  char *p     = (char*)realloc(str->str, n + len + 1);
+  if (!p) return -1;
+  strncpy(p+n, s, len);
+  p[n+len]    = '\0';
+  str->str    = p;
+  str->len    = n+len;
+  return 0;
+}
+
+void ehttp_util_string_clear(ehttp_util_string_t *str) {
+  if (str->str) {
+    str->str[0] = '\0';
+    str->len    = 0;
+  }
+}
+

src/plugins/http/src/httpContext.c

@@ -28,6 +28,22 @@
 #include "httpSql.h"
 #include "httpSession.h"
 
+#include "elog.h"
+
+// dirty tweak
+extern bool httpGetHttpMethod(HttpContext* pContext);
+extern bool httpParseURL(HttpContext* pContext);
+extern bool httpParseHttpVersion(HttpContext* pContext);
+extern bool httpGetDecodeMethod(HttpContext* pContext);
+extern bool httpParseHead(HttpContext* pContext);
+
+static void on_request_line(void *arg, const char *method, const char *target, const char *version, const char *target_raw);
+static void on_status_line(void *arg, const char *version, int status_code, const char *reason_phrase);
+static void on_header_field(void *arg, const char *key, const char *val);
+static void on_body(void *arg, const char *chunk, size_t len);
+static void on_end(void *arg);
+static void on_error(void *arg, int status_code);
+
 static void httpRemoveContextFromEpoll(HttpContext *pContext) {
   HttpThread *pThread = pContext->pThread;
   if (pContext->fd >= 0) {
@@ -149,6 +165,11 @@ void httpReleaseContext(HttpContext *pContext) {
     httpDebug("context:%p, won't be destroyed for cache is already released", pContext);
     // httpDestroyContext((void **)(&ppContext));
   }
+
+  if (pContext->parser.parser) {
+    ehttp_parser_destroy(pContext->parser.parser);
+    pContext->parser.parser = NULL;
+  }
 }
 
 bool httpInitContext(HttpContext *pContext) {
@@ -168,6 +189,20 @@ bool httpInitContext(HttpContext *pContext) {
   memset(pParser, 0, sizeof(HttpParser));
   pParser->pCur = pParser->pLast = pParser->buffer;
 
+  ehttp_parser_callbacks_t callbacks = {
+    on_request_line,
+    on_status_line,
+    on_header_field,
+    on_body,
+    on_end,
+    on_error
+  };
+  ehttp_parser_conf_t conf = {
+    .flush_block_size = 0
+  };
+  pParser->parser = ehttp_parser_create(callbacks, conf, pContext);
+  pParser->inited = 1;
+
   httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, parsed:%d",
           pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, pContext->parsed);
   return true;
@@ -230,3 +265,129 @@ void httpCloseContextByServer(HttpContext *pContext) {
   httpRemoveContextFromEpoll(pContext);
   httpReleaseContext(pContext);
 }
+
+
+
+
+
+static void on_request_line(void *arg, const char *method, const char *target, const char *version, const char *target_raw) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  int avail = sizeof(pParser->buffer) - (pParser->pLast - pParser->buffer);
+  int n = snprintf(pParser->pLast, avail,
+                   "%s %s %s\r\n", method, target_raw, version);
+
+  char *last = pParser->pLast;
+
+  do {
+    if (n>=avail) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_request_line(%s,%s,%s,%s), exceeding buffer size",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, method, target, version, target_raw);
+      break;
+    }
+    pParser->bufsize += n;
+
+    if (!httpGetHttpMethod(pContext)) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_request_line(%s,%s,%s,%s), parse http method failed",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, method, target, version, target_raw);
+      break;
+    }
+    if (!httpParseURL(pContext)) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_request_line(%s,%s,%s,%s), parse http url failed",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, method, target, version, target_raw);
+      break;
+    }
+    if (!httpParseHttpVersion(pContext)) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_request_line(%s,%s,%s,%s), parse http version failed",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, method, target, version, target_raw);
+      break;
+    }
+    if (!httpGetDecodeMethod(pContext)) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_request_line(%s,%s,%s,%s), get decode method failed",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, method, target, version, target_raw);
+      break;
+    }
+
+    last              += n;
+    pParser->pLast     = last;
+    return;
+  } while (0);
+
+  pParser->failed |= EHTTP_CONTEXT_PROCESS_FAILED;
+}
+
+static void on_status_line(void *arg, const char *version, int status_code, const char *reason_phrase) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  pParser->failed |= EHTTP_CONTEXT_PROCESS_FAILED;
+}
+
+static void on_header_field(void *arg, const char *key, const char *val) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  if (pParser->failed) return;
+
+  int avail = sizeof(pParser->buffer) - (pParser->pLast - pParser->buffer);
+  int n = snprintf(pParser->pLast, avail,
+                   "%s: %s\r\n", key, val);
+
+  char *last = pParser->pLast;
+
+  do {
+    if (n>=avail) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_header_field(%s,%s), exceeding buffer size",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, key, val);
+      break;
+    }
+    pParser->bufsize += n;
+    pParser->pCur     = pParser->pLast;
+
+    if (!httpParseHead(pContext)) {
+      httpDebug("context:%p, fd:%d, ip:%s, thread:%s, accessTimes:%d, on_header_field(%s,%s), parse head failed",
+              pContext, pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->accessTimes, key, val);
+      break;
+    }
+
+    last              += n;
+    pParser->pLast     = last;
+    return;
+  } while (0);
+
+  pParser->failed |= EHTTP_CONTEXT_PROCESS_FAILED;
+}
+
+static void on_body(void *arg, const char *chunk, size_t len) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  if (pParser->failed) return;
+
+  if (!pContext->parsed) {
+    pContext->parsed = true;
+  }
+
+  A("not implemented yet");
+}
+
+static void on_end(void *arg) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  if (pParser->failed) return;
+
+  if (!pContext->parsed) {
+    pContext->parsed = true;
+  }
+}
+
+static void on_error(void *arg, int status_code) {
+  HttpContext *pContext = (HttpContext*)arg;
+  HttpParser  *pParser  = &pContext->parser;
+
+  D("==");
+  pParser->failed |= EHTTP_CONTEXT_PARSER_FAILED;
+}
+

src/plugins/http/src/httpServer.c

@@ -26,10 +26,14 @@
 #include "httpResp.h"
 #include "httpUtil.h"
 
+#include "elog.h"
+
 #ifndef EPOLLWAKEUP
  #define EPOLLWAKEUP (1u << 29)
 #endif
 
+static bool ehttpReadData(HttpContext *pContext);
+
 static void httpStopThread(HttpThread* pThread) {
   pThread->stop = true;
 
@@ -134,6 +138,8 @@ static bool httpDecompressData(HttpContext *pContext) {
 }
 
 static bool httpReadData(HttpContext *pContext) {
+  if (1) return ehttpReadData(pContext);
+
   if (!pContext->parsed) {
     httpInitContext(pContext);
   }
@@ -405,3 +411,60 @@ bool httpInitConnect() {
             pServer->serverPort, pServer->numOfThreads);
   return true;
 }
+
+
+
+
+static bool ehttpReadData(HttpContext *pContext) {
+  HttpParser *pParser = &pContext->parser;
+  EQ_ASSERT(!pContext->parsed);
+  if (!pParser->parser) {
+    if (!pParser->inited) {
+      httpInitContext(pContext);
+    }
+    if (!pParser->parser) {
+      return false;
+    }
+  }
+
+  pContext->accessTimes++;
+  pContext->lastAccessTime = taosGetTimestampSec();
+
+  char buf[HTTP_STEP_SIZE+1] = {0};
+  int nread = (int)taosReadSocket(pContext->fd, buf, sizeof(buf));
+  if (nread > 0) {
+    buf[nread] = '\0';
+    if (strstr(buf, "GET ")==buf && !strchr(buf, '\r') && !strchr(buf, '\n')) {
+      D("==half of request line received:\n%s\n==", buf);
+    }
+    if (ehttp_parser_parse(pParser->parser, buf, nread)) {
+      D("==parsing failed==");
+      httpCloseContextByServer(pContext);
+      return false;
+    }
+    if (pContext->parser.failed) {
+      D("==parsing failed: [0x%x]==", pContext->parser.failed);
+      httpNotifyContextClose(pContext);
+      return false;
+    }
+    return pContext->parsed;
+  } else if (nread < 0) {
+    if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
+      httpDebug("context:%p, fd:%d, ip:%s, read from socket error:%d, wait another event",
+                pContext, pContext->fd, pContext->ipstr, errno);
+      return false; // later again
+    } else {
+      httpError("context:%p, fd:%d, ip:%s, read from socket error:%d, close connect",
+                pContext, pContext->fd, pContext->ipstr, errno);
+      D("==releasing because of reading failed==");
+      httpReleaseContext(pContext);
+      return false;
+    }
+  } else {
+    // eof
+    D("==releasing because of remote close/reset==");
+    httpReleaseContext(pContext);
+    return false;
+  }
+}
+