MQTT 插件开发示例

北向应用开发主要包含以下几个部分，最底层的是指令处理层开发，最外层的是应用层开发。

模块文件说明
指令处理层开发command.c command.h common.h heartbeat.c heartbeat.h read_write.c read_write.h插件对指令的解析
应用层开发mqtt_plugin.c mqtt_plugin.h mqtt_util.c mqtt_util.h插件主题框架的实现
插件设置文件mqtt.json插件设置文件的定义

常量说明

常量说明
TOPIC_TYPE_READ主题类型为读
TOPIC_TYPE_WRITE主题类型为写
TOPIC_TYPE_UPLOAD主题类型为上传
TOPIC_TYPE_HEARTBEAT主题类型为心跳包

第一步，指令处理层开发

MQTT 目前实现了上传数据、心跳数据、读 Tags 和写 Tags 的接口。command.c 中定义了 mqtt 请求对应响应的具体处理， heartbeat.c 和 read_write.c 为 command.c 提供需要的函数实现。

函数说明
command_response_handlemqtt 响应处理
command_read_once_response读主题的响应处理
command_read_periodic_response上传主题的响应处理
command_write_response写主题的响应处理
command_heartbeat_response心跳数据的响应处理

第二步，应用层开发

mqtt_util.c 和 mqtt_util.h 文件定义 mqtt_plugin.c 文件中使用的具体函数实现。

南北向驱动层开发中都需要先构建 neu_plugin_intf_funs_t 的结构体，并实现结构体中的每个元素的功能。

c

  1. static const neu_plugin_intf_funs_t plugin_intf_funs = {
  2.     .open    = mqtt_plugin_open,
  3.     .close   = mqtt_plugin_close,
  4.     .init    = mqtt_plugin_init,
  5.     .uninit  = mqtt_plugin_uninit,
  6.     .start   = mqtt_plugin_start,
  7.     .stop    = mqtt_plugin_stop,
  8.     .setting = mqtt_plugin_config,
  9.     .request = mqtt_plugin_request,
  10. };

.open

调用 mqtt_plugin_open 函数，基于 plugin 创建 node 时 neuron 第一个调用的函数，创建插件自己定义的结构体 struct neu_plugin。该结构体在 mqtt_plugin.h 中定义，需要注意的是结构体中的第一个成员一定是 neu_plugin_common_t common，其他成员可根据驱动的具体实现增加。

c

  1. static neu_plugin_t *mqtt_plugin_open(void)
  2. {
  3.     neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
  5.     neu_plugin_common_init(&plugin->common);
  7.     return plugin;
  8. }

.close

调用 mqtt_plugin_close 函数，删除 node 时，neuron 调用的最后一个函数，用于释放由 open 创建的 neu_plugin_t。

c

  1. static int mqtt_plugin_close(neu_plugin_t *plugin)
  2. {
  3.     free(plugin);
  5.     return NEU_ERR_SUCCESS;
  6. }

.init

调用 mqtt_plugin_init 函数，在创建 node 时，neuron 调用完 open 后，紧接着调用的函数。此函数主要做插件内需要初始化的一些资源，mqtt 插件中主要初始化 mqtt 的运行状态及配置。

c

  1. static int mqtt_plugin_init(neu_plugin_t *plugin)
  2. {
  3.     assert(NULL != plugin);
  5.     plugin->routine = NULL;
  6.     plugin->running = false;
  7.     plugin->config  = NULL;
  8.     plugin_cache_init(plugin);
  10.     const char *name = neu_plugin_module.module_name;
  11.     plog_info(plugin, "initialize plugin: %s", name);
  13.     return NEU_ERR_SUCCESS;
  14. }

.uninit

调用 mqtt_plugin_uninit 函数，删除 node 时，neuron 首先调用的函数，此函数主要释放一些在 init 中申请以及初始化的资源。

