Add shared compression for pub/sub

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 */

src/TopicTree.h

@@ -58,15 +58,15 @@ struct Topic {
     /* Terminating wildcard child */
     Topic *terminatingWildcardChild = nullptr;
 
-    /* What we published */
+    /* What we published, {inflated, deflated} */
-    std::map<unsigned int, std::string> messages;
+    std::map<unsigned int, std::pair<std::string, std::string>> messages;
 
     std::set<Subscriber *> subs;
 };
 
 struct TopicTree {
 private:
-    std::function<int(Subscriber *, std::string_view)> cb;
+    std::function<int(Subscriber *, std::pair<std::string_view, std::string_view>)> 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<std::string_view, std::string_view> 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<int(Subscriber *, std::string_view)> cb) {
+    TopicTree(std::function<int(Subscriber *, std::pair<std::string_view, std::string_view>)> 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<std::string_view, std::string_view> 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<uint64_t, std::string> intersectionCache;
+            std::map<uint64_t, std::pair<std::string, std::string>> intersectionCache;
 
             /* Loop over these here */
             std::set<Subscriber *>::iterator it[64];
@@ -332,7 +332,7 @@ public:
                 Subscriber *nextMin = (Subscriber *)UINTPTR_MAX;
 
                 /* The message sets relevant for this intersection */
-                std::map<unsigned int, std::string> *perSubscriberIntersectingTopicMessages[64];
+                std::map<unsigned int, std::pair<std::string, std::string>> *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<unsigned int, std::string> complete;
+                    std::map<unsigned int, std::pair<std::string, std::string>> complete;
                     for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) {
                         complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end());
                     }
 
-                    /* Create the linear cache */
+                    /* Create the linear cache, {inflated, deflated} */
-                    std::string res;
+                    std::pair<std::string, std::string> 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));

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<std::string_view, std::string_view> data) -> int {
         /* We rely on writing to regular asyncSockets */
         auto *asyncSocket = (AsyncSocket<SSL> *) 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<true>(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<true>(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);
     }
 };

src/WebSocketData.h

@@ -28,6 +28,7 @@ namespace uWS {
 
 struct WebSocketData : AsyncSocketData<false>, WebSocketState<true> {
     template <bool, bool> friend struct WebSocketContext;
+    template <bool> friend struct WebSocketContextData;
     template <bool, bool> friend struct WebSocket;
 private:
     std::string fragmentBuffer;