From 55c80a87bb6fda9ad6c4c561167a1cbcbc3467c9 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 8 Aug 2022 17:07:49 +0800 Subject: [PATCH 1/3] feat(restful): add proxy handler --- CMakeLists.txt | 1 + plugins/restful/handle.h | 2 + plugins/restful/proxy.c | 95 ++++++++++++++++++++++++++++++++++++++++ plugins/restful/proxy.h | 40 +++++++++++++++++ plugins/restful/rest.c | 5 +++ 5 files changed, 143 insertions(+) create mode 100644 plugins/restful/proxy.c create mode 100644 plugins/restful/proxy.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 0dff0ea0..c4452f72 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -115,6 +115,7 @@ set(NEURON_SOURCES plugins/restful/version_handle.c plugins/restful/rest.c plugins/restful/http.c + plugins/restful/proxy.c plugins/restful/websocket.c ${PERSIST_SOURCES}) diff --git a/plugins/restful/handle.h b/plugins/restful/handle.h index 86bb4eab..21f6cc3c 100644 --- a/plugins/restful/handle.h +++ b/plugins/restful/handle.h @@ -84,6 +84,7 @@ enum neu_rest_method { enum neu_rest_handler_type { NEU_REST_HANDLER_FUNCTION = 0x0, NEU_REST_HANDLER_DIRECTORY, + NEU_REST_HANDLER_PROXY, }; struct neu_rest_handler { @@ -93,6 +94,7 @@ struct neu_rest_handler { union neu_rest_handler_value { void *handler; char *path; + char *dst_url; } value; }; diff --git a/plugins/restful/proxy.c b/plugins/restful/proxy.c new file mode 100644 index 00000000..0d9c17aa --- /dev/null +++ b/plugins/restful/proxy.c @@ -0,0 +1,95 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#include "proxy.h" +#include "utils/log.h" + +int rest_proxy_handler(const struct neu_rest_handler *rest_handler, + nng_http_handler ** handler) +{ + int ret = 0; + const char * dst = rest_handler->value.dst_url; + nng_url * url = NULL; + nng_http_client *client = NULL; + + ret = nng_url_parse(&url, rest_handler->value.dst_url); + if (0 != ret) { + nlog_error("parse url: `%s` fail", dst); + return ret; + } + + ret = nng_http_client_alloc(&client, url); + if (0 != ret) { + nlog_error("alloc http client for `%s` fail", dst); + goto error; + } + + ret = nng_http_handler_alloc(handler, rest_handler->url, handle_proxy); + if (0 != ret) { + nlog_error("alloc proxy handler for `%s` fail", dst); + goto error; + } + + ret = nng_http_handler_set_tree(*handler); + if (0 != ret) { + nlog_error("proxy handler for `%s` set tree fail", dst); + goto error; + } + + ret = nng_http_handler_set_method(*handler, NULL); + if (0 != ret) { + nlog_error("proxy handler for `%s` set all method fail", dst); + goto error; + } + + ret = nng_http_handler_set_data(*handler, client, NULL); + if (0 != ret) { + nlog_error("proxy handler for `%s` set data fail", dst); + goto error; + } + + return ret; + +error: + if (NULL != url) { + nng_url_free(url); + } + if (NULL != *handler) { + nng_http_handler_free(*handler); + } + if (NULL != client) { + nng_http_client_free(client); + } + return ret; +} + +void handle_proxy(nng_aio *aio) +{ + nng_http_res * res = NULL; + nng_http_req * req = nng_aio_get_input(aio, 0); + nng_http_handler *handler = nng_aio_get_input(aio, 1); + nng_http_client * client = nng_http_handler_get_data(handler); + + if (0 != nng_http_res_alloc(&res)) { + nlog_error("alloc http resp fail"); + return; + } + + nng_http_client_transact(client, req, res, aio); +} diff --git a/plugins/restful/proxy.h b/plugins/restful/proxy.h new file mode 100644 index 00000000..461811d3 --- /dev/null +++ b/plugins/restful/proxy.h @@ -0,0 +1,40 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEU_PLUGIN_REST_PROXY_H +#define NEU_PLUGIN_REST_PROXY_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include +#include + +#include "handle.h" + +int rest_proxy_handler(const struct neu_rest_handler *rest_handler, + nng_http_handler ** handler); +void handle_proxy(nng_aio *aio); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 2cbe3d2e..5444740c 100644 --- a/plugins/restful/rest.c +++ b/plugins/restful/rest.c @@ -31,6 +31,7 @@ #include "handle.h" #include "http.h" #include "plugin_handle.h" +#include "proxy.h" #include "rest.h" #include "rw_handle.h" #include "utils/log.h" @@ -62,6 +63,9 @@ static int rest_add_handler(nng_http_server * server, ret = nng_http_handler_alloc_directory(&handler, rest_handler->url, rest_handler->value.path); break; + case NEU_REST_HANDLER_PROXY: + ret = rest_proxy_handler(rest_handler, &handler); + goto end; } if (ret != 0) { @@ -91,6 +95,7 @@ static int rest_add_handler(nng_http_server * server, break; } +end: if (ret != 0) { return -1; } From 2a1fb73ea172d7631e7068739eeac344cd12fdb3 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 9 Aug 2022 14:11:00 +0800 Subject: [PATCH 2/3] feat(restful): add proxy handler for ekuiper --- plugins/restful/handle.c | 5 + plugins/restful/proxy.c | 199 +++++++++++++++++++++++++++++++++++---- 2 files changed, 184 insertions(+), 20 deletions(-) diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index c292323a..c22327a9 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -109,6 +109,11 @@ struct neu_rest_handler cors_handler[] = { }; struct neu_rest_handler api_handlers[] = { + { + .type = NEU_REST_HANDLER_PROXY, + .url = "/api/v2/ekuiper", + .value.dst_url = "http://127.0.0.1:9081", + }, { .method = NEU_REST_METHOD_POST, .type = NEU_REST_HANDLER_FUNCTION, diff --git a/plugins/restful/proxy.c b/plugins/restful/proxy.c index 0d9c17aa..96cdc309 100644 --- a/plugins/restful/proxy.c +++ b/plugins/restful/proxy.c @@ -18,28 +18,157 @@ **/ #include "proxy.h" +#include "errcodes.h" +#include "http.h" #include "utils/log.h" +#include "utils/utlist.h" +#include "json/neu_json_fn.h" + +struct rest_proxy_ctx; + +typedef struct { + size_t uri_offset; // offset at which destination uri starts + nng_http_client * client; // client to destination + struct rest_proxy_ctx *free_list; // context free list +} rest_proxy_t; + +struct rest_proxy_ctx { + rest_proxy_t * proxy; + nng_aio * aio; + nng_aio * user_aio; // aio to write back response + nng_http_res * res; // http response from the destination + struct rest_proxy_ctx *prev; + struct rest_proxy_ctx *next; +}; + +static void rest_proxy_ctx_free(struct rest_proxy_ctx *ctx) +{ + if (NULL == ctx) { + return; + } + nng_aio_free(ctx->aio); +} + +static int rest_proxy_alloc(rest_proxy_t **proxy, const char *src_url, + const char *dst_url) +{ + int ret = 0; + nng_url *url = NULL; + + rest_proxy_t *p = calloc(1, sizeof(rest_proxy_t)); + if (NULL == p) { + return -1; + } + + ret = nng_url_parse(&url, dst_url); + if (0 != ret) { + free(p); + nlog_error("parse url: `%s` fail", dst_url); + return ret; + } + + ret = nng_http_client_alloc(&p->client, url); + if (0 != ret) { + nng_url_free(url); + free(p); + nlog_error("alloc http client for `%s` fail", dst_url); + return ret; + } + + p->uri_offset = strlen(src_url); + *proxy = p; + nng_url_free(url); + return ret; +} + +static void rest_proxy_free(rest_proxy_t *proxy) +{ + if (NULL == proxy) { + return; + } + nng_http_client_free(proxy->client); + + struct rest_proxy_ctx *el = NULL, *tmp = NULL; + DL_FOREACH_SAFE(proxy->free_list, el, tmp) + { + DL_DELETE(proxy->free_list, el); + nng_aio_free(el->aio); + nng_aio_free(el->user_aio); + free(el); + } + + free(proxy); +} + +static void rest_proxy_cb(void *arg) +{ + int rv = 0; + struct rest_proxy_ctx *ctx = arg; + + nng_aio_set_output(ctx->user_aio, 0, ctx->res); + + if (0 != (rv = nng_aio_result(ctx->aio))) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, { + if (0 != + (rv = nng_http_res_set_status( + ctx->res, NNG_HTTP_STATUS_SERVICE_UNAVAILABLE)) || + 0 != + (rv = nng_http_res_set_header(ctx->res, "Content-Type", + "application/json")) || + 0 != + (rv = nng_http_res_copy_data(ctx->res, result_error, + strlen(result_error)))) { + nng_aio_set_output(ctx->user_aio, 0, NULL); + nng_http_res_free(ctx->res); + } + }); + } + + nng_aio_finish(ctx->user_aio, rv); + + ctx->user_aio = NULL; + ctx->res = NULL; + DL_APPEND(ctx->proxy->free_list, ctx); // put to free list +} + +struct rest_proxy_ctx *rest_proxy_get_ctx(rest_proxy_t *proxy) +{ + struct rest_proxy_ctx *ctx = proxy->free_list; + if (NULL != ctx) { + DL_DELETE(proxy->free_list, ctx); + return ctx; + } + + ctx = calloc(1, sizeof(*ctx)); + if (NULL == ctx) { + nlog_error("alloc proxy ctx fail"); + return NULL; + } + + if (0 != nng_aio_alloc(&ctx->aio, rest_proxy_cb, ctx)) { + nlog_error("alloc proxy ctx aio fail"); + free(ctx); + return NULL; + } + + ctx->proxy = proxy; + return ctx; +} int rest_proxy_handler(const struct neu_rest_handler *rest_handler, nng_http_handler ** handler) { - int ret = 0; - const char * dst = rest_handler->value.dst_url; - nng_url * url = NULL; - nng_http_client *client = NULL; + int ret = 0; + const char * dst = rest_handler->value.dst_url; + nng_url * url = NULL; + rest_proxy_t *proxy = NULL; - ret = nng_url_parse(&url, rest_handler->value.dst_url); + ret = rest_proxy_alloc(&proxy, rest_handler->url, + rest_handler->value.dst_url); if (0 != ret) { - nlog_error("parse url: `%s` fail", dst); return ret; } - ret = nng_http_client_alloc(&client, url); - if (0 != ret) { - nlog_error("alloc http client for `%s` fail", dst); - goto error; - } - ret = nng_http_handler_alloc(handler, rest_handler->url, handle_proxy); if (0 != ret) { nlog_error("alloc proxy handler for `%s` fail", dst); @@ -58,7 +187,8 @@ int rest_proxy_handler(const struct neu_rest_handler *rest_handler, goto error; } - ret = nng_http_handler_set_data(*handler, client, NULL); + ret = nng_http_handler_set_data(*handler, proxy, + (void (*)(void *)) rest_proxy_free); if (0 != ret) { nlog_error("proxy handler for `%s` set data fail", dst); goto error; @@ -73,8 +203,8 @@ error: if (NULL != *handler) { nng_http_handler_free(*handler); } - if (NULL != client) { - nng_http_client_free(client); + if (NULL != proxy) { + rest_proxy_free(proxy); } return ret; } @@ -84,12 +214,41 @@ void handle_proxy(nng_aio *aio) nng_http_res * res = NULL; nng_http_req * req = nng_aio_get_input(aio, 0); nng_http_handler *handler = nng_aio_get_input(aio, 1); - nng_http_client * client = nng_http_handler_get_data(handler); + rest_proxy_t * proxy = nng_http_handler_get_data(handler); - if (0 != nng_http_res_alloc(&res)) { - nlog_error("alloc http resp fail"); - return; + // fix the uri + const char *uri = nng_http_req_get_uri(req); + if (strlen(uri) == proxy->uri_offset) { + uri = "/"; + } else { + uri = uri + proxy->uri_offset; + } + if (0 != nng_http_req_set_uri(req, uri)) { + nlog_error("proxy fail set req uri: `%s`", uri); + goto error; } - nng_http_client_transact(client, req, res, aio); + struct rest_proxy_ctx *ctx = rest_proxy_get_ctx(proxy); + if (NULL == ctx) { + nlog_error("get proxy ctx fail"); + goto error; + } + + if (0 != nng_http_res_alloc(&res)) { + nlog_error("alloc proxy resp fail"); + rest_proxy_ctx_free(ctx); + goto error; + } + + ctx->user_aio = aio; + ctx->res = res; + + nng_http_client_transact(proxy->client, req, ctx->res, ctx->aio); + return; + +error: + NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, { + http_response(aio, error_code.error, result_error); + return; + }); } From 1263d7f636799b40d3812f9209c3815d569d1cff Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 9 Aug 2022 16:10:36 +0800 Subject: [PATCH 3/3] chore: update plugin description length to 512 --- include/neuron/define.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/neuron/define.h b/include/neuron/define.h index f29003c3..539c5f3f 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -28,7 +28,7 @@ #define NEU_NODE_NAME_LEN 32 #define NEU_PLUGIN_NAME_LEN 32 #define NEU_PLUGIN_LIBRARY_LEN 32 -#define NEU_PLUGIN_DESCRIPTION_LEN 256 +#define NEU_PLUGIN_DESCRIPTION_LEN 512 #define NEU_DRIVER_TAG_CACHE_EXPIRE_TIME 30 #define NEU_APP_SUBSCRIBE_MSG_SIZE 4