diff --git a/CMakeLists.txt b/CMakeLists.txt index 8810ed57..5cf39af7 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -194,6 +194,7 @@ target_link_libraries(neuron-base vector-static csptr jansson crypto ssl mqttc y #copy file for run file(COPY ${CMAKE_SOURCE_DIR}/.gitkeep DESTINATION ${CMAKE_BINARY_DIR}/logs) file(COPY ${CMAKE_SOURCE_DIR}/neuron.yaml DESTINATION ${CMAKE_BINARY_DIR}/config) +file(COPY ${CMAKE_SOURCE_DIR}/plugins/ekuiper/ekuiper.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) file(COPY ${CMAKE_SOURCE_DIR}/plugins/modbus/modbus-tcp.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) file(COPY ${CMAKE_SOURCE_DIR}/plugins/mqtt/mqtt.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) file(COPY ${CMAKE_SOURCE_DIR}/private.key DESTINATION ${CMAKE_BINARY_DIR}/config) @@ -204,6 +205,7 @@ include_directories(include/neuron) add_subdirectory(plugins/sample) add_subdirectory(plugins/modbus) add_subdirectory(plugins/mqtt) +add_subdirectory(plugins/ekuiper) # define for simulator add_subdirectory(simulator) diff --git a/default_plugins.json b/default_plugins.json index 1bb489e8..930cb69e 100644 --- a/default_plugins.json +++ b/default_plugins.json @@ -1,5 +1,6 @@ { "plugins": [ + "libplugin-ekuiper.so", "libplugin-mqtt.so", "libplugin-modbus-tcp.so" ] diff --git a/plugins/ekuiper/CMakeLists.txt b/plugins/ekuiper/CMakeLists.txt new file mode 100644 index 00000000..8a6736a6 --- /dev/null +++ b/plugins/ekuiper/CMakeLists.txt @@ -0,0 +1,14 @@ +# LF Edge eKuiper plugin +set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") + +set(src + plugin_ekuiper.c) + +add_library(plugin-ekuiper SHARED ${src}) + +target_include_directories(plugin-ekuiper PRIVATE + ${CMAKE_SOURCE_DIR}/src + ${CMAKE_SOURCE_DIR}/include) + +target_link_libraries(plugin-ekuiper neuron-base nng) +target_link_libraries(plugin-ekuiper ${CMAKE_THREAD_LIBS_INIT}) diff --git a/plugins/ekuiper/ekuiper.json b/plugins/ekuiper/ekuiper.json new file mode 100644 index 00000000..0967ef42 --- /dev/null +++ b/plugins/ekuiper/ekuiper.json @@ -0,0 +1 @@ +{} diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c new file mode 100644 index 00000000..c6d9fbe0 --- /dev/null +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -0,0 +1,193 @@ +/** + * 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 +#include + +#include "errcodes.h" +#include "neuron.h" + +#define EKUIPER_PLUGIN_URL "ipc:///tmp/neuron-ekuiper.ipc" + +const neu_plugin_module_t neu_plugin_module; + +struct neu_plugin { + neu_plugin_common_t common; + nng_socket sock; +}; + +static neu_plugin_t *ekuiper_plugin_open(neu_adapter_t * adapter, + const adapter_callbacks_t *callbacks) +{ + neu_plugin_t *plugin; + + if (adapter == NULL || callbacks == NULL) { + log_error("Open plugin with NULL adapter or callbacks"); + return NULL; + } + + plugin = (neu_plugin_t *) malloc(sizeof(neu_plugin_t)); + if (plugin == NULL) { + log_error("Failed to allocate plugin %s", + neu_plugin_module.module_name); + return NULL; + } + + neu_plugin_common_init(&plugin->common); + plugin->common.adapter = adapter; + plugin->common.adapter_callbacks = callbacks; + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_DISCONNECTED; + + log_info("Success to create plugin: %s", neu_plugin_module.module_name); + return plugin; +} + +static int ekuiper_plugin_close(neu_plugin_t *plugin) +{ + int rv = 0; + + free(plugin); + log_info("Success to free plugin: %s", neu_plugin_module.module_name); + return rv; +} + +static int ekuiper_plugin_init(neu_plugin_t *plugin) +{ + int rv = 0; + + (void) plugin; + + log_info("Initialize plugin: %s", neu_plugin_module.module_name); + return rv; +} + +static int ekuiper_plugin_uninit(neu_plugin_t *plugin) +{ + int rv = 0; + + (void) plugin; + + log_info("Uninitialize plugin: %s", neu_plugin_module.module_name); + return rv; +} + +static int ekuiper_plugin_start(neu_plugin_t *plugin) +{ + int rv = 0; + + rv = nng_pair0_open(&plugin->sock); + if (rv != 0) { + log_error("nng_pair0_open: %s", nng_strerror(rv)); + return NEU_ERR_FAILURE; + } + + if ((rv = nng_listen(plugin->sock, EKUIPER_PLUGIN_URL, NULL, 0)) != 0) { + log_error("nng_listen: %s", nng_strerror(rv)); + return NEU_ERR_FAILURE; + } + + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTED; + return NEU_ERR_SUCCESS; +} + +static int ekuiper_plugin_stop(neu_plugin_t *plugin) +{ + nng_close(plugin->sock); + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_DISCONNECTED; + return NEU_ERR_SUCCESS; +} + +static int ekuiper_plugin_config(neu_plugin_t *plugin, neu_config_t *configs) +{ + int rv = 0; + + (void) plugin; + (void) configs; + + log_info("config plugin: %s", neu_plugin_module.module_name); + return rv; +} + +static int ekuiper_plugin_request(neu_plugin_t *plugin, neu_request_t *req) +{ + int rv = 0; + + if (plugin == NULL || req == NULL) { + log_warn("The plugin pointer or request is NULL"); + return (-1); + } + + log_info("send request to plugin: %s", neu_plugin_module.module_name); + const adapter_callbacks_t *adapter_callbacks; + adapter_callbacks = plugin->common.adapter_callbacks; + (void) adapter_callbacks; + + switch (req->req_type) { + case NEU_REQRESP_TRANS_DATA: { + // handle data + break; + } + + default: + break; + } + return rv; +} + +static int ekuiper_plugin_event_reply(neu_plugin_t * plugin, + neu_event_reply_t *reply) +{ + int rv = 0; + + (void) plugin; + (void) reply; + + log_info("reply event to plugin: %s", neu_plugin_module.module_name); + return rv; +} + +static int ekuiper_plugin_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag) +{ + (void) plugin; + (void) tag; + + return 0; +} + +static const neu_plugin_intf_funs_t plugin_intf_funs = { + .open = ekuiper_plugin_open, + .close = ekuiper_plugin_close, + .init = ekuiper_plugin_init, + .uninit = ekuiper_plugin_uninit, + .start = ekuiper_plugin_start, + .stop = ekuiper_plugin_stop, + .config = ekuiper_plugin_config, + .request = ekuiper_plugin_request, + .validate_tag = ekuiper_plugin_validate_tag, + .event_reply = ekuiper_plugin_event_reply +}; + +const neu_plugin_module_t neu_plugin_module = { + .version = NEURON_PLUGIN_VER_1_0, + .module_name = "ekuiper", + .module_descr = "Neuron and LF Edge eKuiper integration plugin", + .intf_funs = &plugin_intf_funs, + .kind = PLUGIN_KIND_SYSTEM, + .type = NEU_NODE_TYPE_APP, +};