From aaf1f05dcd685bbdeff2e7fa74742251e302cdd0 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 8 Feb 2019 23:23:25 +0100 Subject: [PATCH] Move in TopicTree from uTT --- src/TopicTree.h | 150 +++++++++++++++++++++++++++++++++++++ src/WebSocket.h | 19 +++++ src/WebSocketContextData.h | 8 +- 3 files changed, 174 insertions(+), 3 deletions(-) create mode 100644 src/TopicTree.h diff --git a/src/TopicTree.h b/src/TopicTree.h new file mode 100644 index 0000000..11dc471 --- /dev/null +++ b/src/TopicTree.h @@ -0,0 +1,150 @@ +/* + * Authored by Alex Hultman, 2018-2019. + * Intellectual property of third-party. + + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + + * http://www.apache.org/licenses/LICENSE-2.0 + + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* Every Loop holds one TopicTree */ + +#ifndef TOPICTREE_H +#define TOPICTREE_H + +#include +#include +#include +#include + +class TopicTree { +private: + struct Node : std::map { + Node *get(std::string path) { + std::pair::iterator, bool> p = insert({path, nullptr}); + if (p.second) { + return p.first->second = new Node; + } else { + return p.first->second; + } + } + std::vector> subscribers; + std::string sharedMessage; + }; + + Node *root = new Node; + std::set pubNodes; + +public: + + TopicTree() { + std::cout << "Constructing TopicTree" << std::endl; + } + + /* WebSocket.subscribe will lookup the Loop and subscribe in its tree */ + void subscribe(std::string topic, void *connection, bool *valid) { + Node *curr = root; + for (int i = 0; i < topic.length(); i++) { + int start = i; + while (topic[i] != '/' && i < topic.length()) { + i++; + } + curr = curr->get(topic.substr(start, i - start)); + } + curr->subscribers.push_back({connection, valid}); + + std::cout << "Subscribed to topicTree" << std::endl; + } + + /* WebSocket.publish looks up its tree and publishes to it */ + void publish(std::string topic, char *data, size_t length) { + Node *curr = root; + for (int i = 0; i < topic.length(); i++) { + int start = i; + while (topic[i] != '/' && i < topic.length()) { + i++; + } + std::string path(topic.data() + start, i - start); + + // end wildcard consumes traversal + auto it = curr->find("#"); + if (it != curr->end()) { + curr = it->second; + //matches.push_back(curr); + curr->sharedMessage.append(data, length); + if (curr->subscribers.size()) { + pubNodes.insert(curr); + } + break; + } else { + it = curr->find(path); + if (it == curr->end()) { + it = curr->find("+"); + if (it != curr->end()) { + goto skip; + } + break; + } else { + skip: + curr = it->second; + if (i == topic.length()) { + //matches.push_back(curr); + curr->sharedMessage.append(data, length); + if (curr->subscribers.size()) { + pubNodes.insert(curr); + } + break; + } + } + } + } + + std::cout << "Published to topicTree" << std::endl; + } + + void reset() { + root = new Node; + } + + /* I forgot what this does but probably needs lots of changes anyways */ + void drain(void (*prepareCb)(void *user, char *, size_t), void (*sendCb)(void *, void *), void (*refCb)(void *), void *user) { + if (pubNodes.size()) { + + //if(pubNodes.size() > 1) { + for (Node *topicNode : pubNodes) { + for (std::pair p : topicNode->subscribers) { + if (*p.second) { + refCb(p.first); + } + } + } + //} + + for (Node *topicNode : pubNodes) { + prepareCb(user, (char *) topicNode->sharedMessage.data(), topicNode->sharedMessage.length()); + for (auto it = topicNode->subscribers.begin(); it != topicNode->subscribers.end(); ) { + if (!*it->second) { + it = topicNode->subscribers.erase(it); + } else { + sendCb(user, it->first); + it++; + } + } + + topicNode->sharedMessage.clear(); + } + pubNodes.clear(); + } + } +}; + + +#endif // TOPICTREE_H \ No newline at end of file diff --git a/src/WebSocket.h b/src/WebSocket.h index 21a52f1..e104cd0 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -125,6 +125,25 @@ public: webSocketContextData->closeHandler(this, code, message); } } + + /* Subscribe to a topic according to MQTT rules and syntax */ + void subscribe(std::string_view topic) { + WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_new_socket_context_ext(SSL, + (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this) + ); + + /* Fix this up */ + webSocketContextData->topicTree.subscribe(topic, this, nullptr); + } + + /* Publish a message to a topic according to MQTT rules and syntax */ + void publish(std::string_view topic, std::string_view message) { + WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_new_socket_context_ext(SSL, + (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this) + ); + + webSocketContextData->topicTree.publish(topic, message.data(), message.length()); + } }; } diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 07cc792..de68edd 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -18,12 +18,11 @@ #ifndef WEBSOCKETCONTEXTDATA_H #define WEBSOCKETCONTEXTDATA_H -#include +#include "f2/function2.hpp" #include #include "WebSocketProtocol.h" - -#include "f2/function2.hpp" +#include "TopicTree.h" namespace uWS { @@ -41,6 +40,9 @@ struct WebSocketContextData { /* Settings for this context */ size_t maxPayloadLength = 0; int idleTimeout = 0; + + /* Each websocket context has a topic tree for pub/sub */ + TopicTree topicTree; }; }