GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 04:19:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(restful): add proxy handler for ekuiper

Browse Source
This commit is contained in:
eeff
2022-08-09 14:11:00 +08:00
parent 55c80a87bb
commit 2a1fb73ea1
2 changed files with 184 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/restful/handle.c
+5
View File
@@ -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/proxy.c
+179 -20
View File
@@ -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;
});
}