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

Use of channels to process requests

Browse Source
This commit is contained in:
gc87
2022-01-28 17:29:47 +08:00
committed by Parker Liu
parent c81454fef0
commit 55d69f136b
3 changed files with 89 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command/command.c
+1 -1
View File
@@ -5,7 +5,7 @@
void command_response_handle(mqtt_response_t *response)
{
neu_plugin_t *plugin = (neu_plugin_t *) response->context;
neu_plugin_t *plugin = (neu_plugin_t *) response->plugin;
if (NULL == response->topic_name || NULL == response->payload) {
return;
plugins/mqtt/command/command.h
+18 -1
View File
@@ -33,6 +33,22 @@ extern "C" {
#include "plugin.h"
#include "read_write.h"
enum topic_type {
TOPIC_TYPE_PING = 0,
TOPIC_TYPE_NODE,
TOPIC_TYPE_GCONFIG,
TOPIC_TYPE_TAGS,
TOPIC_TYPE_PLUGIN,
TOPIC_TYPE_SUBSCRIBE,
TOPIC_TYPE_READ,
TOPIC_TYPE_WRITE,
TOPIC_TYPE_TTYS,
TOPIC_TYPE_SCHEMA,
TOPIC_TYPE_SETTING,
TOPIC_TYPE_CTR,
TOPIC_TYPE_STATE,
};
typedef void (*context_add_callback)(neu_plugin_t *plugin, uint32_t req_id,
neu_json_mqtt_t *parse_head, char *result,
bool ready);
@@ -42,7 +58,8 @@ typedef struct {
size_t topic_len;
void * payload;
size_t len;
void * context;
void * plugin;
void * topic_pair;
context_add_callback context_add;
} mqtt_response_t;
plugins/mqtt/plugin_mqtt.c
+70 -33
View File
@@ -59,6 +59,10 @@
#define TOPIC_CTR_RES "neuron/%s/node/ctl/resp"
#define TOPIC_STATE_RES "neuron/%s/node/state/resp"
#define QOS0 0
#define QOS1 1
#define QOS2 2
const neu_plugin_module_t neu_plugin_module;
struct context {
@@ -76,6 +80,7 @@ struct topic_pair {
char * topic_respons;
int qos_request;
int qos_response;
int type;
};
struct neu_plugin {
@@ -208,7 +213,8 @@ static void topic_pair_destory(struct topic_pair *pair)
}
}
static void topic_list_add(neu_list *list, char *request, char *response)
static void topic_list_add(neu_list *list, char *request, int qos_request,
char *response, int qos_response, int type)
{
if (NULL == request) {
return;
@@ -221,8 +227,9 @@ static void topic_list_add(neu_list *list, char *request, char *response)
pair->topic_request = request;
pair->topic_respons = response;
pair->qos_request = 0;
pair->qos_response = 0;
pair->qos_request = qos_request;
pair->qos_response = qos_response;
pair->type = type;
NEU_LIST_NODE_INIT(&pair->node);
neu_list_append(list, pair);
@@ -255,32 +262,57 @@ static void topic_list_cleanup(neu_list *list)
static void topic_list_generate(neu_list *list, char *name)
{
topic_list_add(list, real_topic_generate(TOPIC_PING_REQ, name),
real_topic_generate(TOPIC_STATUS_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_NODE_REQ, name),
real_topic_generate(TOPIC_NODE_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_GCONFIG_REQ, name),
real_topic_generate(TOPIC_GCONFIG_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_TAGS_REQ, name),
real_topic_generate(TOPIC_TAGS_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_PLUGIN_REQ, name),
real_topic_generate(TOPIC_PLUGIN_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_SUBSCRIBE_REQ, name),
real_topic_generate(TOPIC_SUBSCRIBE_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_READ_REQ, name),
real_topic_generate(TOPIC_READ_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_WRITE_REQ, name),
real_topic_generate(TOPIC_WRITE_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_TTYS_REQ, name),
real_topic_generate(TOPIC_TTYS_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_SCHEMA_REQ, name),
real_topic_generate(TOPIC_SCHEMA_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_SETTING_REQ, name),
real_topic_generate(TOPIC_SETTING_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_CTR_REQ, name),
real_topic_generate(TOPIC_CTR_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_STATE_REQ, name),
real_topic_generate(TOPIC_STATE_RES, name));
topic_list_add(list, real_topic_generate(TOPIC_PING_REQ, name), QOS0,
real_topic_generate(TOPIC_STATUS_RES, name), QOS0,
TOPIC_TYPE_PING);
topic_list_add(list, real_topic_generate(TOPIC_NODE_REQ, name), QOS0,
real_topic_generate(TOPIC_NODE_RES, name), QOS0,
TOPIC_TYPE_NODE);
topic_list_add(list, real_topic_generate(TOPIC_GCONFIG_REQ, name), QOS0,
real_topic_generate(TOPIC_GCONFIG_RES, name), QOS0,
TOPIC_TYPE_GCONFIG);
topic_list_add(list, real_topic_generate(TOPIC_TAGS_REQ, name), QOS0,
real_topic_generate(TOPIC_TAGS_RES, name), QOS0,
TOPIC_TYPE_TAGS);
topic_list_add(list, real_topic_generate(TOPIC_PLUGIN_REQ, name), QOS0,
real_topic_generate(TOPIC_PLUGIN_RES, name), QOS0,
TOPIC_TYPE_PLUGIN);
topic_list_add(list, real_topic_generate(TOPIC_SUBSCRIBE_REQ, name), QOS0,
real_topic_generate(TOPIC_SUBSCRIBE_RES, name), QOS0,
TOPIC_TYPE_SUBSCRIBE);
topic_list_add(list, real_topic_generate(TOPIC_READ_REQ, name), QOS0,
real_topic_generate(TOPIC_READ_RES, name), QOS0,
TOPIC_TYPE_READ);
topic_list_add(list, real_topic_generate(TOPIC_WRITE_REQ, name), QOS0,
real_topic_generate(TOPIC_WRITE_RES, name), QOS0,
TOPIC_TYPE_WRITE);
topic_list_add(list, real_topic_generate(TOPIC_TTYS_REQ, name), QOS0,
real_topic_generate(TOPIC_TTYS_RES, name), QOS0,
TOPIC_TYPE_TTYS);
topic_list_add(list, real_topic_generate(TOPIC_SCHEMA_REQ, name), QOS0,
real_topic_generate(TOPIC_SCHEMA_RES, name), QOS0,
TOPIC_TYPE_SCHEMA);
topic_list_add(list, real_topic_generate(TOPIC_SETTING_REQ, name), QOS0,
real_topic_generate(TOPIC_SETTING_RES, name), QOS0,
TOPIC_TYPE_SETTING);
topic_list_add(list, real_topic_generate(TOPIC_CTR_REQ, name), QOS0,
real_topic_generate(TOPIC_CTR_RES, name), QOS0,
TOPIC_TYPE_CTR);
topic_list_add(list, real_topic_generate(TOPIC_STATE_REQ, name), QOS0,
real_topic_generate(TOPIC_STATE_RES, name), QOS0,
TOPIC_TYPE_STATE);
}
static void mqtt_context_add(neu_plugin_t *plugin, uint32_t req_id,
@@ -296,11 +328,15 @@ static void mqtt_response_handle1(const char *topic_name, size_t topic_len,
void *payload, const size_t len,
void *context)
{
neu_plugin_t * plugin = (neu_plugin_t *) context;
struct topic_pair *pair = topic_list_find(&plugin->topic_list, topic_name);
mqtt_response_t response = { .topic_name = topic_name,
.topic_len = topic_len,
.payload = payload,
.len = len,
.context = context,
.plugin = context,
.topic_pair = pair,
.context_add = mqtt_context_add };
command_response_handle(&response);
return;
@@ -312,8 +348,8 @@ static void topic_list_subscribe(neu_list *list, neu_mqtt_client_t *client)
NEU_LIST_FOREACH(list, item)
{
if (NULL != item) {
neu_mqtt_client_subscribe(client, item->topic_request, 0,
mqtt_response_handle1);
neu_mqtt_client_subscribe(client, item->topic_request,
item->qos_request, mqtt_response_handle1);
}
}
}
@@ -383,7 +419,8 @@ static void mqtt_response_handle(const char *topic_name, size_t topic_len,
.topic_len = topic_len,
.payload = payload,
.len = len,
.context = context,
.plugin = context,
.topic_pair = NULL,
.context_add = mqtt_context_add };
command_response_handle(&response);
return;