talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make conceptual pub/sub changes for v19

Browse Source
This commit is contained in:
Alex Hultman
2021-04-10 17:22:16 +02:00
parent 9d8e37d8a4
commit 65d1498774
3 changed files with 59 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/App.h
+31 -1
View File
@@ -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<SSL, true, UserData>::create(Loop::get(), (us_socket_context_t *) httpContext);
/* Add all other WebSocketContextData to this new WebSocketContextData */
for (WebSocketContext<SSL, true, int> *adjacentWebSocketContext : webSocketContexts) {
webSocketContext->getExt()->adjacentWebSocketContextDatas.push_back(adjacentWebSocketContext->getExt());
}
/* Add this WebSocketContextData to all other WebSocketContextData */
for (WebSocketContext<SSL, true, int> *adjacentWebSocketContext : webSocketContexts) {
adjacentWebSocketContext->getExt()->adjacentWebSocketContextDatas.push_back((WebSocketContextData<SSL, int> *) webSocketContext->getExt());
}
/* We need to clear this later on */
webSocketContexts.push_back((WebSocketContext<SSL, true, int> *) webSocketContext);
src/WebSocket.h
+23 -25
View File
@@ -196,8 +196,8 @@ public:
}
}
/* Subscribe to a topic according to MQTT rules and syntax. Returns [numSubscribers, success]. */
std::pair<unsigned int, bool> subscribe(std::string_view topic, bool nonStrict = false) {
/* Subscribe to a topic according to MQTT rules and syntax. Returns success */
/*std::pair<unsigned int, bool>*/ bool subscribe(std::string_view topic, bool nonStrict = false) {
WebSocketContextData<SSL, USERDATA> *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) 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<unsigned int, bool> unsubscribe(std::string_view topic, bool nonStrict = false) {
/* Unsubscribe from a topic, returns true if we were subscribed. */
/*std::pair<unsigned int, bool>*/ bool unsubscribe(std::string_view topic, bool nonStrict = false) {
WebSocketContextData<SSL, USERDATA> *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) 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<SSL, USERDATA> *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) 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<void(std::string_view, unsigned int)> cb) {
void iterateTopics(MoveOnlyFunction<void(std::string_view/*, unsigned int*/)> 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;
}
};
src/WebSocketContextData.h
+5
View File
@@ -20,6 +20,7 @@
#include "MoveOnlyFunction.h"
#include <string_view>
#include <vector>
#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<WebSocketContextData<SSL, int> *> adjacentWebSocketContextDatas;
/* The callbacks for this context */
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *)> openHandler = nullptr;
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, OpCode)> messageHandler = nullptr;