GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-25 11:26:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #681 from fengzeroz/main

Browse Source 
subscribe tag & http aio cancel fn
This commit is contained in:
fengzeroz
2022-08-01 11:05:48 +08:00
committed by GitHub
parent ce4ba04e5e c032f9a4e5
commit 4219110413
5 changed files with 71 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/restful/handle.h
+4 -2
View File
@@ -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), { \
plugins/restful/http.c
+1
View File
@@ -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;
}
src/adapter/driver/cache.c
+42
View File
@@ -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);
}
src/adapter/driver/cache.h
+3
View File
@@ -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
src/adapter/driver/driver.c
+21 -10
View File
@@ -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) {