From fa7e9bd6d262afbf6547dcd49e4db4170c2e2ed7 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 22 Jan 2021 11:59:14 +0800 Subject: [PATCH] TD-1207 --- src/plugins/http/src/httpAuth.c | 8 +-- src/plugins/http/src/httpContext.c | 11 ++- src/plugins/http/src/httpGcHandle.c | 2 +- src/plugins/http/src/httpGcJson.c | 27 ++++---- src/plugins/http/src/httpGzip.c | 92 +++++++++++++------------ src/plugins/http/src/httpHandle.c | 2 +- src/plugins/http/src/httpJson.c | 47 ++++++------- src/plugins/http/src/httpParser.c | 96 +++++++++++++++------------ src/plugins/http/src/httpQueue.c | 14 ++-- src/plugins/http/src/httpResp.c | 5 +- src/plugins/http/src/httpRestHandle.c | 1 - src/plugins/http/src/httpRestJson.c | 23 ++++--- src/plugins/http/src/httpServer.c | 34 +++++----- src/plugins/http/src/httpSession.c | 5 +- src/plugins/http/src/httpSql.c | 10 +-- src/plugins/http/src/httpSystem.c | 6 +- src/plugins/http/src/httpTgHandle.c | 45 +++++++------ src/plugins/http/src/httpUtil.c | 55 +++++++-------- tests/script/general/http/restful.sim | 2 - 19 files changed, 247 insertions(+), 238 deletions(-) diff --git a/src/plugins/http/src/httpAuth.c b/src/plugins/http/src/httpAuth.c index 973f69f24e..2ef5406823 100644 --- a/src/plugins/http/src/httpAuth.c +++ b/src/plugins/http/src/httpAuth.c @@ -26,7 +26,7 @@ int32_t httpParseBasicAuthToken(HttpContext *pContext, char *token, int32_t len) { token[len] = '\0'; int32_t outlen = 0; - char *base64 = (char *)base64_decode(token, len, &outlen); + char * base64 = (char *)base64_decode(token, len, &outlen); if (base64 == NULL || outlen == 0) { httpError("context:%p, fd:%d, basic token:%s parsed error", pContext, pContext->fd, token); free(base64); @@ -49,7 +49,7 @@ int32_t httpParseBasicAuthToken(HttpContext *pContext, char *token, int32_t len) strncpy(pContext->user, base64, (size_t)user_len); pContext->user[user_len] = 0; - char *password = user + 1; + char * password = user + 1; int32_t pass_len = (int32_t)((base64 + outlen) - password); if (pass_len < 1 || pass_len >= HTTP_PASSWORD_LEN) { httpError("context:%p, fd:%d, basic token:%s parse password error", pContext, pContext->fd, token); @@ -66,7 +66,7 @@ int32_t httpParseBasicAuthToken(HttpContext *pContext, char *token, int32_t len) int32_t httpParseTaosdAuthToken(HttpContext *pContext, char *token, int32_t len) { token[len] = '\0'; - int32_t outlen = 0; + int32_t outlen = 0; unsigned char *base64 = base64_decode(token, len, &outlen); if (base64 == NULL || outlen == 0) { httpError("context:%p, fd:%d, taosd token:%s parsed error", pContext, pContext->fd, token); @@ -97,7 +97,7 @@ int32_t httpParseTaosdAuthToken(HttpContext *pContext, char *token, int32_t len) } int32_t httpGenTaosdAuthToken(HttpContext *pContext, char *token, int32_t maxLen) { - char buffer[sizeof(pContext->user) + sizeof(pContext->pass)] = {0}; + char buffer[sizeof(pContext->user) + sizeof(pContext->pass)] = {0}; size_t size = sizeof(pContext->user); tstrncpy(buffer, pContext->user, size); size = sizeof(pContext->pass); diff --git a/src/plugins/http/src/httpContext.c b/src/plugins/http/src/httpContext.c index 266228e7ac..db77c479c0 100644 --- a/src/plugins/http/src/httpContext.c +++ b/src/plugins/http/src/httpContext.c @@ -48,7 +48,7 @@ static void httpDestroyContext(void *data) { httpRemoveContextFromEpoll(pContext); httpReleaseSession(pContext); atomic_sub_fetch_32(&pThread->numOfContexts, 1); - + httpDebug("context:%p, is destroyed, refCount:%d data:%p thread:%s numOfContexts:%d", pContext, pContext->refCount, data, pContext->pThread->label, pContext->pThread->numOfContexts); pContext->pThread = 0; @@ -100,9 +100,7 @@ const char *httpContextStateStr(HttpContextState state) { } } -void httpNotifyContextClose(HttpContext *pContext) { - shutdown(pContext->fd, SHUT_WR); -} +void httpNotifyContextClose(HttpContext *pContext) { shutdown(pContext->fd, SHUT_WR); } bool httpAlterContextState(HttpContext *pContext, HttpContextState srcState, HttpContextState destState) { return (atomic_val_compare_exchange_32(&pContext->state, srcState, destState) == srcState); @@ -123,8 +121,8 @@ HttpContext *httpCreateContext(int32_t fd) { pContext->ppContext = ppContext; httpDebug("context:%p, fd:%d, is created, data:%p", pContext, fd, ppContext); - // set the ref to 0 - taosCacheRelease(tsHttpServer.contextCache, (void**)&ppContext, false); + // set the ref to 0 + taosCacheRelease(tsHttpServer.contextCache, (void **)&ppContext, false); return pContext; } @@ -174,7 +172,6 @@ bool httpInitContext(HttpContext *pContext) { pContext->encodeMethod = NULL; memset(&pContext->singleCmd, 0, sizeof(HttpSqlCmd)); - httpTrace("context:%p, fd:%d, parsed:%d", pContext, pContext->fd, pContext->parsed); return true; } diff --git a/src/plugins/http/src/httpGcHandle.c b/src/plugins/http/src/httpGcHandle.c index 4c10249672..925c74e7cd 100644 --- a/src/plugins/http/src/httpGcHandle.c +++ b/src/plugins/http/src/httpGcHandle.c @@ -192,7 +192,7 @@ bool gcProcessQueryRequest(HttpContext* pContext) { break; } - cJSON* alias = cJSON_GetObjectItem(query, "alias"); + cJSON* alias = cJSON_GetObjectItem(query, "alias"); int32_t aliasBuffer = -1; if (!(alias == NULL || alias->valuestring == NULL || strlen(alias->valuestring) == 0)) { aliasBuffer = httpAddToSqlCmdBuffer(pContext, alias->valuestring); diff --git a/src/plugins/http/src/httpGcJson.c b/src/plugins/http/src/httpGcJson.c index 3a053afd87..8c223a1500 100644 --- a/src/plugins/http/src/httpGcJson.c +++ b/src/plugins/http/src/httpGcJson.c @@ -86,7 +86,7 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, JsonBuf *jsonBuf = httpMallocJsonBuf(pContext); if (jsonBuf == NULL) return false; - int32_t num_fields = taos_num_fields(result); + int32_t num_fields = taos_num_fields(result); TAOS_FIELD *fields = taos_fetch_fields(result); if (num_fields == 0) { return false; @@ -101,7 +101,7 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, // such as select count(*) count(*) from sys.cpu group by ipaddr interval(1d) int32_t dataFields = -1; int32_t groupFields = -1; - bool hasTimestamp = fields[0].type == TSDB_DATA_TYPE_TIMESTAMP; + bool hasTimestamp = fields[0].type == TSDB_DATA_TYPE_TIMESTAMP; if (hasTimestamp) { dataFields = 1; if (num_fields > 2) groupFields = num_fields - 1; @@ -125,15 +125,15 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, cmd->numOfRows--; continue; } - int32_t* length = taos_fetch_lengths(result); + int32_t *length = taos_fetch_lengths(result); // for group by if (groupFields != -1) { - char target[HTTP_GC_TARGET_SIZE] = {0}; + char target[HTTP_GC_TARGET_SIZE] = {0}; int32_t len; - len = snprintf(target,HTTP_GC_TARGET_SIZE,"%s{",aliasBuffer); - for (int32_t i = dataFields + 1; istate) { + switch (gzip->state) { case EHTTP_GZIP_READY: { inflateEnd(gzip->gzip); } break; - default: break; + default: + break; } if (gzip->gzip) { free(gzip->gzip); @@ -60,43 +59,43 @@ static void ehttp_gzip_cleanup(ehttp_gzip_t *gzip) { 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)); +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; + 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)); + gzip->gzip = (z_stream *)calloc(1, sizeof(*gzip->gzip)); if (gzip->conf.get_header) { - gzip->header = (gz_header*)calloc(1, sizeof(*gzip->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->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; + 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. - int32_t ret = inflateInit2(gzip->gzip, 32); // 32/16? 32/16 + MAX_WBITS + // 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. + int32_t 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->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); @@ -105,7 +104,7 @@ ehttp_gzip_t* ehttp_gzip_create_decompressor(ehttp_gzip_conf_t conf, ehttp_gzip_ return NULL; } -ehttp_gzip_t* ehttp_gzip_create_compressor(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) { ehttp_gzip_cleanup(gzip); @@ -129,16 +128,16 @@ int32_t ehttp_gzip_write(ehttp_gzip_t *gzip, const char *buf, int32_t len) { } if (ret != Z_OK && ret != Z_STREAM_END) return -1; - if (gzip->gzip->avail_out>0) { - if (ret!=Z_STREAM_END) continue; + if (gzip->gzip->avail_out > 0) { + if (ret != Z_STREAM_END) continue; } - int32_t len = (int32_t)(gzip->gzip->next_out - (z_const Bytef*)gzip->chunk); + int32_t len = (int32_t)(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; + gzip->gzip->next_out = (z_const Bytef *)gzip->chunk; + gzip->gzip->avail_out = gzip->conf.chunk_size; } return 0; @@ -147,21 +146,20 @@ int32_t ehttp_gzip_write(ehttp_gzip_t *gzip, const char *buf, int32_t len) { int32_t 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; + gzip->gzip->next_in = NULL; + gzip->gzip->avail_in = 0; int32_t ret; ret = inflate(gzip->gzip, Z_FINISH); if (ret != Z_STREAM_END) return -1; - int32_t len = (int32_t)(gzip->gzip->next_out - (z_const Bytef*)gzip->chunk); + int32_t len = (int32_t)(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; + gzip->gzip->next_out = NULL; + gzip->gzip->avail_out = 0; return 0; } - diff --git a/src/plugins/http/src/httpHandle.c b/src/plugins/http/src/httpHandle.c index 7c56507514..ad79e24061 100644 --- a/src/plugins/http/src/httpHandle.c +++ b/src/plugins/http/src/httpHandle.c @@ -48,7 +48,7 @@ bool httpProcessData(HttpContext* pContext) { /* * httpCloseContextByApp has been called when parsing the error */ - //httpCloseContextByApp(pContext); + // httpCloseContextByApp(pContext); } else { httpProcessRequest(pContext); } diff --git a/src/plugins/http/src/httpJson.c b/src/plugins/http/src/httpJson.c index 132553ffe1..b120496898 100644 --- a/src/plugins/http/src/httpJson.c +++ b/src/plugins/http/src/httpJson.c @@ -44,20 +44,21 @@ int32_t httpWriteBufByFd(struct HttpContext* pContext, const char* buf, int32_t int32_t writeLen = 0; do { - if (pContext->fd > 2){ + if (pContext->fd > 2) { len = (int32_t)taosSend(pContext->fd, buf + writeLen, (size_t)(sz - writeLen), MSG_NOSIGNAL); - } - else { + } else { return sz; } if (len < 0) { - httpDebug("context:%p, fd:%d, socket write errno:%d:%s, times:%d", pContext, pContext->fd, errno, strerror(errno), countWait); + httpDebug("context:%p, fd:%d, socket write errno:%d:%s, times:%d", pContext, pContext->fd, errno, strerror(errno), + countWait); if (++countWait > HTTP_WRITE_RETRY_TIMES) break; taosMsleep(HTTP_WRITE_WAIT_TIME_MS); continue; } else if (len == 0) { - httpDebug("context:%p, fd:%d, socket write errno:%d:%s, connect already closed", pContext, pContext->fd, errno, strerror(errno)); + httpDebug("context:%p, fd:%d, socket write errno:%d:%s, connect already closed", pContext, pContext->fd, errno, + strerror(errno)); break; } else { countWait = 0; @@ -80,7 +81,7 @@ int32_t httpWriteBuf(struct HttpContext* pContext, const char* buf, int32_t sz) return writeSz; } -int32_t httpWriteBufNoTrace(struct HttpContext *pContext, const char *buf, int32_t sz) { +int32_t httpWriteBufNoTrace(struct HttpContext* pContext, const char* buf, int32_t sz) { int32_t writeSz = httpWriteBufByFd(pContext, buf, sz); if (writeSz != sz) { httpError("context:%p, fd:%d, dataSize:%d, writeSize:%d, failed to send response", pContext, pContext->fd, sz, @@ -92,8 +93,8 @@ int32_t httpWriteBufNoTrace(struct HttpContext *pContext, const char *buf, int32 int32_t httpWriteJsonBufBody(JsonBuf* buf, bool isTheLast) { int32_t remain = 0; - char sLen[24]; - int32_t srcLen = (int32_t) (buf->lst - buf->buf); + char sLen[24]; + int32_t srcLen = (int32_t)(buf->lst - buf->buf); if (buf->pContext->fd <= 0) { httpTrace("context:%p, fd:%d, write json body error", buf->pContext, buf->pContext->fd); @@ -113,21 +114,21 @@ int32_t httpWriteJsonBufBody(JsonBuf* buf, bool isTheLast) { httpTrace("context:%p, fd:%d, no data need dump", buf->pContext, buf->pContext->fd); return 0; // there is no data to dump. } else { - int32_t len = sprintf(sLen, "%d\r\n", srcLen); - httpTrace("context:%p, fd:%d, write body, chunkSize:%d, response:\n%s", buf->pContext, buf->pContext->fd, - srcLen, buf->buf); + int32_t len = sprintf(sLen, "%x\r\n", srcLen); + httpTrace("context:%p, fd:%d, write body, chunkSize:%d, response:\n%s", buf->pContext, buf->pContext->fd, srcLen, + buf->buf); httpWriteBufNoTrace(buf->pContext, sLen, len); remain = httpWriteBufNoTrace(buf->pContext, buf->buf, srcLen); } } else { - char compressBuf[JSON_BUFFER_SIZE] = {0}; + char compressBuf[JSON_BUFFER_SIZE] = {0}; int32_t compressBufLen = JSON_BUFFER_SIZE; int32_t ret = httpGzipCompress(buf->pContext, buf->buf, srcLen, compressBuf, &compressBufLen, isTheLast); if (ret == 0) { if (compressBufLen > 0) { int32_t len = sprintf(sLen, "%x\r\n", compressBufLen); - httpTrace("context:%p, fd:%d, write body, chunkSize:%d, compressSize:%d, last:%d, response:\n%s", - buf->pContext, buf->pContext->fd, srcLen, compressBufLen, isTheLast, buf->buf); + httpTrace("context:%p, fd:%d, write body, chunkSize:%d, compressSize:%d, last:%d, response:\n%s", buf->pContext, + buf->pContext->fd, srcLen, compressBufLen, isTheLast, buf->buf); httpWriteBufNoTrace(buf->pContext, sLen, len); remain = httpWriteBufNoTrace(buf->pContext, (const char*)compressBuf, compressBufLen); } else { @@ -154,8 +155,8 @@ void httpWriteJsonBufHead(JsonBuf* buf) { buf->pContext->fd = -1; } - char msg[1024] = {0}; - int32_t len = -1; + char msg[1024] = {0}; + int32_t len = -1; if (buf->pContext->parser->acceptEncodingGzip == 0 || !tsHttpEnableCompress) { len = sprintf(msg, httpRespTemplate[HTTP_RESPONSE_CHUNKED_UN_COMPRESS], httpVersionStr[buf->pContext->parser->httpVersion], @@ -256,16 +257,16 @@ void httpJsonInt64(JsonBuf* buf, int64_t num) { } void httpJsonTimestamp(JsonBuf* buf, int64_t t, bool us) { - char ts[35] = {0}; - struct tm *ptm; - int32_t precision = 1000; + char ts[35] = {0}; + struct tm* ptm; + int32_t precision = 1000; if (us) { precision = 1000000; } time_t tt = t / precision; ptm = localtime(&tt); - int32_t length = (int32_t) strftime(ts, 35, "%Y-%m-%d %H:%M:%S", ptm); + int32_t length = (int32_t)strftime(ts, 35, "%Y-%m-%d %H:%M:%S", ptm); if (us) { length += snprintf(ts + length, 8, ".%06" PRId64, t % precision); } else { @@ -276,9 +277,9 @@ void httpJsonTimestamp(JsonBuf* buf, int64_t t, bool us) { } void httpJsonUtcTimestamp(JsonBuf* buf, int64_t t, bool us) { - char ts[40] = {0}; - struct tm *ptm; - int32_t precision = 1000; + char ts[40] = {0}; + struct tm* ptm; + int32_t precision = 1000; if (us) { precision = 1000000; } diff --git a/src/plugins/http/src/httpParser.c b/src/plugins/http/src/httpParser.c index 331ca079fa..5ebe9df96f 100644 --- a/src/plugins/http/src/httpParser.c +++ b/src/plugins/http/src/httpParser.c @@ -130,7 +130,7 @@ static int32_t httpAppendString(HttpString *str, const char *s, int32_t len) { static void httpClearString(HttpString *str) { if (str->str) { str->str[0] = '\0'; - str->pos = 0; + str->pos = 0; } } @@ -237,7 +237,6 @@ static int32_t httpOnParseHeaderField(HttpParser *parser, const char *key, const } httpTrace("context:%p, fd:%d, keepAlive:%d", pContext, pContext->fd, pContext->parser->keepAlive); } - #if 0 else if (0 == strcasecmp(key, "Content-Encoding")) { if (0 == strcmp(val, "gzip")) { @@ -246,7 +245,7 @@ static int32_t httpOnParseHeaderField(HttpParser *parser, const char *key, const } return 0; } - #endif +#endif else if (0 == strcasecmp(key, "Transfer-Encoding") || 0 == strcasecmp(key, "Content-Encoding")) { if (strstr(val, "gzip")) { @@ -349,7 +348,7 @@ static int32_t httpOnBody(HttpParser *parser, const char *chunk, int32_t len) { newSize = MIN(newSize, HTTP_BUFFER_SIZE); buf->str = realloc(buf->str, newSize); buf->size = newSize; - + if (buf->str == NULL) { httpError("context:%p, fd:%d, failed parse body, realloc %d failed", pContext, pContext->fd, buf->size); httpOnError(parser, 0, TSDB_CODE_HTTP_NO_ENOUGH_MEMORY); @@ -410,9 +409,7 @@ static int32_t httpPopStack(HttpParser *parser) { return 0; } -static void httpClearStack(HttpStack *stack) { - stack->pos = 0; -} +static void httpClearStack(HttpStack *stack) { stack->pos = 0; } static int32_t httpCleanupStack(HttpStack *stack) { free(stack->stacks); @@ -451,7 +448,7 @@ void httpInitParser(HttpParser *parser) { free(parser->key); parser->key = NULL; free(parser->val); parser->val = NULL; free(parser->authContent); parser->authContent = NULL; - + httpClearStack(&parser->stacks); httpClearString(&parser->str); httpClearString(&parser->body); @@ -497,7 +494,7 @@ void httpDestroyParser(HttpParser *parser) { free(parser->key); parser->key = NULL; free(parser->val); parser->val = NULL; free(parser->authContent); parser->authContent = NULL; - + httpCleanupStack(&parser->stacks); httpCleanupString(&parser->str); httpCleanupString(&parser->body); @@ -513,25 +510,36 @@ void httpDestroyParser(HttpParser *parser) { free(parser); } -#define is_token(c) (strchr("!#$%&'*+-.^_`|~", c) || isdigit(c) || isalpha(c)) +#define is_token(c) (strchr("!#$%&'*+-.^_`|~", c) || isdigit(c) || isalpha(c)) char *httpDecodeUrl(const char *enc) { - int32_t ok = 1; + int32_t ok = 1; HttpString str = {0}; while (*enc) { char *p = strchr(enc, '%'); if (!p) break; int32_t hex, cnt; - int32_t n = sscanf(p+1, "%2x%n", &hex, &cnt); - if (n!=1 && cnt !=2) { ok = 0; break; } - if (httpAppendString(&str, enc, (int32_t)(p-enc))) { ok = 0; break; } + int32_t n = sscanf(p + 1, "%2x%n", &hex, &cnt); + if (n != 1 && cnt != 2) { + ok = 0; + break; + } + if (httpAppendString(&str, enc, (int32_t)(p - enc))) { + ok = 0; + break; + } char c = (char)hex; - if (httpAppendString(&str, &c, 1)) { ok = 0; break; } - enc = p+3; + if (httpAppendString(&str, &c, 1)) { + ok = 0; + break; + } + enc = p + 3; } char *dec = NULL; if (ok && *enc) { - if (httpAppendString(&str, enc, (int32_t)strlen(enc))) { ok = 0; } + if (httpAppendString(&str, enc, (int32_t)strlen(enc))) { + ok = 0; + } } if (ok) { dec = str.str; @@ -542,13 +550,13 @@ char *httpDecodeUrl(const char *enc) { } static void httpOnData(ehttp_gzip_t *gzip, void *arg, const char *buf, int32_t len) { - HttpParser *parser = (HttpParser*)arg; + HttpParser *parser = (HttpParser *)arg; httpOnBody(parser, buf, len); } static int32_t httpParserOnBegin(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (c == 'G' || c == 'P' || c == 'H' || c == 'D' || c == 'C' || c == 'O' || c == 'T') { if (httpAppendString(&parser->str, &c, 1)) { @@ -570,7 +578,7 @@ static int32_t httpParserOnBegin(HttpParser *parser, HTTP_PARSER_STATE state, co static int32_t httpParserOnRquestOrResponse(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (parser->str.pos == 1) { if (c == 'T' && parser->str.str[0] == 'H') { @@ -608,7 +616,7 @@ static int32_t httpParserOnRquestOrResponse(HttpParser *parser, HTTP_PARSER_STAT static int32_t httpParserOnMethod(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (isalnum(c) || strchr("!#$%&'*+-.^_`|~", c)) { if (httpAppendString(&parser->str, &c, 1)) { @@ -637,7 +645,7 @@ static int32_t httpParserOnMethod(HttpParser *parser, HTTP_PARSER_STATE state, c static int32_t httpParserOnTarget(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (!isspace(c) && c != '\r' && c != '\n') { if (httpAppendString(&parser->str, &c, 1)) { @@ -664,7 +672,7 @@ static int32_t httpParserOnTarget(HttpParser *parser, HTTP_PARSER_STATE state, c static int32_t httpParserOnVersion(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { const char *prefix = "HTTP/1."; int32_t len = (int32_t)strlen(prefix); @@ -727,7 +735,7 @@ static int32_t httpParserOnVersion(HttpParser *parser, HTTP_PARSER_STATE state, static int32_t httpParserOnSp(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (c == ' ') { httpPopStack(parser); @@ -742,7 +750,7 @@ static int32_t httpParserOnSp(HttpParser *parser, HTTP_PARSER_STATE state, const static int32_t httpParserOnStatusCode(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (isdigit(c)) { if (httpAppendString(&parser->str, &c, 1)) { @@ -767,7 +775,7 @@ static int32_t httpParserOnStatusCode(HttpParser *parser, HTTP_PARSER_STATE stat static int32_t httpParserOnReasonPhrase(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (c == '\r') { parser->reasonPhrase = strdup(parser->str.str); @@ -808,10 +816,10 @@ static int32_t httpParserPostProcess(HttpParser *parser) { static int32_t httpParserOnCrlf(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { - const char *s = "\r\n"; - int32_t len = (int32_t)strlen(s); + const char *s = "\r\n"; + int32_t len = (int32_t)strlen(s); if (s[parser->str.pos] != c) { httpError("context:%p, fd:%d, parser state:%d, unexpected char:[%c]%02x", pContext, pContext->fd, state, c, c); ok = -1; @@ -838,7 +846,7 @@ static int32_t httpParserOnCrlf(HttpParser *parser, HTTP_PARSER_STATE state, con static int32_t httpParserOnHeader(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (c == '\r') { httpPopStack(parser); @@ -876,7 +884,7 @@ static int32_t httpParserOnHeader(HttpParser *parser, HTTP_PARSER_STATE state, c static int32_t httpParserOnHeaderKey(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (isalnum(c) || strchr("!#$%&'*+-.^_`|~", c)) { if (httpAppendString(&parser->str, &c, 1)) { @@ -888,7 +896,7 @@ static int32_t httpParserOnHeaderKey(HttpParser *parser, HTTP_PARSER_STATE state break; } if (c == ':') { - parser->key = strdup(parser->str.str); + parser->key = strdup(parser->str.str); if (!parser->key) { httpError("context:%p, fd:%d, parser state:%d, char:[%c]%02x, oom", pContext, pContext->fd, state, c, c); ok = -1; @@ -908,7 +916,7 @@ static int32_t httpParserOnHeaderKey(HttpParser *parser, HTTP_PARSER_STATE state static int32_t httpParserOnHeaderVal(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (c != '\r' && c != '\n' && (!isspace(c) || parser->str.pos > 0)) { if (httpAppendString(&parser->str, &c, 1)) { @@ -935,10 +943,10 @@ static int32_t httpParserOnHeaderVal(HttpParser *parser, HTTP_PARSER_STATE state static int32_t httpParserOnChunkSize(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; - int32_t bytes; - int32_t len; - int32_t n; + int32_t ok = 0; + int32_t bytes; + int32_t len; + int32_t n; do { if (isxdigit(c)) { if (httpAppendString(&parser->str, &c, 1)) { @@ -985,7 +993,7 @@ static int32_t httpParserOnChunkSize(HttpParser *parser, HTTP_PARSER_STATE state static int32_t httpParserOnChunk(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { if (httpAppendString(&parser->str, &c, 1)) { httpError("context:%p, fd:%d, parser state:%d, char:[%c]%02x, oom", pContext, pContext->fd, state, c, c); @@ -1019,7 +1027,7 @@ static int32_t httpParserOnChunk(HttpParser *parser, HTTP_PARSER_STATE state, co static int32_t httpParserOnEnd(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) { HttpContext *pContext = parser->pContext; - int32_t ok = 0; + int32_t ok = 0; do { ok = -1; httpError("context:%p, fd:%d, parser state:%d, unexpected char:[%c]%02x", pContext, pContext->fd, state, c, c); @@ -1029,8 +1037,8 @@ static int32_t httpParserOnEnd(HttpParser *parser, HTTP_PARSER_STATE state, cons } static int32_t httpParseChar(HttpParser *parser, const char c, int32_t *again) { - HttpContext *pContext = parser->pContext; - int32_t ok = 0; + HttpContext * pContext = parser->pContext; + int32_t ok = 0; HTTP_PARSER_STATE state = httpTopStack(parser); do { if (state == HTTP_PARSER_BEGIN) { @@ -1119,9 +1127,9 @@ static int32_t httpParseChar(HttpParser *parser, const char c, int32_t *again) { int32_t httpParseBuf(HttpParser *parser, const char *buf, int32_t len) { HttpContext *pContext = parser->pContext; - const char *p = buf; - int32_t ret = 0; - int32_t i = 0; + const char * p = buf; + int32_t ret = 0; + int32_t i = 0; while (i < len) { int32_t again = 0; diff --git a/src/plugins/http/src/httpQueue.c b/src/plugins/http/src/httpQueue.c index 7ae54717f3..aebba97fb8 100644 --- a/src/plugins/http/src/httpQueue.c +++ b/src/plugins/http/src/httpQueue.c @@ -38,16 +38,16 @@ typedef struct { } SHttpWorkerPool; typedef struct { - void * param; - void * result; - int32_t code; - int32_t rows; + void * param; + void * result; + int32_t code; + int32_t rows; FHttpResultFp fp; } SHttpResult; static SHttpWorkerPool tsHttpPool; -static taos_qset tsHttpQset; -static taos_queue tsHttpQueue; +static taos_qset tsHttpQset; +static taos_queue tsHttpQueue; void httpDispatchToResultQueue(void *param, TAOS_RES *result, int32_t code, int32_t rows, FHttpResultFp fp) { if (tsHttpQueue != NULL) { @@ -105,7 +105,7 @@ static bool httpAllocateResultQueue() { httpDebug("http result worker:%d is launched, total:%d", pWorker->workerId, tsHttpPool.num); } - httpInfo("http result queue is opened"); + httpInfo("http result queue is opened"); return true; } diff --git a/src/plugins/http/src/httpResp.c b/src/plugins/http/src/httpResp.c index 72604e79b7..37eef2bfad 100644 --- a/src/plugins/http/src/httpResp.c +++ b/src/plugins/http/src/httpResp.c @@ -160,8 +160,9 @@ void httpSendTaosdInvalidSqlErrorResp(HttpContext *pContext, char *errMsg) { if (temp[i] == '\"') { temp[i] = '\''; } else if (temp[i] == '\n') { - temp[i] = ' '; - } else {} + temp[i] = ' '; + } else { + } } httpSendErrorRespImp(pContext, httpCode, "Bad Request", TSDB_CODE_TSC_INVALID_SQL & 0XFFFF, temp); diff --git a/src/plugins/http/src/httpRestHandle.c b/src/plugins/http/src/httpRestHandle.c index 112137b3df..8999fb879e 100644 --- a/src/plugins/http/src/httpRestHandle.c +++ b/src/plugins/http/src/httpRestHandle.c @@ -95,7 +95,6 @@ bool restProcessSqlRequest(HttpContext* pContext, int32_t timestampFmt) { return false; } - /* * for async test * diff --git a/src/plugins/http/src/httpRestJson.c b/src/plugins/http/src/httpRestJson.c index a5b156bffc..baa61117be 100644 --- a/src/plugins/http/src/httpRestJson.c +++ b/src/plugins/http/src/httpRestJson.c @@ -83,7 +83,8 @@ void restStartSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result) httpJsonToken(jsonBuf, JsonArrStt); } -bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, int32_t numOfRows, int32_t timestampFormat) { +bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, int32_t numOfRows, + int32_t timestampFormat) { JsonBuf *jsonBuf = httpMallocJsonBuf(pContext); if (jsonBuf == NULL) return false; @@ -95,7 +96,7 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, if (row == NULL) { continue; } - int32_t* length = taos_fetch_lengths(result); + int32_t *length = taos_fetch_lengths(result); // data row array begin httpJsonItemToken(jsonBuf); @@ -131,15 +132,17 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, break; case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_NCHAR: - httpJsonStringForTransMean(jsonBuf, (char*)row[i], length[i]); + httpJsonStringForTransMean(jsonBuf, (char *)row[i], length[i]); break; case TSDB_DATA_TYPE_TIMESTAMP: if (timestampFormat == REST_TIMESTAMP_FMT_LOCAL_STRING) { - httpJsonTimestamp(jsonBuf, *((int64_t *)row[i]), taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO); + httpJsonTimestamp(jsonBuf, *((int64_t *)row[i]), + taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO); } else if (timestampFormat == REST_TIMESTAMP_FMT_TIMESTAMP) { httpJsonInt64(jsonBuf, *((int64_t *)row[i])); } else { - httpJsonUtcTimestamp(jsonBuf, *((int64_t *)row[i]), taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO); + httpJsonUtcTimestamp(jsonBuf, *((int64_t *)row[i]), + taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO); } break; default: @@ -148,8 +151,8 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, } // data row array end - httpJsonToken(jsonBuf, JsonArrEnd); - cmd->numOfRows ++; + httpJsonToken(jsonBuf, JsonArrEnd); + cmd->numOfRows++; if (pContext->fd <= 0) { httpError("context:%p, fd:%d, user:%s, conn closed, abort retrieve", pContext, pContext->fd, pContext->user); @@ -168,15 +171,15 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, } bool restBuildSqlTimestampJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, int32_t numOfRows) { - return restBuildSqlJson(pContext,cmd, result, numOfRows, REST_TIMESTAMP_FMT_TIMESTAMP); + return restBuildSqlJson(pContext, cmd, result, numOfRows, REST_TIMESTAMP_FMT_TIMESTAMP); } bool restBuildSqlLocalTimeStringJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, int32_t numOfRows) { - return restBuildSqlJson(pContext,cmd, result, numOfRows, REST_TIMESTAMP_FMT_LOCAL_STRING); + return restBuildSqlJson(pContext, cmd, result, numOfRows, REST_TIMESTAMP_FMT_LOCAL_STRING); } bool restBuildSqlUtcTimeStringJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result, int32_t numOfRows) { - return restBuildSqlJson(pContext,cmd, result, numOfRows, REST_TIMESTAMP_FMT_UTC_STRING); + return restBuildSqlJson(pContext, cmd, result, numOfRows, REST_TIMESTAMP_FMT_UTC_STRING); } void restStopSqlJson(HttpContext *pContext, HttpSqlCmd *cmd) { diff --git a/src/plugins/http/src/httpServer.c b/src/plugins/http/src/httpServer.c index 505b658743..c4a03d44ea 100644 --- a/src/plugins/http/src/httpServer.c +++ b/src/plugins/http/src/httpServer.c @@ -27,19 +27,21 @@ static bool httpReadData(HttpContext *pContext); -static void httpStopThread(HttpThread* pThread) { +static void httpStopThread(HttpThread *pThread) { pThread->stop = true; // signal the thread to stop, try graceful method first, // and use pthread_cancel when failed - struct epoll_event event = { .events = EPOLLIN }; - eventfd_t fd = eventfd(1, 0); + struct epoll_event event = {.events = EPOLLIN}; + eventfd_t fd = eventfd(1, 0); if (fd == -1) { - httpError("%s, failed to create eventfd, will call pthread_cancel instead, which may result in data corruption: %s", pThread->label, strerror(errno)); + httpError("%s, failed to create eventfd, will call pthread_cancel instead, which may result in data corruption: %s", + pThread->label, strerror(errno)); pThread->stop = true; pthread_cancel(pThread->thread); } else if (epoll_ctl(pThread->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) { - httpError("%s, failed to call epoll_ctl, will call pthread_cancel instead, which may result in data corruption: %s", pThread->label, strerror(errno)); + httpError("%s, failed to call epoll_ctl, will call pthread_cancel instead, which may result in data corruption: %s", + pThread->label, strerror(errno)); pthread_cancel(pThread->thread); } @@ -61,7 +63,7 @@ void httpCleanUpConnect() { } for (int32_t i = 0; i < pServer->numOfThreads; ++i) { - HttpThread* pThread = pServer->pThreads + i; + HttpThread *pThread = pServer->pThreads + i; if (pThread != NULL) { httpStopThread(pThread); } @@ -71,8 +73,8 @@ void httpCleanUpConnect() { } static void httpProcessHttpData(void *param) { - HttpServer *pServer = &tsHttpServer; - HttpThread *pThread = (HttpThread *)param; + HttpServer * pServer = &tsHttpServer; + HttpThread * pThread = (HttpThread *)param; HttpContext *pContext; int32_t fdNum; @@ -92,8 +94,8 @@ static void httpProcessHttpData(void *param) { pContext = httpGetContext(events[i].data.ptr); if (pContext == NULL) { httpError("context:%p, is already released, close connect", events[i].data.ptr); - //epoll_ctl(pThread->pollFd, EPOLL_CTL_DEL, events[i].data.fd, NULL); - //taosClose(events[i].data.fd); + // epoll_ctl(pThread->pollFd, EPOLL_CTL_DEL, events[i].data.fd, NULL); + // taosClose(events[i].data.fd); continue; } @@ -200,7 +202,7 @@ static void *httpAcceptHttpConnection(void *arg) { taosCloseSocket(connFd); continue; } -#endif +#endif taosKeepTcpAlive(connFd); taosSetNonblocking(connFd, 1); @@ -210,15 +212,15 @@ static void *httpAcceptHttpConnection(void *arg) { pContext = httpCreateContext(connFd); if (pContext == NULL) { - httpError("fd:%d, ip:%s:%u, no enough resource to allocate http context", connFd, taosInetNtoa(clientAddr.sin_addr), - htons(clientAddr.sin_port)); + httpError("fd:%d, ip:%s:%u, no enough resource to allocate http context", connFd, + taosInetNtoa(clientAddr.sin_addr), htons(clientAddr.sin_port)); taosCloseSocket(connFd); continue; } pContext->pThread = pThread; sprintf(pContext->ipstr, "%s:%u", taosInetNtoa(clientAddr.sin_addr), htons(clientAddr.sin_port)); - + struct epoll_event event; event.events = EPOLLIN | EPOLLPRI | EPOLLWAKEUP | EPOLLERR | EPOLLHUP | EPOLLRDHUP; event.data.ptr = pContext; @@ -276,7 +278,7 @@ bool httpInitConnect() { if (pthread_create(&(pThread->thread), &thattr, (void *)httpProcessHttpData, (void *)(pThread)) != 0) { httpError("http thread:%s, failed to create HTTP process data thread, reason:%s", pThread->label, strerror(errno)); - pthread_mutex_destroy(&(pThread->threadMutex)); + pthread_mutex_destroy(&(pThread->threadMutex)); return false; } pthread_attr_destroy(&thattr); @@ -307,7 +309,7 @@ static bool httpReadData(HttpContext *pContext) { } if (pParser->parsed) { - httpDebug("context:%p, fd:%d, not in ready state, parsed:%d", pContext, pContext->fd, pParser->parsed); + httpDebug("context:%p, fd:%d, not in ready state, parsed:%d", pContext, pContext->fd, pParser->parsed); return false; } diff --git a/src/plugins/http/src/httpSession.c b/src/plugins/http/src/httpSession.c index 35ce0160b2..2e1ee7df2f 100644 --- a/src/plugins/http/src/httpSession.c +++ b/src/plugins/http/src/httpSession.c @@ -35,7 +35,8 @@ void httpCreateSession(HttpContext *pContext, void *taos) { session.refCount = 1; int32_t len = snprintf(session.id, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass); - pContext->session = taosCachePut(server->sessionCache, session.id, len, &session, sizeof(HttpSession), tsHttpSessionExpire * 1000); + pContext->session = + taosCachePut(server->sessionCache, session.id, len, &session, sizeof(HttpSession), tsHttpSessionExpire * 1000); // void *temp = pContext->session; // taosCacheRelease(server->sessionCache, (void **)&temp, false); @@ -56,7 +57,7 @@ static void httpFetchSessionImp(HttpContext *pContext) { HttpServer *server = &tsHttpServer; pthread_mutex_lock(&server->serverMutex); - char sessionId[HTTP_SESSION_ID_LEN]; + char sessionId[HTTP_SESSION_ID_LEN]; int32_t len = snprintf(sessionId, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass); pContext->session = taosCacheAcquireByKey(server->sessionCache, sessionId, len); diff --git a/src/plugins/http/src/httpSql.c b/src/plugins/http/src/httpSql.c index cc8e9e86e3..4e9b54b7bd 100644 --- a/src/plugins/http/src/httpSql.c +++ b/src/plugins/http/src/httpSql.c @@ -64,7 +64,7 @@ void httpProcessMultiSqlRetrieveCallBackImp(void *param, TAOS_RES *result, int32 } taos_free_result(result); - + if (singleCmd->cmdReturnType == HTTP_CMD_RETURN_TYPE_WITH_RETURN && encode->stopJsonFp) { (encode->stopJsonFp)(pContext, singleCmd); } @@ -82,7 +82,7 @@ void httpProcessMultiSqlCallBackImp(void *param, TAOS_RES *result, int32_t code, HttpContext *pContext = (HttpContext *)param; if (pContext == NULL) return; - HttpSqlCmds *multiCmds = pContext->multiCmds; + HttpSqlCmds * multiCmds = pContext->multiCmds; HttpEncodeMethod *encode = pContext->encodeMethod; HttpSqlCmd *singleCmd = multiCmds->cmds + multiCmds->pos; @@ -269,8 +269,8 @@ void httpProcessSingleSqlCallBackImp(void *param, TAOS_RES *result, int32_t code pContext->user, tstrerror(code), pObj, taos_errstr(pObj)); httpSendTaosdInvalidSqlErrorResp(pContext, taos_errstr(pObj)); } else { - httpError("context:%p, fd:%d, user:%s, query error, code:%s, sqlObj:%p", pContext, pContext->fd, - pContext->user, tstrerror(code), pObj); + httpError("context:%p, fd:%d, user:%s, query error, code:%s, sqlObj:%p", pContext, pContext->fd, pContext->user, + tstrerror(code), pObj); httpSendErrorResp(pContext, code); } taos_free_result(result); @@ -381,7 +381,7 @@ void httpExecCmd(HttpContext *pContext) { void httpProcessRequestCb(void *param, TAOS_RES *result, int32_t code) { HttpContext *pContext = param; taos_free_result(result); - + if (pContext == NULL) return; if (code < 0) { diff --git a/src/plugins/http/src/httpSystem.c b/src/plugins/http/src/httpSystem.c index df39c080a5..4721451529 100644 --- a/src/plugins/http/src/httpSystem.c +++ b/src/plugins/http/src/httpSystem.c @@ -110,10 +110,8 @@ void httpCleanUpSystem() { pthread_mutex_destroy(&tsHttpServer.serverMutex); tfree(tsHttpServer.pThreads); tsHttpServer.pThreads = NULL; - + tsHttpServer.status = HTTP_SERVER_CLOSED; } -int32_t httpGetReqCount() { - return atomic_exchange_32(&tsHttpServer.requestNum, 0); -} +int32_t httpGetReqCount() { return atomic_exchange_32(&tsHttpServer.requestNum, 0); } diff --git a/src/plugins/http/src/httpTgHandle.c b/src/plugins/http/src/httpTgHandle.c index 7a26cae97c..c1d006ff5a 100644 --- a/src/plugins/http/src/httpTgHandle.c +++ b/src/plugins/http/src/httpTgHandle.c @@ -137,8 +137,8 @@ void tgInitSchemas(int32_t size) { } void tgParseSchemaMetric(cJSON *metric) { - STgSchema schema = {0}; - bool parsedOk = true; + STgSchema schema = {0}; + bool parsedOk = true; // name cJSON *name = cJSON_GetObjectItem(metric, "name"); @@ -186,7 +186,7 @@ void tgParseSchemaMetric(cJSON *metric) { schema.tbName = calloc(tbnameLen + 1, 1); strcpy(schema.tbName, tbname->valuestring); - // fields + // fields cJSON *fields = cJSON_GetObjectItem(metric, "fields"); if (fields == NULL) { goto ParseEnd; @@ -227,14 +227,14 @@ ParseEnd: } } -int32_t tgParseSchema(const char *content, char*fileName) { +int32_t tgParseSchema(const char *content, char *fileName) { cJSON *root = cJSON_Parse(content); if (root == NULL) { httpError("failed to parse telegraf schema file:%s, invalid json format, content:%s", fileName, content); return -1; } int32_t size = 0; - cJSON *metrics = cJSON_GetObjectItem(root, "metrics"); + cJSON * metrics = cJSON_GetObjectItem(root, "metrics"); if (metrics != NULL) { size = cJSON_GetArraySize(metrics); if (size <= 0) { @@ -277,7 +277,7 @@ int32_t tgReadSchema(char *fileName) { rewind(fp); char * content = (char *)calloc(contentSize + 1, 1); int32_t result = (int32_t)fread(content, 1, contentSize, fp); - + if (result != contentSize) { httpError("failed to read telegraf schema file:%s", fileName); fclose(fp); @@ -296,7 +296,7 @@ int32_t tgReadSchema(char *fileName) { } void tgInitHandle(HttpServer *pServer) { - char fileName[TSDB_FILENAME_LEN*2] = {0}; + char fileName[TSDB_FILENAME_LEN * 2] = {0}; sprintf(fileName, "%s/taos.telegraf.cfg", configDir); if (tgReadSchema(fileName) <= 0) { tgFreeSchemas(); @@ -308,9 +308,7 @@ void tgInitHandle(HttpServer *pServer) { httpAddMethod(pServer, &tgDecodeMethod); } -void tgCleanupHandle() { - tgFreeSchemas(); -} +void tgCleanupHandle() { tgFreeSchemas(); } bool tgGetUserFromUrl(HttpContext *pContext) { HttpParser *pParser = pContext->parser; @@ -357,7 +355,7 @@ char *tgGetStableName(char *stname, cJSON *fields, int32_t fieldsSize) { bool schemaMatched = true; for (int32_t f = 0; f < schema->fieldNum; ++f) { char *fieldName = schema->fields[f]; - bool fieldMatched = false; + bool fieldMatched = false; for (int32_t i = 0; i < fieldsSize; i++) { cJSON *field = cJSON_GetArrayItem(fields, i); @@ -469,9 +467,9 @@ bool tgProcessSingleMetric(HttpContext *pContext, cJSON *metric, char *db) { } /* - * tag size may be larget than TSDB_COL_NAME_LEN - * we keep the first TSDB_COL_NAME_LEN bytes - */ + * tag size may be larget than TSDB_COL_NAME_LEN + * we keep the first TSDB_COL_NAME_LEN bytes + */ if (0) { if (strlen(tag->string) >= TSDB_COL_NAME_LEN) { httpSendErrorResp(pContext, TSDB_CODE_HTTP_TG_TAG_NAME_SIZE); @@ -540,9 +538,9 @@ bool tgProcessSingleMetric(HttpContext *pContext, cJSON *metric, char *db) { return false; } /* - * tag size may be larget than TSDB_COL_NAME_LEN - * we keep the first TSDB_COL_NAME_LEN bytes - */ + * tag size may be larget than TSDB_COL_NAME_LEN + * we keep the first TSDB_COL_NAME_LEN bytes + */ if (0) { if (strlen(field->string) >= TSDB_COL_NAME_LEN) { httpSendErrorResp(pContext, TSDB_CODE_HTTP_TG_FIELD_NAME_SIZE); @@ -578,8 +576,8 @@ bool tgProcessSingleMetric(HttpContext *pContext, cJSON *metric, char *db) { table_cmd->cmdType = HTTP_CMD_TYPE_INSERT; // order by tag name - cJSON *orderedTags[TG_MAX_SORT_TAG_SIZE] = {0}; - int32_t orderTagsLen = 0; + cJSON * orderedTags[TG_MAX_SORT_TAG_SIZE] = {0}; + int32_t orderTagsLen = 0; for (int32_t i = 0; i < tagsSize; ++i) { cJSON *tag = cJSON_GetArrayItem(tags, i); orderedTags[orderTagsLen++] = tag; @@ -603,7 +601,8 @@ bool tgProcessSingleMetric(HttpContext *pContext, cJSON *metric, char *db) { if (tsTelegrafUseFieldNum == 0) { table_cmd->stable = stable_cmd->stable = httpAddToSqlCmdBuffer(pContext, "%s", stname); } else { - table_cmd->stable = stable_cmd->stable = httpAddToSqlCmdBuffer(pContext, "%s_%d_%d", stname, fieldsSize, orderTagsLen); + table_cmd->stable = stable_cmd->stable = + httpAddToSqlCmdBuffer(pContext, "%s_%d_%d", stname, fieldsSize, orderTagsLen); } table_cmd->stable = stable_cmd->stable = httpShrinkTableName(pContext, table_cmd->stable, httpGetCmdsString(pContext, table_cmd->stable)); @@ -627,9 +626,11 @@ bool tgProcessSingleMetric(HttpContext *pContext, cJSON *metric, char *db) { // table name if (tsTelegrafUseFieldNum == 0) { - table_cmd->table = stable_cmd->table = httpAddToSqlCmdBufferNoTerminal(pContext, "%s_%s", stname, host->valuestring); + table_cmd->table = stable_cmd->table = + httpAddToSqlCmdBufferNoTerminal(pContext, "%s_%s", stname, host->valuestring); } else { - table_cmd->table = stable_cmd->table = httpAddToSqlCmdBufferNoTerminal(pContext, "%s_%d_%d_%s", stname, fieldsSize, orderTagsLen, host->valuestring); + table_cmd->table = stable_cmd->table = + httpAddToSqlCmdBufferNoTerminal(pContext, "%s_%d_%d_%s", stname, fieldsSize, orderTagsLen, host->valuestring); } for (int32_t i = 0; i < orderTagsLen; ++i) { cJSON *tag = orderedTags[i]; diff --git a/src/plugins/http/src/httpUtil.c b/src/plugins/http/src/httpUtil.c index 51333d2118..17d61e9e3d 100644 --- a/src/plugins/http/src/httpUtil.c +++ b/src/plugins/http/src/httpUtil.c @@ -160,8 +160,7 @@ bool httpMallocMultiCmds(HttpContext *pContext, int32_t cmdSize, int32_t bufferS free(multiCmds->cmds); multiCmds->cmds = (HttpSqlCmd *)malloc((size_t)cmdSize * sizeof(HttpSqlCmd)); if (multiCmds->cmds == NULL) { - httpError("context:%p, fd:%d, user:%s, malloc cmds:%d error", pContext, pContext->fd, - pContext->user, cmdSize); + httpError("context:%p, fd:%d, user:%s, malloc cmds:%d error", pContext, pContext->fd, pContext->user, cmdSize); return false; } multiCmds->maxSize = (int16_t)cmdSize; @@ -350,38 +349,40 @@ char *httpGetCmdsString(HttpContext *pContext, int32_t pos) { } int32_t httpGzipDeCompress(char *srcData, int32_t nSrcData, char *destData, int32_t *nDestData) { - int32_t err = 0; + int32_t err = 0; z_stream gzipStream = {0}; static char dummyHead[2] = { - 0x8 + 0x7 * 0x10, - (((0x8 + 0x7 * 0x10) * 0x100 + 30) / 31 * 31) & 0xFF, + 0x8 + 0x7 * 0x10, + (((0x8 + 0x7 * 0x10) * 0x100 + 30) / 31 * 31) & 0xFF, }; - gzipStream.zalloc = (alloc_func) 0; - gzipStream.zfree = (free_func) 0; - gzipStream.opaque = (voidpf) 0; - gzipStream.next_in = (Bytef *) srcData; + gzipStream.zalloc = (alloc_func)0; + gzipStream.zfree = (free_func)0; + gzipStream.opaque = (voidpf)0; + gzipStream.next_in = (Bytef *)srcData; gzipStream.avail_in = 0; - gzipStream.next_out = (Bytef *) destData; + gzipStream.next_out = (Bytef *)destData; if (inflateInit2(&gzipStream, 47) != Z_OK) { return -1; } while (gzipStream.total_out < *nDestData && gzipStream.total_in < nSrcData) { - gzipStream.avail_in = gzipStream.avail_out = nSrcData; //1 + gzipStream.avail_in = gzipStream.avail_out = nSrcData; // 1 if ((err = inflate(&gzipStream, Z_NO_FLUSH)) == Z_STREAM_END) { break; } if (err != Z_OK) { if (err == Z_DATA_ERROR) { - gzipStream.next_in = (Bytef *) dummyHead; + gzipStream.next_in = (Bytef *)dummyHead; gzipStream.avail_in = sizeof(dummyHead); if ((err = inflate(&gzipStream, Z_NO_FLUSH)) != Z_OK) { return -2; } - } else return -3; + } else { + return -3; + } } } @@ -394,23 +395,25 @@ int32_t httpGzipDeCompress(char *srcData, int32_t nSrcData, char *destData, int3 } int32_t httpGzipCompressInit(HttpContext *pContext) { - pContext->gzipStream.zalloc = (alloc_func) 0; - pContext->gzipStream.zfree = (free_func) 0; - pContext->gzipStream.opaque = (voidpf) 0; - if (deflateInit2(&pContext->gzipStream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, MAX_WBITS + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK) { + pContext->gzipStream.zalloc = (alloc_func)0; + pContext->gzipStream.zfree = (free_func)0; + pContext->gzipStream.opaque = (voidpf)0; + if (deflateInit2(&pContext->gzipStream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, MAX_WBITS + 16, 8, Z_DEFAULT_STRATEGY) != + Z_OK) { return -1; } return 0; } -int32_t httpGzipCompress(HttpContext *pContext, char *srcData, int32_t nSrcData, char *destData, int32_t *nDestData, bool isTheLast) { +int32_t httpGzipCompress(HttpContext *pContext, char *srcData, int32_t nSrcData, char *destData, int32_t *nDestData, + bool isTheLast) { int32_t err = 0; - int32_t lastTotalLen = (int32_t) (pContext->gzipStream.total_out); - pContext->gzipStream.next_in = (Bytef *) srcData; - pContext->gzipStream.avail_in = (uLong) nSrcData; - pContext->gzipStream.next_out = (Bytef *) destData; - pContext->gzipStream.avail_out = (uLong) (*nDestData); + int32_t lastTotalLen = (int32_t)(pContext->gzipStream.total_out); + pContext->gzipStream.next_in = (Bytef *)srcData; + pContext->gzipStream.avail_in = (uLong)nSrcData; + pContext->gzipStream.next_out = (Bytef *)destData; + pContext->gzipStream.avail_out = (uLong)(*nDestData); while (pContext->gzipStream.avail_in != 0) { if (deflate(&pContext->gzipStream, Z_FULL_FLUSH) != Z_OK) { @@ -442,12 +445,12 @@ int32_t httpGzipCompress(HttpContext *pContext, char *srcData, int32_t nSrcData, } } - *nDestData = (int32_t) (pContext->gzipStream.total_out) - lastTotalLen; + *nDestData = (int32_t)(pContext->gzipStream.total_out) - lastTotalLen; return 0; } -bool httpUrlMatch(HttpContext* pContext, int32_t pos, char* cmp) { - HttpParser* pParser = pContext->parser; +bool httpUrlMatch(HttpContext *pContext, int32_t pos, char *cmp) { + HttpParser *pParser = pContext->parser; if (pos < 0 || pos >= HTTP_MAX_URL) { return false; diff --git a/tests/script/general/http/restful.sim b/tests/script/general/http/restful.sim index a06e899d93..fdde975238 100644 --- a/tests/script/general/http/restful.sim +++ b/tests/script/general/http/restful.sim @@ -1,12 +1,10 @@ system sh/stop_dnodes.sh -sleep 2000 system sh/deploy.sh -n dnode1 -i 1 system sh/cfg.sh -n dnode1 -c wallevel -v 0 system sh/cfg.sh -n dnode1 -c http -v 1 system sh/cfg.sh -n dnode1 -c httpEnableRecordSql -v 1 system sh/exec.sh -n dnode1 -s start -sleep 2000 sql connect print ============================ dnode1 start