GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 15:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Adapt web server to neuron adapter

Browse Source
This commit is contained in:
alvin1221
2021-08-04 14:53:19 +08:00
committed by Parker Liu
parent b0aae692fd
commit f0fbf9481b
4 changed files with 353 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.clang-format
+2
View File
@@ -12,3 +12,5 @@ PointerAlignment: Right
SpaceAfterCStyleCast: true
AllowShortFunctionsOnASingleLine: Inline
BreakBeforeBinaryOperators: None
AlignAfterOpenBracket: true
CMakeLists.txt
+2 -1
View File
@@ -30,7 +30,8 @@ set(NEURON_SOURCES
src/core/plugin_manager.c
src/core/datatag_table.c
src/adapter/adapter.c
src/adapter/web_server.c)
# src/adapter/web_server.c
)
set(NEURON_INCLUDE_DIRECTORIES include src deps)
add_executable(neuron)
src/adapter/web_server.c
+349 -21
View File
@@ -17,23 +17,29 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#define INPROC_URL "inproc://rest"
#define INPROC_URL "inproc://web_server"
#define REST_URL "http://0.0.0.0:%u/api"
#define DEFAULT_PORT 7000
#include <nng/nng.h>
#include <nng/protocol/pair1/pair.h>
#include <nng/protocol/reqrep0/rep.h>
#include <nng/protocol/reqrep0/req.h>
#include <nng/supplemental/http/http.h>
#include <nng/supplemental/util/platform.h>
#include <ctype.h>
#include <dlfcn.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "adapter_internal.h"
#include "neu_log.h"
#include "neu_panic.h"
#include "neu_plugin.h"
#include "web_server.h"
// utility function
@@ -73,7 +79,9 @@ nng_socket req_sock;
nng_mtx * job_lock;
rest_job *job_freelist;
nng_mtx *mtx_log;
nng_thread *inproc_thr;
nng_mtx * mtx_log;
FILE * weblog;
static void rest_job_cb(void *arg);
@@ -128,7 +136,7 @@ static void rest_http_fatal(rest_job *job, const char *fmt, int rv)
job->http_res = NULL;
job->http_aio = NULL;
snprintf(buf, sizeof(buf), fmt, nng_strerror(rv));
log_fatal("%s",buf);
log_fatal("%s", buf);
nng_http_res_set_status(res, NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR);
nng_http_res_set_reason(res, buf);
nng_aio_set_output(aio, 0, res);
@@ -371,10 +379,10 @@ void inproc_server(void *arg)
log_debug("token: %.*s", hmsg.token_len, hmsg.token);
log_debug("data: %.*s", hmsg.data_len, hmsg.data);
// TODO complete the logic
// TODO logic process area
if (strstr((const char *) hmsg.data, "func")) {
nng_msg_clear(msg);
char *res = "get 'func'";
char *res = "reply 'func'";
nng_msg_append(msg, res, strlen(res));
if ((rv = nng_sendmsg(s, msg, 0)) != 0) {
fatal("inproc sendmsg", rv);
@@ -407,35 +415,355 @@ static void log_lock(bool lock, void *udata)
nng_mtx_unlock(LOCK);
}
int adapter_loop(int argc, char **argv)
typedef enum adapter_state {
ADAPTER_STATE_NULL,
ADAPTER_STATE_IDLE,
ADAPTER_STATE_STARTING,
ADAPTER_STATE_RUNNING,
ADAPTER_STATE_STOPPING,
} adapter_state_e;
struct neu_adapter {
uint32_t id;
adapter_type_e type;
nng_mtx * mtx;
adapter_state_e state;
bool stop;
char * name;
neu_manager_t * manager;
nng_pipe pipe;
nng_socket sock;
nng_thread * thrd;
uint32_t new_req_id;
char * plugin_lib_name;
void * plugin_lib; // handle of dynamic lib
neu_plugin_module_t *plugin_module;
neu_plugin_t * plugin;
adapter_callbacks_t cb_funs;
};
static uint32_t adapter_get_req_id(neu_adapter_t *adapter)
{
int rv;
nng_thread *inproc_thr;
uint16_t port = 0;
uint32_t req_id;
req_id = adapter->new_req_id++;
return req_id;
}
static void adapter_loop(void *arg)
{
int rv;
neu_adapter_t *adapter;
adapter = (neu_adapter_t *) arg;
const char *manager_url;
manager_url = neu_manager_get_url(adapter->manager);
rv = nng_dial(adapter->sock, manager_url, NULL, 0);
if (rv != 0) {
neu_panic("The adapter can't dial to %s", manager_url);
}
nng_mtx_lock(adapter->mtx);
adapter->state = ADAPTER_STATE_IDLE;
nng_mtx_unlock(adapter->mtx);
const char *adapter_str = "adapter started";
nng_msg * out_msg;
size_t msg_size;
msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1);
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t *msg_ptr;
char * buf_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, MSG_EVENT_STATUS_STRING, msg_size);
buf_ptr = msg_get_buf_ptr(msg_ptr);
memcpy(buf_ptr, adapter_str, strlen(adapter_str));
buf_ptr[strlen(adapter_str)] = 0;
nng_sendmsg(adapter->sock, out_msg, 0);
}
while (1) {
nng_msg *msg;
nng_mtx_lock(adapter->mtx);
if (adapter->stop) {
adapter->state = ADAPTER_STATE_NULL;
nng_mtx_unlock(adapter->mtx);
log_info("Exit loop of the adapter(%s)", adapter->name);
break;
}
nng_mtx_unlock(adapter->mtx);
rv = nng_recvmsg(adapter->sock, &msg, 0);
if (rv != 0) {
log_warn("Manage pipe no message received");
continue;
}
message_t *pay_msg;
pay_msg = nng_msg_body(msg);
switch (msg_get_type(pay_msg)) {
case MSG_CONFIG_INFO_STRING: {
char *buf_ptr;
buf_ptr = msg_get_buf_ptr(pay_msg);
log_info("Received string: %s", buf_ptr);
break;
}
case MSG_CMD_START_READ: {
const neu_plugin_intf_funs_t *intf_funs;
neu_request_t req;
uint32_t req_code;
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_type = NEU_REQRESP_READ;
req.buf_len = sizeof(uint32_t);
req_code = 1;
req.buf = (char *) &req_code;
intf_funs->request(adapter->plugin, &req);
break;
}
case MSG_CMD_STOP_READ: {
break;
}
case MSG_CMD_EXIT_LOOP: {
uint32_t exit_code;
exit_code = *(uint32_t *) msg_get_buf_ptr(pay_msg);
log_info("adapter(%s) exit loop by exit_code=%d", adapter->name,
exit_code);
nng_mtx_lock(adapter->mtx);
adapter->state = ADAPTER_STATE_NULL;
adapter->stop = true;
nng_mtx_unlock(adapter->mtx);
break;
}
default:
log_warn("Receive a not supported message(type: %d)",
msg_get_type(pay_msg));
break;
}
nng_msg_free(msg);
}
return;
}
static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp)
{
int rv = 0;
if (adapter == NULL || resp == NULL) {
log_warn("The adapter or response is NULL");
return (-1);
}
log_info("Get response from plugin");
switch (resp->resp_type) {
case NEU_REQRESP_MOVE_BUF: {
core_databuf_t *databuf;
databuf = core_databuf_new_with_buf(resp->buf, resp->buf_len);
// for debug
log_debug("Get respose buf: %s", core_databuf_get_ptr(databuf));
nng_msg *msg;
size_t msg_size;
msg_size = msg_external_data_get_size();
rv = nng_msg_alloc(&msg, msg_size);
if (rv == 0) {
message_t *pay_msg;
pay_msg = (message_t *) nng_msg_body(msg);
msg_external_data_init(pay_msg, MSG_DATA_NEURON_DATABUF, databuf);
nng_sendmsg(adapter->sock, msg, 0);
}
core_databuf_put(databuf);
break;
}
default:
break;
}
return rv;
}
static int adapter_event_notify(
neu_adapter_t *adapter, neu_event_notify_t *event)
{
int rv = 0;
log_info("Get event notify from plugin");
return rv;
}
static const adapter_callbacks_t callback_funs = { .response = adapter_response,
.event_notify = adapter_event_notify };
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
{
neu_adapter_t *adapter;
nng_mtx_alloc(&mtx_log);
log_set_lock(log_lock, mtx_log);
log_set_level(LOG_DEBUG);
FILE *logfile = fopen("web-server.log", "a");
weblog = fopen("web-server.log", "a");
// log_set_quiet(true);
log_add_fp(logfile, LOG_DEBUG);
log_add_fp(weblog, LOG_DEBUG);
adapter = malloc(sizeof(neu_adapter_t));
if (adapter == NULL) {
log_error("Out of memeory for create adapter");
return NULL;
}
int rv;
adapter->state = ADAPTER_STATE_NULL;
adapter->stop = false;
if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) {
log_error("Can't allocate mutex for adapter");
free(adapter);
return NULL;
}
adapter->id = info->id;
adapter->type = info->type;
adapter->name = strdup(info->name);
// adapter->plugin_lib_name = strdup(info->plugin_lib_name);
adapter->new_req_id = 0;
if (adapter->name == NULL /*|| adapter->plugin_lib_name == NULL*/) {
if (adapter->name != NULL) {
free(adapter->name);
}
if (adapter->plugin_lib_name != NULL) {
free(adapter->plugin_lib_name);
}
nng_mtx_free(adapter->mtx);
free(adapter);
log_error("Failed duplicate string for create adapter");
return NULL;
}
/** no need plugin
void * handle;
neu_plugin_module_t *plugin_module;
handle = load_plugin_library(adapter->plugin_lib_name, &plugin_module);
if (handle == NULL) {
neu_panic("Can't to load library(%s) for plugin(%s)",
adapter->plugin_lib_name, adapter->name);
}
adapter->plugin_lib = handle;
adapter->plugin_module = plugin_module;
neu_plugin_t *plugin;
plugin_module = adapter->plugin_module;
plugin = plugin_module->intf_funs->open(adapter, &callback_funs);
if (plugin == NULL) {
neu_panic("Can't to open plugin(%s)", plugin_module->module_name);
}
*/
rv = nng_pair1_open(&adapter->sock);
if (rv != 0) {
neu_panic("The adapter(%s) can't open pipe", adapter->name);
}
// adapter->plugin = plugin;
return adapter;
}
void neu_adapter_destroy(neu_adapter_t *adapter)
{
if (adapter == NULL) {
neu_panic("Destroied adapter is NULL");
}
nng_close(adapter->sock);
// adapter->plugin_module->intf_funs->close(adapter->plugin);
// unload_plugin_library(adapter->plugin_lib);
if (adapter->name != NULL) {
free(adapter->name);
}
// if (adapter->plugin_lib_name != NULL) {
// free(adapter->plugin_lib_name);
// }
nng_mtx_free(adapter->mtx);
free(adapter);
fclose(weblog);
nng_mtx_free(mtx_log);
return;
}
int neu_adapter_start(neu_adapter_t *adapter, neu_manager_t *manager)
{
int rv = 0;
uint16_t port = 0;
if (manager == NULL) {
log_error("Start adapter with NULL manager");
return (-1);
}
// const neu_plugin_intf_funs_t *intf_funs;
// intf_funs = adapter->plugin_module->intf_funs;
// intf_funs->init(adapter->plugin);
adapter->manager = manager;
rv = nng_thread_create(&adapter->thrd, adapter_loop, adapter);
if (rv != 0) {
fatal("Cannot start web adapter", rv);
}
rv = nng_thread_create(&inproc_thr, inproc_server, NULL);
if (rv != 0) {
fatal("cannot start inproc server", rv);
fatal("Cannot start inproc server", rv);
}
if (getenv("PORT") != NULL) {
port = (uint16_t) atoi(getenv("PORT"));
}
port = port ? port : 8888;
port = port ? port : DEFAULT_PORT;
log_debug(REST_URL, port);
rest_start(port);
// This runs forever. The inproc_thr never exits, so we
// just block behind its condition variable.
return rv;
}
int neu_adapter_stop(neu_adapter_t *adapter, neu_manager_t *manager)
{
int rv = 0;
log_info("Stop the adapter(%s)", adapter->name);
nng_mtx_lock(adapter->mtx);
adapter->stop = true;
nng_mtx_unlock(adapter->mtx);
nng_thread_destroy(adapter->thrd);
nng_thread_destroy(inproc_thr);
fclose(logfile);
nng_mtx_free(mtx_log);
// const neu_plugin_intf_funs_t *intf_funs;
// intf_funs = adapter->plugin_module->intf_funs;
// intf_funs->uninit(adapter->plugin);
return rv;
}
const char *neu_adapter_get_name(neu_adapter_t *adapter)
{
return (const char *) adapter->name;
}
neu_manager_t *neu_adapter_get_manager(neu_adapter_t *adapter)
{
return (neu_manager_t *) adapter->manager;
}
return 0;
nng_socket neu_adapter_get_sock(neu_adapter_t *adapter)
{
return adapter->sock;
}
src/adapter/web_server.h
-1
View File
@@ -21,5 +21,4 @@
#define __WEB_SERVER_H__
#endif // __WEB_SERVER_H__