MQTT Application Development Example

Northbound application development mainly includes the following parts. The bottom layer is the development of the instruction processing layer, and the outermost layer is the development of the application layer.

ModulesDocumentationInstructions
Instruction processing layer developmentcommand.c command.h common.h heartbeat.c heartbeat.h read_write.c read_write.hPlugin parsing of directives
Application layer developmentmqtt_plugin.c mqtt_plugin.h mqtt_util.c mqtt_util.hImplementation of plugin theme framework
Plugin Settings Filemqtt.jsonDefinition of Plugin Settings File

Constant Description

ConstantDescription
TOPIC_TYPE_READTopic type is read
TOPIC_TYPE_WRITETopic type is write
TOPIC_TYPE_UPLOADTopic type is upload
TOPIC_TYPE_HEARTBEATTopic type is heartbeat

Step 1 Instruction Processing Layer Development

MQTT currently implements the interfaces for uploading data, heartbeat data, reading Tags and writing Tags. The specific processing of the response corresponding to the mqtt request is defined in command.c, and heartbeat.c and read_write.c provide the required function implementation for command.c.

FunctionDescription
command_response_handleMQTT response processing
command_read_once_responseResponse handling for read topics
command_read_periodic_responseResponse handling for uploading topics
command_write_responseResponse handling fot write topics
command_heartbeat_responseResponse processing of heartbeat data

Step 2 Driver Layer Development

The mqtt_util.c and mqtt_util.h files define the specific function implementations used in the mqtt.c file.

In the development of the north-south driver layer,it is necessary to constrict the structure of neu_plugin_intf_funs_t first, and realize the function of each element in the structure.

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

Call the mqtt_plugin_open function, and create a structure struct neu_plugin defined by the plugin itself based on the first function called by neuron when the plugin creates a node. The structure is defined in mqtt_plugin.h. It should be noted that the first member of the structure must be neu_plugin_common_t common, and other members can be added according to the specific implementation of the driver.

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

The mqtt_plugin_close function is called, the last function called by neuron when the node is deleted, to release the neu_plugin_t created by open.

c

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

.init

Call the mqtt_plugin_init function, which is called immediately after neuron calls open when creating a node. This function is mainly used for some resources that need to be initialized in the plug-in. The mqtt plug-in mainly initializes the running status and configuration of 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

Call the mqtt_plugin_uninit function, the first function called by neuron when the node is deleted. This function mainly releases some resources applied and initialized in 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

Call the mqtt_plugin_start function. When the user clicks start on the neuron node page, neuron will call this function to notify the plug-in to start running and start connecting to the device. If the configuration is incorrect, it will return the node setting invalid error.

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

Call the mqtt_plugin_stop function, when the user clicks stop on the neuron node page, neuron will call this function, stop notifies the plug-in to stop running, and closes the connection between the plug-in and 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

Call the mqtt_plugin_config function, which is used by the user when making settings on the neuron node setting page. The parameters set by the node will be presented in json (for the configuration of the json file, please refer to Plugin Setting File ), and neuron will notify the plugin to set through this function. The mqtt_plugin_config function first parses and saves the configuration information, and then establishes the connection.

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

Call the mqtt_plugin_request function to handle the response according to the request type.

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. }

Step 3 Plugin Setting File

The mqtt.json file sets the application configuration parameters. The field descriptions for each parameter of the mqtt plugin are as follows.

ParametersDescription
nameThe page displays the name of the parameter
descriptionThe specific description of this parameter
typeThe type of the parameter, currently commonly used two types: int and string
attributeThe attributes of this parameter, there are only two optional and required, namely required and optional
defaultSet the default value of this parameter
validThe range that this parameter can fill in, use length for string type, max and min for int type
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. }