project builds / gross hack to disable compression code path / ws connect -x works but test fails

CMakeLists.txt

@@ -36,8 +36,6 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXNetSystem.cpp
     ixwebsocket/IXSelectInterrupt.cpp
     ixwebsocket/IXSelectInterruptFactory.cpp
-    ixwebsocket/IXSelectInterruptPipe.cpp
-    ixwebsocket/IXSetThreadName.cpp
     ixwebsocket/IXSocket.cpp
     ixwebsocket/IXSocketConnect.cpp
     ixwebsocket/IXSocketFactory.cpp
@@ -60,7 +58,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
@@ -71,7 +68,6 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXProgressCallback.h
     ixwebsocket/IXSelectInterrupt.h
     ixwebsocket/IXSelectInterruptFactory.h
-    ixwebsocket/IXSelectInterruptPipe.h
     ixwebsocket/IXSetThreadName.h
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketConnect.h
@@ -102,6 +98,23 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXWebSocketVersion.h
 )
 
+if (UNIX)
+    # Linux, Mac, iOS, Android
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptPipe.cpp )
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptPipe.h )
+endif()
+
+# Platform specific code
+if (APPLE)
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/apple/IXSetThreadName_apple.cpp)
+elseif (WIN32)
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/windows/IXSetThreadName_windows.cpp)
+elseif (${CMAKE_SYSTEM_NAME} MATCHES "FreeBSD")
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/freebsd/IXSetThreadName_freebsd.cpp)
+else()
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/linux/IXSetThreadName_linux.cpp)
+endif()
+
 option(USE_TLS "Enable TLS support" FALSE)
 
 if (USE_TLS)

docs/CHANGELOG.md

@@ -1,18 +1,6 @@
 # Changelog
 All changes to this project will be documented in this file.
 
-## [9.9.2] - 2020-07-10
-
-(socket server) bump default max connection count from 32 to 128
-
-## [9.9.1] - 2020-07-10
-
-(snake) implement super simple stream sql expression support in snake server
-
-## [9.9.0] - 2020-07-08
-
-(socket+websocket+http+redis+snake servers) expose the remote ip and remote port when a new connection is made
-
 ## [9.8.6] - 2020-07-06
 
 (cmake) change the way zlib and openssl are searched

docs/usage.md

