diff --git a/src/relay.cpp b/src/relay.cpp new file mode 100644 index 000000000..3daf1012c --- /dev/null +++ b/src/relay.cpp @@ -0,0 +1,243 @@ +/** Simple WebSocket relay facilitating client-to-client connections. + * + * @author Steffen Vogel + * @copyright 2018, Institute for Automation of Complex Power Systems, EONERC + * @license GNU General Public License (version 3) + * + * VILLASnode + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * 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 General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + *********************************************************************************/ + +#include +#include +#include +#include +#include + +#include + +#include + +#include + +/** The libwebsockets server context. */ +static lws_context *context; + +/** The libwebsockets vhost. */ +static lws_vhost *vhost; + +/* Forward declarations */ +lws_callback_function protocol_cb; +struct Session; +struct Connection; + +static std::map> sessions; + +class Frame : public std::vector { }; + +class Session { +public: + typedef std::string Identifier; + + + static std::shared_ptr get(Identifier sid) + { + auto it = sessions.find(sid); + if (it == sessions.end()) { + auto s = std::make_shared(sid); + + sessions[sid] = s; + + return s; + } + else + return it->second; + } + + Session(Identifier sid) : + identifier(sid) + { } + + ~Session() + { } + + Identifier identifier; + + std::vector> connections; +}; + +class Connection { +public: + lws *wsi; + + std::shared_ptr currentFrame; + + std::deque> outgoingFrames; + + std::shared_ptr session; + + Connection() : + currentFrame(nullptr) + { } +}; + +/** List of libwebsockets protocols. */ +lws_protocols protocols[] = { + { + .name = "live", + .callback = protocol_cb, + .per_session_data_size = sizeof(Connection), + .rx_buffer_size = 0 + }, + { NULL /* terminator */ } +}; + +/** List of libwebsockets extensions. */ +static const lws_extension extensions[] = { + { + "permessage-deflate", + lws_extension_callback_pm_deflate, + "permessage-deflate" + }, + { + "deflate-frame", + lws_extension_callback_pm_deflate, + "deflate_frame" + }, + { NULL /* terminator */ } +}; + +extern log *global_log; + +static void logger(int level, const char *msg) { + int len = strlen(msg); + if (strchr(msg, '\n')) + len -= 1; + + /* Decrease severity for some errors. */ + if (strstr(msg, "Unable to open") == msg) + level = LLL_WARN; + + switch (level) { + case LLL_ERR: log_print(global_log, CLR_RED("Web "), "%.*s", len, msg); break; + case LLL_WARN: log_print(global_log, CLR_YEL("Web "), "%.*s", len, msg); break; + case LLL_INFO: log_print(global_log, CLR_WHT("Web "), "%.*s", len, msg); break; + default: log_print(global_log, "Web ", "%.*s", len, msg); break; + } +} + +int protocol_cb(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) +{ + Connection *c = reinterpret_cast(user); + + switch (reason) { + + case LWS_CALLBACK_ESTABLISHED: { + char uri[64]; + + new (c) Connection(); + + c->wsi = wsi; + + /* We use the URI to associate this connection to a session + * Example: ws://example.com/node_1 + * Will select the session with the name 'node_1' + */ + + /* Get path of incoming request */ + lws_hdr_copy(wsi, uri, sizeof(uri), WSI_TOKEN_GET_URI); + if (strlen(uri) <= 0) { + lws_close_reason(wsi, LWS_CLOSE_STATUS_PROTOCOL_ERR, (unsigned char *) "Invalid URL", strlen("Invalid URL")); + return -1; + } + + Session::Identifier sid = &uri[1]; + + auto s = Session::get(sid); + + c->session = s; + s->connections.push_back(std::shared_ptr(c)); + + break; + } + + case LWS_CALLBACK_CLOSED: + + c->~Connection(); + break; + + case LWS_CALLBACK_SERVER_WRITEABLE: + while (c->outgoingFrames.size() > 0) { + std::shared_ptr fr = c->outgoingFrames.front(); + + lws_write(wsi, fr->data(), fr->size(), LWS_WRITE_BINARY); + + c->outgoingFrames.pop_front(); + } + break; + + case LWS_CALLBACK_RECEIVE: + if (!c->currentFrame.get() || lws_is_first_fragment(wsi)) + c->currentFrame = std::make_shared(); + + c->currentFrame->insert(c->currentFrame->end(), (uint8_t *) in, (uint8_t *) in + len); + + if (lws_is_final_fragment(wsi)) { + for (auto cc : c->session->connections) { + if (cc.get() == c) + continue; + + cc->outgoingFrames.push_back(c->currentFrame); + + lws_callback_on_writable(cc->wsi); + } + } + + break; + + default: + break; + } + + return 0; +} + +int main(int argc, char *argv[]) +{ + lws_set_log_level((1 << LLL_COUNT) - 1, logger); + + /* Start server */ + lws_context_creation_info ctx_info = { 0 }; + + ctx_info.options = LWS_SERVER_OPTION_EXPLICIT_VHOSTS | LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT; + ctx_info.gid = -1; + ctx_info.uid = -1; + ctx_info.protocols = protocols; + ctx_info.extensions = extensions; + ctx_info.port = 8088; + + context = lws_create_context(&ctx_info); + if (context == NULL) + error("WebSocket: failed to initialize server context"); + + vhost = lws_create_vhost(context, &ctx_info); + if (vhost == NULL) + error("WebSocket: failed to initialize virtual host"); + + for (;;) + lws_service(context, 100); + + return 0; +}