GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 14:07:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(mqtt): plog to mqtt

Browse Source
This commit is contained in:
gc87
2022-06-17 18:46:30 +08:00
parent 5e16738a23
commit 41dec9b209
6 changed files with 83 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command/command.c
+12 -9
View File
@@ -22,6 +22,7 @@
#include "read_write.h"
#include "utils/log.h"
#include "../mqtt.h"
#include "command.h"
// Read tags with group config
@@ -81,7 +82,8 @@ void command_response_handle(mqtt_response_t *response)
// neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt,
// neu_json_encode_mqtt_resp, &ret_str);
nlog_error("json parsing mqtt failed");
plog_error(plugin, "json parsing mqtt failed");
free(json_str);
return;
}
@@ -103,26 +105,27 @@ void command_response_handle(mqtt_response_t *response)
break;
}
default:
nlog_error("invalid topic type");
plog_error(plugin, "invalid topic type");
break;
}
free(json_str);
}
char *command_read_once_response(neu_reqresp_head_t * head,
char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_read_group_t *data, int format)
{
return command_rw_read_once_response(head, data, format);
return NULL;
return command_rw_read_once_response(plugin, head, data, format);
}
char *command_read_periodic_response(neu_reqresp_trans_data_t *data, int format)
char *command_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data, int format)
{
return command_rw_read_periodic_response(data, format);
return command_rw_read_periodic_response(plugin, data, format);
}
char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data)
char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data)
{
return command_rw_write_response(head, data);
return command_rw_write_response(plugin, head, data);
}
plugins/mqtt/command/command.h
+5 -3
View File
@@ -46,13 +46,15 @@ typedef struct {
void command_response_handle(mqtt_response_t *response);
char *command_read_once_response(neu_reqresp_head_t * head,
char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_read_group_t *data, int format);
char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format);
char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data);
char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data);
#ifdef __cplusplus
}
plugins/mqtt/command/read_write.c
+16 -9
View File
@@ -22,6 +22,7 @@
#include <stdlib.h>
#include <sys/time.h>
#include "../mqtt.h"
#include "read_write.h"
#include "utils/log.h"
@@ -36,7 +37,7 @@ static uint64_t current_time()
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req)
{
nlog_info("read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
req->node);
neu_reqresp_head_t header = { 0 };
@@ -145,7 +146,8 @@ static void clean_read_response_json(neu_json_read_resp_t *json)
free(json->tags);
}
char *command_rw_read_once_response(neu_reqresp_head_t * head,
char *command_rw_read_once_response(neu_plugin_t * plugin,
neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format)
{
UNUSED(format);
@@ -155,7 +157,8 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head,
char * json_str = NULL;
neu_json_read_resp_t json = { 0 };
nlog_info("read resp uuid: %s", ((neu_json_mqtt_t *) head->ctx)->uuid);
plog_info(plugin, "read resp uuid: %s",
((neu_json_mqtt_t *) head->ctx)->uuid);
wrap_read_response_json(tags, len, &json);
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, head->ctx,
neu_json_encode_mqtt_resp, &json_str);
@@ -164,9 +167,12 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head,
return json_str;
}
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_rw_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format)
{
UNUSED(plugin);
neu_resp_tag_value_t * tags = data->tags;
uint16_t len = data->n_tag;
char * json_str = NULL;
@@ -196,8 +202,8 @@ char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *req)
{
nlog_info("write uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
req->node);
plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
req->group, req->node);
neu_reqresp_head_t header = { 0 };
neu_req_write_tag_t cmd = { 0 };
@@ -239,13 +245,14 @@ int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
return 0;
}
char *command_rw_write_response(neu_reqresp_head_t *head,
neu_resp_error_t * data)
char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data)
{
neu_json_error_resp_t error = { .error = data->error };
char * json_str = NULL;
nlog_info("write resp uuid: %s", ((neu_json_mqtt_t *) head->ctx)->uuid);
plog_info(plugin, "write resp uuid: %s",
((neu_json_mqtt_t *) head->ctx)->uuid);
neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, head->ctx,
neu_json_encode_mqtt_resp, &json_str);
neu_json_decode_mqtt_req_free(head->ctx);
plugins/mqtt/command/read_write.h
+6 -4
View File
@@ -31,17 +31,19 @@ extern "C" {
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req);
char *command_rw_read_once_response(neu_reqresp_head_t * head,
char *command_rw_read_once_response(neu_plugin_t * plugin,
neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format);
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_rw_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format);
int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *req);
char *command_rw_write_response(neu_reqresp_head_t *head,
neu_resp_error_t * data);
char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data);
#ifdef __cplusplus
}
plugins/mqtt/mqtt.c
+36 -29
View File
@@ -18,15 +18,25 @@
**/
#include <assert.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/time.h>
#include "command/command.h"
#include "mqtt.h"
#include "neuron.h"
#include "util.h"
#define INTERVAL 100000U
#define TIMEOUT 3000U
#define TOPIC_READ_REQ "neuron/%s/read/req"
#define TOPIC_WRITE_REQ "neuron/%s/write/req"
#define TOPIC_READ_RES "neuron/%s/read/resp"
#define TOPIC_WRITE_RES "neuron/%s/write/resp"
#define TOPIC_UPLOAD_RES "neuron/%s/upload"
#define QOS0 0
#define QOS1 1
#define QOS2 2
const neu_plugin_module_t neu_plugin_module;
struct topic_pair {
@@ -37,20 +47,15 @@ struct topic_pair {
int type;
};
typedef struct mqtt_routine {
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
} mqtt_routine_t;
struct neu_plugin {
neu_plugin_common_t common;
bool running;
mqtt_routine_t * routine;
char * config;
};
static neu_plugin_t *plugin_log = NULL;
static char *topics_format(char *format, char *name)
{
char temp[256] = { '\0' };
@@ -73,8 +78,8 @@ static void topics_add(UT_array *topics, char *request, int qos_request,
pair->qos_response = qos_response;
pair->type = type;
utarray_push_back(topics, &pair);
nlog_info("add topic-req:%s, topic-res:%s, count:%d", pair->topic_request,
pair->topic_response, utarray_len(topics));
plog_info(plugin_log, "add topic-req:%s, topic-res:%s, count:%d",
pair->topic_request, pair->topic_response, utarray_len(topics));
}
static void topics_cleanup(UT_array *topics)
@@ -84,8 +89,8 @@ static void topics_cleanup(UT_array *topics)
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
nlog_info("remove topic-req:%s, topic-res:%s", (*pair)->topic_request,
(*pair)->topic_response);
plog_info(plugin_log, "remove topic-req:%s, topic-res:%s",
(*pair)->topic_request, (*pair)->topic_response);
if (NULL != (*pair)->topic_request) {
free((*pair)->topic_request);
}
@@ -257,6 +262,7 @@ static neu_plugin_t *mqtt_plugin_open(void)
{
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
plugin_log = plugin; // for plog
return plugin;
}
@@ -267,6 +273,7 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "success to free plugin:%s", name);
plugin_log = NULL;
free(plugin);
return NEU_ERR_SUCCESS;
}
@@ -382,13 +389,13 @@ static neu_err_code_e write_response(neu_plugin_t * plugin,
return NEU_ERR_MQTT_FAILURE;
}
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_write_response(head, write_data);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_write_response(plugin, head, write_data);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
@@ -414,10 +421,10 @@ static neu_err_code_e read_response(neu_plugin_t * plugin,
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_read_once_response(head, read_data, routine->option.format);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
char * json_str = command_read_once_response(plugin, head, read_data,
routine->option.format);
const char *topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
@@ -443,8 +450,8 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_read_periodic_response(trans_data, routine->option.format);
char *json_str = command_read_periodic_response(plugin, trans_data,
routine->option.format);
if (NULL == json_str) {
return NEU_ERR_MQTT_FAILURE;
}
plugins/mqtt/mqtt.h
+8 -13
View File
@@ -25,21 +25,16 @@ extern "C" {
#endif
#include <neuron.h>
#include <stdlib.h>
#define UNUSED(x) (void) (x)
#define INTERVAL 100000U
#define TIMEOUT 3000U
typedef struct mqtt_routine mqtt_routine_t;
#define TOPIC_READ_REQ "neuron/%s/read/req"
#define TOPIC_WRITE_REQ "neuron/%s/write/req"
#define TOPIC_READ_RES "neuron/%s/read/resp"
#define TOPIC_WRITE_RES "neuron/%s/write/resp"
#define TOPIC_UPLOAD_RES "neuron/%s/upload"
#define QOS0 0
#define QOS1 1
#define QOS2 2
struct neu_plugin {
neu_plugin_common_t common;
bool running;
mqtt_routine_t * routine;
char * config;
};
#ifdef __cplusplus
}