@@ -257,31 +257,28 @@ ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
     [&server](std::shared_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;
-
         webSocket->setOnMessageCallback(
             [webSocket, connectionState, &server](const ix::WebSocketMessagePtr msg)
             {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
-                    std::cout << "New connection" << std::endl;
+                    std::cerr << "New connection" << std::endl;
 
                     // A connection state object is available, and has a default id
                     // You can subclass ConnectionState and pass an alternate factory
                     // to override it. It is useful if you want to store custom
                     // attributes per connection (authenticated bool flag, attributes, etc...)
-                    std::cout << "id: " << connectionState->getId() << std::endl;
+                    std::cerr << "id: " << connectionState->getId() << std::endl;
 
                     // The uri the client did connect to.
-                    std::cout << "Uri: " << msg->openInfo.uri << std::endl;
+                    std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
 
-                    std::cout << "Headers:" << std::endl;
+                    std::cerr << "Headers:" << std::endl;
                     for (auto it : msg->openInfo.headers)
                     {
-                        std::cout << it.first << ": " << it.second << std::endl;
+                        std::cerr << it.first << ": " << it.second << std::endl;
                     }
                 }
                 else if (msg->type == ix::WebSocketMessageType::Message)
@@ -420,14 +417,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
-           << " "
-           << request->method
+        ss << request->method
            << " "
            << request->uri;
 

ixredis/ixredis/IXRedisServer.cpp

@@ -45,11 +45,8 @@ 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);
-
         _connectedClientsCount++;
 
         while (!_stopHandlingConnections)

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/CMakeLists.txt

@@ -7,14 +7,12 @@ set (IXSNAKE_SOURCES
     ixsnake/IXSnakeServer.cpp
     ixsnake/IXSnakeProtocol.cpp
     ixsnake/IXAppConfig.cpp
-    ixsnake/IXStreamSql.cpp
 )
 
 set (IXSNAKE_HEADERS
     ixsnake/IXSnakeServer.h
     ixsnake/IXSnakeProtocol.h
     ixsnake/IXAppConfig.h
-    ixsnake/IXStreamSql.h
 )
 
 add_library(ixsnake STATIC

ixsnake/ixsnake/IXAppConfig.h

@@ -33,9 +33,6 @@ namespace snake
         // Misc
         bool verbose;
         bool disablePong;
-
-        // If non empty, every published message gets republished to a given channel
-        std::string republishChannel;
     };
 
     bool isAppKeyValid(const AppConfig& appConfig, std::string appkey);

ixsnake/ixsnake/IXSnakeConnectionState.h

@@ -30,7 +30,6 @@ namespace snake
         {
             return _appkey;
         }
-
         void setAppkey(const std::string& appkey)
         {
             _appkey = appkey;
@@ -40,7 +39,6 @@ namespace snake
         {
             return _role;
         }
-
         void setRole(const std::string& role)
         {
             _role = role;

ixsnake/ixsnake/IXSnakeProtocol.cpp

@@ -8,7 +8,6 @@
 
 #include "IXAppConfig.h"
 #include "IXSnakeConnectionState.h"
-#include "IXStreamSql.h"
 #include "nlohmann/json.hpp"
 #include <iostream>
 #include <ixcore/utils/IXCoreLogger.h>
@@ -92,7 +91,6 @@ namespace snake
 
     void handlePublish(std::shared_ptr<SnakeConnectionState> state,
                        std::shared_ptr<ix::WebSocket> ws,
-                       const AppConfig& appConfig,
                        const nlohmann::json& pdu)
     {
         std::vector<std::string> channels;
@@ -117,12 +115,6 @@ namespace snake
             return;
         }
 
-        // add an extra channel if the config has one specified
-        if (!appConfig.republishChannel.empty())
-        {
-            channels.push_back(appConfig.republishChannel);
-        }
-
         for (auto&& channel : channels)
         {
             std::stringstream ss;
@@ -188,25 +180,12 @@ namespace snake
             }
         }
 
-        std::string filterStr;
-        if (pdu["body"].find("filter") != pdu["body"].end())
-        {
-            std::string filterStr = pdu["body"]["filter"];
-        }
-
-        std::unique_ptr<StreamSql> streamSql = std::make_unique<StreamSql>(filterStr);
-
         int id = 0;
-        auto callback = [ws, &id, &subscriptionId, &streamSql](const std::string& messageStr) {
+        auto callback = [ws, &id, &subscriptionId](const std::string& messageStr) {
             auto msg = nlohmann::json::parse(messageStr);
 
             msg = msg["body"]["message"];
 
-            if (streamSql->valid() && !streamSql->match(msg))
-            {
-                return;
-            }
-
             nlohmann::json response = {
                 {"action", "rtm/subscription/data"},
                 {"id", id++},
@@ -300,7 +279,7 @@ namespace snake
         }
         else if (action == "rtm/publish")
         {
-            handlePublish(state, ws, appConfig, pdu);
+            handlePublish(state, ws, pdu);
         }
         else if (action == "rtm/subscribe")
         {

ixsnake/ixsnake/IXSnakeServer.cpp

@@ -61,19 +61,16 @@ namespace snake
 
         _server.setOnConnectionCallback(
             [this](std::shared_ptr<ix::WebSocket> webSocket,
-                   std::shared_ptr<ix::ConnectionState> connectionState,
-                   std::unique_ptr<ix::ConnectionInfo> connectionInfo) {
+                   std::shared_ptr<ix::ConnectionState> connectionState) {
                 auto state = std::dynamic_pointer_cast<SnakeConnectionState>(connectionState);
-                auto remoteIp = connectionInfo->remoteIp;
-            
+
                 webSocket->setOnMessageCallback(
-                    [this, webSocket, state, remoteIp](const ix::WebSocketMessagePtr& msg) {
+                    [this, webSocket, state](const ix::WebSocketMessagePtr& msg) {
                         std::stringstream ss;
                         ix::LogLevel logLevel = ix::LogLevel::Debug;
                         if (msg->type == ix::WebSocketMessageType::Open)
                         {
                             ss << "New connection" << std::endl;
-                            ss << "remote ip: " << remoteIp << std::endl;
                             ss << "id: " << state->getId() << std::endl;
                             ss << "Uri: " << msg->openInfo.uri << std::endl;
                             ss << "Headers:" << std::endl;

ixsnake/ixsnake/IXStreamSql.cpp

@@ -1,63 +0,0 @@
-/*
- *  IXStreamSql.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- *
- *  Super simple hacked up version of a stream sql expression,
- *  that only supports non nested field evaluation
- */
-
-#include "IXStreamSql.h"
-#include <sstream>
-#include <iostream>
-
-namespace snake
-{
-    StreamSql::StreamSql(const std::string& sqlFilter)
-        : _valid(false)
-    {
-        std::string token;
-        std::stringstream tokenStream(sqlFilter);
-        std::vector<std::string> tokens;
-
-        // Split by ' '
-        while (std::getline(tokenStream, token, ' '))
-        {
-            tokens.push_back(token);
-        }
-
-        _valid = tokens.size() == 8;
-        if (!_valid) return;
-
-        _field = tokens[5];
-        _operator = tokens[6];
-        _value = tokens[7];
-
-        // remove single quotes
-        _value = _value.substr(1, _value.size() - 2);
-
-        if (_operator == "LIKE")
-        {
-            _value = _value.substr(1, _value.size() - 2);
-        }
-    }
-
-    bool StreamSql::valid() const
-    {
-        return _valid;
-    }
-
-    bool StreamSql::match(const nlohmann::json& msg)
-    {
-        if (!_valid) return false;
-
-        if (msg.find(_field) == msg.end())
-        {
-            return false;
-        }
-
-        std::string value = msg[_field];
-        return value == _value;
-    }
-
-} // namespace snake

ixsnake/ixsnake/IXStreamSql.h

@@ -1,29 +0,0 @@
-/*
- *  IXStreamSql.h
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-
-#pragma once
-
-#include <string>
-#include "nlohmann/json.hpp"
-
-namespace snake
-{
-    class StreamSql
-    {
-    public:
-        StreamSql(const std::string& sqlFilter = std::string());
-        ~StreamSql() = default;
-
-        bool match(const nlohmann::json& msg);
-        bool valid() const;
-
-    private:
-        std::string _field;
-        std::string _operator;
-        std::string _value;
-        bool _valid;
-    };
-}

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/IXHttpClient.h

@@ -103,9 +103,9 @@ namespace ix
         std::thread _thread;
 
         std::unique_ptr<Socket> _socket;
-        std::recursive_mutex _mutex; // to protect accessing the _socket (only one socket per
-                                     // client) the mutex needs to be recursive as this function
-                                     // might be called recursively to follow HTTP redirections
+        std::recursive_mutex _mutex; // to protect accessing the _socket (only one socket per client)
+                                     // the mutex needs to be recursive as this function might
+                                     // be called recursively to follow HTTP redirections
 
         SocketTLSOptions _tlsOptions;
 

ixwebsocket/IXHttpServer.cpp

@@ -9,11 +9,11 @@
 #include "IXNetSystem.h"
 #include "IXSocketConnect.h"
 #include "IXUserAgent.h"
-#include <cstring>
 #include <fstream>
 #include <sstream>
 #include <vector>
 #include <zlib.h>
+#include <cstring>
 
 namespace
 {
@@ -50,11 +50,8 @@ namespace
         const int windowBits = 15;
         const int GZIP_ENCODING = 16;
 
-        deflateInit2(&zs,
-                     Z_DEFAULT_COMPRESSION,
-                     Z_DEFLATED,
-                     windowBits | GZIP_ENCODING,
-                     8,
+        deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
+                     windowBits | GZIP_ENCODING, 8,
                      Z_DEFAULT_STRATEGY);
 
         zs.next_in = (Bytef*) str.data();
@@ -72,12 +69,13 @@ namespace
 
             ret = deflate(&zs, Z_FINISH);
 
-            if (outstring.size() < zs.total_out)
+            if(outstring.size() < zs.total_out)
             {
                 // append the block to the output string
-                outstring.append(outbuffer, zs.total_out - outstring.size());
+                outstring.append(outbuffer,
+                                 zs.total_out - outstring.size());
             }
-        } while (ret == Z_OK);
+        } while(ret == Z_OK);
 
         deflateEnd(&zs);
 
@@ -115,8 +113,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++;
 
@@ -125,9 +122,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");
@@ -147,8 +142,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 == "/")
                 {
@@ -178,8 +172,7 @@ namespace ix
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
+                ss << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri << " " << content.size();
                 logInfo(ss.str());
 
@@ -205,15 +198,13 @@ namespace ix
         setOnConnectionCallback(
             [this,
              redirectUrl](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 << 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/IXSelectInterruptPipe.cpp

@@ -5,10 +5,8 @@
  */
 
 //
-// On UNIX we use pipes to wake up select. There is no way to do that
-// on Windows so this file is compiled out on Windows.
+// On macOS we use UNIX pipes to wake up select.
 //
-#ifndef _WIN32
 
 #include "IXSelectInterruptPipe.h"
 
@@ -146,5 +144,3 @@ namespace ix
         return _fildes[kPipeReadIndex];
     }
 } // namespace ix
-
-#endif // !_WIN32

ixwebsocket/IXSetThreadName.cpp

@@ -1,81 +0,0 @@
-/*
- *  IXSetThreadName.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 2020 Machine Zone, Inc. All rights reserved.
- */
-#include "IXSetThreadName.h"
-
-// unix systems
-#if defined(__APPLE__) || defined(__linux__) || defined(BSD)
-# include <pthread.h>
-#endif
-
-// freebsd needs this header as well
-#if defined(BSD)
-# include <pthread_np.h>
-#endif
-
-// Windows
-#ifdef _WIN32
-# include <Windows.h>
-#endif
-
-namespace ix
-{
-#ifdef _WIN32
-    const DWORD MS_VC_EXCEPTION = 0x406D1388;
-
-#pragma pack(push, 8)
-    typedef struct tagTHREADNAME_INFO
-    {
-        DWORD dwType;     // Must be 0x1000.
-        LPCSTR szName;    // Pointer to name (in user addr space).
-        DWORD dwThreadID; // Thread ID (-1=caller thread).
-        DWORD dwFlags;    // Reserved for future use, must be zero.
-    } THREADNAME_INFO;
-#pragma pack(pop)
-
-    void SetThreadName(DWORD dwThreadID, const char* threadName)
-    {
-        THREADNAME_INFO info;
-        info.dwType = 0x1000;
-        info.szName = threadName;
-        info.dwThreadID = dwThreadID;
-        info.dwFlags = 0;
-
-        __try
-        {
-            RaiseException(
-                MS_VC_EXCEPTION, 0, sizeof(info) / sizeof(ULONG_PTR), (ULONG_PTR*) &info);
-        }
-        __except (EXCEPTION_EXECUTE_HANDLER)
-        {
-        }
-    }
-#endif
-
-    void setThreadName(const std::string& name)
-    {
-#if defined(__APPLE__)
-        //
-        // Apple reserves 16 bytes for its thread names
-        // Notice that the Apple version of pthread_setname_np
-        // does not take a pthread_t argument
-        //
-        pthread_setname_np(name.substr(0, 63).c_str());
-#elif defined(__linux__)
-        //
-        // Linux only reserves 16 bytes for its thread names
-        // See prctl and PR_SET_NAME property in
-        // http://man7.org/linux/man-pages/man2/prctl.2.html
-        //
-        pthread_setname_np(pthread_self(), name.substr(0, 15).c_str());
-#elif defined(_WIN32)
-        SetThreadName(-1, name.c_str());
-#elif defined(BSD)
-        pthread_set_name_np(pthread_self(), name.substr(0, 15).c_str());
-#else
-        // ... assert here ?
-#endif
-    }
-} // namespace ix

ixwebsocket/IXSocketServer.cpp

@@ -22,7 +22,7 @@ namespace ix
     const int SocketServer::kDefaultPort(8080);
     const std::string SocketServer::kDefaultHost("127.0.0.1");
     const int SocketServer::kDefaultTcpBacklog(5);
-    const size_t SocketServer::kDefaultMaxConnections(128);
+    const size_t SocketServer::kDefaultMaxConnections(32);
     const int SocketServer::kDefaultAddressFamily(AF_INET);
 
     SocketServer::SocketServer(
@@ -276,7 +276,6 @@ namespace ix
             }
 
             // Accept a connection.
-            // FIXME: Is this working for ipv6 ?
             struct sockaddr_in client; // client address information
             int clientFd;              // socket connected to client
             socklen_t addressLen = sizeof(client);
@@ -308,45 +307,6 @@ namespace ix
                 continue;
             }
 
-            std::unique_ptr<ConnectionInfo> connectionInfo;
-
-            if (_addressFamily == AF_INET)
-            {
-                char remoteIp[INET_ADDRSTRLEN];
-                if (inet_ntop(AF_INET, &client.sin_addr, remoteIp, INET_ADDRSTRLEN) == nullptr)
-                {
-                    int err = Socket::getErrno();
-                    std::stringstream ss;
-                    ss << "SocketServer::run() error calling inet_ntop (ipv4): " << err << ", "
-                       << strerror(err);
-                    logError(ss.str());
-
-                    Socket::closeSocket(clientFd);
-
-                    continue;
-                }
-
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
-            }
-            else // AF_INET6
-            {
-                char remoteIp[INET6_ADDRSTRLEN];
-                if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp, INET6_ADDRSTRLEN) == nullptr)
-                {
-                    int err = Socket::getErrno();
-                    std::stringstream ss;
-                    ss << "SocketServer::run() error calling inet_ntop (ipv6): " << err << ", "
-                       << strerror(err);
-                    logError(ss.str());
-
-                    Socket::closeSocket(clientFd);
-
-                    continue;
-                }
-
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
-            }
-
             std::shared_ptr<ConnectionState> connectionState;
             if (_connectionStateFactory)
             {
@@ -382,7 +342,7 @@ namespace ix
             _connectionsThreads.push_back(std::make_pair(
                 connectionState,
                 std::thread(
-                    &SocketServer::handleConnection, this, std::move(socket), connectionState, std::move(connectionInfo))));
+                    &SocketServer::handleConnection, this, std::move(socket), connectionState)));
         }
     }
 

ixwebsocket/IXSocketServer.h

@@ -6,7 +6,6 @@
 
 #pragma once
 
-#include "IXConnectionInfo.h"
 #include "IXConnectionState.h"
 #include "IXSocketTLSOptions.h"
 #include <atomic>
@@ -103,8 +102,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/IXUrlParser.cpp

@@ -32,8 +32,8 @@
 #include "IXUrlParser.h"
 
 #include <algorithm>
-#include <cstdlib>
 #include <cstring>
+#include <cstdlib>
 
 namespace
 {

ixwebsocket/IXWebSocket.cpp

@@ -429,6 +429,16 @@ namespace ix
         return (binary) ? sendBinary(data, onProgressCallback) : sendText(data, onProgressCallback);
     }
 
+    WebSocketSendInfo WebSocket::sendBinary(const std::vector<uint8_t>& data,
+                                            const OnProgressCallback& onProgressCallback)
+    {
+        if (!isConnected()) return WebSocketSendInfo(false);
+        std::lock_guard<std::mutex> lock(_writeMutex);
+        auto webSocketSendInfo = _ws.sendBinary(data, onProgressCallback);
+        WebSocket::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize, false);
+        return webSocketSendInfo;
+    }
+
     WebSocketSendInfo WebSocket::sendBinary(const std::string& text,
                                             const OnProgressCallback& onProgressCallback)
     {

ixwebsocket/IXWebSocket.h

@@ -74,8 +74,11 @@ namespace ix
         WebSocketSendInfo send(const std::string& data,
                                bool binary = false,
                                const OnProgressCallback& onProgressCallback = nullptr);
+
         WebSocketSendInfo sendBinary(const std::string& text,
                                      const OnProgressCallback& onProgressCallback = nullptr);
+        WebSocketSendInfo sendBinary(const std::vector<uint8_t>& data,
+                                     const OnProgressCallback& onProgressCallback = nullptr);
         WebSocketSendInfo sendText(const std::string& text,
                                    const OnProgressCallback& onProgressCallback = nullptr);
         WebSocketSendInfo ping(const std::string& text);

ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp

@@ -75,26 +75,22 @@ namespace ix
         return compressData(in, out);
     }
 
-    bool WebSocketPerMessageDeflateCompressor::compress(const std::string& in,
-                                                        std::vector<uint8_t>& out)
+    bool WebSocketPerMessageDeflateCompressor::compress(const std::string& in, std::vector<uint8_t>& out)
     {
         return compressData(in, out);
     }
 
-    bool WebSocketPerMessageDeflateCompressor::compress(const std::vector<uint8_t>& in,
-                                                        std::string& out)
+    bool WebSocketPerMessageDeflateCompressor::compress(const std::vector<uint8_t>& in, std::string& out)
     {
         return compressData(in, out);
     }
 
-    bool WebSocketPerMessageDeflateCompressor::compress(const std::vector<uint8_t>& in,
-                                                        std::vector<uint8_t>& out)
+    bool WebSocketPerMessageDeflateCompressor::compress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out)
     {
         return compressData(in, out);
     }
 
