Update neu_manager to forward node setting event to persist adapter

1. Forward set node setting event in neu_manager loop
  2. Handle set node setting event in persist adapter loop
This commit is contained in:
eeff
2022-01-14 19:10:22 +08:00
committed by Parker Liu
parent 96e2829d92
commit e1cd0906db
3 changed files with 171 additions and 13 deletions
+21
View File
@@ -496,6 +496,27 @@ static void adapter_loop(void *arg)
break;
}
case MSG_EVENT_SET_NODE_SETTING: {
neu_cmd_set_node_setting_t *cmd = msg_get_buf_ptr(pay_msg);
neu_persister_t * persister = persister_singleton_get();
char * adapter_name = NULL;
rv = neu_manager_get_node_name_by_id(adapter->manager, cmd->node_id,
&adapter_name);
if (0 != rv) {
free((char *) cmd->setting);
break;
}
rv = neu_persister_store_adapter_setting(persister, adapter_name,
cmd->setting);
if (0 != rv) {
log_error("%s fail to store adapter:%s setting:%s",
adapter->name, adapter_name, cmd->setting);
}
free((char *) cmd->setting);
free(adapter_name);
break;
}
case MSG_EVENT_ADD_PLUGIN:
// fall through
+5 -2
View File
@@ -1326,6 +1326,9 @@ static void manager_loop(void *arg)
case MSG_EVENT_DEL_NODE:
// fall through
case MSG_EVENT_SET_NODE_SETTING:
// fall through
case MSG_EVENT_ADD_PLUGIN:
// fall through
@@ -1349,8 +1352,8 @@ static void manager_loop(void *arg)
char *out_buf = msg_get_buf_ptr(msg_ptr);
memcpy(out_buf, name, len);
nng_msg_set_pipe(out_msg, persist_pipe);
log_info("Forward node/plugin event %d to %s pipe: %d",
msg_type, DEFAULT_PERSIST_ADAPTER_NAME, persist_pipe);
log_info("Forward event %d to %s pipe: %d", msg_type,
DEFAULT_PERSIST_ADAPTER_NAME, persist_pipe);
nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
}
break;
+145 -11
View File
@@ -40,19 +40,123 @@
#include "neu_vector.h"
#include "json/neu_json_fn.h"
#if defined _WIN32 || defined __CYGWIN__
#define PATH_SEP_CHAR '\\'
#else
#define PATH_SEP_CHAR '/'
#endif
#define PATH_MAX_SIZE 128
/**
* Escape special characters in path string,
* '%' -> '%%', '/' => '%-', '.' -> '%.'
*
* @param buf output buffer,
* @param size size of output buffer
* @param path path string
*
* @return if `buf` is NULL, return number of bytes needed to store the result.
* otherwise, return the number of bytes written excluding the
* terminating null byte, `size` indicates overflow.
*/
static int path_escape(char *buf, size_t size, const char *path)
{
size_t i = 0;
char c = '\0';
if (NULL != buf) {
for (const char *s = path; (c = *s++) && i < size; ++i) {
if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
c = (PATH_SEP_CHAR == c) ? '-' : c;
buf[i++] = '%';
if (size == i) {
break;
}
}
buf[i] = c;
}
if (i < size) {
buf[i] = '\0';
} else if (i > 0) {
buf[i - 1] = '\0';
}
} else {
for (const char *s = path; (c = *s++); ++i) {
if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
++i;
}
}
}
return i;
}
/**
* Concatenate a path string to another.
*
* @param dst destination path string buffer
* @param len destination path string len, not greater than size
* @param size destination path buffer size
* @param src path string
*
* @return length of the result path string excluding the terminating NULL
* byte, `size` indicates overflow.
*/
static int path_cat(char *dst, size_t len, size_t size, const char *src)
{
size_t i = len;
if (0 < i && i < size && (PATH_SEP_CHAR != dst[i - 1])) {
dst[i++] = PATH_SEP_CHAR;
}
if (*src && PATH_SEP_CHAR == *src) {
++src;
}
while (i < size && (dst[i] = *src++)) {
++i;
}
if (i == size && i > 0) {
dst[i - 1] = '\0';
}
return i;
}
/**
* Escape and concatenate a path string to another.
*
* @param dst destination path string buffer
* @param len destination path string len, not greater than size
* @param size destination path buffer size
* @param src path string
*
* @return length of the result path string excluding the terminating NULL
* byte, `size` indicates overflow.
*/
static int path_cat_escaped(char *dst, size_t len, size_t size, const char *src)
{
size_t i = len;
if (0 < i && i < size && (PATH_SEP_CHAR != dst[i - 1])) {
dst[i++] = PATH_SEP_CHAR;
dst[i] = '\0';
}
size_t n = path_escape(dst + i, size - i, src);
return i + n;
}
typedef struct neu_persister {
const char *persist_dir;
const char *adapters_fname;
const char *plugins_fname;
} neu_persister_t;
typedef neu_json_node_req_node_t neu_persist_adapter_info_t;
typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
typedef neu_json_datatag_req_tag_t neu_persist_datatag_info_t;
typedef neu_json_group_configs_req_t neu_persist_group_config_info_t;
typedef neu_json_subscriptions_req_subscription_t
neu_persist_subscription_info_t;
static int create_dir(char *dir_name)
{
int rv = mkdir(dir_name, 0777);
@@ -138,6 +242,19 @@ error_dir_name:
return NULL;
}
static inline int persister_adapter_dir(char *buf, size_t size,
neu_persister_t *persister,
const char * adapter_name)
{
size_t n = path_cat(buf, 0, size, persister->persist_dir);
if (size == n) {
return n;
}
n = path_cat_escaped(buf, n, size, adapter_name);
return n;
}
static int write_file_string(const char *fn, const char *s)
{
char tmp[PATH_MAX] = { 0 };
@@ -602,10 +719,27 @@ int neu_persister_store_adapter_setting(neu_persister_t *persister,
const char * adapter_name,
const char * setting)
{
(void *) persister;
(void *) adapter_name;
(void *) setting;
return 0;
char path[PATH_MAX_SIZE] = { 0 };
int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
if (sizeof(path) == n) {
log_error("persister path too long: %s", path);
return -1;
}
if (0 != create_dir(path)) {
log_error("persister failed to create dir: %s", path);
return -1;
}
n = path_cat(path, n, sizeof(path), "settings.json");
if (sizeof(path) == n) {
log_error("persister path too long: %s", path);
return -1;
}
int rv = write_file_string(path, setting);
return rv;
}
int neu_persister_delete_adapter_setting(neu_persister_t *persister,