(ws cobra to sentry) game is picked in a fair manner, so that all games get the same share of sent events

docs/CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [7.8.4] - 2019-12-27
+
+(ws cobra to sentry) game is picked in a fair manner, so that all games get the same share of sent events
+
+## [7.8.3] - 2019-12-27
+
+(ws cobra to sentry) refactor queue related code into a class
+
 ## [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

ixwebsocket/IXWebSocketVersion.h

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

ws/ws_cobra_to_sentry.cpp

@@ -15,9 +15,81 @@
 #include <sstream>
 #include <thread>
 #include <vector>
+#include <map>
 
 namespace ix
 {
+    class QueueManager
+    {
+    public:
+        QueueManager(size_t maxQueueSize, std::atomic<bool> &stop) : _maxQueueSize(maxQueueSize), _stop(stop) {}
+
+        Json::Value pop();
+        void add(Json::Value msg);
+
+    private:
+        std::map<std::string, std::queue<Json::Value>> _queues;
+        std::mutex _mutex;
+        std::condition_variable _condition;
+        size_t _maxQueueSize;
+        std::atomic<bool>& _stop;
+    };
+
+    Json::Value QueueManager::pop()
+    {
+        std::unique_lock<std::mutex> lock(_mutex);
+
+        if (_queues.empty())
+        {
+            Json::Value val;
+            return val;
+        }
+
+        std::vector<std::string> games;
+        for (auto it : _queues)
+        {
+            games.push_back(it.first);
+        }
+
+        std::random_shuffle(games.begin(), games.end());
+        std::string game = games[0];
+
+        spdlog::info("Sending event for game '{}'", game);
+
+        _condition.wait(lock, [this] { return !_stop; });
+
+        if (_queues[game].empty())
+        {
+            Json::Value val;
+            return val;
+        }
+
+        auto msg = _queues[game].front();
+        _queues[game].pop();
+        return msg;
+    }
+
+    void QueueManager::add(Json::Value msg)
+    {
+        std::unique_lock<std::mutex> lock(_mutex);
+
+        std::string game;
+        if (msg.isMember("device") && msg["device"].isMember("game"))
+        {
+            game = msg["device"]["game"].asString();
+        }
+
+        if (game.empty()) return;
+
+        // if the sending is not fast enough there is no point
+        // in queuing too many events.
+        if (_queues[game].size() < _maxQueueSize)
+        {
+            _queues[game].push(msg);
+            _condition.notify_one();
+        }
+    }
+
     int ws_cobra_to_sentry_main(const std::string& appkey,
                                 const std::string& endpoint,
                                 const std::string& rolename,
@@ -47,9 +119,7 @@ namespace ix
         std::atomic<bool> stop(false);
         std::atomic<bool> throttled(false);
 
-        std::condition_variable condition;
-        std::mutex conditionVariableMutex;
-        std::queue<Json::Value> queue;
+        QueueManager queueManager(maxQueueSize, stop);
 
         auto timer = [&sentCount, &receivedCount] {
             while (true)
@@ -63,9 +133,7 @@ namespace ix
 
         std::thread t1(timer);
 
-        auto sentrySender = [&condition,
-                             &conditionVariableMutex,
-                             &queue,
+        auto sentrySender = [&queueManager,
                              verbose,
                              &errorSending,
                              &sentCount,
@@ -76,15 +144,9 @@ namespace ix
 
             while (true)
             {
-                Json::Value msg;
+                Json::Value msg = queueManager.pop();
 
-                {
-                    std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                    condition.wait(lock, [&queue, &stop] { return !queue.empty() && !stop; });
-
-                    msg = queue.front();
-                    queue.pop();
-                }
+                if (stop) return;
 
                 auto ret = sentryClient.send(msg, verbose);
                 HttpResponsePtr response = ret.first;
@@ -175,14 +237,12 @@ namespace ix
                                verbose,
                                &throttled,
                                &receivedCount,
-                               &condition,
-                               &conditionVariableMutex,
-                               &maxQueueSize,
-                               &queue](ix::CobraConnectionEventType eventType,
-                                       const std::string& errMsg,
-                                       const ix::WebSocketHttpHeaders& headers,
-                                       const std::string& subscriptionId,
-                                       CobraConnection::MsgId msgId) {
+                               &queueManager](
+                                        ix::CobraConnectionEventType eventType,
+                                        const std::string& errMsg,
+                                        const ix::WebSocketHttpHeaders& headers,
+                                        const std::string& subscriptionId,
+                                        CobraConnection::MsgId msgId) {
             if (eventType == ix::CobraConnection_EventType_Open)
             {
                 spdlog::info("Subscriber connected");
@@ -205,10 +265,7 @@ namespace ix
                                 verbose,
                                 &throttled,
                                 &receivedCount,
-                                &condition,
-                                &conditionVariableMutex,
-                                &maxQueueSize,
-                                &queue](const Json::Value& msg) {
+                                &queueManager](const Json::Value& msg) {
                                    if (verbose)
                                    {
                                        spdlog::info(jsonWriter.write(msg));
@@ -217,23 +274,11 @@ namespace ix
                                    // If we cannot send to sentry fast enough, drop the message
                                    if (throttled)
                                    {
-                                       condition.notify_one();
                                        return;
                                    }
 
                                    ++receivedCount;
-
-                                   {
-                                       std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                                       // 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();
+                                   queueManager.add(msg);
                                });
             }
             else if (eventType == ix::CobraConnection_EventType_Subscribed)