-    template<typename T, typename S>
-    bool WebSocketPerMessageDeflateCompressor::compressData(const T& in, S& out)
+    template<typename T, typename S> bool WebSocketPerMessageDeflateCompressor::compressData(const T& in, S& out)
     {
         //
         // 7.2.1.  Compression

ixwebsocket/IXWebSocketPerMessageDeflateCodec.h

@@ -26,10 +26,8 @@ namespace ix
         bool compress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out);
 
     private:
-        template<typename T, typename S>
-        bool compressData(const T& in, S& out);
-        template<typename T>
-        bool endsWithEmptyUnCompressedBlock(const T& value);
+        template<typename T, typename S> bool compressData(const T& in, S& out);
+        template<typename T> bool endsWithEmptyUnCompressedBlock(const T& value);
 
         int _flush;
         size_t _compressBufferSize;

ixwebsocket/IXWebSocketServer.cpp

@@ -72,13 +72,12 @@ 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>();
-        _onConnectionCallback(webSocket, connectionState, std::move(connectionInfo));
+        _onConnectionCallback(webSocket, connectionState);
 
         webSocket->disableAutomaticReconnection();
 

ixwebsocket/IXWebSocketServer.h

@@ -23,8 +23,7 @@ namespace ix
     {
     public:
         using OnConnectionCallback =
-            std::function<void(std::shared_ptr<WebSocket>, std::shared_ptr<ConnectionState>,
-                               std::unique_ptr<ConnectionInfo> connectionInfo)>;
+            std::function<void(std::shared_ptr<WebSocket>, std::shared_ptr<ConnectionState>)>;
 
         WebSocketServer(int port = SocketServer::kDefaultPort,
                         const std::string& host = SocketServer::kDefaultHost,
@@ -61,8 +60,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) final;
         virtual size_t getConnectedClientsCount() final;
     };
 } // namespace ix