c

  1. static int mqtt_plugin_uninit(neu_plugin_t *plugin)
  2. {
  3.     assert(NULL != plugin);
  5.     plugin_stop_running(plugin);
  6.     plugin_cache_uninit(plugin);
  7.     plugin_config_free(plugin);
  9.     const char *name = neu_plugin_module.module_name;
  10.     plog_info(plugin, "uninitialize plugin: %s", name);
  11.     return NEU_ERR_SUCCESS;
  12. }

.start

调用 mqtt_plugin_start 函数，用户在 neuron node 页面，点击开始时，neuron 会调用此函数，通知插件开始运行，以及开始连接设备等，如果配置不正确，将会返回节点设置无效的错误。

mqtt-start

c

  1. static int mqtt_plugin_start(neu_plugin_t *plugin)
  2. {
  3.     assert(NULL != plugin);
  5.     int rc = plguin_start_running(plugin);
  6.     if (0 != rc) {
  7.         return NEU_ERR_NODE_SETTING_INVALID;
  8.     }
  10.     return NEU_ERR_SUCCESS;
  11. }

.stop

调用 mqtt_plugin_stop 函数，用户在 neuron node 页面，点击停止时，neuron 会调用此函数，stop 通知插件停止运行，关闭插件与 neuron 之间的连接。

mqtt-stop

c

  1. static int mqtt_plugin_stop(neu_plugin_t *plugin)
  2. {
  3.     assert(NULL != plugin);
  5.     plugin_stop_running(plugin);
  6.     return NEU_ERR_SUCCESS;
  7. }

.setting

调用 mqtt_plugin_config 函数，用户在 neuron node 设置页面进行设置时使用，node 设置的参数将通过 json 方式呈现（json 文件的配置请参考 插件设置文件 ），neuron 将通过此函数通知插件进行设置。mqtt_plugin_config 函数首先会解析并保存配置信息，然后建立连接。

mqtt-config

c

  1. static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
  2. {
  3.     plog_info(plugin, "config: %s", config);
  5.     return NEU_ERR_SUCCESS;
  6. }

.request

调用 mqtt_plugin_request 函数，根据请求类型对应响应处理。

c

  1. static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
  2.                                void *data)
  3. {
  4.     assert(NULL != plugin);
  5.     assert(NULL != head);
  6.     assert(NULL != data);
  8.     neu_err_code_e error = NEU_ERR_SUCCESS;
  10.     switch (head->type) {
  11.     case NEU_RESP_ERROR:
  12.         error = write_response(plugin, head, data);
  13.         break;
  14.     case NEU_RESP_READ_GROUP:
  15.         error = read_response(plugin, head, data);
  16.         break;
  17.     case NEU_REQRESP_TRANS_DATA:
  18.         error = trans_data(plugin, data);
  19.         break;
  20.     case NEU_REQRESP_NODES_STATE: {
  21.         error = node_state_send(plugin, head, data);
  22.         break;
  23.     }
  24.     case NEU_REQRESP_NODE_DELETED:
  25.         break;
  26.     default:
  27.         error = NEU_ERR_MQTT_FAILURE;
  28.         break;
  29.     }
  31.     return error;
  32. }

第三步，插件设置文件

mqtt.json 文件设置应用配置参数，配置 mqtt 插件每个参数的字段说明如下所示。

参数说明
name页面显示该参数的名称
description该参数的具体说明
type该参数的类型，目前常用的是 int 和 string 两种类型
attribute该参数的属性，只有两种可选和必选，即 required 和 optional
default设置该参数的默认值
valid该参数可填写的范围，string 类型用 length，int 类型使用 max 和 min
map用于设置选项框

