diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 21e49aa3..9a38cfa6 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -80,6 +80,8 @@ struct neu_adapter_driver { struct group *groups; }; +static void report_to_app(neu_adapter_driver_t *driver, group_t *group, + struct sockaddr_in dst); static int report_callback(void *usr_data); static int read_callback(void *usr_data); static int write_callback(void *usr_data); @@ -1370,6 +1372,54 @@ UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver, return tags; } +static void report_to_app(neu_adapter_driver_t *driver, group_t *group, + struct sockaddr_in dst) +{ + neu_reqresp_head_t header = { + .type = NEU_REQRESP_TRANS_DATA, + }; + + UT_array *tags = + neu_adapter_driver_get_read_tag(group->driver, group->name); + + neu_reqresp_trans_data_t *data = + calloc(1, sizeof(neu_reqresp_trans_data_t)); + + data->driver = strdup(group->driver->adapter.name); + data->group = strdup(group->name); + utarray_new(data->tags, neu_resp_tag_value_meta_icd()); + + read_group(global_timestamp, + neu_group_get_interval(group->group) * + NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + neu_adapter_get_tag_cache_type(&driver->adapter), driver->cache, + group->name, tags, data->tags); + + nlog_info("report group: %s, all tags: %d, report tags: %d", group->name, + utarray_len(tags), utarray_len(data->tags)); + if (utarray_len(data->tags) > 0) { + pthread_mutex_lock(&group->apps_mtx); + + data->ctx = calloc(1, sizeof(neu_reqresp_trans_data_ctx_t)); + data->ctx->index = 1; + + pthread_mutex_init(&data->ctx->mtx, NULL); + + if (driver->adapter.cb_funs.responseto(&driver->adapter, &header, data, + dst) != 0) { + neu_trans_data_free(data); + } + + pthread_mutex_unlock(&group->apps_mtx); + } else { + utarray_free(data->tags); + free(data->group); + free(data->driver); + } + utarray_free(tags); + free(data); +} + static int report_callback(void *usr_data) { group_t * group = (group_t *) usr_data; @@ -1922,6 +1972,8 @@ void neu_adapter_driver_subscribe(neu_adapter_driver_t *driver, utarray_push_back(find->apps, &sub_app); pthread_mutex_unlock(&find->apps_mtx); + + report_to_app(driver, find, sub_app.addr); } void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver,