GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-24 02:39:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #636 from gc87/branch-mqtt

Browse Source 
Add heartbeat support for mqtt plugin
This commit is contained in:
fengzeroz
2022-07-15 17:21:22 +08:00
committed by GitHub
parent fa87dd7a58 9d3b411627
commit f628c9153b
20 changed files with 884 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/connection/mqtt_c_client.c
+56 -29
View File
@@ -592,29 +592,48 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
utarray_new(client->array, &tuple_icd);
struct reconnect_state *state = &client->state;
state->hostname = strdup(client->option->host);
state->port = strdup(client->option->port);
state->client_id = strdup(client->option->clientid);
state->connection = strdup(client->option->connection);
state->username = strdup(client->option->username);
state->password = strdup(client->option->password);
state->keepalive = client->option->keepalive_interval;
state->clean = client->option->clean_session;
state->ca_file = strdup(client->option->ca);
state->cert_file = strdup(client->option->cert);
state->key_file = strdup(client->option->key);
state->client = client;
state->send_buf_sz = sizeof(state->send_buf);
state->recv_buf_sz = sizeof(state->recv_buf);
state->cert = NULL;
state->key = NULL;
state->ssl_ctx = NULL;
state->sock_fd = NULL;
state->running = true;
state->working = true;
pthread_mutex_init(&state->sock_state_mutex, NULL);
pthread_mutex_init(&state->running_mutex, NULL);
pthread_mutex_init(&state->working_mutex, NULL);
if (NULL != client->option->clientid) {
state->client_id = strdup(client->option->clientid);
}
if (NULL != client->option->connection) {
state->connection = strdup(client->option->connection);
}
if (NULL != client->option->host) {
state->hostname = strdup(client->option->host);
}
if (NULL != client->option->port) {
state->port = strdup(client->option->port);
}
if (NULL != client->option->username) {
state->username = strdup(client->option->username);
}
if (NULL != client->option->password) {
state->password = strdup(client->option->password);
}
if (NULL != client->option->ca) {
state->ca_file = strdup(client->option->ca);
}
if (NULL != client->option->cert) {
state->cert_file = strdup(client->option->cert);
}
if (NULL != client->option->key) {
state->key_file = strdup(client->option->key);
}
if (NULL != client->option->keypass) {
state->keypass = strdup(client->option->keypass);
} else {
state->keypass = NULL;
}
if (NULL != client->option->will_topic) {
state->will_topic = strdup(option->will_topic);
@@ -624,12 +643,20 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
}
}
if (NULL != client->option->keypass &&
0 < strlen(client->option->keypass)) {
state->keypass = strdup(client->option->keypass);
} else {
state->keypass = NULL;
}
state->client = client;
state->send_buf_sz = sizeof(state->send_buf);
state->recv_buf_sz = sizeof(state->recv_buf);
state->cert = NULL;
state->key = NULL;
state->ssl_ctx = NULL;
state->sock_fd = NULL;
state->running = true;
state->working = true;
state->keepalive = client->option->keepalive_interval;
state->clean = client->option->clean_session;
pthread_mutex_init(&state->sock_state_mutex, NULL);
pthread_mutex_init(&state->running_mutex, NULL);
pthread_mutex_init(&state->working_mutex, NULL);
client->mqtt.publish_response_callback_state = state;
return client;
src/parser/neu_json_mqtt.c
+61
View File
@@ -22,6 +22,7 @@
* It was automatically generated by `json-autotype`.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -85,3 +86,63 @@ int neu_json_encode_mqtt_resp(void *json_object, void *param)
return ret;
}
int neu_json_encode_state_header_resp(void *json_object, void *param)
{
int ret = 0;
neu_json_states_head_t *resp = (neu_json_states_head_t *) param;
neu_json_elem_t resp_elems[] = { {
.name = "version",
.t = NEU_JSON_STR,
.v.val_str = resp->version,
},
{
.name = "timestamp",
.t = NEU_JSON_INT,
.v.val_int = resp->timpstamp,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}
int neu_json_encode_states_resp(void *json_object, void *param)
{
int ret = 0;
neu_json_states_t *resp = (neu_json_states_t *) param;
void * state_array = neu_json_array();
neu_json_node_state_t *states = resp->states;
for (int i = 0; i < resp->n_state; i++) {
neu_json_elem_t state_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
.v.val_str = states[i].node,
},
{
.name = "link",
.t = NEU_JSON_INT,
.v.val_int = states[i].link,
},
{
.name = "running",
.t = NEU_JSON_INT,
.v.val_int = states[i].running,
} };
state_array = neu_json_encode_array(state_array, state_elems,
NEU_JSON_ELEM_SIZE(state_elems));
}
neu_json_elem_t resp_elems[] = { {
.name = "states",
.t = NEU_JSON_OBJECT,
.v.val_object = state_array,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}