From 65d1498774af82b201ad15e42a43554a583de147 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sat, 10 Apr 2021 17:22:16 +0200 Subject: [PATCH] Make conceptual pub/sub changes for v19 --- src/App.h | 32 ++++++++++++++++++++++++- src/WebSocket.h | 48 ++++++++++++++++++-------------------- src/WebSocketContextData.h | 5 ++++ 3 files changed, 59 insertions(+), 26 deletions(-) diff --git a/src/App.h b/src/App.h index 779d195..e9d9741 100644 --- a/src/App.h +++ b/src/App.h @@ -96,13 +96,33 @@ public: httpContext->filter(std::move(filterHandler)); } - /* Publishes a message to all websocket contexts */ + /* Publishes a message to all websocket contexts - conceptually as if publishing to the one single + * TopicTree of this app (technically there are many TopicTrees, however the concept is that one + * app has one conceptual Topic tree) */ void publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress = false) { for (auto *webSocketContext : webSocketContexts) { webSocketContext->getExt()->publish(topic, message, opCode, compress); } } + /* Returns number of subscribers for this topic, or 0 for failure. + * This function should probably be optimized a lot in future releases, + * it could be O(1) with a hash map of fullnames and their counts. */ + unsigned int numSubscribers(std::string_view topic) { + unsigned int subscribers = 0; + + for (auto *webSocketContext : webSocketContexts) { + auto *webSocketContextData = webSocketContext->getExt(); + + Topic *t = webSocketContextData->lookupTopic(topic); + if (t) { + subscribers += t->subs.size(); + } + } + + return subscribers; + } + ~TemplatedApp() { /* Let's just put everything here */ if (httpContext) { @@ -183,6 +203,16 @@ public: /* Every route has its own websocket context with its own behavior and user data type */ auto *webSocketContext = WebSocketContext::create(Loop::get(), (us_socket_context_t *) httpContext); + /* Add all other WebSocketContextData to this new WebSocketContextData */ + for (WebSocketContext *adjacentWebSocketContext : webSocketContexts) { + webSocketContext->getExt()->adjacentWebSocketContextDatas.push_back(adjacentWebSocketContext->getExt()); + } + + /* Add this WebSocketContextData to all other WebSocketContextData */ + for (WebSocketContext *adjacentWebSocketContext : webSocketContexts) { + adjacentWebSocketContext->getExt()->adjacentWebSocketContextDatas.push_back((WebSocketContextData *) webSocketContext->getExt()); + } + /* We need to clear this later on */ webSocketContexts.push_back((WebSocketContext *) webSocketContext); diff --git a/src/WebSocket.h b/src/WebSocket.h index 771e01c..64f864f 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -196,8 +196,8 @@ public: } } - /* Subscribe to a topic according to MQTT rules and syntax. Returns [numSubscribers, success]. */ - std::pair subscribe(std::string_view topic, bool nonStrict = false) { + /* Subscribe to a topic according to MQTT rules and syntax. Returns success */ + /*std::pair*/ bool subscribe(std::string_view topic, bool nonStrict = false) { WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); @@ -208,18 +208,20 @@ public: webSocketData->subscriber = new Subscriber(this); } - return webSocketContextData->topicTree.subscribe(topic, webSocketData->subscriber, nonStrict); + /* Cannot return numSubscribers as this is only for this particular websocket context */ + return webSocketContextData->topicTree.subscribe(topic, webSocketData->subscriber, nonStrict).second; } - /* Unsubscribe from a topic, returns true if we were subscribed. Returns [numSubscribers, success]. */ - std::pair unsubscribe(std::string_view topic, bool nonStrict = false) { + /* Unsubscribe from a topic, returns true if we were subscribed. */ + /*std::pair*/ bool unsubscribe(std::string_view topic, bool nonStrict = false) { WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) ); WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); - return webSocketContextData->topicTree.unsubscribe(topic, webSocketData->subscriber, nonStrict); + /* Cannot return numSubscribers as this is only for this particular websocket context */ + return webSocketContextData->topicTree.unsubscribe(topic, webSocketData->subscriber, nonStrict).second; } /* Returns whether this socket is subscribed to the specified topic */ @@ -236,25 +238,11 @@ public: return false; } - /* Returns number of subscribers for this topic, or 0 for failure */ - unsigned int numSubscribers(std::string_view topic) { - WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, - (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this) - ); - - Topic *t = webSocketContextData->lookupTopic(topic); - if (t) { - return t->subs.size(); - } - - return 0; - } - - /* Iterates all topics of this WebSocket. Every topic is represented by [name, numSubscribers]. + /* Iterates all topics of this WebSocket. Every topic is represented by its full name. * Can be called in close handler. It is possible to modify the subscription list while * inside the callback ONLY IF not modifying the topic passed to the callback. * Topic names are valid only for the duration of the callback. */ - void iterateTopics(MoveOnlyFunction cb) { + void iterateTopics(MoveOnlyFunction cb) { WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this); if (webSocketData->subscriber) { @@ -262,14 +250,14 @@ public: /* Lock this topic so that nobody may unsubscribe from it during this callback */ t->locked = true; - cb(t->fullName, (unsigned int) t->subs.size()); + cb(t->fullName/*, (unsigned int) t->subs.size()*/); t->locked = false; } } } - /* Publish a message to a topic according to MQTT rules and syntax. Returns [numSubscribers, success]. + /* Publish a message to a topic according to MQTT rules and syntax. Returns success. * We, the WebSocket, must be subscribed to the topic itself and if so - no message will be sent to ourselves. * Use App::publish for an unconditional publish that simply publishes to whomever might be subscribed. */ bool publish(std::string_view topic, std::string_view message, OpCode opCode = OpCode::TEXT, bool compress = false) { @@ -285,7 +273,17 @@ public: } /* Publish as sender, does not receive its own messages even if subscribed to relevant topics */ - return webSocketContextData->publish(topic, message, opCode, compress, webSocketData->subscriber); + bool success = webSocketContextData->publish(topic, message, opCode, compress, webSocketData->subscriber); + + /* Loop over all websocket contexts for this App */ + if (success) { + /* Success is really only determined by the first publish. We must be subscribed to the topic. */ + for (auto *adjacentWebSocketContextData : webSocketContextData->adjacentWebSocketContextDatas) { + adjacentWebSocketContextData->publish(topic, message, opCode, compress); + } + } + + return success; } }; diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 269bf70..1449bf4 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -20,6 +20,7 @@ #include "MoveOnlyFunction.h" #include +#include #include "WebSocketProtocol.h" #include "TopicTree.h" @@ -46,6 +47,10 @@ private: }; public: + /* All WebSocketContextData holds a list to all other WebSocketContextData in this app. + * We cannot type it USERDATA since different WebSocketContextData can have different USERDATA. */ + std::vector *> adjacentWebSocketContextDatas; + /* The callbacks for this context */ MoveOnlyFunction *)> openHandler = nullptr; MoveOnlyFunction *, std::string_view, OpCode)> messageHandler = nullptr;