xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

TD-1207

This commit is contained in:
Shengliang Guan 2021-01-22 11:59:14 +08:00
parent 2079533b50
commit fa7e9bd6d2
19 changed files with 247 additions and 238 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/plugins/http/src/httpAuth.c
View File

@ -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);

11
src/plugins/http/src/httpContext.c
View File

@ -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;
}

2
src/plugins/http/src/httpGcHandle.c
View File

@ -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);

27
src/plugins/http/src/httpGcJson.c
View File

@ -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; i<num_fields; i++){
switch (fields[i].type) {
len = snprintf(target, HTTP_GC_TARGET_SIZE, "%s{", aliasBuffer);
for (int32_t i = dataFields + 1; i < num_fields; i++) {
switch (fields[i].type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, "%s:%d", fields[i].name, *((int8_t *)row[i]));
@ -155,9 +155,9 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
if (row[i]!= NULL){
if (row[i] != NULL) {
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, "%s:", fields[i].name);
memcpy(target + len, (char *) row[i], length[i]);
memcpy(target + len, (char *)row[i], length[i]);
len = (int32_t)strlen(target);
}
break;
@ -165,10 +165,9 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, "%s:%s", fields[i].name, "-");
break;
}
if(i < num_fields - 1 ){
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, ", ");
if (i < num_fields - 1) {
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, ", ");
}
}
len += snprintf(target + len, HTTP_GC_TARGET_SIZE - len, "}");
@ -217,10 +216,10 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
httpJsonStringForTransMean(jsonBuf, (char*)row[i], fields[i].bytes);
httpJsonStringForTransMean(jsonBuf, (char *)row[i], fields[i].bytes);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
if (precision == TSDB_TIME_PRECISION_MILLI) { //ms
if (precision == TSDB_TIME_PRECISION_MILLI) { // ms
httpJsonInt64(jsonBuf, *((int64_t *)row[i]));
} else {
httpJsonInt64(jsonBuf, *((int64_t *)row[i]) / 1000);

92
src/plugins/http/src/httpGzip.c
View File

@ -25,25 +25,24 @@ typedef enum {
} 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;
int32_t state;
ehttp_gzip_conf_t conf;
ehttp_gzip_callbacks_t callbacks;
void * arg;
z_stream * gzip;
gz_header * header;
char * chunk;
int32_t state;
};
static void dummy_on_data(ehttp_gzip_t *gzip, void *arg, const char *buf, int32_t len) {
}
static void dummy_on_data(ehttp_gzip_t *gzip, void *arg, const char *buf, int32_t len) {}
static void ehttp_gzip_cleanup(ehttp_gzip_t *gzip) {
switch(gzip->state) {
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;
}

2
src/plugins/http/src/httpHandle.c
View File

@ -48,7 +48,7 @@ bool httpProcessData(HttpContext* pContext) {
/*
* httpCloseContextByApp has been called when parsing the error
*/
//httpCloseContextByApp(pContext);
// httpCloseContextByApp(pContext);
} else {
httpProcessRequest(pContext);
}

47
src/plugins/http/src/httpJson.c
View File

@ -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;
}

96
src/plugins/http/src/httpParser.c
View File

@ -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;

14
src/plugins/http/src/httpQueue.c
View File

@ -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;
}

5
src/plugins/http/src/httpResp.c
View File

@ -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);

1
src/plugins/http/src/httpRestHandle.c
View File

@ -95,7 +95,6 @@ bool restProcessSqlRequest(HttpContext* pContext, int32_t timestampFmt) {
return false;
}
/*
* for async test
*

23
src/plugins/http/src/httpRestJson.c
View File

@ -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) {

34
src/plugins/http/src/httpServer.c
View File

@ -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;
}

5
src/plugins/http/src/httpSession.c
View File

@ -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);

10
src/plugins/http/src/httpSql.c
View File

@ -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) {

6
src/plugins/http/src/httpSystem.c
View File

@ -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); }

45
src/plugins/http/src/httpTgHandle.c
View File

@ -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];

55
src/plugins/http/src/httpUtil.c
View File

@ -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;

2
tests/script/general/http/restful.sim
View File

@ -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