json

  1. {
  2.     "upload-topic": {
  3.         "name": "upload topic",
  4.         "description": "User defined upload topic",
  5.         "type": "string",
  6.         "attribute": "required",
  7.         "default": "/neuron/${node-name}/upload",
  8.         "valid": {
  9.             "length": 255
  10.         }
  11.     },
  12.     "heartbeat-topic": {
  13.         "name": "heartbeat topic",
  14.         "description": "User defined heartbeat topic",
  15.         "type": "string",
  16.         "attribute": "required",
  17.         "default": "/neuron/${node-name}/heartbeat",
  18.         "valid": {
  19.             "length": 255
  20.         }
  21.     },
  22.     "format": {
  23.         "name": "upload format",
  24.         "description": "The json format of the data reported in the upload topic. In the values mode, all items are contained in the values object or the errors object, respectively. In tags mode, all items are contained in an array",
  25.         "attribute": "optional",
  26.         "type": "map",
  27.         "default": 0,
  28.         "valid": {
  29.             "map": [
  30.                 {
  31.                     "key": "format-values",
  32.                     "value": 0
  33.                 },
  34.                 {
  35.                     "key": "format-tags",
  36.                     "value": 1
  37.                 }
  38.             ]
  39.         }
  40.     },
  41.     "cache": {
  42.         "name": "cache size(MB)",
  43.         "description": "The maximum byte limit in MB for the data backlog when an MQTT connection exception occurs",
  44.         "type": "int",
  45.         "attribute": "optional",
  46.         "default": 64,
  47.         "valid": {
  48.             "min": 1,
  49.             "max": 256 
  50.         }
  51.     },
  52.     "ssl": {
  53.         "name": "ssl",
  54.         "description": "Enable SSL connection",
  55.         "attribute": "optional",
  56.         "type": "bool",
  57.         "default": false,
  58.         "valid": {}
  59.     },
  60.     "host": {
  61.         "name": "host",
  62.         "description": "MQTT broker host",
  63.         "attribute": "required",
  64.         "type": "string",
  65.         "default": "broker.emqx.io",
  66.         "valid": {
  67.             "length": 255
  68.         }
  69.     },
  70.     "port": {
  71.         "name": "port",
  72.         "description": "MQTT broker port",
  73.         "attribute": "required",
  74.         "type": "int",
  75.         "default": 1883,
  76.         "valid": {
  77.             "min": 1024,
  78.             "max": 65535
  79.         }
  80.     },
  81.     "username": {
  82.         "name": "username",
  83.         "description": "User name",
  84.         "attribute": "optional",
  85.         "type": "string",
  86.         "default": "",
  87.         "valid": {
  88.             "length": 255
  89.         }
  90.     },
  91.     "password": {
  92.         "name": "password",
  93.         "description": "Password",
  94.         "attribute": "optional",
  95.         "type": "string",
  96.         "default": "",
  97.         "valid": {
  98.             "length":255 
  99.         }
  100.     },
  101.     "ca": {
  102.         "name": "CA",
  103.         "description": "CA certificate file",
  104.         "attribute": "required",
  105.         "type": "file",
  106.         "condition": {
  107.             "field": "ssl",
  108.             "value": true
  109.         },
  110.         "default": "",
  111.         "valid": {
  112.             "length": 81960
  113.         }
  114.     },
  115.     "cert": {
  116.         "name": "client cert",
  117.         "description": "client x509 certificate file",
  118.         "attribute": "optional",
  119.         "type": "file",
  120.         "condition": {
  121.             "field": "ssl",
  122.             "value": true
  123.         },
  124.         "default": "",
  125.         "valid": {
  126.             "length": 81960
  127.         }
  128.     },
  129.     "key": {
  130.         "name": "client key",
  131.         "description": "client key file",
  132.         "attribute": "optional",
  133.         "type": "file",
  134.         "condition": {
  135.             "field": "ssl",
  136.             "value": true
  137.         },
  138.         "default": "",
  139.         "valid": {
  140.             "length": 81960
  141.         }
  142.     },
  143.     "keypass": {
  144.         "name": "keypass",
  145.         "description": "key password",
  146.         "attribute": "optional",
  147.         "type": "string",
  148.         "condition": {
  149.             "field": "ssl",
  150.             "value": true
  151.         },
  152.         "default": "",
  153.         "valid": {
  154.             "length": 256
  155.         }
  156.     }
  157. }