diff --git a/examples/BroadcastingEchoServer.cpp b/examples/BroadcastingEchoServer.cpp index 3a7b325..7280f05 100644 --- a/examples/BroadcastingEchoServer.cpp +++ b/examples/BroadcastingEchoServer.cpp @@ -29,7 +29,7 @@ int main() { } /* Simply broadcast every single message we get */ - ws->publish("broadcast", message, opCode); + ws->publish("broadcast", message, opCode, true); }, .drain = [](auto *ws) { /* Check getBufferedAmount here */ diff --git a/src/TopicTree.h b/src/TopicTree.h index 2ca4f10..0f039bb 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -58,15 +58,15 @@ struct Topic { /* Terminating wildcard child */ Topic *terminatingWildcardChild = nullptr; - /* What we published */ - std::map messages; + /* What we published, {inflated, deflated} */ + std::map> messages; std::set subs; }; struct TopicTree { private: - std::function cb; + std::function)> cb; Topic *root = new Topic; @@ -120,7 +120,7 @@ private: } /* Should be getData and commit? */ - void publish(Topic *iterator, size_t start, size_t stop, std::string_view topic, std::string_view message) { + void publish(Topic *iterator, size_t start, size_t stop, std::string_view topic, std::pair message) { /* If we already have 64 triggered topics make sure to drain it here */ if (numTriggeredTopics == 64) { drain(); @@ -167,7 +167,7 @@ private: public: - TopicTree(std::function cb) { + TopicTree(std::function)> cb) { this->cb = cb; } @@ -226,7 +226,7 @@ public: } } - void publish(std::string_view topic, std::string_view message) { + void publish(std::string_view topic, std::pair message) { publish(root, 0, 0, topic, message); messageId++; } @@ -316,7 +316,7 @@ public: if (min != (Subscriber *)UINTPTR_MAX) { /* Up to 64 triggered Topics per batch */ - std::map intersectionCache; + std::map> intersectionCache; /* Loop over these here */ std::set::iterator it[64]; @@ -332,7 +332,7 @@ public: Subscriber *nextMin = (Subscriber *)UINTPTR_MAX; /* The message sets relevant for this intersection */ - std::map *perSubscriberIntersectingTopicMessages[64]; + std::map> *perSubscriberIntersectingTopicMessages[64]; int numPerSubscriberIntersectingTopicMessages = 0; uint64_t intersection = 0; @@ -363,18 +363,19 @@ public: } /* Generate cache for intersection */ - if (intersectionCache[intersection].length() == 0) { + if (intersectionCache[intersection].first.length() == 0) { /* Build the union in order without duplicates */ - std::map complete; + std::map> complete; for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) { complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end()); } - /* Create the linear cache */ - std::string res; + /* Create the linear cache, {inflated, deflated} */ + std::pair res; for (auto &p : complete) { - res.append(p.second); + res.first.append(p.second.first); + res.second.append(p.second.second); } cb(min, intersectionCache[intersection] = std::move(res)); diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index ac94868..3bd6558 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -23,6 +23,7 @@ #include "WebSocketProtocol.h" #include "TopicTree.h" +#include "WebSocketData.h" namespace uWS { @@ -56,15 +57,23 @@ struct WebSocketContextData { Loop::get()->removePreHandler(this); } - WebSocketContextData() : topicTree([this](Subscriber *s, std::string_view data) -> int { + WebSocketContextData() : topicTree([this](Subscriber *s, std::pair data) -> int { /* We rely on writing to regular asyncSockets */ auto *asyncSocket = (AsyncSocket *) s->user; /* Check if we now have too much backpressure (todo: don't buffer up before check) */ if (!maxBackpressure || (unsigned int) asyncSocket->getBufferedAmount() < maxBackpressure) { - + /* Pick uncompressed data track */ + std::string_view selectedData = data.first; + + /* Are we using compression? Fine, pick the compressed data track */ + WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); + if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { + selectedData = data.second; + } + /* Note: this assumes we are not corked, as corking will swallow things and fail later on */ - auto [written, failed] = asyncSocket->write(data.data(), (int) data.length()); + auto [written, failed] = asyncSocket->write(selectedData.data(), (int) selectedData.length()); if (!failed) { asyncSocket->timeout(this->idleTimeout); } @@ -97,7 +106,27 @@ struct WebSocketContextData { char *dst = (char *) malloc(protocol::messageFrameSize(message.size())); size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), opCode, message.length(), false); - topicTree.publish(topic, std::string_view(dst, dst_length)); + if (compress) { + /* Loop data holds shared compressor */ + LoopData *loopData = (LoopData *) us_loop_ext((us_loop_t *) Loop::get()); + + /* Compress it */ + std::string_view compressedMessage = loopData->deflationStream->deflate(loopData->zlibContext, message, true); + + /* Frame it */ + char *dst_compressed = (char *) malloc(protocol::messageFrameSize(compressedMessage.size())); + size_t dst_compressed_length = protocol::formatMessage(dst_compressed, compressedMessage.data(), compressedMessage.length(), opCode, compressedMessage.length(), true); + + /* Always publish the shortest one in any case */ + topicTree.publish(topic, {std::string_view(dst, dst_length), dst_compressed_length >= dst_length ? std::string_view(dst, dst_length) : std::string_view(dst_compressed, dst_compressed_length)}); + + /* We don't care for allocation here */ + ::free(dst_compressed); + } else { + /* If not compressing, put same message on both tracks */ + topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}); + } + ::free(dst); } }; diff --git a/src/WebSocketData.h b/src/WebSocketData.h index a51c1b1..c733981 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -28,6 +28,7 @@ namespace uWS { struct WebSocketData : AsyncSocketData, WebSocketState { template friend struct WebSocketContext; + template friend struct WebSocketContextData; template friend struct WebSocket; private: std::string fragmentBuffer;