add a github action to publish a docker container for ws

faye-websocket-ruby to receive messages as fast as possible

.github/workflows/docker.yml

@@ -0,0 +1,66 @@
+name: docker
+
+# When its time to do a release do a build for amd64
+# and push all of them to Docker Hub.
+# Only trigger on semver shaped tags.
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+jobs:
+  login:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Prepare
+        id: prep
+        run: |
+          DOCKER_IMAGE=machinezone/ws
+          VERSION=edge
+          if [[ $GITHUB_REF == refs/tags/* ]]; then
+            VERSION=${GITHUB_REF#refs/tags/v}
+          fi
+          if [ "${{ github.event_name }}" = "schedule" ]; then
+            VERSION=nightly
+          fi
+          TAGS="${DOCKER_IMAGE}:${VERSION}"
+          if [[ $VERSION =~ ^[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}$ ]]; then
+            TAGS="$TAGS,${DOCKER_IMAGE}:latest"
+          fi
+          echo ::set-output name=tags::${TAGS}
+
+      - name: Set up Docker Buildx
+        id: buildx
+        uses: docker/setup-buildx-action@master
+
+      - name: Cache Docker layers
+        uses: actions/cache@v2
+        with:
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-
+
+      - name: Login to GitHub Package Registry
+        uses: docker/login-action@v1
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.GHCR_TOKEN }}
+
+      - name: Build and push
+        id: docker_build
+        uses: docker/build-push-action@v2-build-push
+        with:
+          builder: ${{ steps.buildx.outputs.name }}
+          context: .
+          file: ./Dockerfile
+          target: prod
+          platforms: linux/amd64
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.prep.outputs.tags }}
+          cache-from: type=local,src=/tmp/.buildx-cache
+          cache-to: type=local,dest=/tmp/.buildx-cache

CMakeLists.txt

@@ -62,7 +62,6 @@ set( IXWEBSOCKET_SOURCES
 set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXBench.h
     ixwebsocket/IXCancellationRequest.h
-    ixwebsocket/IXConnectionInfo.h
     ixwebsocket/IXConnectionState.h
     ixwebsocket/IXDNSLookup.h
     ixwebsocket/IXExponentialBackoff.h

docs/CHANGELOG.md

@@ -2,6 +2,34 @@
 
 All changes to this project will be documented in this file.
 
+## [10.3.2] - 2020-08-31
+
+(ws + cobra bots) add a cobra_to_cobra ws subcommand to subscribe to a channel and republish received events to a different channel
+
+## [10.3.1] - 2020-08-28
+
+(socket servers) merge the ConnectionInfo class with the ConnectionState one, which simplify all the server apis
+
+## [10.3.0] - 2020-08-26
+
+(ws) set the main thread name, to help with debugging in XCode, gdb, lldb etc...
+
+## [10.2.9] - 2020-08-19
+
+(ws) cobra to python bot / take a module python name as argument foo.bar.baz instead of a path foo/bar/baz.py
+
+## [10.2.8] - 2020-08-19
+
+(ws) on Linux with mbedtls, when the system ca certs are specified (the default) pick up sensible OS supplied paths (tested with CentOS and Alpine)
+
+## [10.2.7] - 2020-08-18
+
+(ws push_server) on the server side, stop sending and close the connection when the remote end has disconnected
+
+## [10.2.6] - 2020-08-17
+
+(ixwebsocket) replace std::unique_ptr<unsigned char[]> with std::array for some fixed arrays (which are in C++11)
+
 ## [10.2.5] - 2020-08-15
 
 (ws) merge all ws_*.cpp files into a single one to speedup compilation

docs/usage.md

@@ -280,10 +280,9 @@ ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
     [&server](std::weak_ptr<WebSocket> webSocket,
-              std::shared_ptr<ConnectionState> connectionState,
-              std::unique_ptr<ConnectionInfo> connectionInfo)
+              std::shared_ptr<ConnectionState> connectionState)
     {
-        std::cout << "Remote ip: " << connectionInfo->remoteIp << std::endl;
+        std::cout << "Remote ip: " << connectionState->remoteIp << std::endl;
 
         auto ws = webSocket.lock();
         if (ws)
@@ -359,13 +358,12 @@ The webSocket reference is guaranteed to be always valid ; by design the callbac
 ix::WebSocketServer server(port);
 
 server.setOnClientMessageCallback(std::shared_ptr<ConnectionState> connectionState,
-                                  ConnectionInfo& connectionInfo,
                                   WebSocket& webSocket,
                                   const WebSocketMessagePtr& msg)
 {
-    // The ConnectionInfo object contains information about the connection,
+    // The ConnectionState object contains information about the connection,
     // at this point only the client ip address and the port.
-    std::cout << "Remote ip: " << connectionInfo.remoteIp << std::endl;
+    std::cout << "Remote ip: " << connectionState->getRemoteIp();
 
     if (msg->type == ix::WebSocketMessageType::Open)
     {
@@ -519,12 +517,11 @@ If you want to handle how requests are processed, implement the setOnConnectionC
 ```cpp
 setOnConnectionCallback(
     [this](HttpRequestPtr request,
-           std::shared_ptr<ConnectionState> /*connectionState*/,
-           std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr
+           std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr
     {
         // Build a string for the response
         std::stringstream ss;
-        ss << connectionInfo->remoteIp
+        ss << connectionState->getRemoteIp();
            << " "
            << request->method
            << " "

ixbots/CMakeLists.txt

@@ -5,6 +5,7 @@
 
 set (IXBOTS_SOURCES
     ixbots/IXCobraBot.cpp
+    ixbots/IXCobraToCobraBot.cpp
     ixbots/IXCobraToSentryBot.cpp
     ixbots/IXCobraToStatsdBot.cpp
     ixbots/IXCobraToStdoutBot.cpp
@@ -16,6 +17,7 @@ set (IXBOTS_SOURCES
 set (IXBOTS_HEADERS
     ixbots/IXCobraBot.h
     ixbots/IXCobraBotConfig.h
+    ixbots/IXCobraToCobraBot.h
     ixbots/IXCobraToSentryBot.h
     ixbots/IXCobraToStatsdBot.h
     ixbots/IXCobraToStdoutBot.h

ixbots/ixbots/IXCobraToCobraBot.cpp

@@ -0,0 +1,43 @@
+/*
+ *  IXCobraToCobraBot.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXCobraToCobraBot.h"
+
+#include "IXCobraBot.h"
+#include <ixcobra/IXCobraMetricsPublisher.h>
+#include <sstream>
+
+namespace ix
+{
+    int64_t cobra_to_cobra_bot(const ix::CobraBotConfig& cobraBotConfig,
+                               const std::string& republishChannel,
+                               const std::string& publisherRolename,
+                               const std::string& publisherRolesecret)
+    {
+        CobraBot bot;
+
+        CobraMetricsPublisher cobraMetricsPublisher;
+        CobraConfig cobraPublisherConfig = cobraBotConfig.cobraConfig;
+        cobraPublisherConfig.rolename = publisherRolename;
+        cobraPublisherConfig.rolesecret = publisherRolesecret;
+        cobraMetricsPublisher.configure(cobraPublisherConfig, republishChannel);
+
+        bot.setOnBotMessageCallback(
+            [&republishChannel, &cobraMetricsPublisher](const Json::Value& msg,
+                                                        const std::string& /*position*/,
+                                                        std::atomic<bool>& /*throttled*/,
+                                                        std::atomic<bool>& /*fatalCobraError*/,
+                                                        std::atomic<uint64_t>& sentCount) -> void {
+                Json::Value msgWithNoId(msg);
+                msgWithNoId.removeMember("id");
+            
+                cobraMetricsPublisher.push(republishChannel, msg);
+                sentCount++;
+            });
+
+        return bot.run(cobraBotConfig);
+    }
+} // namespace ix

ixbots/ixbots/IXCobraToCobraBot.h

@@ -0,0 +1,20 @@
+/*
+ *  IXCobraToCobraBot.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+#pragma once
+
+#include <cstdint>
+#include <ixbots/IXStatsdClient.h>
+#include "IXCobraBotConfig.h"
+#include <stddef.h>
+#include <string>
+
+namespace ix
+{
+    int64_t cobra_to_cobra_bot(const ix::CobraBotConfig& config,
+                               const std::string& republishChannel,
+                               const std::string& publisherRolename,
+                               const std::string& publisherRolesecret);
+} // namespace ix

ixbots/ixbots/IXCobraToPythonBot.cpp

@@ -102,7 +102,7 @@ namespace ix
 {
     int64_t cobra_to_python_bot(const ix::CobraBotConfig& config,
                                 StatsdClient& statsdClient,
-                                const std::string& scriptPath)
+                                const std::string& moduleName)
     {
 #ifndef IXBOTS_USE_PYTHON
         CoreLogger::error("Command is disabled. "
@@ -113,10 +113,7 @@ namespace ix
         Py_InitializeEx(0); // 0 arg so that we do not install signal handlers 
                             // which prevent us from using Ctrl-C
 
-        size_t lastIndex = scriptPath.find_last_of("."); 
-        std::string modulePath = scriptPath.substr(0, lastIndex);
-
-        PyObject* pyModuleName = PyUnicode_DecodeFSDefault(modulePath.c_str());
+        PyObject* pyModuleName = PyUnicode_DecodeFSDefault(moduleName.c_str());
 
         if (pyModuleName == nullptr)
         {

ixbots/ixbots/IXCobraToPythonBot.h

@@ -15,5 +15,5 @@ namespace ix
 {
     int64_t cobra_to_python_bot(const ix::CobraBotConfig& config,
                                 StatsdClient& statsdClient,
-                                const std::string& scriptPath);
+                                const std::string& moduleName);
 } // namespace ix

ixcobra/ixcobra/IXCobraMetricsThreadedPublisher.cpp

@@ -24,6 +24,7 @@ namespace ix
     {
         _cobra_connection.setEventCallback([](const CobraEventPtr& event) {
             std::stringstream ss;
+            ix::LogLevel logLevel = LogLevel::Info;
 
             if (event->type == ix::CobraEventType::Open)
             {
@@ -41,6 +42,7 @@ namespace ix
             else if (event->type == ix::CobraEventType::Error)
             {
                 ss << "Error: " << event->errMsg;
+                logLevel = ix::LogLevel::Error;
             }
             else if (event->type == ix::CobraEventType::Closed)
             {
@@ -57,6 +59,7 @@ namespace ix
             else if (event->type == ix::CobraEventType::Published)
             {
                 ss << "Published message " << event->msgId << " acked";
+                logLevel = ix::LogLevel::Debug;
             }
             else if (event->type == ix::CobraEventType::Pong)
             {
@@ -65,17 +68,20 @@ namespace ix
             else if (event->type == ix::CobraEventType::HandshakeError)
             {
                 ss << "Handshake error: " << event->errMsg;
+                logLevel = ix::LogLevel::Error;
             }
             else if (event->type == ix::CobraEventType::AuthenticationError)
             {
                 ss << "Authentication error: " << event->errMsg;
+                logLevel = ix::LogLevel::Error;
             }
             else if (event->type == ix::CobraEventType::SubscriptionError)
             {
                 ss << "Subscription error: " << event->errMsg;
+                logLevel = ix::LogLevel::Error;
             }
 
-            CoreLogger::log(ss.str().c_str());
+            CoreLogger::log(ss.str().c_str(), logLevel);
         });
     }
 

ixredis/ixredis/IXRedisServer.cpp

@@ -45,10 +45,9 @@ namespace ix
     }
 
     void RedisServer::handleConnection(std::unique_ptr<Socket> socket,
-                                       std::shared_ptr<ConnectionState> connectionState,
-                                       std::unique_ptr<ConnectionInfo> connectionInfo)
+                                       std::shared_ptr<ConnectionState> connectionState)
     {
-        logInfo("New connection from remote ip " + connectionInfo->remoteIp);
+        logInfo("New connection from remote ip " + connectionState->getRemoteIp());
 
         _connectedClientsCount++;
 

ixredis/ixredis/IXRedisServer.h

@@ -44,8 +44,7 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket>,
-                                      std::shared_ptr<ConnectionState> connectionState,
-                                      std::unique_ptr<ConnectionInfo> connectionInfo) final;
+                                      std::shared_ptr<ConnectionState> connectionState) final;
         virtual size_t getConnectedClientsCount() final;
 
         bool startsWith(const std::string& str, const std::string& start);

ixsnake/ixsnake/IXSnakeServer.cpp

@@ -61,11 +61,10 @@ namespace snake
 
         _server.setOnClientMessageCallback(
             [this](std::shared_ptr<ix::ConnectionState> connectionState,
-                   ix::ConnectionInfo& connectionInfo,
                    ix::WebSocket& webSocket,
                    const ix::WebSocketMessagePtr& msg) {
                 auto state = std::dynamic_pointer_cast<SnakeConnectionState>(connectionState);
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
             
                 std::stringstream ss;
                 ss << "[" << state->getId() << "] ";

ixwebsocket/IXConnectionInfo.h

@@ -1,25 +0,0 @@
-/*
- *  IXConnectionInfo.h
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-
-#pragma once
-
-#include <string>
-
-namespace ix
-{
-    struct ConnectionInfo
-    {
-        std::string remoteIp;
-        int remotePort;
-
-        ConnectionInfo(const std::string& r = std::string(), int p = 0)
-            : remoteIp(r)
-            , remotePort(p)
-        {
-            ;
-        }
-    };
-} // namespace ix

ixwebsocket/IXConnectionState.cpp

@@ -50,4 +50,24 @@ namespace ix
             _onSetTerminatedCallback();
         }
     }
+
+    const std::string& ConnectionState::getRemoteIp()
+    {
+        return _remoteIp;
+    }
+
+    int ConnectionState::getRemotePort()
+    {
+        return _remotePort;
+    }
+
+    void ConnectionState::setRemoteIp(const std::string& remoteIp)
+    {
+        _remoteIp = remoteIp;
+    }
+
+    void ConnectionState::setRemotePort(int remotePort)
+    {
+        _remotePort = remotePort;
+    }
 } // namespace ix

ixwebsocket/IXConnectionState.h

@@ -28,11 +28,17 @@ namespace ix
         void setTerminated();
         bool isTerminated() const;
 
+        const std::string& getRemoteIp();
+        int getRemotePort();
+
         static std::shared_ptr<ConnectionState> createConnectionState();
 
     private:
         void setOnSetTerminatedCallback(const OnSetTerminatedCallback& callback);
 
+        void setRemoteIp(const std::string& remoteIp);
+        void setRemotePort(int remotePort);
+
     protected:
         std::atomic<bool> _terminated;
         std::string _id;
@@ -40,6 +46,9 @@ namespace ix
 
         static std::atomic<uint64_t> _globalId;
 
+        std::string _remoteIp;
+        int _remotePort;
+
         friend class SocketServer;
     };
 } // namespace ix

ixwebsocket/IXHttpClient.cpp

@@ -10,6 +10,7 @@
 #include "IXUrlParser.h"
 #include "IXUserAgent.h"
 #include "IXWebSocketHttpHeaders.h"
+#include <array>
 #include <assert.h>
 #include <cstring>
 #include <iomanip>
@@ -700,14 +701,12 @@ namespace ix
         inflateState.next_in = (unsigned char*) (const_cast<char*>(in.data()));
 
         const int kBufferSize = 1 << 14;
-
-        std::unique_ptr<unsigned char[]> compressBuffer =
-            std::make_unique<unsigned char[]>(kBufferSize);
+        std::array<unsigned char, kBufferSize> compressBuffer;
 
         do
         {
             inflateState.avail_out = (uInt) kBufferSize;
-            inflateState.next_out = compressBuffer.get();
+            inflateState.next_out = &compressBuffer.front();
 
             int ret = inflate(&inflateState, Z_SYNC_FLUSH);
 
@@ -717,7 +716,7 @@ namespace ix
                 return false;
             }
 
-            out.append(reinterpret_cast<char*>(compressBuffer.get()),
+            out.append(reinterpret_cast<char*>(&compressBuffer.front()),
                        kBufferSize - inflateState.avail_out);
         } while (inflateState.avail_out == 0);
 

ixwebsocket/IXHttpServer.cpp

@@ -120,8 +120,7 @@ namespace ix
     }
 
     void HttpServer::handleConnection(std::unique_ptr<Socket> socket,
-                                      std::shared_ptr<ConnectionState> connectionState,
-                                      std::unique_ptr<ConnectionInfo> connectionInfo)
+                                      std::shared_ptr<ConnectionState> connectionState)
     {
         _connectedClientsCount++;
 
@@ -130,8 +129,7 @@ namespace ix
 
         if (std::get<0>(ret))
         {
-            auto response =
-                _onConnectionCallback(std::get<2>(ret), connectionState, std::move(connectionInfo));
+            auto response = _onConnectionCallback(std::get<2>(ret), connectionState);
             if (!Http::sendResponse(response, socket))
             {
                 logError("Cannot send response");
@@ -151,8 +149,7 @@ namespace ix
     {
         setOnConnectionCallback(
             [this](HttpRequestPtr request,
-                   std::shared_ptr<ConnectionState> /*connectionState*/,
-                   std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
+                   std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr {
                 std::string uri(request->uri);
                 if (uri.empty() || uri == "/")
                 {
@@ -184,8 +181,8 @@ namespace ix
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionState->getRemoteIp() << ":" << connectionState->getRemotePort()
+                   << " " << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri << " " << content.size();
                 logInfo(ss.str());
 
@@ -209,16 +206,16 @@ namespace ix
         // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Redirections
         //
         setOnConnectionCallback(
-            [this, redirectUrl](HttpRequestPtr request,
-                                std::shared_ptr<ConnectionState> /*connectionState*/,
-                                std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
+            [this,
+             redirectUrl](HttpRequestPtr request,
+                          std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionState->getRemoteIp() << ":" << connectionState->getRemotePort()
+                   << " " << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
                 logInfo(ss.str());
 

ixwebsocket/IXHttpServer.h

@@ -23,9 +23,7 @@ namespace ix
     {
     public:
         using OnConnectionCallback =
-            std::function<HttpResponsePtr(HttpRequestPtr,
-                                          std::shared_ptr<ConnectionState>,
-                                          std::unique_ptr<ConnectionInfo> connectionInfo)>;
+            std::function<HttpResponsePtr(HttpRequestPtr, std::shared_ptr<ConnectionState>)>;
 
         HttpServer(int port = SocketServer::kDefaultPort,
                    const std::string& host = SocketServer::kDefaultHost,
@@ -46,8 +44,7 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket>,
-                                      std::shared_ptr<ConnectionState> connectionState,
-                                      std::unique_ptr<ConnectionInfo> connectionInfo) final;
+                                      std::shared_ptr<ConnectionState> connectionState) final;
         virtual size_t getConnectedClientsCount() final;
 
         void setDefaultConnectionCallback();

ixwebsocket/IXSocketServer.cpp

@@ -332,12 +332,13 @@ namespace ix
             }
 
             // Retrieve connection info, the ip address of the remote peer/client)
-            std::unique_ptr<ConnectionInfo> connectionInfo;
+            std::string remoteIp;
+            int remotePort;
 
             if (_addressFamily == AF_INET)
             {
-                char remoteIp[INET_ADDRSTRLEN];
-                if (inet_ntop(AF_INET, &client.sin_addr, remoteIp, INET_ADDRSTRLEN) == nullptr)
+                char remoteIp4[INET_ADDRSTRLEN];
+                if (inet_ntop(AF_INET, &client.sin_addr, remoteIp4, INET_ADDRSTRLEN) == nullptr)
                 {
                     int err = Socket::getErrno();
                     std::stringstream ss;
@@ -350,12 +351,13 @@ namespace ix
                     continue;
                 }
 
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
+                remotePort = client.sin_port;
+                remoteIp = remoteIp4;
             }
             else // AF_INET6
             {
-                char remoteIp[INET6_ADDRSTRLEN];
-                if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp, INET6_ADDRSTRLEN) == nullptr)
+                char remoteIp6[INET6_ADDRSTRLEN];
+                if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp6, INET6_ADDRSTRLEN) == nullptr)
                 {
                     int err = Socket::getErrno();
                     std::stringstream ss;
@@ -368,7 +370,8 @@ namespace ix
                     continue;
                 }
 
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
+                remotePort = client.sin_port;
+                remoteIp = remoteIp6;
             }
 
             std::shared_ptr<ConnectionState> connectionState;
@@ -377,6 +380,8 @@ namespace ix
                 connectionState = _connectionStateFactory();
             }
             connectionState->setOnSetTerminatedCallback([this] { onSetTerminatedCallback(); });
+            connectionState->setRemoteIp(remoteIp);
+            connectionState->setRemotePort(remotePort);
 
             if (_stop) return;
 
@@ -404,13 +409,10 @@ namespace ix
 
             // Launch the handleConnection work asynchronously in its own thread.
             std::lock_guard<std::mutex> lock(_connectionsThreadsMutex);
-            _connectionsThreads.push_back(
-                std::make_pair(connectionState,
-                               std::thread(&SocketServer::handleConnection,
-                                           this,
-                                           std::move(socket),
-                                           connectionState,
-                                           std::move(connectionInfo))));
+            _connectionsThreads.push_back(std::make_pair(
+                connectionState,
+                std::thread(
+                    &SocketServer::handleConnection, this, std::move(socket), connectionState)));
         }
     }
 

