Merge pull request #701 from eeff/eeff/dev

eKuiper api proxy

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

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
 

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,

plugins/restful/handle.h

@@ -80,6 +80,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 {
@@ -89,6 +90,7 @@ struct neu_rest_handler {
     union neu_rest_handler_value {
         void *handler;
         char *path;
+        char *dst_url;
     } value;
 };
 

plugins/restful/proxy.c

@@ -0,0 +1,254 @@
+/**
+ * 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 "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;
+    rest_proxy_t *proxy = NULL;
+
+    ret = rest_proxy_alloc(&proxy, rest_handler->url,
+                           rest_handler->value.dst_url);
+    if (0 != ret) {
+        return ret;
+    }
+
+    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, proxy,
+                                    (void (*)(void *)) rest_proxy_free);
+    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 != proxy) {
+        rest_proxy_free(proxy);
+    }
+    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);
+    rest_proxy_t *    proxy   = nng_http_handler_get_data(handler);
+
+    // 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;
+    }
+
+    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;
+    });
+}

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 <nng/nng.h>
+#include <nng/supplemental/http/http.h>
+
+#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

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