ixwebsocket/IXWebSocketTransport.cpp

@@ -751,11 +751,28 @@ namespace ix
         return static_cast<unsigned>(seconds);
     }
 
-    template<class T>
     WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
-                                                   const T& message,
-                                                   bool compress,
-                                                   const OnProgressCallback& onProgressCallback)
+                               const std::string& message,
+                               bool compress,
+                               const OnProgressCallback& onProgressCallback)
+    {
+        return sendRawData(type, message, compress, onProgressCallback);
+    }
+
+    WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
+                               const std::vector<uint8_t>& message,
+                               bool compress,
+                               const OnProgressCallback& onProgressCallback)
+    {
+        return sendRawData(type, message, compress, onProgressCallback);
+    }
+
+
+    template<class T>
+    WebSocketSendInfo WebSocketTransport::sendRawData(wsheader_type::opcode_type type,
+                                                      const T& message,
+                                                      bool compress,
+                                                      const OnProgressCallback& onProgressCallback)
     {
         if (_readyState != ReadyState::OPEN && _readyState != ReadyState::CLOSING)
         {
@@ -766,12 +783,14 @@ namespace ix
         size_t wireSize = message.size();
         bool compressionError = false;
 
-        auto message_begin = message.cbegin();
-        auto message_end = message.cend();
+        auto message_begin = message.begin();
+        auto message_end = message.end();
 
+#if 0
         if (compress)
         {
-            if (!_perMessageDeflate->compress(message, _compressedMessage))
+            T compressedMessage;
+            if (!_perMessageDeflate->compress(message, compressedMessage))
             {
                 bool success = false;
                 compressionError = true;
@@ -780,11 +799,12 @@ namespace ix
                 return WebSocketSendInfo(success, compressionError, payloadSize, wireSize);
             }
             compressionError = false;
-            wireSize = _compressedMessage.size();
+            wireSize = compressedMessage.size();
 
-            message_begin = _compressedMessage.cbegin();
-            message_end = _compressedMessage.cend();
+            message_begin = compressedMessage.begin();
+            message_end = compressedMessage.end();
         }
+#endif
 
         {
             std::lock_guard<std::mutex> lock(_txbufMutex);
@@ -810,8 +830,8 @@ namespace ix
             //
             auto steps = wireSize / kChunkSize;
 
-            std::string::const_iterator begin = message_begin;
-            std::string::const_iterator end = message_end;
+            auto begin = message_begin;
+            auto end = message_end;
 
             for (uint64_t i = 0; i < steps; ++i)
             {
@@ -964,6 +984,14 @@ namespace ix
         return info;
     }
 
+    WebSocketSendInfo WebSocketTransport::sendBinary(const std::vector<uint8_t>& message,
+                                                     const OnProgressCallback& onProgressCallback)
+
+    {
+        return sendData(
+            wsheader_type::BINARY_FRAME, message, _enablePerMessageDeflate, onProgressCallback);
+    }
+
     WebSocketSendInfo WebSocketTransport::sendBinary(const std::string& message,
                                                      const OnProgressCallback& onProgressCallback)
 

ixwebsocket/IXWebSocketTransport.h

@@ -86,6 +86,9 @@ namespace ix
         WebSocketInitResult connectToSocket(std::unique_ptr<Socket> socket, int timeoutSecs);
 
         PollResult poll();
+
+        WebSocketSendInfo sendBinary(const std::vector<uint8_t>& message,
+                                     const OnProgressCallback& onProgressCallback);
         WebSocketSendInfo sendBinary(const std::string& message,
                                      const OnProgressCallback& onProgressCallback);
         WebSocketSendInfo sendText(const std::string& message,
@@ -190,7 +193,7 @@ namespace ix
         std::atomic<bool> _enablePerMessageDeflate;
 
         std::string _decompressedMessage;
-        std::string _compressedMessage;
+        std::vector<uint8_t> _compressedMessage;
 
         // Used to control TLS connection behavior
         SocketTLSOptions _socketTLSOptions;
@@ -239,15 +242,28 @@ namespace ix
         bool sendOnSocket();
         bool receiveFromSocket();
 
-        template<class T>
         WebSocketSendInfo sendData(wsheader_type::opcode_type type,
-                                   const T& message,
+                                   const std::string& message,
                                    bool compress,
                                    const OnProgressCallback& onProgressCallback = nullptr);
 
+        WebSocketSendInfo sendData(wsheader_type::opcode_type type,
+                                   const std::vector<uint8_t>& message,
+                                   bool compress,
+                                   const OnProgressCallback& onProgressCallback = nullptr);
+
+        template<class T>
+        WebSocketSendInfo sendRawData(wsheader_type::opcode_type type,
+                                      const T& message,
+                                      bool compress,
+                                      const OnProgressCallback& onProgressCallback = nullptr);
+
         template<class Iterator>
-        bool sendFragment(
-            wsheader_type::opcode_type type, bool fin, Iterator begin, Iterator end, bool compress);
+        bool sendFragment(wsheader_type::opcode_type type,
+                          bool fin,
+                          Iterator begin,
+                          Iterator end,
+                          bool compress);
 
         void emitMessage(MessageKind messageKind,
                          const std::string& message,

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "9.9.2"
+#define IX_WEBSOCKET_VERSION "9.8.6"

ixwebsocket/apple/IXSetThreadName_apple.cpp

@@ -0,0 +1,20 @@
+/*
+ *  IXSetThreadName_apple.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+#include "../IXSetThreadName.h"
+#include <pthread.h>
+
+namespace ix
+{
+    void setThreadName(const std::string& name)
+    {
+        //
+        // Apple reserves 16 bytes for its thread names
+        // Notice that the Apple version of pthread_setname_np
+        // does not take a pthread_t argument
+        //
+        pthread_setname_np(name.substr(0, 63).c_str());
+    }
+} // namespace ix

ixwebsocket/freebsd/IXSetThreadName_freebsd.cpp

@@ -0,0 +1,16 @@
+/*
+ *  IXSetThreadName_freebsd.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+#include "../IXSetThreadName.h"
+#include <pthread.h>
+#include <pthread_np.h>
+
+namespace ix
+{
+    void setThreadName(const std::string& name)
+    {
+        pthread_set_name_np(pthread_self(), name.substr(0, 15).c_str());
+    }
+} // namespace ix

ixwebsocket/linux/IXSetThreadName_linux.cpp

@@ -0,0 +1,20 @@
+/*
+ *  IXSetThreadName_linux.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+#include "../IXSetThreadName.h"
+#include <pthread.h>
+
+namespace ix
+{
+    void setThreadName(const std::string& name)
+    {
+        //
+        // Linux only reserves 16 bytes for its thread names
+        // See prctl and PR_SET_NAME property in
+        // http://man7.org/linux/man-pages/man2/prctl.2.html
+        //
+        pthread_setname_np(pthread_self(), name.substr(0, 15).c_str());
+    }
+} // namespace ix

ixwebsocket/windows/IXSetThreadName_windows.cpp

@@ -0,0 +1,46 @@
+/*
+ *  IXSetThreadName_windows.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+#include "../IXSetThreadName.h"
+
+#include <Windows.h>
+
+namespace ix
+{
+    const DWORD MS_VC_EXCEPTION = 0x406D1388;
+
+#pragma pack(push, 8)
+    typedef struct tagTHREADNAME_INFO
+    {
+        DWORD dwType;     // Must be 0x1000.
+        LPCSTR szName;    // Pointer to name (in user addr space).
+        DWORD dwThreadID; // Thread ID (-1=caller thread).
+        DWORD dwFlags;    // Reserved for future use, must be zero.
+    } THREADNAME_INFO;
+#pragma pack(pop)
+
+    void SetThreadName(DWORD dwThreadID, const char* threadName)
+    {
+        THREADNAME_INFO info;
+        info.dwType = 0x1000;
+        info.szName = threadName;
+        info.dwThreadID = dwThreadID;
+        info.dwFlags = 0;
+
+        __try
+        {
+            RaiseException(
+                MS_VC_EXCEPTION, 0, sizeof(info) / sizeof(ULONG_PTR), (ULONG_PTR*) &info);
+        }
+        __except (EXCEPTION_EXECUTE_HANDLER)
+        {
+        }
+    }
+
+    void setThreadName(const std::string& name)
+    {
+        SetThreadName(-1, name.c_str());
+    }
+} // namespace ix

makefile

@@ -104,10 +104,8 @@ test_server:
 # env TEST=Websocket_server make test
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
-build_test:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. ; ninja install)
-
-test: build_test
+test:
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; ninja install)
 	(cd test ; python2.7 run.py -r)
 
 test_make:

test/CMakeLists.txt

@@ -42,7 +42,6 @@ set (SOURCES
 
   IXSocketTest.cpp
   IXSocketConnectTest.cpp
-  # IXWebSocketLeakTest.cpp # commented until we have a fix for #224
   IXWebSocketServerTest.cpp
   IXWebSocketTestConnectionDisconnection.cpp
   IXUrlParserTest.cpp
@@ -56,8 +55,6 @@ set (SOURCES
   IXSentryClientTest.cpp
   IXWebSocketChatTest.cpp
   IXWebSocketBroadcastTest.cpp
-  IXWebSocketPerMessageDeflateCompressorTest.cpp
-  IXStreamSqlTest.cpp
 )
 
 # Some unittest don't work on windows yet

test/IXCobraToSentryBotTest.cpp

@@ -12,8 +12,8 @@
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcobra/IXCobraMetricsPublisher.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixredis/IXRedisServer.h>
 #include <ixsentry/IXSentryClient.h>
+#include <ixredis/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>
@@ -95,15 +95,13 @@ 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 << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
 
                 if (request->method == "POST")

test/IXCobraToStatsdBotTest.cpp

@@ -12,8 +12,8 @@
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcobra/IXCobraMetricsPublisher.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixredis/IXRedisServer.h>
 #include <ixsentry/IXSentryClient.h>
+#include <ixredis/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>

test/IXCobraToStdoutBotTest.cpp

@@ -12,8 +12,8 @@
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcobra/IXCobraMetricsPublisher.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixredis/IXRedisServer.h>
 #include <ixsentry/IXSentryClient.h>
+#include <ixredis/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>
@@ -92,9 +92,6 @@ TEST_CASE("Cobra_to_stdout_bot", "[cobra_bots]")
         cobraBotConfig.enableHeartbeat = false;
         bool quiet = false;
 
-        cobraBotConfig.filter =
-            std::string("select * from `") + channel + "` where id = 'sms_metric_A_id'";
-
         // We could try to capture the output ... not sure how.
         bool fluentd = true;
 

test/IXHttpServerTest.cpp

@@ -67,8 +67,7 @@ TEST_CASE("http server", "[httpd]")
 
 TEST_CASE("http server redirection", "[httpd_redirect]")
 {
-    SECTION(
-        "Connect to a local HTTP server, with redirection enabled, but we do not follow redirects")
+    SECTION("Connect to a local HTTP server, with redirection enabled, but we do not follow redirects")
     {
         int port = getFreePort();
         ix::HttpServer server(port, "127.0.0.1");

test/IXStreamSqlTest.cpp

@@ -1,42 +0,0 @@
-/*
- *  IXStreamSqlTest.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone. All rights reserved.
- */
-
-#include "IXTest.h"
-#include "catch.hpp"
-#include <iostream>
-#include <ixsnake/IXStreamSql.h>
-#include <string.h>
-
-using namespace ix;
-
-namespace ix
-{
-    TEST_CASE("stream_sql", "[streamsql]")
-    {
-        SECTION("expression A")
-        {
-            snake::StreamSql streamSql(
-                "select * from subscriber_republished_v1_neo where session LIKE '%123456%'");
-
-            nlohmann::json msg = {{"session", "123456"}, {"id", "foo_id"}, {"timestamp", 12}};
-
-            CHECK(streamSql.match(msg));
-        }
-
-        SECTION("expression A")
-        {
-            snake::StreamSql streamSql("select * from `subscriber_republished_v1_neo` where "
-                                       "session = '30091320ed8d4e50b758f8409b83bed7'");
-
-            nlohmann::json msg = {{"session", "30091320ed8d4e50b758f8409b83bed7"},
-                                  {"id", "foo_id"},
-                                  {"timestamp", 12}};
-
-            CHECK(streamSql.match(msg));
-        }
-    }
-
-} // namespace ix

test/IXTest.cpp

@@ -85,15 +85,12 @@ namespace ix
     bool startWebSocketEchoServer(ix::WebSocketServer& server)
     {
         server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState,
-                                                 std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
+                                                 std::shared_ptr<ConnectionState> connectionState) {
             webSocket->setOnMessageCallback(
-                [webSocket, connectionState, remoteIp, &server](const ix::WebSocketMessagePtr& msg) {
+                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
                     if (msg->type == ix::WebSocketMessageType::Open)
                     {
                         TLogger() << "New connection";
-                        TLogger() << "Remote ip: " << remoteIp;
                         TLogger() << "Uri: " << msg->openInfo.uri;
                         TLogger() << "Headers:";
                         for (auto it : msg->openInfo.headers)

test/IXWebSocketBroadcastTest.cpp

@@ -191,16 +191,13 @@ namespace
 
         server.setOnConnectionCallback([&server, &connectionId](
                                            std::shared_ptr<ix::WebSocket> webSocket,
-                                           std::shared_ptr<ConnectionState> connectionState,
-                                           std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
-            webSocket->setOnMessageCallback([webSocket, connectionState, remoteIp, &connectionId, &server](
+                                           std::shared_ptr<ConnectionState> connectionState) {
+            webSocket->setOnMessageCallback([webSocket, connectionState, &connectionId, &server](
                                                 const ix::WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";
                     connectionState->computeId();
-                    TLogger() << "remote ip: " << remoteIp;
                     TLogger() << "id: " << connectionState->getId();
                     TLogger() << "Uri: " << msg->openInfo.uri;
                     TLogger() << "Headers:";

test/IXWebSocketChatTest.cpp

@@ -100,6 +100,7 @@ namespace
         }
 
         _webSocket.setUrl(url);
+        _webSocket.disablePerMessageDeflate();
 
         std::stringstream ss;
         log(std::string("Connecting to url: ") + url);
@@ -188,22 +189,20 @@ namespace
 
     void WebSocketChat::sendMessage(const std::string& text)
     {
-        _webSocket.sendBinary(encodeMessage(text));
+        auto msg = encodeMessage(text);
+        std::vector<uint8_t> data(text.begin(), text.end());
+        _webSocket.sendBinary(data);
     }
 
     bool startServer(ix::WebSocketServer& server)
     {
         server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState,
-
-                                                 std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
+                                                 std::shared_ptr<ConnectionState> connectionState) {
             webSocket->setOnMessageCallback(
-                [webSocket, connectionState, remoteIp, &server](const ix::WebSocketMessagePtr& msg) {
+                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
                     if (msg->type == ix::WebSocketMessageType::Open)
                     {
                         TLogger() << "New connection";
-                        TLogger() << "remote ip: " << remoteIp;
                         TLogger() << "id: " << connectionState->getId();
                         TLogger() << "Uri: " << msg->openInfo.uri;
                         TLogger() << "Headers:";

test/IXWebSocketCloseTest.cpp

@@ -171,12 +171,9 @@ namespace
         server.setOnConnectionCallback(
             [&receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](
                 std::shared_ptr<ix::WebSocket> webSocket,
-                std::shared_ptr<ConnectionState> connectionState,
-                std::unique_ptr<ConnectionInfo> connectionInfo) {
-                auto remoteIp = connectionInfo->remoteIp;
+                std::shared_ptr<ConnectionState> connectionState) {
                 webSocket->setOnMessageCallback([webSocket,
                                                  connectionState,
-                                                 remoteIp,
                                                  &receivedCloseCode,
                                                  &receivedCloseReason,
                                                  &receivedCloseRemote,
@@ -184,7 +181,6 @@ namespace
                     if (msg->type == ix::WebSocketMessageType::Open)
                     {
                         TLogger() << "New server connection";
-                        TLogger() << "remote ip: " << remoteIp;
                         TLogger() << "id: " << connectionState->getId();
                         TLogger() << "Uri: " << msg->openInfo.uri;
                         TLogger() << "Headers:";

test/IXWebSocketLeakTest.cpp

@@ -1,182 +0,0 @@
-/*
- *  IXWebSocketServer.cpp
- *  Author: Benjamin Sergeant, @marcelkauf
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-
-#include "IXTest.h"
-
-#include "catch.hpp"
-#include <memory>
-#include <sstream>
-
-#include <ixwebsocket/IXWebSocket.h>
-#include <ixwebsocket/IXWebSocketServer.h>
-
-using namespace ix;
-
-namespace
-{
-    class WebSocketClient
-    {
-    public:
-        WebSocketClient(int port);
-        void start();
-        void stop();
-        bool isReady() const;
-        bool hasConnectionError() const;
-
-    private:
-        ix::WebSocket _webSocket;
-        int _port;
-        std::atomic<bool> _connectionError;
-    };
-
-    WebSocketClient::WebSocketClient(int port)
-        : _port(port)
-        , _connectionError(false)
-    {
-    }
-
-    bool WebSocketClient::hasConnectionError() const
-    {
-        return _connectionError;
-    }
-
-    bool WebSocketClient::isReady() const
-    {
-        return _webSocket.getReadyState() == ix::ReadyState::Open;
-    }
-
-    void WebSocketClient::stop()
-    {
-        _webSocket.stop();
-    }
-
-    void WebSocketClient::start()
-    {
-        std::string url;
-        {
-            std::stringstream ss;
-            ss << "ws://localhost:" << _port << "/";
-
-            url = ss.str();
-        }
-
-        _webSocket.setUrl(url);
-        _webSocket.disableAutomaticReconnection();
-
-        std::stringstream ss;
-        log(std::string("Connecting to url: ") + url);
-
-        _webSocket.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg)
-        {
-            std::stringstream ss;
-            if (msg->type == ix::WebSocketMessageType::Open)
-            {
-                log("client connected");
-            }
-            else if (msg->type == ix::WebSocketMessageType::Close)
-            {
-                log("client disconnected");
-            }
-            else if (msg->type == ix::WebSocketMessageType::Error)
-            {
-                _connectionError = true;
-                log("error");
-            }
-            else if (msg->type == ix::WebSocketMessageType::Pong)
-            {
-                log("pong");
-            }
-            else if (msg->type == ix::WebSocketMessageType::Ping)
-            {
-                log("ping");
-            }
-            else if (msg->type == ix::WebSocketMessageType::Message)
-            {
-                log("message");
-            }
-            else
-            {
-                log("invalid type");
-            }
-        });
-
-        _webSocket.start();
-    }
-} // namespace
-
-TEST_CASE("Websocket leak test")
-{
-    SECTION("Websocket destructor is called when closing the connection.")
-    {
-        // stores the server websocket in order to check the use_count
-        std::shared_ptr<WebSocket> webSocketPtr;
-
-        {
-            int port = getFreePort();
-            WebSocketServer server(port);
-
-            server.setOnConnectionCallback([&webSocketPtr](std::shared_ptr<ix::WebSocket> webSocket,
-                                                           std::shared_ptr<ConnectionState> connectionState,
-                                                           std::unique_ptr<ConnectionInfo> connectionInfo)
-           {
-                // original ptr in WebSocketServer::handleConnection and the callback argument
-                REQUIRE(webSocket.use_count() == 2);
-                webSocket->setOnMessageCallback([&webSocketPtr, webSocket, connectionState](const ix::WebSocketMessagePtr& msg)
-                {
-                    if (msg->type == ix::WebSocketMessageType::Open)
-                    {
-                        log(std::string("New connection id: ") + connectionState->getId());
-                        // original ptr in WebSocketServer::handleConnection, captured ptr of this callback, and ptr in WebSocketServer::_clients
-                        REQUIRE(webSocket.use_count() == 3);
-                        webSocketPtr = std::shared_ptr<WebSocket>(webSocket);
-                        REQUIRE(webSocket.use_count() == 4);
-                    }
-                    else if (msg->type == ix::WebSocketMessageType::Close)
-                    {
-                        log(std::string("Client closed connection id: ") + connectionState->getId());
-                    }
-                    else
-                    {
-                        log(std::string(msg->str));
-                    }
-                });
-                // original ptr in WebSocketServer::handleConnection, argument of this callback, and captured ptr in websocket callback
-                REQUIRE(webSocket.use_count() == 3);
-            });
-
-            server.listen();
-            server.start();
-
-            WebSocketClient webSocketClient(port);
-            webSocketClient.start();
-
-            while (true)
-            {
-                REQUIRE(!webSocketClient.hasConnectionError());
-                if (webSocketClient.isReady()) break;
-                ix::msleep(10);
-            }
-
-            REQUIRE(server.getClients().size() == 1);
-            // same value as in Open-handler above
-            REQUIRE(webSocketPtr.use_count() == 4);
-
-            ix::msleep(500);
-            webSocketClient.stop();
-            ix::msleep(500);
-            REQUIRE(server.getClients().size() == 0);
-
-            // websocket should only be referenced by webSocketPtr but is still used by the websocket callback
-            REQUIRE(webSocketPtr.use_count() == 1);
-            webSocketPtr->setOnMessageCallback(nullptr);
-            // websocket should only be referenced by webSocketPtr
-            REQUIRE(webSocketPtr.use_count() == 1);
-            server.stop();
-        }
-        // websocket should only be referenced by webSocketPtr
-        REQUIRE(webSocketPtr.use_count() == 1);
-    }
-}

test/IXWebSocketPerMessageDeflateCompressorTest.cpp

@@ -1,76 +0,0 @@
-/*
- *  IXWebSocketPerMessageDeflateCodecTest.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone. All rights reserved.
- *
- *  make build_test && build/test/ixwebsocket_unittest per-message-deflate-codec
- */
-
-#include "IXTest.h"
-#include "catch.hpp"
-#include <iostream>
-#include <ixwebsocket/IXWebSocketPerMessageDeflateCodec.h>
-#include <string.h>
-
-using namespace ix;
-
-namespace ix
-{
-    std::string compressAndDecompress(const std::string& a)
-    {
-        std::string b, c;
-
-        WebSocketPerMessageDeflateCompressor compressor;
-        compressor.init(11, true);
-        compressor.compress(a, b);
-
-        WebSocketPerMessageDeflateDecompressor decompressor;
-        decompressor.init(11, true);
-        decompressor.decompress(b, c);
-
-        return c;
-    }
-
-    std::string compressAndDecompressVector(const std::string& a)
-    {
-        std::string b, c;
-
-        std::vector<uint8_t> vec(a.begin(), a.end());
-
-        WebSocketPerMessageDeflateCompressor compressor;
-        compressor.init(11, true);
-        compressor.compress(vec, b);
-
-        WebSocketPerMessageDeflateDecompressor decompressor;
-        decompressor.init(11, true);
-        decompressor.decompress(b, c);
-
-        return c;
-    }
-
-    TEST_CASE("per-message-deflate-codec", "[zlib]")
-    {
-        SECTION("string api")
-        {
-            REQUIRE(compressAndDecompress("") == "");
-            REQUIRE(compressAndDecompress("foo") == "foo");
-            REQUIRE(compressAndDecompress("bar") == "bar");
-            REQUIRE(compressAndDecompress("asdcaseqw`21897dehqwed") == "asdcaseqw`21897dehqwed");
-            REQUIRE(compressAndDecompress("/usr/local/include/ixwebsocket/IXSocketAppleSSL.h") ==
-                    "/usr/local/include/ixwebsocket/IXSocketAppleSSL.h");
-        }
-
-        SECTION("vector api")
-        {
-            REQUIRE(compressAndDecompressVector("") == "");
-            REQUIRE(compressAndDecompressVector("foo") == "foo");
-            REQUIRE(compressAndDecompressVector("bar") == "bar");
-            REQUIRE(compressAndDecompressVector("asdcaseqw`21897dehqwed") ==
-                    "asdcaseqw`21897dehqwed");
-            REQUIRE(
-                compressAndDecompressVector("/usr/local/include/ixwebsocket/IXSocketAppleSSL.h") ==
-                "/usr/local/include/ixwebsocket/IXSocketAppleSSL.h");
-        }
-    }
-
-} // namespace ix

test/IXWebSocketServerTest.cpp

@@ -35,16 +35,13 @@ namespace ix
 
         server.setOnConnectionCallback([&server, &connectionId](
                                            std::shared_ptr<ix::WebSocket> webSocket,
-                                           std::shared_ptr<ConnectionState> connectionState,
-                                           std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
-            webSocket->setOnMessageCallback([webSocket, connectionState, remoteIp, &connectionId, &server](
+                                           std::shared_ptr<ConnectionState> connectionState) {
+            webSocket->setOnMessageCallback([webSocket, connectionState, &connectionId, &server](
                                                 const ix::WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";
                     connectionState->computeId();
-                    TLogger() << "remote ip: " << remoteIp;
                     TLogger() << "id: " << connectionState->getId();
                     TLogger() << "Uri: " << msg->openInfo.uri;
                     TLogger() << "Headers:";

test/IXWebSocketSubProtocolTest.cpp

@@ -18,15 +18,12 @@ bool startServer(ix::WebSocketServer& server, std::string& subProtocols)
 {
     server.setOnConnectionCallback(
         [&server, &subProtocols](std::shared_ptr<ix::WebSocket> webSocket,
-                                 std::shared_ptr<ConnectionState> connectionState,
-                                 std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
-            webSocket->setOnMessageCallback([webSocket, connectionState, remoteIp, &server, &subProtocols](
+                                 std::shared_ptr<ConnectionState> connectionState) {
+            webSocket->setOnMessageCallback([webSocket, connectionState, &server, &subProtocols](
                                                 const ix::WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";
-                    TLogger() << "remote ip: " << remoteIp;
                     TLogger() << "id: " << connectionState->getId();
                     TLogger() << "Uri: " << msg->openInfo.uri;
                     TLogger() << "Headers:";

third_party/cpp-linenoise/linenoise.hpp

@@ -122,8 +122,6 @@
 
 #pragma once
 
-#include <string>
-
 namespace linenoise 
 {
     bool Readline(const char *prompt, std::string& line);

ws/ws.cpp

@@ -11,13 +11,13 @@
 
 #include <cli11/CLI11.hpp>
 #include <fstream>
-#include <ixbots/IXCobraMetricsToRedisBot.h>
 #include <ixbots/IXCobraToPythonBot.h>
 #include <ixbots/IXCobraToSentryBot.h>
 #include <ixbots/IXCobraToStatsdBot.h>
 #include <ixbots/IXCobraToStdoutBot.h>
-#include <ixcore/utils/IXCoreLogger.h>
+#include <ixbots/IXCobraMetricsToRedisBot.h>
 #include <ixredis/IXRedisClient.h>
+#include <ixcore/utils/IXCoreLogger.h>
 #include <ixsentry/IXSentryClient.h>
 #include <ixwebsocket/IXNetSystem.h>
 #include <ixwebsocket/IXSocket.h>
@@ -122,7 +122,6 @@ int main(int argc, char** argv)
     std::string key;
     std::string logfile;
     std::string scriptPath;
-    std::string republishChannel;
     ix::SocketTLSOptions tlsOptions;
     ix::CobraConfig cobraConfig;
     ix::CobraBotConfig cobraBotConfig;
@@ -194,7 +193,8 @@ int main(int argc, char** argv)
             "--limit_received_events", cobraBotConfig.limitReceivedEvents, "Max events per minute");
         app->add_option(
             "--max_events_per_minute", cobraBotConfig.maxEventsPerMinute, "Max events per minute");
-        app->add_option("--batch_size", cobraBotConfig.batchSize, "Subscription batch size");
+        app->add_option(
+            "--batch_size", cobraBotConfig.batchSize, "Subscription batch size");
     };
 
     app.add_flag("--version", version, "Print ws version");
@@ -358,8 +358,7 @@ int main(int argc, char** argv)
     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("--script", scriptPath, "Python script path")->check(CLI::ExistingPath);
     cobra2python->add_option("--pidfile", pidfile, "Pid file");
     addTLSOptions(cobra2python);
     addCobraBotConfig(cobra2python);
@@ -392,7 +391,6 @@ int main(int argc, char** argv)
     snakeApp->add_option("--redis_password", redisPassword, "Redis password");
     snakeApp->add_option("--apps_config_path", appsConfigPath, "Path to auth data")
         ->check(CLI::ExistingPath);
-    snakeApp->add_option("--republish_channel", republishChannel, "Republish channel");
     snakeApp->add_flag("-v", verbose, "Verbose");
     snakeApp->add_flag("-d", disablePong, "Disable Pongs");
     addTLSOptions(snakeApp);
@@ -606,7 +604,8 @@ 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, scriptPath);
         }
     }
     else if (app.got_subcommand("cobra_to_sentry"))
@@ -621,12 +620,14 @@ int main(int argc, char** argv)
         ix::RedisClient redisClient;
         if (!redisClient.connect(hostname, redisPort))
         {
-            spdlog::error("Cannot connect to redis host {}:{}", redisHosts, redisPort);
+            spdlog::error("Cannot connect to redis host {}:{}",
+                          redisHosts, redisPort);
             return 1;
         }
         else
         {
-            ret = (int) ix::cobra_metrics_to_redis_bot(cobraBotConfig, redisClient, verbose);
+            ret = (int) ix::cobra_metrics_to_redis_bot(
+                cobraBotConfig, redisClient, verbose);
         }
     }
     else if (app.got_subcommand("snake"))
@@ -639,8 +640,7 @@ int main(int argc, char** argv)
                                 verbose,
                                 appsConfigPath,
                                 tlsOptions,
-                                disablePong,
-                                republishChannel);
+                                disablePong);
     }
     else if (app.got_subcommand("httpd"))
     {

ws/ws.h

@@ -64,7 +64,9 @@ namespace ix
                      bool disablePerMessageDeflate,
                      const ix::SocketTLSOptions& tlsOptions);
 
-    int ws_redis_cli_main(const std::string& hostname, int port, const std::string& password);
+    int ws_redis_cli_main(const std::string& hostname,
+                          int port,
+                          const std::string& password);
 
     int ws_redis_publish_main(const std::string& hostname,
                               int port,
@@ -103,8 +105,7 @@ namespace ix
                       bool verbose,
                       const std::string& appsConfigPath,
                       const ix::SocketTLSOptions& tlsOptions,
-                      bool disablePong,
-                      const std::string& republishChannel);
+                      bool disablePong);
 
     int ws_httpd_main(int port,
                       const std::string& hostname,

ws/ws_broadcast_server.cpp

@@ -21,15 +21,12 @@ namespace ix
         server.setTLSOptions(tlsOptions);
 
         server.setOnConnectionCallback([&server](std::shared_ptr<WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState,
-                                                 std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
-            webSocket->setOnMessageCallback([webSocket, connectionState, remoteIp, &server](
+                                                 std::shared_ptr<ConnectionState> connectionState) {
+            webSocket->setOnMessageCallback([webSocket, connectionState, &server](
                                                 const WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("New connection");
-                    spdlog::info("remote ip: {}", remoteIp);
                     spdlog::info("id: {}", connectionState->getId());
                     spdlog::info("Uri: {}", msg->openInfo.uri);
                     spdlog::info("Headers:");

ws/ws_connect.cpp

@@ -6,12 +6,12 @@
 
 #include "IXBench.h"
 #include "linenoise.hpp"
-#include <iostream>
 #include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXSocketTLSOptions.h>
 #include <ixwebsocket/IXWebSocket.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
+#include <iostream>
 
 
 namespace ix
@@ -216,7 +216,8 @@ namespace ix
     {
         if (_binaryMode)
         {
-            _webSocket.sendBinary(text);
+            std::vector<uint8_t> data(text.begin(), text.end());
+            _webSocket.sendBinary(data);
         }
         else
         {

ws/ws_echo_server.cpp

@@ -44,15 +44,12 @@ namespace ix
 
         server.setOnConnectionCallback(
             [greetings](std::shared_ptr<ix::WebSocket> webSocket,
-                        std::shared_ptr<ConnectionState> connectionState,
-                        std::unique_ptr<ConnectionInfo> connectionInfo) {
-                auto remoteIp = connectionInfo->remoteIp;
+                        std::shared_ptr<ConnectionState> connectionState) {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, remoteIp, greetings](const WebSocketMessagePtr& msg) {
+                    [webSocket, connectionState, greetings](const WebSocketMessagePtr& msg) {
                         if (msg->type == ix::WebSocketMessageType::Open)
                         {
                             spdlog::info("New connection");
-                            spdlog::info("remote ip: {}", remoteIp);
                             spdlog::info("id: {}", connectionState->getId());
                             spdlog::info("Uri: {}", msg->openInfo.uri);
                             spdlog::info("Headers:");

ws/ws_proxy_server.cpp

@@ -56,18 +56,15 @@ namespace ix
 
         server.setOnConnectionCallback([remoteUrl,
                                         verbose](std::shared_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;
 
             // Server connection
-            state->webSocket().setOnMessageCallback([webSocket, state, remoteIp, verbose](
+            state->webSocket().setOnMessageCallback([webSocket, state, verbose](
                                                         const WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("New connection to remote server");
-                    spdlog::info("remote ip: {}", remoteIp);
                     spdlog::info("id: {}", state->getId());
                     spdlog::info("Uri: {}", msg->openInfo.uri);
                     spdlog::info("Headers:");

ws/ws_redis_cli.cpp

@@ -4,15 +4,17 @@
  *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
  */
 
-#include "linenoise.hpp"
-#include <iostream>
 #include <ixredis/IXRedisClient.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
+#include <iostream>
+#include "linenoise.hpp"
 
 namespace ix
 {
-    int ws_redis_cli_main(const std::string& hostname, int port, const std::string& password)
+    int ws_redis_cli_main(const std::string& hostname,
+                          int port,
+                          const std::string& password)
     {
         RedisClient redisClient;
         if (!redisClient.connect(hostname, port))
@@ -69,7 +71,9 @@ namespace ix
             {
                 if (response.first != RespType::String)
                 {
-                    std::cout << "(" << redisClient.getRespTypeDescription(response.first) << ")"
+                    std::cout << "("
+                              << redisClient.getRespTypeDescription(response.first)
+                              << ")"
                               << " ";
                 }
 

ws/ws_snake.cpp

@@ -45,8 +45,7 @@ namespace ix
                       bool verbose,
                       const std::string& appsConfigPath,
                       const SocketTLSOptions& socketTLSOptions,
-                      bool disablePong,
-                      const std::string& republishChannel)
+                      bool disablePong)
     {
         snake::AppConfig appConfig;
         appConfig.port = port;
@@ -56,7 +55,6 @@ namespace ix
         appConfig.redisPassword = redisPassword;
         appConfig.socketTLSOptions = socketTLSOptions;
         appConfig.disablePong = disablePong;
-        appConfig.republishChannel = republishChannel;
 
         // Parse config file
         auto str = readAsString(appsConfigPath);

ws/ws_transfer.cpp

@@ -20,15 +20,12 @@ namespace ix
         server.setTLSOptions(tlsOptions);
 
         server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState,
-                                                 std::unique_ptr<ConnectionInfo> connectionInfo) {
-            auto remoteIp = connectionInfo->remoteIp;
-            webSocket->setOnMessageCallback([webSocket, connectionState, remoteIp, &server](
+                                                 std::shared_ptr<ConnectionState> connectionState) {
+            webSocket->setOnMessageCallback([webSocket, connectionState, &server](
                                                 const WebSocketMessagePtr& msg) {
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     spdlog::info("ws_transfer: New connection");
-                    spdlog::info("remote ip: {}", remoteIp);
                     spdlog::info("id: {}", connectionState->getId());
                     spdlog::info("Uri: {}", msg->openInfo.uri);
                     spdlog::info("Headers:");