diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 451cf25..1c416cd 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -31,6 +31,7 @@ template struct AsyncSocket { template friend struct HttpContext; template friend struct WebSocketContext; + template friend struct WebSocketContextData; friend struct TopicTree; protected: diff --git a/src/TopicTreeDraft.h b/src/TopicTreeDraft.h index dc84158..6ee839a 100644 --- a/src/TopicTreeDraft.h +++ b/src/TopicTreeDraft.h @@ -6,6 +6,8 @@ #include #include +namespace uWS { + /* A Subscriber is an extension of a socket */ struct Subscriber { /* List of all our subscriptions (subscribersNextSubscription) */ @@ -215,6 +217,10 @@ public: /* Drain the tree by emitting what to send with every Subscriber */ void drain(/*std::function cb*/) { + if (!numTriggeredTopics) { + return; + } + /* Up to 64 triggered Topics per batch */ std::map intersectionCache; @@ -285,6 +291,13 @@ public: min = nextMin; } + + /* Clear messages of triggered Topics */ + for (int i = 0; i < numTriggeredTopics; i++) { + triggeredTopics[i]->messages.clear(); + triggeredTopics[i]->triggered = false; + } + numTriggeredTopics = 0; } void print(Topic *root = nullptr, int indentation = 1) { @@ -302,3 +315,5 @@ public: } } }; + +} \ No newline at end of file diff --git a/src/WebSocket.h b/src/WebSocket.h index 8f047ef..1cfb369 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -128,7 +128,7 @@ public: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll(this); + webSocketContextData->topicTree.unsubscribeAll((Subscriber *) this); } /* Subscribe to a topic according to MQTT rules and syntax */ @@ -138,9 +138,7 @@ public: ); /* Fix this up */ - bool *valid = new bool; - *valid = true; - webSocketContextData->topicTree.subscribe(std::string(topic), this, valid); + webSocketContextData->topicTree.subscribe(topic, (Subscriber *) this); } /* Publish a message to a topic according to MQTT rules and syntax */ @@ -153,7 +151,7 @@ public: 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); + webSocketContextData->topicTree.publish(topic, std::string_view(dst, dst_length)); } }; diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 52abd78..8dac930 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -237,7 +237,7 @@ private: } /* Make sure to unsubscribe from any pub/sub node at exit */ - webSocketContextData->topicTree.unsubscribeAll(s); + webSocketContextData->topicTree.unsubscribeAll((Subscriber *) s); } /* Destruct in-placed data struct */ diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 7e2fc27..bae3451 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -22,7 +22,7 @@ #include #include "WebSocketProtocol.h" -#include "TopicTree.h" +#include "TopicTreeDraft.h" namespace uWS { @@ -43,6 +43,23 @@ struct WebSocketContextData { /* Each websocket context has a topic tree for pub/sub */ TopicTree topicTree; + + WebSocketContextData() : topicTree([](Subscriber *s, std::string_view data) -> int { + //std::cout << "Skickar data: " << data << " på sub: " << s << std::endl; + + + auto *asyncSocket = (AsyncSocket *) s; + + asyncSocket->write(data.data(), data.length()); + + return 0; + }) { + + Loop::get()->addPostHandler([this](Loop *loop) { + + topicTree.drain(); + }); + } }; } diff --git a/uSockets b/uSockets index e2c093c..12a235b 160000 --- a/uSockets +++ b/uSockets @@ -1 +1 @@ -Subproject commit e2c093cb7857fa8dc5716470613c46bad508007a +Subproject commit 12a235bcfe5e0cc5923aef495840fa8cd566bf3b