fix(mqtt): add config validate

plugins/mqtt/mqtt_plugin.c

@@ -368,8 +368,13 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
     assert(NULL != plugin);
     assert(NULL != config);
 
+    if (NEU_ERR_SUCCESS != mqtt_option_validate(plugin, config)) {
+        return NEU_ERR_NODE_SETTING_INVALID;
+    }
+
     plugin_config_save(plugin, config);
 
+    // Routine not running
     if (!plugin->running) {
         return NEU_ERR_SUCCESS;
     }
@@ -377,7 +382,7 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
     plugin_stop_running(plugin);
     int rc = plguin_start_running(plugin);
     if (0 != rc) {
-        return NEU_ERR_NODE_SETTING_INVALID;
+        return NEU_ERR_MQTT_INIT_FAILURE;
     }
 
     plog_info(plugin, "config plugin: %s", neu_plugin_module.module_name);

plugins/mqtt/mqtt_util.c

@@ -313,4 +313,18 @@ const char *mqtt_option_error(int error)
     } else {
         return mqtt_option_error_strs[9];
     }
+}
+
+int mqtt_option_validate(neu_plugin_t *plugin, const char *config)
+{
+    neu_mqtt_option_t option = { 0 };
+    int               rc = mqtt_option_init(plugin, (char *) config, &option);
+    if (0 != rc) {
+        plog_error(plugin, "%s", mqtt_option_error(rc));
+        mqtt_option_uninit(plugin, &option);
+        return NEU_ERR_NODE_SETTING_INVALID;
+    }
+
+    mqtt_option_uninit(plugin, &option);
+    return NEU_ERR_SUCCESS;
 }

plugins/mqtt/mqtt_util.h

@@ -32,6 +32,8 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config,
 
 const char *mqtt_option_error(int error);
 
+int mqtt_option_validate(neu_plugin_t *plugin, const char *config);
+
 void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option);
 
 #ifdef __cplusplus