diff --git a/src/Loop.h b/src/Loop.h index 13b951e..7dc553b 100644 --- a/src/Loop.h +++ b/src/Loop.h @@ -21,12 +21,8 @@ /* The loop is lazily created per-thread and run with uWS::run() */ #include "LoopData.h" - #include - - - #include namespace uWS { @@ -62,6 +58,11 @@ private: if (loopData->postHandler) { loopData->postHandler((Loop *) loop); } + + /* We should move over to using only these */ + for (auto &f : loopData->postHandlers) { + f((Loop *) loop); + } } Loop() = delete; @@ -109,6 +110,13 @@ public: us_loop_free((us_loop *) this); } + /* We want to have multiple of these */ + void addPostHandler(fu2::unique_function &&handler) { + LoopData *loopData = (LoopData *) us_loop_ext((us_loop *) this); + + loopData->postHandlers.emplace_back(std::move(handler)); + } + /* Set postCb callback */ void setPostHandler(fu2::unique_function &&handler) { LoopData *loopData = (LoopData *) us_loop_ext((us_loop *) this); diff --git a/src/LoopData.h b/src/LoopData.h index d704958..3419654 100644 --- a/src/LoopData.h +++ b/src/LoopData.h @@ -40,6 +40,9 @@ private: fu2::unique_function postHandler, preHandler; + /* Move over to these later on */ + std::vector> postHandlers; + public: ~LoopData() { /* If we have had App.ws called with compression we need to clear this */ diff --git a/src/TopicTree.h b/src/TopicTree.h index 11dc471..4686df1 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -15,7 +15,9 @@ * limitations under the License. */ -/* Every Loop holds one TopicTree */ +/* Every WebSocketContext holds one TopicTree */ +#include "Loop.h" +#include "WebSocket.h" #ifndef TOPICTREE_H #define TOPICTREE_H @@ -25,6 +27,8 @@ #include #include +namespace uWS { + class TopicTree { private: struct Node : std::map { @@ -46,7 +50,42 @@ private: public: TopicTree() { - std::cout << "Constructing TopicTree" << std::endl; + /* Dynamically hook us up with the Loop post handler */ + Loop::defaultLoop()->addPostHandler([this](Loop *loop) { + + if (!pubNodes.size()) { + return; + } + + // messages need to be prepared twice: compressed and non compressed + // if using dedicated compression, don't prepare + + // user should be something like a std::string with formatted content + std::string *preparedMessage = new std::string; + + // prepare, send, ref, user + drain([](void *user, char *sharedMessage, size_t sharedMessageLength) { + + //std::cout << "Preparing " << std::string_view(sharedMessage, sharedMessageLength) << std::endl; + + std::string *preparedMessage = (std::string *) user; + preparedMessage->append(sharedMessage, sharedMessageLength); + + }, [](void *user, void *ws) { + + /* This would be where we send the preformatted pre-compessed message in user */ + + //std::cout << "Sending " << std::endl; + + WebSocket *webSocket = (WebSocket *) ws; // assumes non-SSL + + std::string *preparedMessage = (std::string *) user; + webSocket->send(*preparedMessage, OpCode::TEXT, false); + + }, [](void *ws) { + std::cout << "Refing" << std::endl; + }, preparedMessage); + }); } /* WebSocket.subscribe will lookup the Loop and subscribe in its tree */ @@ -116,6 +155,9 @@ public: /* 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) { + + std::cout << "pubNodes: " << pubNodes.size() << std::endl; + if (pubNodes.size()) { //if(pubNodes.size() > 1) { @@ -128,6 +170,8 @@ public: } //} + std::cout << "Now we are here" << std::endl; + for (Node *topicNode : pubNodes) { prepareCb(user, (char *) topicNode->sharedMessage.data(), topicNode->sharedMessage.length()); for (auto it = topicNode->subscribers.begin(); it != topicNode->subscribers.end(); ) { @@ -145,6 +189,6 @@ public: } } }; - +} #endif // TOPICTREE_H \ No newline at end of file diff --git a/src/WebSocket.h b/src/WebSocket.h index e104cd0..20a2c62 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -133,7 +133,9 @@ public: ); /* Fix this up */ - webSocketContextData->topicTree.subscribe(topic, this, nullptr); + bool *valid = new bool; + *valid = true; + webSocketContextData->topicTree.subscribe(std::string(topic), this, valid); } /* Publish a message to a topic according to MQTT rules and syntax */ @@ -142,7 +144,7 @@ public: (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this) ); - webSocketContextData->topicTree.publish(topic, message.data(), message.length()); + webSocketContextData->topicTree.publish(std::string(topic), (char *) message.data(), message.length()); } };