diff --git a/ft/resource/error.resource b/ft/resource/error.resource index e9479879..a0e9ea4c 100644 --- a/ft/resource/error.resource +++ b/ft/resource/error.resource @@ -54,4 +54,5 @@ ${NEU_ERR_PLUGIN_WRITE_FAILURE} 3001 ${NEU_ERR_PLUGIN_DISCONNECTED} 3002 ${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ} 3003 ${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE} 3004 -${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007 \ No newline at end of file +${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007 +${NEU_ERR_PLUGIN_TAG_EXPIRED} 3008 \ No newline at end of file diff --git a/include/neuron/define.h b/include/neuron/define.h index afa5e38f..edde2277 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -29,6 +29,7 @@ #define NEU_PLUGIN_NAME_LEN 32 #define NEU_PLUGIN_LIBRARY_LEN 32 #define NEU_PLUGIN_DESCRIPTION_LEN 256 +#define NEU_DRIVER_TAG_CACHE_EXPIRE_TIME 5 typedef enum neu_plugin_kind { NEU_PLUGIN_KIND_STATIC = 0, diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index bec49014..3a28cef8 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -85,6 +85,7 @@ typedef enum { NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004, NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007, + NEU_ERR_PLUGIN_TAG_EXPIRED = 3008, NEU_ERR_MQTT_FAILURE = 4000, NEU_ERR_MQTT_NO_CERTFILESET = 4001, diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 6215ae1f..2a76ba3e 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -42,6 +42,7 @@ typedef struct neu_plugin_common { neu_node_link_state_e link_state; uint16_t tag_size; + int64_t timestamp; zlog_category_t *log; } neu_plugin_common_t; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 2e003587..58958519 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -46,6 +47,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); +static int update_timestamp(void *usr_data); static const adapter_callbacks_t callback_funs = { .command = adapter_command, @@ -54,9 +56,14 @@ static const adapter_callbacks_t callback_funs = { neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start) { - int rv = 0; - neu_adapter_t * adapter = NULL; - neu_event_io_param_t param = { 0 }; + int rv = 0; + neu_adapter_t * adapter = NULL; + neu_event_io_param_t param = { 0 }; + neu_event_timer_param_t timer_param = { + .second = 0, + .millisecond = 100, + .cb = update_timestamp, + }; switch (info->module->type) { case NEU_NA_TYPE_DRIVER: @@ -126,6 +133,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start) rv = nng_dial(adapter->sock, neu_manager_get_url(), &adapter->dialer, 0); assert(rv == 0); + timer_param.usr_data = (void *) adapter; + adapter->timer = neu_event_add_timer(adapter->events, timer_param); + nlog_info("Success to create adapter: %s", adapter->name); return adapter; @@ -604,6 +614,7 @@ int neu_adapter_uninit(neu_adapter_t *adapter) neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); } + neu_event_del_timer(adapter->events, adapter->timer); adapter->module->intf_funs->close(adapter->plugin); nlog_info("Stop the adapter(%s)", adapter->name); @@ -744,4 +755,18 @@ inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, nng_msg *msg = neu_msg_gen(header, data); nng_sendmsg(adapter->sock, msg, 0); +} + +static int update_timestamp(void *usr_data) +{ + neu_adapter_t *adapter = (neu_adapter_t *) usr_data; + struct timeval tv = { 0 }; + uint64_t timestamp = 0; + + gettimeofday(&tv, NULL); + timestamp = tv.tv_sec * 1000 + tv.tv_usec / 1000; + adapter->timestamp = timestamp; + + neu_plugin_to_plugin_common(adapter->plugin)->timestamp = timestamp; + return 0; } \ No newline at end of file diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 0736d339..8e75e5d5 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -34,6 +34,8 @@ struct neu_adapter { char *name; char *setting; + int64_t timestamp; + neu_node_running_state_e state; neu_persister_t * persister; @@ -46,9 +48,10 @@ struct neu_adapter { nng_socket sock; nng_dialer dialer; - neu_events_t * events; - neu_event_io_t *nng_io; - int recv_fd; + neu_events_t * events; + neu_event_io_t * nng_io; + neu_event_timer_t *timer; + int recv_fd; }; typedef void (*adapter_handler)(neu_adapter_t * adapter, diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 94fe207e..c8132727 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -53,9 +53,9 @@ struct neu_driver_cache { }; static void update_tag_error(neu_driver_cache_t *cache, const char *group, - const char *tag, int error); + const char *tag, int64_t timestamp, int error); static void update_group_error(neu_driver_cache_t *cache, const char *group, - int error); + int64_t timestamp, int error); inline static tkey_t to_key(const char *group, const char *tag) { @@ -100,11 +100,10 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group, const char *tag, int64_t timestamp, int32_t error) { - (void) timestamp; if (tag == NULL) { - update_group_error(cache, group, error); + update_group_error(cache, group, timestamp, error); } else { - update_tag_error(cache, group, tag, error); + update_tag_error(cache, group, tag, timestamp, error); } } @@ -198,7 +197,7 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, } static void update_tag_error(neu_driver_cache_t *cache, const char *group, - const char *tag, int error) + const char *tag, int64_t timestamp, int error) { struct elem *elem = NULL; tkey_t key = to_key(group, tag); @@ -213,13 +212,14 @@ static void update_tag_error(neu_driver_cache_t *cache, const char *group, HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem); } - elem->error = error; + elem->error = error; + elem->timestamp = timestamp; nng_mtx_unlock(cache->mtx); } static void update_group_error(neu_driver_cache_t *cache, const char *group, - int error) + int64_t timestamp, int error) { struct elem *elem = NULL, *tmp = NULL; @@ -227,7 +227,8 @@ static void update_group_error(neu_driver_cache_t *cache, const char *group, HASH_ITER(hh, cache->table, elem, tmp) { if (strcmp(elem->key.group, group) == 0) { - elem->error = error; + elem->timestamp = timestamp; + elem->error = error; } } diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 401f1d26..843d9fa2 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -61,7 +61,8 @@ struct neu_adapter_driver { static int report_callback(void *usr_data); static int read_callback(void *usr_data); -static void read_group(neu_driver_cache_t *cache, const char *group, +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); static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value); @@ -119,9 +120,11 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, } if (value.type == NEU_TYPE_ERROR) { - neu_driver_cache_error(driver->cache, group, tag, 0, value.value.i32); + neu_driver_cache_error(driver->cache, group, tag, + driver->adapter.timestamp, value.value.i32); } else { - neu_driver_cache_update(driver->cache, group, tag, 0, n_byte, + neu_driver_cache_update(driver->cache, group, tag, + driver->adapter.timestamp, n_byte, value.value.bytes); } } @@ -190,21 +193,30 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { - if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_NODE_NOT_RUNNING); - return; - } - - neu_req_read_group_t *cmd = (neu_req_read_group_t *) &req[1]; - neu_resp_read_group_t resp = { 0 }; - UT_array * tags = - neu_group_get_read_tag(find_group(driver, cmd->group)->group); + neu_req_read_group_t *cmd = (neu_req_read_group_t *) &req[1]; + neu_resp_read_group_t resp = { 0 }; + neu_group_t * group = find_group(driver, cmd->group)->group; + UT_array * tags = neu_group_get_read_tag(group); resp.n_tag = utarray_len(tags); resp.tags = calloc(1, utarray_len(tags) * sizeof(neu_resp_tag_value_t)); - read_group(driver->cache, cmd->group, tags, resp.tags); + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + utarray_foreach(tags, neu_datatag_t *, tag) + { + int index = utarray_eltidx(tags, tag); + strcpy(resp.tags[index].tag, tag->name); + resp.tags[index].value.type = NEU_TYPE_ERROR; + resp.tags[index].value.value.i32 = NEU_ERR_NODE_NOT_RUNNING; + } + + } else { + read_group(driver->adapter.timestamp, + neu_group_get_interval(group) * + NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + driver->cache, cmd->group, tags, resp.tags); + } + strcpy(resp.driver, cmd->driver); strcpy(resp.group, cmd->group); @@ -517,7 +529,10 @@ static int report_callback(void *usr_data) strcpy(data->driver, group->driver->adapter.name); strcpy(data->group, group->name); data->n_tag = utarray_len(tags); - read_group(group->driver->cache, group->name, tags, + read_group(group->driver->adapter.timestamp, + neu_group_get_interval(group->group) * + NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + group->driver->cache, group->name, tags, (neu_resp_tag_value_t *) &data[1]); group->driver->adapter.cb_funs.response(&group->driver->adapter, &header, @@ -580,7 +595,8 @@ static int read_callback(void *usr_data) return 0; } -static void read_group(neu_driver_cache_t *cache, const char *group, +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) { utarray_foreach(tags, neu_datatag_t *, tag) @@ -602,7 +618,12 @@ static void read_group(neu_driver_cache_t *cache, const char *group, continue; } - datas[index].value.type = tag->type; - datas[index].value.value = value.value; + if ((timestamp - value.timestamp) > timeout) { + datas[index].value.type = NEU_TYPE_ERROR; + datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_EXPIRED; + } else { + datas[index].value.type = tag->type; + datas[index].value.value = value.value; + } } }