(cobra bots) add a utility class to factor out the common bots features (heartbeat) and move cobra to sentry bot to use it

docs/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [9.3.1] - 2020-04-16
+
+(cobra bots) add a utility class to factor out the common bots features (heartbeat) and move cobra to sentry bot to use it
+
 ## [9.3.0] - 2020-04-15
 
 (websocket) add a positive number to the heartbeat message sent, incremented each time the heartbeat is sent

ixbots/CMakeLists.txt

@@ -4,6 +4,7 @@
 #
 
 set (IXBOTS_SOURCES
+    ixbots/IXCobraBot.cpp
     ixbots/IXCobraToSentryBot.cpp
     ixbots/IXCobraToStatsdBot.cpp
     ixbots/IXQueueManager.cpp
@@ -11,6 +12,7 @@ set (IXBOTS_SOURCES
 )
 
 set (IXBOTS_HEADERS
+    ixbots/IXCobraBot.h
     ixbots/IXCobraToSentryBot.h
     ixbots/IXCobraToStatsdBot.h
     ixbots/IXQueueManager.h

ixbots/ixbots/IXCobraBot.cpp

@@ -0,0 +1,247 @@
+/*
+ *  IXCobraBot.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXCobraBot.h"
+#include "IXQueueManager.h"
+
+#include <algorithm>
+#include <chrono>
+#include <ixcobra/IXCobraConnection.h>
+#include <spdlog/spdlog.h>
+#include <sstream>
+#include <thread>
+#include <vector>
+
+namespace ix
+{
+    int64_t CobraBot::run(const CobraConfig& config,
+                          const std::string& channel,
+                          const std::string& filter,
+                          const std::string& position,
+                          bool verbose,
+                          size_t maxQueueSize,
+                          bool enableHeartbeat,
+                          int runtime)
+    {
+        ix::CobraConnection conn;
+        conn.configure(config);
+        conn.connect();
+
+        Json::FastWriter jsonWriter;
+        std::atomic<uint64_t> sentCount(0);
+        std::atomic<uint64_t> receivedCount(0);
+        std::atomic<bool> stop(false);
+        std::atomic<bool> throttled(false);
+        std::atomic<bool> fatalCobraError(false);
+
+        QueueManager queueManager(maxQueueSize);
+
+        auto timer = [&sentCount, &receivedCount, &stop] {
+            while (!stop)
+            {
+                spdlog::info("messages received {} sent {}", receivedCount, sentCount);
+
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+            }
+
+            spdlog::info("timer thread done");
+        };
+
+        std::thread t1(timer);
+
+        auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
+            std::string state("na");
+
+            if (!enableHeartbeat) return;
+
+            while (!stop)
+            {
+                std::stringstream ss;
+                ss << "messages received " << receivedCount;
+                ss << "messages sent " << sentCount;
+
+                std::string currentState = ss.str();
+
+                if (currentState == state)
+                {
+                    spdlog::error("no messages received or sent for 1 minute, exiting");
+                    exit(1);
+                }
+                state = currentState;
+
+                auto duration = std::chrono::minutes(1);
+                std::this_thread::sleep_for(duration);
+            }
+
+            spdlog::info("heartbeat thread done");
+        };
+
+        std::thread t2(heartbeat);
+
+        auto sender =
+            [this, &queueManager, verbose, &sentCount, &stop, &throttled] {
+
+                while (true)
+                {
+                    Json::Value msg = queueManager.pop();
+
+                    if (stop) break;
+                    if (msg.isNull()) continue;
+
+                    if (_onBotMessageCallback && _onBotMessageCallback(msg, verbose, throttled))
+                    {
+                        // That might be too noisy
+                        if (verbose)
+                        {
+                            spdlog::info("cobra bot: sending succesfull");
+                        }
+                        ++sentCount;
+                    }
+                    else
+                    {
+                        spdlog::error("cobra bot: error sending");
+                    }
+
+                    if (stop) break;
+                }
+
+                spdlog::info("sender thread done");
+            };
+
+        std::thread t3(sender);
+
+        conn.setEventCallback([&conn,
+                               &channel,
+                               &filter,
+                               &position,
+                               &jsonWriter,
+                               verbose,
+                               &throttled,
+                               &receivedCount,
+                               &fatalCobraError,
+                               &queueManager](const CobraEventPtr& event)
+       {
+            if (event->type == ix::CobraEventType::Open)
+            {
+                spdlog::info("Subscriber connected");
+
+                for (auto&& it : event->headers)
+                {
+                    spdlog::info("{}: {}", it.first, it.second);
+                }
+            }
+            else if (event->type == ix::CobraEventType::Closed)
+            {
+                spdlog::info("Subscriber closed: {}", event->errMsg);
+            }
+            else if (event->type == ix::CobraEventType::Authenticated)
+            {
+                spdlog::info("Subscriber authenticated");
+                conn.subscribe(channel,
+                               filter,
+                               position,
+                               [&jsonWriter, verbose, &throttled, &receivedCount, &queueManager](
+                                   const Json::Value& msg, const std::string& position) {
+                                   if (verbose)
+                                   {
+                                       spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
+                                   }
+
+                                   // If we cannot send to sentry fast enough, drop the message
+                                   if (throttled)
+                                   {
+                                       return;
+                                   }
+
+                                   ++receivedCount;
+                                   queueManager.add(msg);
+                               });
+            }
+            else if (event->type == ix::CobraEventType::Subscribed)
+            {
+                spdlog::info("Subscriber: subscribed to channel {}", event->subscriptionId);
+            }
+            else if (event->type == ix::CobraEventType::UnSubscribed)
+            {
+                spdlog::info("Subscriber: unsubscribed from channel {}", event->subscriptionId);
+            }
+            else if (event->type == ix::CobraEventType::Error)
+            {
+                spdlog::error("Subscriber: error {}", event->errMsg);
+            }
+            else if (event->type == ix::CobraEventType::Published)
+            {
+                spdlog::error("Published message hacked: {}", event->msgId);
+            }
+            else if (event->type == ix::CobraEventType::Pong)
+            {
+                spdlog::info("Received websocket pong: {}", event->errMsg);
+            }
+            else if (event->type == ix::CobraEventType::HandshakeError)
+            {
+                spdlog::error("Subscriber: Handshake error: {}", event->errMsg);
+                fatalCobraError = true;
+            }
+            else if (event->type == ix::CobraEventType::AuthenticationError)
+            {
+                spdlog::error("Subscriber: Authentication error: {}", event->errMsg);
+                fatalCobraError = true;
+            }
+            else if (event->type == ix::CobraEventType::SubscriptionError)
+            {
+                spdlog::error("Subscriber: Subscription error: {}", event->errMsg);
+                fatalCobraError = true;
+            }
+        });
+
+        // Run forever
+        if (runtime == -1)
+        {
+            while (true)
+            {
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+
+                if (fatalCobraError) break;
+            }
+        }
+        // Run for a duration, used by unittesting now
+        else
+        {
+            for (int i = 0 ; i < runtime; ++i)
+            {
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+
+                if (fatalCobraError) break;
+            }
+        }
+
+        //
+        // Cleanup.
+        // join all the bg threads and stop them.
+        //
+        conn.disconnect();
+        stop = true;
+
+        // progress thread
+        t1.join();
+
+        // heartbeat thread
+        if (t2.joinable()) t2.join();
+
+        // sentry sender thread
+        t3.join();
+
+        return fatalCobraError ? -1 : (int64_t) sentCount;
+    }
+
+    void CobraBot::setOnBotMessageCallback(const OnBotMessageCallback& callback)
+    {
+        _onBotMessageCallback = callback;
+    }
+}

ixbots/ixbots/IXCobraBot.h

@@ -0,0 +1,42 @@
+/*
+ *  IXCobraBot.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <ixcobra/IXCobraConfig.h>
+#include <stddef.h>
+#include <json/json.h>
+#include <mutex>
+#include <condition_variable>
+#include <queue>
+#include <map>
+
+namespace ix
+{
+    using OnBotMessageCallback = std::function<bool(const Json::Value&,
+                                                    const bool verbose,
+                                                    std::atomic<bool>&)>;
+
+    class CobraBot
+    {
+    public:
+        CobraBot() = default;
+
+        int64_t run(const CobraConfig& config,
+                    const std::string& channel,
+                    const std::string& filter,
+                    const std::string& position,
+                    bool verbose,
+                    size_t maxQueueSize,
+                    bool enableHeartbeat,
+                    int runtime);
+
+        void setOnBotMessageCallback(const OnBotMessageCallback& callback);
+
+    private:
+        OnBotMessageCallback _onBotMessageCallback;
+    };
+}

ixbots/ixbots/IXCobraToSentryBot.cpp

@@ -5,6 +5,7 @@
  */
 
 #include "IXCobraToSentryBot.h"
+#include "IXCobraBot.h"
 #include "IXQueueManager.h"
 
 #include <chrono>
@@ -27,276 +28,87 @@ namespace ix
                             bool enableHeartbeat,
                             int runtime)
     {
-        ix::CobraConnection conn;
+        CobraBot bot;
-        conn.configure(config);
+        bot.setOnBotMessageCallback([&sentryClient](const Json::Value& msg,
-        conn.connect();
+                                                    const bool verbose,
+                                                    std::atomic<bool>& throttled) -> bool {
+            auto ret = sentryClient.send(msg, verbose);
+            HttpResponsePtr response = ret.first;
 
-        Json::FastWriter jsonWriter;
+            if (!response)
-        std::atomic<uint64_t> sentCount(0);
-        std::atomic<uint64_t> receivedCount(0);
-        std::atomic<bool> errorSending(false);
-        std::atomic<bool> stop(false);
-        std::atomic<bool> throttled(false);
-        std::atomic<bool> fatalCobraError(false);
-
-        QueueManager queueManager(maxQueueSize);
-
-        auto timer = [&sentCount, &receivedCount, &stop] {
-            while (!stop)
             {
-                spdlog::info("messages received {} sent {}", receivedCount, sentCount);
+                spdlog::warn("Null HTTP Response");
-
+                return false;
-                auto duration = std::chrono::seconds(1);
-                std::this_thread::sleep_for(duration);
             }
 
-            spdlog::info("timer thread done");
+            if (verbose)
-        };
-
-        std::thread t1(timer);
-
-        auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat] {
-            std::string state("na");
-
-            if (!enableHeartbeat) return;
-
-            while (!stop)
             {
-                std::stringstream ss;
+                for (auto it : response->headers)
-                ss << "messages received " << receivedCount;
-                ss << "messages sent " << sentCount;
-
-                std::string currentState = ss.str();
-
-                if (currentState == state)
-                {
-                    spdlog::error("no messages received or sent for 1 minute, exiting");
-                    exit(1);
-                }
-                state = currentState;
-
-                auto duration = std::chrono::minutes(1);
-                std::this_thread::sleep_for(duration);
-            }
-
-            spdlog::info("heartbeat thread done");
-        };
-
-        std::thread t2(heartbeat);
-
-        auto sentrySender =
-            [&queueManager, verbose, &errorSending, &sentCount, &stop, &throttled, &sentryClient] {
-
-                while (true)
-                {
-                    Json::Value msg = queueManager.pop();
-
-                    if (stop) break;
-                    if (msg.isNull()) continue;
-
-                    auto ret = sentryClient.send(msg, verbose);
-                    HttpResponsePtr response = ret.first;
-
-                    if (!response)
-                    {
-                        spdlog::warn("Null HTTP Response");
-                        continue;
-                    }
-
-                    if (verbose)
-                    {
-                        for (auto it : response->headers)
-                        {
-                            spdlog::info("{}: {}", it.first, it.second);
-                        }
-
-                        spdlog::info("Upload size: {}", response->uploadSize);
-                        spdlog::info("Download size: {}", response->downloadSize);
-
-                        spdlog::info("Status: {}", response->statusCode);
-                        if (response->errorCode != HttpErrorCode::Ok)
-                        {
-                            spdlog::info("error message: {}", response->errorMsg);
-                        }
-
-                        if (response->headers["Content-Type"] != "application/octet-stream")
-                        {
-                            spdlog::info("payload: {}", response->payload);
-                        }
-                    }
-
-                    if (response->statusCode != 200)
-                    {
-                        spdlog::error("Error sending data to sentry: {}", response->statusCode);
-                        spdlog::error("Body: {}", ret.second);
-                        spdlog::error("Response: {}", response->payload);
-                        errorSending = true;
-
-                        // Error 429 Too Many Requests
-                        if (response->statusCode == 429)
-                        {
-                            auto retryAfter = response->headers["Retry-After"];
-                            std::stringstream ss;
-                            ss << retryAfter;
-                            int seconds;
-                            ss >> seconds;
-
-                            if (!ss.eof() || ss.fail())
-                            {
-                                seconds = 30;
-                                spdlog::warn("Error parsing Retry-After header. "
-                                             "Using {} for the sleep duration",
-                                             seconds);
-                            }
-
-                            spdlog::warn("Error 429 - Too Many Requests. ws will sleep "
-                                         "and retry after {} seconds",
-                                         retryAfter);
-
-                            throttled = true;
-                            auto duration = std::chrono::seconds(seconds);
-                            std::this_thread::sleep_for(duration);
-                            throttled = false;
-                        }
-                    }
-                    else
-                    {
-                        ++sentCount;
-                    }
-
-                    if (stop) break;
-                }
-
-                spdlog::info("sentrySender thread done");
-            };
-
-        std::thread t3(sentrySender);
-
-        conn.setEventCallback([&conn,
-                               &channel,
-                               &filter,
-                               &position,
-                               &jsonWriter,
-                               verbose,
-                               &throttled,
-                               &receivedCount,
-                               &fatalCobraError,
-                               &queueManager](const CobraEventPtr& event)
-       {
-            if (event->type == ix::CobraEventType::Open)
-            {
-                spdlog::info("Subscriber connected");
-
-                for (auto&& it : event->headers)
                 {
                     spdlog::info("{}: {}", it.first, it.second);
                 }
-            }
-            else if (event->type == ix::CobraEventType::Closed)
-            {
-                spdlog::info("Subscriber closed: {}", event->errMsg);
-            }
-            else if (event->type == ix::CobraEventType::Authenticated)
-            {
-                spdlog::info("Subscriber authenticated");
-                conn.subscribe(channel,
-                               filter,
-                               position,
-                               [&jsonWriter, verbose, &throttled, &receivedCount, &queueManager](
-                                   const Json::Value& msg, const std::string& position) {
-                                   if (verbose)
-                                   {
-                                       spdlog::info("Subscriber received message {} -> {}", position, jsonWriter.write(msg));
-                                   }
 
-                                   // If we cannot send to sentry fast enough, drop the message
+                spdlog::info("Upload size: {}", response->uploadSize);
-                                   if (throttled)
+                spdlog::info("Download size: {}", response->downloadSize);
-                                   {
-                                       return;
-                                   }
 
-                                   ++receivedCount;
+                spdlog::info("Status: {}", response->statusCode);
-                                   queueManager.add(msg);
+                if (response->errorCode != HttpErrorCode::Ok)
-                               });
+                {
+                    spdlog::info("error message: {}", response->errorMsg);
+                }
+
+                if (response->headers["Content-Type"] != "application/octet-stream")
+                {
+                    spdlog::info("payload: {}", response->payload);
+                }
             }
-            else if (event->type == ix::CobraEventType::Subscribed)
+
+            bool success = response->statusCode == 200;
+
+            if (!success)
             {
-                spdlog::info("Subscriber: subscribed to channel {}", event->subscriptionId);
+                spdlog::error("Error sending data to sentry: {}", response->statusCode);
-            }
+                spdlog::error("Body: {}", ret.second);
-            else if (event->type == ix::CobraEventType::UnSubscribed)
+                spdlog::error("Response: {}", response->payload);
-            {
+
-                spdlog::info("Subscriber: unsubscribed from channel {}", event->subscriptionId);
+                // Error 429 Too Many Requests
-            }
+                if (response->statusCode == 429)
-            else if (event->type == ix::CobraEventType::Error)
+                {
-            {
+                    auto retryAfter = response->headers["Retry-After"];
-                spdlog::error("Subscriber: error {}", event->errMsg);
+                    std::stringstream ss;
-            }
+                    ss << retryAfter;
-            else if (event->type == ix::CobraEventType::Published)
+                    int seconds;
-            {
+                    ss >> seconds;
-                spdlog::error("Published message hacked: {}", event->msgId);
+
-            }
+                    if (!ss.eof() || ss.fail())
-            else if (event->type == ix::CobraEventType::Pong)
+                    {
-            {
+                        seconds = 30;
-                spdlog::info("Received websocket pong: {}", event->errMsg);
+                        spdlog::warn("Error parsing Retry-After header. "
-            }
+                                     "Using {} for the sleep duration",
-            else if (event->type == ix::CobraEventType::HandshakeError)
+                                     seconds);
-            {
+                    }
-                spdlog::error("Subscriber: Handshake error: {}", event->errMsg);
+
-                fatalCobraError = true;
+                    spdlog::warn("Error 429 - Too Many Requests. ws will sleep "
-            }
+                                 "and retry after {} seconds",
-            else if (event->type == ix::CobraEventType::AuthenticationError)
+                                 retryAfter);
-            {
+
-                spdlog::error("Subscriber: Authentication error: {}", event->errMsg);
+                    throttled = true;
-                fatalCobraError = true;
+                    auto duration = std::chrono::seconds(seconds);
-            }
+                    std::this_thread::sleep_for(duration);
-            else if (event->type == ix::CobraEventType::SubscriptionError)
+                    throttled = false;
-            {
+                }
-                spdlog::error("Subscriber: Subscription error: {}", event->errMsg);
-                fatalCobraError = true;
             }
+
+            return success;
         });
 
-        // Run forever
+        return bot.run(config,
-        if (runtime == -1)
+                       channel,
-        {
+                       filter,
-            while (true)
+                       position,
-            {
+                       verbose,
-                auto duration = std::chrono::seconds(1);
+                       maxQueueSize,
-                std::this_thread::sleep_for(duration);
+                       enableHeartbeat,
-
+                       runtime);
-                if (strict && errorSending) break;
-                if (fatalCobraError) break;
-            }
-        }
-        // Run for a duration, used by unittesting now
-        else
-        {
-            for (int i = 0 ; i < runtime; ++i)
-            {
-                auto duration = std::chrono::seconds(1);
-                std::this_thread::sleep_for(duration);
-
-                if (strict && errorSending) break;
-                if (fatalCobraError) break;
-            }
-        }
-
-        //
-        // Cleanup.
-        // join all the bg threads and stop them.
-        //
-        conn.disconnect();
-        stop = true;
-
-        // progress thread
-        t1.join();
-
-        // heartbeat thread
-        if (t2.joinable()) t2.join();
-
-        // sentry sender thread
-        t3.join();
-
-        return ((strict && errorSending) || fatalCobraError) ? -1 : (int) sentCount;
     }
 } // namespace ix

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "9.3.0"
+#define IX_WEBSOCKET_VERSION "9.3.1"