ixwebsocket/IXSocketServer.h

@@ -6,7 +6,6 @@
 
 #pragma once
 
-#include "IXConnectionInfo.h"
 #include "IXConnectionState.h"
 #include "IXSelectInterrupt.h"
 #include "IXSocketTLSOptions.h"
@@ -105,8 +104,7 @@ namespace ix
         ConnectionStateFactory _connectionStateFactory;
 
         virtual void handleConnection(std::unique_ptr<Socket>,
-                                      std::shared_ptr<ConnectionState> connectionState,
-                                      std::unique_ptr<ConnectionInfo> connectionInfo) = 0;
+                                      std::shared_ptr<ConnectionState> connectionState) = 0;
         virtual size_t getConnectedClientsCount() = 0;
 
         // Returns true if all connection threads are joined

ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp

@@ -16,8 +16,6 @@ namespace
     // is treated as a char* and the null termination (\x00) makes it
     // look like an empty string.
     const std::string kEmptyUncompressedBlock = std::string("\x00\x00\xff\xff", 4);
-
-    const int kBufferSize = 1 << 14;
 } // namespace
 
 namespace ix
@@ -26,7 +24,6 @@ namespace ix
     // Compressor
     //
     WebSocketPerMessageDeflateCompressor::WebSocketPerMessageDeflateCompressor()
-        : _compressBufferSize(kBufferSize)
     {
 #ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_deflateState, 0, sizeof(_deflateState));
@@ -57,8 +54,6 @@ namespace ix
 
         if (ret != Z_OK) return false;
 
-        _compressBuffer = std::make_unique<unsigned char[]>(_compressBufferSize);
-
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
@@ -145,14 +140,14 @@ namespace ix
         do
         {
             // Output to local buffer
-            _deflateState.avail_out = (uInt) _compressBufferSize;
-            _deflateState.next_out = _compressBuffer.get();
+            _deflateState.avail_out = (uInt) _compressBuffer.size();
+            _deflateState.next_out = &_compressBuffer.front();
 
             deflate(&_deflateState, _flush);
 
-            output = _compressBufferSize - _deflateState.avail_out;
+            output = _compressBuffer.size() - _deflateState.avail_out;
 
-            out.insert(out.end(), _compressBuffer.get(), _compressBuffer.get() + output);
+            out.insert(out.end(), _compressBuffer.begin(), _compressBuffer.begin() + output);
         } while (_deflateState.avail_out == 0);
 
         if (endsWithEmptyUnCompressedBlock(out))
@@ -170,7 +165,6 @@ namespace ix
     // Decompressor
     //
     WebSocketPerMessageDeflateDecompressor::WebSocketPerMessageDeflateDecompressor()
-        : _compressBufferSize(kBufferSize)
     {
 #ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_inflateState, 0, sizeof(_inflateState));
@@ -198,8 +192,6 @@ namespace ix
 
         if (ret != Z_OK) return false;
 
-        _compressBuffer = std::make_unique<unsigned char[]>(_compressBufferSize);
-
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
@@ -232,8 +224,8 @@ namespace ix
 
         do
         {
-            _inflateState.avail_out = (uInt) _compressBufferSize;
-            _inflateState.next_out = _compressBuffer.get();
+            _inflateState.avail_out = (uInt) _compressBuffer.size();
+            _inflateState.next_out = &_compressBuffer.front();
 
             int ret = inflate(&_inflateState, Z_SYNC_FLUSH);
 
@@ -242,8 +234,8 @@ namespace ix
                 return false; // zlib error
             }
 
-            out.append(reinterpret_cast<char*>(_compressBuffer.get()),
-                       _compressBufferSize - _inflateState.avail_out);
+            out.append(reinterpret_cast<char*>(&_compressBuffer.front()),
+                       _compressBuffer.size() - _inflateState.avail_out);
         } while (_inflateState.avail_out == 0);
 
         return true;

ixwebsocket/IXWebSocketPerMessageDeflateCodec.h

@@ -9,7 +9,7 @@
 #ifdef IXWEBSOCKET_USE_ZLIB
 #include "zlib.h"
 #endif
-#include <memory>
+#include <array>
 #include <string>
 #include <vector>
 
@@ -34,8 +34,7 @@ namespace ix
         bool endsWithEmptyUnCompressedBlock(const T& value);
 
         int _flush;
-        size_t _compressBufferSize;
-        std::unique_ptr<unsigned char[]> _compressBuffer;
+        std::array<unsigned char, 1 << 14> _compressBuffer;
 
 #ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _deflateState;
@@ -53,8 +52,7 @@ namespace ix
 
     private:
         int _flush;
-        size_t _compressBufferSize;
-        std::unique_ptr<unsigned char[]> _compressBuffer;
+        std::array<unsigned char, 1 << 14> _compressBuffer;
 
 #ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _inflateState;

ixwebsocket/IXWebSocketProxyServer.cpp

@@ -56,10 +56,9 @@ namespace ix
 
         server.setOnConnectionCallback(
             [remoteUrl, remoteUrlsMapping](std::weak_ptr<ix::WebSocket> webSocket,
-                                           std::shared_ptr<ConnectionState> connectionState,
-                                           std::unique_ptr<ConnectionInfo> connectionInfo) {
+                                           std::shared_ptr<ConnectionState> connectionState) {
                 auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
-                auto remoteIp = connectionInfo->remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
 
                 // Server connection
                 state->webSocket().setOnMessageCallback(

ixwebsocket/IXWebSocketServer.cpp

@@ -77,15 +77,14 @@ namespace ix
     }
 
     void WebSocketServer::handleConnection(std::unique_ptr<Socket> socket,
-                                           std::shared_ptr<ConnectionState> connectionState,
-                                           std::unique_ptr<ConnectionInfo> connectionInfo)
+                                           std::shared_ptr<ConnectionState> connectionState)
     {
         setThreadName("WebSocketServer::" + connectionState->getId());
 
         auto webSocket = std::make_shared<WebSocket>();
         if (_onConnectionCallback)
         {
-            _onConnectionCallback(webSocket, connectionState, std::move(connectionInfo));
+            _onConnectionCallback(webSocket, connectionState);
 
             if (!webSocket->isOnMessageCallbackRegistered())
             {
@@ -99,9 +98,8 @@ namespace ix
         else if (_onClientMessageCallback)
         {
             webSocket->setOnMessageCallback(
-                [this, &ws = *webSocket.get(), connectionState, &ci = *connectionInfo.get()](
-                    const WebSocketMessagePtr& msg) {
-                    _onClientMessageCallback(connectionState, ci, ws, msg);
+                [this, &ws = *webSocket.get(), connectionState](const WebSocketMessagePtr& msg) {
+                    _onClientMessageCallback(connectionState, ws, msg);
                 });
         }
         else

ixwebsocket/IXWebSocketServer.h

@@ -23,14 +23,10 @@ namespace ix
     {
     public:
         using OnConnectionCallback =
-            std::function<void(std::weak_ptr<WebSocket>,
-                               std::shared_ptr<ConnectionState>,
-                               std::unique_ptr<ConnectionInfo> connectionInfo)>;
+            std::function<void(std::weak_ptr<WebSocket>, std::shared_ptr<ConnectionState>)>;
 
-        using OnClientMessageCallback = std::function<void(std::shared_ptr<ConnectionState>,
-                                                           ConnectionInfo&,
-                                                           WebSocket&,
-                                                           const WebSocketMessagePtr&)>;
+        using OnClientMessageCallback = std::function<void(
+            std::shared_ptr<ConnectionState>, WebSocket&, const WebSocketMessagePtr&)>;
 
         WebSocketServer(int port = SocketServer::kDefaultPort,
                         const std::string& host = SocketServer::kDefaultHost,
@@ -69,8 +65,7 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket> socket,
-                                      std::shared_ptr<ConnectionState> connectionState,
-                                      std::unique_ptr<ConnectionInfo> connectionInfo);
+                                      std::shared_ptr<ConnectionState> connectionState);
         virtual size_t getConnectedClientsCount() final;
     };
 } // namespace ix

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "10.2.5"
+#define IX_WEBSOCKET_VERSION "10.3.2"

test/IXCobraToSentryBotTest.cpp

@@ -95,15 +95,14 @@ TEST_CASE("Cobra_to_sentry_bot", "[cobra_bots]")
 
         sentryServer.setOnConnectionCallback(
             [](HttpRequestPtr request,
-               std::shared_ptr<ConnectionState> /*connectionState*/,
-               std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
+               std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionState->getRemoteIp() << ":" << connectionState->getRemotePort()
+                   << " " << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
 
                 if (request->method == "POST")

test/IXTest.cpp

@@ -85,11 +85,10 @@ namespace ix
     bool startWebSocketEchoServer(ix::WebSocketServer& server)
     {
         server.setOnClientMessageCallback(
-            [&server](std::shared_ptr<ConnectionState> /*connectionState*/,
-                      ConnectionInfo& connectionInfo,
+            [&server](std::shared_ptr<ConnectionState> connectionState,
                       WebSocket& webSocket,
                       const ix::WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";

test/IXWebSocketBroadcastTest.cpp

@@ -191,11 +191,9 @@ namespace
 
         server.setOnClientMessageCallback(
             [&server, &connectionId](std::shared_ptr<ConnectionState> connectionState,
-                                     ConnectionInfo& connectionInfo,
                                      WebSocket& webSocket,
                                      const ix::WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
-
+                auto remoteIp = connectionState->getRemoteIp();
 
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {

test/IXWebSocketChatTest.cpp

@@ -195,10 +195,9 @@ namespace
     {
         server.setOnClientMessageCallback(
             [&server](std::shared_ptr<ConnectionState> connectionState,
-                      ConnectionInfo& connectionInfo,
                       WebSocket& webSocket,
                       const ix::WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";
@@ -286,27 +285,27 @@ TEST_CASE("Websocket_chat", "[websocket_chat]")
         int attempts = 0;
         while (chatA.getReceivedMessagesCount() != 3 || chatB.getReceivedMessagesCount() != 3)
         {
-            REQUIRE(attempts++ < 10);
+            CHECK(attempts++ < 10);
             ix::msleep(1000);
         }
 
         chatA.stop();
         chatB.stop();
 
-        REQUIRE(chatA.getReceivedMessagesCount() == 3);
-        REQUIRE(chatB.getReceivedMessagesCount() == 3);
+        CHECK(chatA.getReceivedMessagesCount() == 3);
+        CHECK(chatB.getReceivedMessagesCount() == 3);
 
-        REQUIRE(chatB.getReceivedMessages()[0] == "from A1");
-        REQUIRE(chatB.getReceivedMessages()[1] == "from A2");
-        REQUIRE(chatB.getReceivedMessages()[2] == "from A3");
+        CHECK(chatB.getReceivedMessages()[0] == "from A1");
+        CHECK(chatB.getReceivedMessages()[1] == "from A2");
+        CHECK(chatB.getReceivedMessages()[2] == "from A3");
 
-        REQUIRE(chatA.getReceivedMessages()[0] == "from B1");
-        REQUIRE(chatA.getReceivedMessages()[1] == "from B2");
-        REQUIRE(chatA.getReceivedMessages()[2].size() == bigMessage.size());
+        CHECK(chatA.getReceivedMessages()[0] == "from B1");
+        CHECK(chatA.getReceivedMessages()[1] == "from B2");
+        CHECK(chatA.getReceivedMessages()[2].size() == bigMessage.size());
 
         // Give us 1000ms for the server to notice that clients went away
         ix::msleep(1000);
-        REQUIRE(server.getClients().size() == 0);
+        CHECK(server.getClients().size() == 0);
 
         ix::reportWebSocketTraffic();
     }

test/IXWebSocketCloseTest.cpp

@@ -171,10 +171,9 @@ namespace
         server.setOnClientMessageCallback(
             [&receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](
                 std::shared_ptr<ConnectionState> connectionState,
-                ConnectionInfo& connectionInfo,
                 WebSocket& /*webSocket*/,
                 const ix::WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New server connection";

test/IXWebSocketServerTest.cpp

@@ -35,11 +35,9 @@ namespace ix
 
         server.setOnClientMessageCallback(
             [&server, &connectionId](std::shared_ptr<ConnectionState> connectionState,
-                                     ConnectionInfo& connectionInfo,
                                      WebSocket& webSocket,
                                      const ix::WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
-
+                auto remoteIp = connectionState->getRemoteIp();
 
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {

test/IXWebSocketSubProtocolTest.cpp

@@ -18,10 +18,9 @@ bool startServer(ix::WebSocketServer& server, std::string& subProtocols)
 {
     server.setOnClientMessageCallback(
         [&server, &subProtocols](std::shared_ptr<ConnectionState> connectionState,
-                                 ConnectionInfo& connectionInfo,
                                  WebSocket& webSocket,
                                  const ix::WebSocketMessagePtr& msg) {
-            auto remoteIp = connectionInfo.remoteIp;
+            auto remoteIp = connectionState->getRemoteIp();
             if (msg->type == ix::WebSocketMessageType::Open)
             {
                 TLogger() << "New connection";

test/compatibility/ruby/README.md

@@ -0,0 +1,6 @@
+```
+export GEM_HOME=$HOME/local/gems
+bundle install faye-websocket
+```
+
+https://stackoverflow.com/questions/486995/ruby-equivalent-of-virtualenv

test/compatibility/ruby/devnull_client.rb

@@ -0,0 +1,59 @@
+#
+# $ ruby --version
+# ruby 2.6.3p62 (2019-04-16 revision 67580) [universal.x86_64-darwin19]
+# 
+# Install a gem locally by setting GEM_HOME
+# https://stackoverflow.com/questions/486995/ruby-equivalent-of-virtualenv
+# export GEM_HOME=$HOME/local/gems
+# bundle install faye-websocket
+#
+# In a different terminal, start a push server:
+# $ ws push_server -q
+#
+# $ ruby devnull_client.rb
+# [:open]
+# Connected to server
+# messages received per second: 115926
+# messages received per second: 119156
+# messages received per second: 119156
+# messages received per second: 119157
+# messages received per second: 119156
+# messages received per second: 119156
+# messages received per second: 119157
+# messages received per second: 119156
+# messages received per second: 119156
+# messages received per second: 119157
+# messages received per second: 119156
+# messages received per second: 119157
+# messages received per second: 119156
+# ^C[:close, 1006, ""]
+# 
+require 'faye/websocket'
+require 'eventmachine'
+
+EM.run {
+  ws = Faye::WebSocket::Client.new('ws://127.0.0.1:8008')
+
+  counter = 0
+
+  EM.add_periodic_timer(1) do
+    print "messages received per second: #{counter}\n"
+    counter = 0  # reset counter
+  end
+
+  ws.on :open do |event|
+    p [:open]
+    print "Connected to server\n"
+  end
+
+  ws.on :message do |event|
+    # Uncomment the next line to validate that we receive something correct
+    # p [:message, event.data]
+    counter += 1
+  end
+
+  ws.on :close do |event|
+    p [:close, event.code, event.reason]
+    ws = nil
+  end
+}

ws/ws.cpp

@@ -18,6 +18,7 @@
 #include <fstream>
 #include <iostream>
 #include <ixbots/IXCobraMetricsToRedisBot.h>
+#include <ixbots/IXCobraToCobraBot.h>
 #include <ixbots/IXCobraToPythonBot.h>
 #include <ixbots/IXCobraToSentryBot.h>
 #include <ixbots/IXCobraToStatsdBot.h>
@@ -120,6 +121,12 @@ namespace
             return str.substr(0, n) + "...";
         }
     }
+
+    bool fileExists(const std::string& fileName)
+    {
+        std::ifstream infile(fileName);
+        return infile.good();
+    }
 } // namespace
 
 namespace ix
@@ -406,10 +413,9 @@ namespace ix
 
         server.setOnClientMessageCallback(
             [&server](std::shared_ptr<ConnectionState> connectionState,
-                      ConnectionInfo& connectionInfo,
                       WebSocket& webSocket,
                       const WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("New connection");
@@ -1234,10 +1240,9 @@ namespace ix
 
         server.setOnClientMessageCallback(
             [greetings](std::shared_ptr<ConnectionState> connectionState,
-                        ConnectionInfo& connectionInfo,
                         WebSocket& webSocket,
                         const WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("New connection");
@@ -1668,10 +1673,9 @@ namespace ix
 
         server.setOnClientMessageCallback(
             [greetings, &sendMsg](std::shared_ptr<ConnectionState> connectionState,
-                                  ConnectionInfo& connectionInfo,
                                   WebSocket& webSocket,
                                   const WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("New connection");
@@ -1692,7 +1696,13 @@ namespace ix
                     bool binary = false;
                     while (true)
                     {
-                        webSocket.send(sendMsg, binary);
+                        auto sendInfo = webSocket.send(sendMsg, binary);
+                        if (!sendInfo.success)
+                        {
+                            spdlog::info("Error sending message, closing connection");
+                            webSocket.close();
+                            break;
+                        }
                     }
                 }
                 else if (msg->type == ix::WebSocketMessageType::Close)
@@ -2601,10 +2611,9 @@ namespace ix
 
         server.setOnClientMessageCallback(
             [&server](std::shared_ptr<ConnectionState> connectionState,
-                      ConnectionInfo& connectionInfo,
                       WebSocket& webSocket,
                       const WebSocketMessagePtr& msg) {
-                auto remoteIp = connectionInfo.remoteIp;
+                auto remoteIp = connectionState->getRemoteIp();
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("ws_transfer: New connection");
@@ -2716,6 +2725,7 @@ namespace ix
 
 int main(int argc, char** argv)
 {
+    ix::setThreadName("ws main thread");
     ix::initNetSystem();
 
     ix::CoreLogger::LogFunc logFunc = [](const char* msg, ix::LogLevel level) {
@@ -2802,8 +2812,10 @@ int main(int argc, char** argv)
     std::string project;
     std::string key;
     std::string logfile;
-    std::string scriptPath;
+    std::string moduleName;
     std::string republishChannel;
+    std::string publisherRolename;
+    std::string publisherRolesecret;
     std::string sendMsg("hello world");
     ix::SocketTLSOptions tlsOptions;
     ix::CobraConfig cobraConfig;
@@ -3068,13 +3080,23 @@ int main(int argc, char** argv)
     addTLSOptions(cobra2statsd);
     addCobraBotConfig(cobra2statsd);
 
+    CLI::App* cobra2cobra = app.add_subcommand("cobra_to_cobra", "Cobra to Cobra");
+    cobra2cobra->fallthrough();
+    cobra2cobra->add_option("--republish", republishChannel, "Republish channel");
+    cobra2cobra->add_option("--publisher_rolename", publisherRolename, "Publisher Role name")
+        ->required();
+    cobra2cobra->add_option("--publisher_rolesecret", publisherRolesecret, "Publisher Role secret")
+        ->required();
+    cobra2cobra->add_flag("-q", quiet, "Quiet");
+    addTLSOptions(cobra2cobra);
+    addCobraBotConfig(cobra2cobra);
+
     CLI::App* cobra2python = app.add_subcommand("cobra_to_python", "Cobra to python");
     cobra2python->fallthrough();
     cobra2python->add_option("--host", hostname, "Statsd host");
     cobra2python->add_option("--port", statsdPort, "Statsd port");
     cobra2python->add_option("--prefix", prefix, "Statsd prefix");
-    cobra2python->add_option("--script", scriptPath, "Python script path")
-        ->check(CLI::ExistingPath);
+    cobra2python->add_option("--module", moduleName, "Python module");
     cobra2python->add_option("--pidfile", pidfile, "Pid file");
     addTLSOptions(cobra2python);
     addCobraBotConfig(cobra2python);
@@ -3177,11 +3199,27 @@ int main(int argc, char** argv)
 
     if (tlsOptions.isUsingSystemDefaults())
     {
-#ifdef __APPLE__
+#if defined(__APPLE__)
 #if defined(IXWEBSOCKET_USE_MBED_TLS) || defined(IXWEBSOCKET_USE_OPEN_SSL)
         // We could try to load some system certs as well, but this is easy enough
         tlsOptions.caFile = "/etc/ssl/cert.pem";
 #endif
+#elif defined(__linux__)
+#if defined(IXWEBSOCKET_USE_MBED_TLS)
+        std::vector<std::string> caFiles = {
+            "/etc/ssl/certs/ca-bundle.crt",       // CentOS
+            "/etc/ssl/certs/ca-certificates.crt", // Alpine
+        };
+
+        for (auto&& caFile : caFiles)
+        {
+            if (fileExists(caFile))
+            {
+                tlsOptions.caFile = caFile;
+                break;
+            }
+        }
+#endif
 #endif
     }
 
@@ -3361,7 +3399,7 @@ int main(int argc, char** argv)
         }
         else
         {
-            ret = (int) ix::cobra_to_python_bot(cobraBotConfig, statsdClient, scriptPath);
+            ret = (int) ix::cobra_to_python_bot(cobraBotConfig, statsdClient, moduleName);
         }
     }
     else if (app.got_subcommand("cobra_to_sentry"))
@@ -3384,6 +3422,11 @@ int main(int argc, char** argv)
             ret = (int) ix::cobra_metrics_to_redis_bot(cobraBotConfig, redisClient, verbose);
         }
     }
+    else if (app.got_subcommand("cobra_to_cobra"))
+    {
+        ret = (int) ix::cobra_to_cobra_bot(
+            cobraBotConfig, republishChannel, publisherRolename, publisherRolesecret);
+    }
     else if (app.got_subcommand("snake"))
     {
         ret = ix::ws_snake_main(port,