(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,7 +1,15 @@
 # Changelog
 All changes to this project will be documented in this file.
 
-## [7.7.0] - 2019-12-24
+## [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)
+
+## [7.8.0] - 2019-12-24
 
 (ws client) all commands use spdlog instead of std::cerr or std::cout for logging
 

ixwebsocket/IXWebSocketVersion.h

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

test/CMakeLists.txt

@@ -23,6 +23,8 @@ include_directories(
   ../ws
 )
 
+add_definitions(-DSPDLOG_COMPILED_LIB=1)
+
 find_package(JsonCpp)
 if (NOT JSONCPP_FOUND)
   include_directories(../third_party/jsoncpp)
@@ -98,4 +100,6 @@ target_link_libraries(ixwebsocket_unittest ixcrypto)
 target_link_libraries(ixwebsocket_unittest ixcore)
 target_link_libraries(ixwebsocket_unittest ixsentry)
 
+target_link_libraries(ixwebsocket_unittest spdlog)
+
 install(TARGETS ixwebsocket_unittest DESTINATION bin)

ws/CMakeLists.txt

@@ -25,6 +25,8 @@ include_directories(ws ../third_party/statsd-client-cpp/src)
 include_directories(ws ../third_party/spdlog/include)
 include_directories(ws ../third_party/cpp-linenoise)
 
+add_definitions(-DSPDLOG_COMPILED_LIB=1)
+
 if (UNIX)
   set( STATSD_CLIENT_SOURCES ../third_party/statsd-client-cpp/src/statsd_client.cpp)
 endif()
@@ -72,6 +74,8 @@ target_link_libraries(ws ixcrypto)
 target_link_libraries(ws ixcore)
 target_link_libraries(ws ixsentry)
 
+target_link_libraries(ws spdlog)
+
 if(NOT APPLE AND NOT USE_MBED_TLS)
   find_package(OpenSSL REQUIRED)
   add_definitions(${OPENSSL_DEFINITIONS})

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();