diff --git a/ft/resource/error.resource b/ft/resource/error.resource index 2d75e1ab..8e5c08af 100644 --- a/ft/resource/error.resource +++ b/ft/resource/error.resource @@ -16,6 +16,8 @@ ${NEU_ERR_FILE_NOT_EXIST} 1011 ${NEU_ERR_INVALID_PASSWORD_LEN} 1012 ${NEU_ERR_DUPLICATE_PASSWORD} 1013 ${NEU_ERR_COMMAND_EXECUTION_FAILED} 1014 +${NEU_ERR_IP_ADDRESS_INVALID} 1015 +${NEU_ERR_IP_ADDRESS_IN_USE} 1016 ${NEU_ERR_NODE_EXIST} 2002 ${NEU_ERR_NODE_NOT_EXIST} 2003 diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index ed4ecc3a..789e8cc1 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -45,6 +45,8 @@ typedef enum { NEU_ERR_INVALID_PASSWORD_LEN = 1012, NEU_ERR_DUPLICATE_PASSWORD = 1013, NEU_ERR_COMMAND_EXECUTION_FAILED = 1014, + NEU_ERR_IP_ADDRESS_INVALID = 1015, + NEU_ERR_IP_ADDRESS_IN_USE = 1016, NEU_ERR_NODE_EXIST = 2002, NEU_ERR_NODE_NOT_EXIST = 2003, diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 9325651c..27cfe3d0 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -17,6 +17,10 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ +#include +#include +#include + #include "errcodes.h" #include "neuron.h" #include "utils/asprintf.h" @@ -99,6 +103,7 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin) { int rv = 0; + nng_close(plugin->sock); nng_aio_free(plugin->recv_aio); nng_mtx_free(plugin->mtx); free(plugin->host); @@ -127,7 +132,14 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin) if ((rv = nng_listen(plugin->sock, url, NULL, 0)) != 0) { plog_error(plugin, "nng_listen: %s", nng_strerror(rv)); nng_close(plugin->sock); - return NEU_ERR_EINTERNAL; + if (NNG_EADDRINVAL == rv) { + rv = NEU_ERR_IP_ADDRESS_INVALID; + } else if (NNG_EADDRINUSE == rv) { + rv = NEU_ERR_IP_ADDRESS_IN_USE; + } else { + rv = NEU_ERR_EINTERNAL; + } + return rv; } nng_recv_aio(plugin->sock, plugin->recv_aio); @@ -163,6 +175,12 @@ static int parse_config(neu_plugin_t *plugin, const char *setting, goto error; } + struct in_addr addr; + if (0 == inet_aton(host.v.val_str, &addr)) { + plog_error(plugin, "inet_aton fail: %s", host.v.val_str); + goto error; + } + // port, required if (0 == port.v.val_int || port.v.val_int > 65535) { plog_error(plugin, "setting invalid port: %" PRIi64, port.v.val_int); @@ -182,6 +200,41 @@ error: return -1; } +static inline int check_url_listenable(neu_plugin_t *plugin, const char *url, + const char *host, uint16_t port) +{ + if (NULL != plugin->host && // already configured and + port == plugin->port && // port is the same, then if + (0 == strcmp(plugin->host, host) // 1. host is the same + || 0 == strcmp("0.0.0.0", host) // 2. to bind to any address + || 0 == strcmp("0.0.0.0", plugin->host) // 3. bound to any address + )) { + // early return, no need check url is listenable + return 0; + } + + nng_socket sock = NNG_SOCKET_INITIALIZER; + int rv = nng_pair0_open(&sock); + if (0 != rv) { + plog_error(plugin, "nng_pair0_open: %s", nng_strerror(rv)); + return NEU_ERR_EINTERNAL; + } + + if (0 != (rv = nng_listen(sock, url, NULL, 0))) { + plog_error(plugin, "nng_listen: %s", nng_strerror(rv)); + if (NNG_EADDRINVAL == rv) { + rv = NEU_ERR_IP_ADDRESS_INVALID; + } else if (NNG_EADDRINUSE == rv) { + rv = NEU_ERR_IP_ADDRESS_IN_USE; + } else { + rv = NEU_ERR_EINTERNAL; + } + } + + nng_close(sock); + return rv; +} + static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting) { int rv = 0; @@ -201,6 +254,10 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting) goto error; } + if (0 != (rv = check_url_listenable(plugin, url, host, port))) { + goto error; + } + plog_notice(plugin, "config success"); free(plugin->host); diff --git a/src/utils/http.c b/src/utils/http.c index de2f9f99..ee3a5d12 100644 --- a/src/utils/http.c +++ b/src/utils/http.c @@ -321,6 +321,8 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content) break; case NEU_ERR_BODY_IS_WRONG: case NEU_ERR_PARAM_IS_WRONG: + case NEU_ERR_IP_ADDRESS_INVALID: + case NEU_ERR_IP_ADDRESS_IN_USE: case NEU_ERR_NODE_SETTING_INVALID: case NEU_ERR_NODE_NOT_ALLOW_UPDATE: case NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE: