(ws cobra to sentry) bound the queue size used to hold up cobra messages before they are sent to sentry. Default queue size is a 100 messages. Without such limit the program runs out of memory when a subscriber receive a lot of messages that cannot make it to sentry

docs/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [7.8.2] - 2019-12-25
+
+(ws cobra to sentry) bound the queue size used to hold up cobra messages before they are sent to sentry. Default queue size is a 100 messages. Without such limit the program runs out of memory when a subscriber receive a lot of messages that cannot make it to sentry
+
 ## [7.8.1] - 2019-12-25
 
 (ws client) use correct compilation defines so that spdlog is not used as a header only library (reduce binary size and increase compilation speed)

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "7.7.1"
+#define IX_WEBSOCKET_VERSION "7.8.2"

ws/ws.cpp

@@ -105,6 +105,7 @@ int main(int argc, char** argv)
     int count = 1;
     int jobs = 4;
     uint32_t maxWaitBetweenReconnectionRetries;
+    size_t maxQueueSize = 100;
 
     auto addTLSOptions = [&tlsOptions, &verifyNone](CLI::App* app) {
         app->add_option(
@@ -268,6 +269,7 @@ int main(int argc, char** argv)
     cobra2sentry->add_option("--rolesecret", rolesecret, "Role secret")->required();
     cobra2sentry->add_option("--dsn", dsn, "Sentry DSN");
     cobra2sentry->add_option("--jobs", jobs, "Number of thread sending events to Sentry");
+    cobra2sentry->add_option("--queue_size", maxQueueSize, "Size of the queue to hold messages before they are sent to Sentry");
     cobra2sentry->add_option("channel", channel, "Channel")->required();
     cobra2sentry->add_flag("-v", verbose, "Verbose");
     cobra2sentry->add_flag("-s", strict, "Strict mode. Error out when sending to sentry fails");
@@ -455,6 +457,7 @@ int main(int argc, char** argv)
                                           verbose,
                                           strict,
                                           jobs,
+                                          maxQueueSize,
                                           tlsOptions);
     }
     else if (app.got_subcommand("cobra_metrics_to_redis"))

ws/ws.h

@@ -119,6 +119,7 @@ namespace ix
                                 bool verbose,
                                 bool strict,
                                 int jobs,
+                                size_t maxQueueSize,
                                 const ix::SocketTLSOptions& tlsOptions);
 
     int ws_cobra_metrics_to_redis(const std::string& appkey,

ws/ws_cobra_to_sentry.cpp

@@ -28,6 +28,7 @@ namespace ix
                                 bool verbose,
                                 bool strict,
                                 int jobs,
+                                size_t maxQueueSize,
                                 const ix::SocketTLSOptions& tlsOptions)
     {
         ix::CobraConnection conn;
@@ -176,6 +177,7 @@ namespace ix
                                &receivedCount,
                                &condition,
                                &conditionVariableMutex,
+                               &maxQueueSize,
                                &queue](ix::CobraConnectionEventType eventType,
                                        const std::string& errMsg,
                                        const ix::WebSocketHttpHeaders& headers,
@@ -205,6 +207,7 @@ namespace ix
                                 &receivedCount,
                                 &condition,
                                 &conditionVariableMutex,
+                                &maxQueueSize,
                                 &queue](const Json::Value& msg) {
                                    if (verbose)
                                    {
@@ -222,7 +225,12 @@ namespace ix
 
                                    {
                                        std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                                       queue.push(msg);
+                                       // if the sending is not fast enough there is no point
+                                       // in queuing too many events.
+                                       if (queue.size() < maxQueueSize)
+                                       {
+                                           queue.push(msg);
+                                       }
                                    }
 
                                    condition.notify_one();