diff --git a/plugins/restful/handle.h b/plugins/restful/handle.h index c03e321c..86bb4eab 100644 --- a/plugins/restful/handle.h +++ b/plugins/restful/handle.h @@ -32,8 +32,6 @@ size_t req_data_size = 0; \ req_type *req = NULL; \ \ - nng_aio_set_timeout(aio, 3000); \ - nng_aio_defer(aio, http_timeout, NULL); \ if (http_get_body((aio), (void **) &req_data, &req_data_size) == 0 && \ decode_fun(req_data, &req) == 0) { \ { func }; \ @@ -47,6 +45,8 @@ #define REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(aio, req_type, decode_fun, \ func) \ { \ + nng_aio_set_timeout(aio, 3000); \ + nng_aio_defer(aio, http_timeout, NULL); \ char *jwt = (char *) http_get_header(aio, (char *) "Authorization"); \ \ NEU_JSON_RESPONSE_ERROR(neu_jwt_validate(jwt), { \ @@ -61,6 +61,8 @@ #define VALIDATE_JWT(aio) \ { \ + nng_aio_set_timeout(aio, 3000); \ + nng_aio_defer(aio, http_timeout, NULL); \ char *jwt = (char *) http_get_header(aio, (char *) "Authorization"); \ \ NEU_JSON_RESPONSE_ERROR(neu_jwt_validate(jwt), { \ diff --git a/plugins/restful/http.c b/plugins/restful/http.c index d929a8ae..6b5bc092 100644 --- a/plugins/restful/http.c +++ b/plugins/restful/http.c @@ -409,6 +409,7 @@ void http_timeout(nng_aio *aio, void *arg, int err) http_internal_error(aio, "{\"error\": \"canceled\"}"); break; default: + nlog_error("aio error: %d", err); http_internal_error(aio, "{\"error\": \"unknown\"}"); break; } diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 40ee7358..703fe392 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -39,6 +39,7 @@ struct elem { int32_t error; int64_t timestamp; + bool changed; uint16_t n_byte; uint8_t *bytes; @@ -122,9 +123,15 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, elem->n_byte = n_byte; elem->bytes = calloc(elem->n_byte, 1); + elem->changed = true; + HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem); } + if (elem->error != 0) { + elem->changed = true; + } + elem->error = 0; elem->timestamp = timestamp; @@ -133,11 +140,19 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, assert(n_byte <= NEU_VALUE_SIZE); elem->n_byte = n_byte; elem->bytes = calloc(elem->n_byte, n_byte); + + elem->changed = true; } else { if (n_byte != elem->n_byte) { elem->n_byte = n_byte; elem->bytes = realloc(elem->bytes, n_byte); memset(elem->bytes, 0, elem->n_byte); + + elem->changed = true; + } else { + if (memcmp(elem->bytes, bytes, elem->n_byte) != 0) { + elem->changed = true; + } } } @@ -170,6 +185,32 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, return ret; } +int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group, + const char * tag, + neu_driver_cache_value_t *value) +{ + struct elem *elem = NULL; + int ret = -1; + tkey_t key = to_key(group, tag); + + nng_mtx_lock(cache->mtx); + HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem); + + if (elem != NULL && elem->changed) { + value->error = elem->error; + value->n_byte = elem->n_byte; + value->timestamp = elem->timestamp; + memcpy(value->value.bytes, elem->bytes, elem->n_byte); + + elem->changed = false; + ret = 0; + } + + nng_mtx_unlock(cache->mtx); + + return ret; +} + void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, const char *tag) { @@ -208,6 +249,7 @@ static void update_tag_error(neu_driver_cache_t *cache, const char *group, } elem->error = error; elem->timestamp = timestamp; + elem->changed = true; nng_mtx_unlock(cache->mtx); } diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index 4f5e4c63..1110aca6 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -47,5 +47,8 @@ typedef struct { int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, const char *tag, neu_driver_cache_value_t *value); +int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group, + const char * tag, + neu_driver_cache_value_t *value); #endif diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 565af111..4e40f341 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -62,8 +62,9 @@ struct neu_adapter_driver { static int report_callback(void *usr_data); static int read_callback(void *usr_data); static void read_group(int64_t timestamp, int64_t timeout, - neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_t *datas); + neu_driver_cache_t *cache, bool report, + const char *group, UT_array *tags, + neu_resp_tag_value_t *datas); static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value); static void write_response(neu_adapter_t *adapter, void *r, neu_error error); @@ -241,7 +242,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, read_group(driver->adapter.timestamp, neu_group_get_interval(group) * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - driver->cache, cmd->group, tags, resp.tags); + driver->cache, false, cmd->group, tags, resp.tags); } strcpy(resp.driver, cmd->driver); @@ -594,7 +595,7 @@ static int report_callback(void *usr_data) read_group(group->driver->adapter.timestamp, neu_group_get_interval(group->group) * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - group->driver->cache, group->name, tags, + group->driver->cache, true, group->name, tags, (neu_resp_tag_value_t *) &data[1]); group->driver->adapter.cb_funs.response(&group->driver->adapter, &header, @@ -658,8 +659,9 @@ static int read_callback(void *usr_data) } static void read_group(int64_t timestamp, int64_t timeout, - neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_t *datas) + neu_driver_cache_t *cache, bool report, + const char *group, UT_array *tags, + neu_resp_tag_value_t *datas) { utarray_foreach(tags, neu_datatag_t *, tag) { @@ -668,10 +670,19 @@ static void read_group(int64_t timestamp, int64_t timeout, strcpy(datas[index].tag, tag->name); - if (neu_driver_cache_get(cache, group, tag->name, &value) != 0) { - datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; - continue; + if (report && neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) { + if (neu_driver_cache_get_changed(cache, group, tag->name, &value) != + 0) { + datas[index].value.type = NEU_TYPE_ERROR; + datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + continue; + } + } else { + if (neu_driver_cache_get(cache, group, tag->name, &value) != 0) { + datas[index].value.type = NEU_TYPE_ERROR; + datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + continue; + } } if (value.error != 0) {