GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-19 05:51:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix the timing problem of request context

Browse Source 
- Thread lock handling for running status flags
This commit is contained in:
gc87
2021-10-28 10:14:07 +08:00
committed by Parker Liu
parent dc018ca805
commit da3746d135
4 changed files with 33 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command/command_rw.c
+8 -10
View File
@@ -24,7 +24,7 @@
#include "command_rw.h"
int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req)
neu_parse_read_req_t *req, uint32_t req_id)
{
log_info("READ uuid:%s, node id:%u", mqtt->uuid, req->node_id);
int rc = common_has_node(plugin, req->node_id);
@@ -40,9 +40,8 @@ int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
return -2;
}
uint32_t req_id = neu_plugin_get_event_id(plugin);
neu_plugin_send_read_cmd(plugin, req_id, req->node_id, config);
return req_id;
return 0;
}
static int wrap_read_response_json_object(neu_fixed_array_t * array,
@@ -230,7 +229,7 @@ char *command_read_cycle_response(neu_plugin_t * plugin,
static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
const char * group_config_name,
neu_data_val_t *write_val)
neu_data_val_t *write_val, uint32_t req_id)
{
int rc = common_has_node(plugin, dest_node_id);
if (0 != rc) {
@@ -244,13 +243,12 @@ static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
return -2;
}
uint32_t req_id = neu_plugin_get_event_id(plugin);
neu_plugin_send_write_cmd(plugin, req_id, dest_node_id, config, write_val);
return req_id;
return 0;
}
int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req)
neu_parse_write_req_t *write_req, uint32_t req_id)
{
log_info("WRITE uuid:%s, group config name:%s", mqtt->uuid,
write_req->group_config_name);
@@ -312,10 +310,10 @@ int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
}
neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array);
int req_id = write_command(plugin, write_req->node_id,
write_req->group_config_name, write_val);
write_command(plugin, write_req->node_id, write_req->group_config_name,
write_val, req_id);
neu_dvalue_free(write_val);
return req_id;
return 0;
}
static int wrap_write_response_json_object(neu_fixed_array_t * array,
plugins/mqtt/command/command_rw.h
+2 -2
View File
@@ -29,14 +29,14 @@ extern "C" {
#include "common.h"
int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req);
neu_parse_read_req_t *req, uint32_t req_id);
char *command_read_once_response(neu_plugin_t * plugin,
neu_parse_mqtt_t *parse_header,
neu_data_val_t * resp_val);
char *command_read_cycle_response(neu_plugin_t * plugin,
neu_data_val_t *resp_val);
int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req);
neu_parse_write_req_t *write_req, uint32_t req_id);
char *command_write_response(neu_plugin_t * plugin,
neu_parse_mqtt_t *parse_header,
neu_data_val_t * resp_val);
plugins/mqtt/mqttc_client.c
+17 -1
View File
@@ -54,6 +54,7 @@ struct mqttc_client {
uint8_t send_buf[SEND_BUF_SIZE];
uint8_t recv_buf[RECV_BUF_SIZE];
pthread_t client_daemon;
pthread_mutex_t mutex;
void * user_data;
};
@@ -207,7 +208,17 @@ static void *mqtt_refresher(void *context)
{
mqttc_client_t * client = (mqttc_client_t *) context;
struct mqtt_client *mqtt = &client->mqtt;
while (client->running) {
bool run_flag = true;
while (1) {
pthread_mutex_lock(&client->mutex);
run_flag = client->running;
pthread_mutex_unlock(&client->mutex);
if (!run_flag) {
break;
}
mqtt_sync((struct mqtt_client *) mqtt);
usleep(INTERVAL);
}
@@ -224,6 +235,7 @@ mqttc_client_t *mqttc_client_create(option_t *option, void *context)
}
memset(client, 0, sizeof(mqttc_client_t));
pthread_mutex_init(&client->mutex, NULL);
client->option = option;
client->running = true;
client->user_data = context;
@@ -354,6 +366,7 @@ client_error_e mqttc_client_disconnect(mqttc_client_t *client)
client_error_e mqttc_client_destroy(mqttc_client_t *client)
{
pthread_mutex_destroy(&client->mutex);
free(client);
return ClientSuccess;
}
@@ -492,7 +505,10 @@ client_error_e mqttc_client_close(mqttc_client_t *client)
return ClientIsNULL;
}
pthread_mutex_lock(&client->mutex);
client->running = false;
pthread_mutex_unlock(&client->mutex);
pthread_join(client->client_daemon, NULL);
client_error_e rc = mqttc_client_disconnect(client);
rc = mqttc_client_destroy(client);
plugins/mqtt/plugin_mqtt.c
+6 -2
View File
@@ -198,10 +198,12 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
neu_parse_read_req_t *req = NULL;
rc = neu_parse_decode_read(json_str, &req);
if (0 == rc) {
int req_id = command_read_once_request(plugin, mqtt, req);
uint32_t req_id = neu_plugin_get_event_id(plugin);
if (0 < req_id) {
context_list_add(&plugin->context_list, req_id, mqtt);
}
command_read_once_request(plugin, mqtt, req, req_id);
neu_parse_decode_read_free(req);
}
break;
@@ -210,10 +212,12 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
neu_parse_write_req_t *req = NULL;
rc = neu_parse_decode_write(json_str, &req);
if (0 == rc) {
int req_id = command_write_request(plugin, mqtt, req);
uint32_t req_id = neu_plugin_get_event_id(plugin);
if (0 < req_id) {
context_list_add(&plugin->context_list, req_id, mqtt);
}
command_write_request(plugin, mqtt, req, req_id);
neu_parse_decode_write_free(req);
}
break;