diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index d293a82..52032c4 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -30,6 +30,7 @@ template struct AsyncSocket { template friend struct HttpContext; template friend struct WebSocketContext; + friend class TopicTree; protected: /* Get loop data for socket */ diff --git a/src/Loop.h b/src/Loop.h index 7dc553b..506f240 100644 --- a/src/Loop.h +++ b/src/Loop.h @@ -50,19 +50,24 @@ private: if (loopData->preHandler) { loopData->preHandler((Loop *) loop); } + + /* trying this one here */ + for (auto &f : loopData->postHandlers) { + f((Loop *) loop); + } } static void postCb(us_loop *loop) { LoopData *loopData = (LoopData *) us_loop_ext(loop); - if (loopData->postHandler) { - loopData->postHandler((Loop *) loop); - } - /* We should move over to using only these */ for (auto &f : loopData->postHandlers) { f((Loop *) loop); } + + if (loopData->postHandler) { + loopData->postHandler((Loop *) loop); + } } Loop() = delete; diff --git a/src/TopicTree.h b/src/TopicTree.h index 4686df1..6b9a19a 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -17,7 +17,7 @@ /* Every WebSocketContext holds one TopicTree */ #include "Loop.h" -#include "WebSocket.h" +#include "AsyncSocket.h" #ifndef TOPICTREE_H #define TOPICTREE_H @@ -29,6 +29,9 @@ namespace uWS { + // publishing to a node, then another node, then another node should prioritize draining that way + // sending and publishing will interleave undefined, they are separate streams + class TopicTree { private: struct Node : std::map { @@ -40,13 +43,24 @@ private: return p.first->second; } } + /* Every subscriber should hold some backpressure cursor */ std::vector> subscribers; std::string sharedMessage; - }; - Node *root = new Node; + /* We need backpressure stored */ + /* vector */ + std::string backpressure; + } *root = new Node; + + /* Nodes that hold something to send this iteration */ std::set pubNodes; + /* Settings */ + bool mergePublishedMessages = false; + + /* Where we store prepared messages to send */ + //std::string preparedMessage; + public: TopicTree() { @@ -57,34 +71,32 @@ public: return; } - // messages need to be prepared twice: compressed and non compressed - // if using dedicated compression, don't prepare + /* We say that all senders get their own message as well, for now being */ - // user should be something like a std::string with formatted content - std::string *preparedMessage = new std::string; + for (Node *topicNode : pubNodes) { + for (auto [ws, valid] : topicNode->subscribers) { + AsyncSocket *asyncSocket = (AsyncSocket *) ws; // assumes non-SSL for now - // prepare, send, ref, user - drain([](void *user, char *sharedMessage, size_t sharedMessageLength) { + /* Writing optionally raw data */ + auto [written, failed] = asyncSocket->write(topicNode->sharedMessage.data(), topicNode->sharedMessage.length(), true, 0); - //std::cout << "Preparing " << std::string_view(sharedMessage, sharedMessageLength) << std::endl; - std::string *preparedMessage = (std::string *) user; - preparedMessage->append(sharedMessage, sharedMessageLength); + /* Every subscriber to a topicNode will have int backpressure cursor to this room */ - }, [](void *user, void *ws) { + /* How far we wrote will be stored in the WebSocket's Pub/sub block and drained before any other sending (we need to fail sending if already sending pubsub) */ - /* This would be where we send the preformatted pre-compessed message in user */ + /* All messages not fully sent, will be stored in the topictree with an index so that websocket can refer to it by two index: what buffer, what offset */ + /* If total backpressure of the topictree is larger than a set limit we close all the slow receivers */ - //std::cout << "Sending " << std::endl; + /* It is also possible to move topictree backpressure to the websockets themselves, if only one */ + } - WebSocket *webSocket = (WebSocket *) ws; // assumes non-SSL + /* If not all sockets managed to send this message, move it to backpressure */ + topicNode->sharedMessage.clear(); + } + pubNodes.clear(); - std::string *preparedMessage = (std::string *) user; - webSocket->send(*preparedMessage, OpCode::TEXT, false); - }, [](void *ws) { - std::cout << "Refing" << std::endl; - }, preparedMessage); }); } @@ -99,8 +111,6 @@ public: 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 */ @@ -145,48 +155,6 @@ public: } } } - - 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) { - - std::cout << "pubNodes: " << pubNodes.size() << std::endl; - - if (pubNodes.size()) { - - //if(pubNodes.size() > 1) { - for (Node *topicNode : pubNodes) { - for (std::pair p : topicNode->subscribers) { - if (*p.second) { - refCb(p.first); - } - } - } - //} - - 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(); ) { - if (!*it->second) { - it = topicNode->subscribers.erase(it); - } else { - sendCb(user, it->first); - it++; - } - } - - topicNode->sharedMessage.clear(); - } - pubNodes.clear(); - } } }; } diff --git a/src/WebSocket.h b/src/WebSocket.h index 20a2c62..16037d2 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -144,7 +144,11 @@ public: (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this) ); - webSocketContextData->topicTree.publish(std::string(topic), (char *) message.data(), message.length()); + /* We frame the message right here and only pass raw bytes to the pub/subber */ + char dst[1024]; + size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), OpCode::TEXT, message.length(), false); + + webSocketContextData->topicTree.publish(std::string(topic), dst, dst_length); } };