enable unittest

* Added test for websocket leak

* Fixed test

.github/workflows/unittest_linux.yml

@@ -1,13 +0,0 @@
-name: linux
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  linux:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: make test_make
-      run: make test_make

.github/workflows/unittest_mac_tsan_mbedtls.yml

@@ -1,15 +0,0 @@
-name: mac_tsan_mbedtls
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  mac_tsan_mbedtls:
-    runs-on: macOS-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: install mbedtls
-      run: brew install mbedtls
-    - name: make test
-      run: make test_tsan_mbedtls

.github/workflows/unittest_mac_tsan_openssl.yml

@@ -1,15 +0,0 @@
-name: mac_tsan_openssl
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  mac_tsan_openssl:
-    runs-on: macOS-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: install openssl
-      run: brew install openssl@1.1
-    - name: make test
-      run: make test_tsan_openssl

.github/workflows/unittest_mac_tsan_sectransport.yml

@@ -1,13 +0,0 @@
-name: mac_tsan_sectransport
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  mac_tsan_sectransport:
-    runs-on: macOS-latest
-    steps:
-    - uses: actions/checkout@v1
-    - name: make test_tsan
-      run: make test_tsan

.github/workflows/unittest_uwp.yml

@@ -1,40 +0,0 @@
-name: uwp
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  uwp:
-    runs-on: windows-latest
-    steps:
-    - uses: actions/checkout@v1
-    - uses: seanmiddleditch/gha-setup-vsdevenv@master
-    - run: |
-        vcpkg install zlib:x64-uwp
-    - run: |
-        mkdir build
-        cd build
-        cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_SYSTEM_NAME=WindowsStore -DCMAKE_SYSTEM_VERSION="10.0" -DCMAKE_CXX_COMPILER=cl.exe -DUSE_TEST=1 ..
-    - run: cmake --build build
-
-#
-#   Windows with OpenSSL is working but disabled as it takes 13 minutes (10 for openssl) to build with vcpkg
-#
-#   windows_openssl:
-#     runs-on: windows-latest
-#     steps:
-#     - uses: actions/checkout@v1
-#     - uses: seanmiddleditch/gha-setup-vsdevenv@master
-#     - run: |
-#         vcpkg install zlib:x64-windows
-#         vcpkg install openssl:x64-windows
-#     - run: |
-#         mkdir build
-#         cd build
-#         cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_CXX_COMPILER=cl.exe -DUSE_OPEN_SSL=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 ..
-#     - run: cmake --build build
-# 
-#     # Running the unittest does not work, the binary cannot be found
-#     #- run: ../build/test/ixwebsocket_unittest.exe
-#     # working-directory: test

.github/workflows/unittest_windows.yml

@@ -10,10 +10,10 @@ jobs:
     steps:
     - uses: actions/checkout@v1
     - uses: seanmiddleditch/gha-setup-vsdevenv@master
-    - run: |
-        vcpkg install zlib:x64-windows
     - run: |
         mkdir build
         cd build
-        cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_CXX_COMPILER=cl.exe -DUSE_WS=1 -DUSE_TEST=1 ..
+        cmake -DCMAKE_CXX_COMPILER=cl.exe -DUSE_WS=1 -DUSE_TEST=1 -DUSE_ZLIB=0 ..
     - run: cmake --build build
+    - run: ../build/test/ixwebsocket_unittest.exe
+      working-directory: test

CMakeLists.txt

@@ -30,12 +30,15 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXConnectionState.cpp
     ixwebsocket/IXDNSLookup.cpp
     ixwebsocket/IXExponentialBackoff.cpp
+    ixwebsocket/IXGetFreePort.cpp
     ixwebsocket/IXHttp.cpp
     ixwebsocket/IXHttpClient.cpp
     ixwebsocket/IXHttpServer.cpp
     ixwebsocket/IXNetSystem.cpp
     ixwebsocket/IXSelectInterrupt.cpp
     ixwebsocket/IXSelectInterruptFactory.cpp
+    ixwebsocket/IXSelectInterruptPipe.cpp
+    ixwebsocket/IXSetThreadName.cpp
     ixwebsocket/IXSocket.cpp
     ixwebsocket/IXSocketConnect.cpp
     ixwebsocket/IXSocketFactory.cpp
@@ -51,6 +54,7 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXWebSocketPerMessageDeflate.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
+    ixwebsocket/IXWebSocketProxyServer.cpp
     ixwebsocket/IXWebSocketServer.cpp
     ixwebsocket/IXWebSocketTransport.cpp
 )
@@ -58,9 +62,11 @@ set( IXWEBSOCKET_SOURCES
 set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXBench.h
     ixwebsocket/IXCancellationRequest.h
+    ixwebsocket/IXConnectionInfo.h
     ixwebsocket/IXConnectionState.h
     ixwebsocket/IXDNSLookup.h
     ixwebsocket/IXExponentialBackoff.h
+    ixwebsocket/IXGetFreePort.h
     ixwebsocket/IXHttp.h
     ixwebsocket/IXHttpClient.h
     ixwebsocket/IXHttpServer.h
@@ -68,6 +74,7 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXProgressCallback.h
     ixwebsocket/IXSelectInterrupt.h
     ixwebsocket/IXSelectInterruptFactory.h
+    ixwebsocket/IXSelectInterruptPipe.h
     ixwebsocket/IXSetThreadName.h
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketConnect.h
@@ -92,29 +99,13 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXWebSocketPerMessageDeflate.h
     ixwebsocket/IXWebSocketPerMessageDeflateCodec.h
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.h
+    ixwebsocket/IXWebSocketProxyServer.h
     ixwebsocket/IXWebSocketSendInfo.h
     ixwebsocket/IXWebSocketServer.h
     ixwebsocket/IXWebSocketTransport.h
     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)
@@ -199,10 +190,18 @@ if (USE_TLS)
   endif()
 endif()
 
-# Use ZLIB_ROOT CMake variable if you need to use your own zlib
-find_package(ZLIB REQUIRED)
-include_directories(${ZLIB_INCLUDE_DIRS})
-target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})
+option(USE_ZLIB "Enable zlib support" TRUE)
+
+if (USE_ZLIB)
+  # Use ZLIB_ROOT CMake variable if you need to use your own zlib
+  find_package(ZLIB)
+  if (ZLIB_FOUND)
+    include_directories(${ZLIB_INCLUDE_DIRS})
+    target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})
+
+    target_compile_definitions(ixwebsocket PUBLIC IXWEBSOCKET_USE_ZLIB)
+  endif()
+endif()
 
 if (WIN32)
   target_link_libraries(ixwebsocket wsock32 ws2_32 shlwapi)

docker/Dockerfile.ubuntu_groovy

@@ -0,0 +1,23 @@
+# Build time
+FROM ubuntu:groovy as build
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update
+
+RUN apt-get -y install g++ libssl-dev libz-dev make python ninja-build
+RUN apt-get -y install cmake
+RUN apt-get -y install gdb
+
+COPY . /opt
+WORKDIR /opt
+
+#
+# To use the container interactively for debugging/building
+# 1. Build with
+#    CMD ["ls"]
+# 2. Run with
+#    docker run --entrypoint sh -it docker-game-eng-dev.addsrv.com/ws:9.10.6 
+#
+
+RUN ["make", "test"]
+# CMD ["ls"]

docs/CHANGELOG.md

@@ -1,6 +1,78 @@
 # Changelog
 All changes to this project will be documented in this file.
 
+## [10.1.1] - 2020-07-29
+
+(websocket client) onProgressCallback not called for short messages on a websocket (fix #233)
+
+## [10.1.0] - 2020-07-29
+
+(websocket client) heartbeat is not sent at the requested frequency (fix #232)
+
+## [10.0.3] - 2020-07-28
+
+compiler warning fixes
+
+## [10.0.2] - 2020-07-28
+
+(ixcobra) CobraConnection: unsubscribe from all subscriptions when disconnecting
+
+## [10.0.1] - 2020-07-27
+
+(socket utility) move ix::getFreePort to ixwebsocket library
+
+## [10.0.0] - 2020-07-25
+
+(ixwebsocket server) change legacy api with 2 nested callbacks, so that the first api takes a weak_ptr<WebSocket> as its first argument
+
+## [9.10.7] - 2020-07-25
+
+(ixwebsocket) add WebSocketProxyServer, from ws. Still need to make the interface better.
+
+## [9.10.6] - 2020-07-24
+
+(ws) port broadcast_server sub-command to the new server API
+
+## [9.10.5] - 2020-07-24
+
+(unittest) port most unittests to the new server API
+
+## [9.10.3] - 2020-07-24
+
+(ws) port ws transfer to the new server API
+
+## [9.10.2] - 2020-07-24
+
+(websocket client) reset WebSocketTransport onClose callback in the WebSocket destructor
+
+## [9.10.1] - 2020-07-24
+
+(websocket server) reset client websocket callback when the connection is closed
+
+## [9.10.0] - 2020-07-23
+
+(websocket server) add a new simpler API to handle client connections / that API does not trigger a memory leak while the previous one did
+
+## [9.9.3] - 2020-07-17
+
+(build) merge platform specific files which were used to have different implementations for setting a thread name into a single file, to make it easier to include every source files and build the ixwebsocket library (fix #226)
+
+## [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
+
 ## [9.8.5] - 2020-07-06
 
 (cobra python bots) remove the test which stop the bot when events do not follow cobra metrics system schema with an id and a device entry

docs/usage.md

@@ -246,6 +246,10 @@ uint32_t m = webSocket.getMaxWaitBetweenReconnectionRetries();
 
 ## WebSocket server API
 
+### Legacy api
+
+This api was actually changed to take a weak_ptr<WebSocket> as the first argument to setOnConnectionCallback ; previously it would take a shared_ptr<WebSocket> which was creating cycles and then memory leaks problems.
+
 ```cpp
 #include <ixwebsocket/IXWebSocketServer.h>
 
@@ -256,29 +260,35 @@ uint32_t m = webSocket.getMaxWaitBetweenReconnectionRetries();
 ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
-    [&server](std::shared_ptr<WebSocket> webSocket,
-              std::shared_ptr<ConnectionState> connectionState)
+    [&server](std::weak_ptr<WebSocket> webSocket,
+              std::shared_ptr<ConnectionState> connectionState,
+              std::unique_ptr<ConnectionInfo> connectionInfo)
     {
-        webSocket->setOnMessageCallback(
+        std::cout << "Remote ip: " << connectionInfo->remoteIp << std::endl;
+
+        auto ws = webSocket.lock();
+        if (ws)
+        {
+            ws->setOnMessageCallback(
                 [webSocket, connectionState, &server](const ix::WebSocketMessagePtr msg)
                 {
                     if (msg->type == ix::WebSocketMessageType::Open)
                     {
-                    std::cerr << "New connection" << std::endl;
+                        std::cout << "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::cerr << "id: " << connectionState->getId() << std::endl;
+                        std::cout << "id: " << connectionState->getId() << std::endl;
 
                         // The uri the client did connect to.
-                    std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
+                        std::cout << "Uri: " << msg->openInfo.uri << std::endl;
 
-                    std::cerr << "Headers:" << std::endl;
+                        std::cout << "Headers:" << std::endl;
                         for (auto it : msg->openInfo.headers)
                         {
-                        std::cerr << it.first << ": " << it.second << std::endl;
+                            std::cout << it.first << ": " << it.second << std::endl;
                         }
                     }
                     else if (msg->type == ix::WebSocketMessageType::Message)
@@ -287,7 +297,12 @@ server.setOnConnectionCallback(
                         // All connected clients are available in an std::set. See the broadcast cpp example.
                         // Second parameter tells whether we are sending the message in binary or text mode.
                         // Here we send it in the same mode as it was received.
-                    webSocket->send(msg->str, msg->binary);
+                        auto ws = webSocket.lock();
+                        if (ws)
+                        {
+                            ws->send(msg->str, msg->binary);
+                        }
+                    }
                 }
             }
         );
@@ -309,6 +324,74 @@ server.wait();
 
 ```
 
+### New api
+
+The new API does not require to use 2 nested callbacks, which is a bit annoying. The real fix is that there was a memory leak due to a shared_ptr cycle, due to passing down a shared_ptr<WebSocket> down to the callbacks.
+
+The webSocket reference is guaranteed to be always valid ; by design the callback will never be invoked with a null webSocket object.
+
+```cpp
+#include <ixwebsocket/IXWebSocketServer.h>
+
+...
+
+// Run a server on localhost at a given port.
+// Bound host name, max connections and listen backlog can also be passed in as parameters.
+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,
+    // at this point only the client ip address and the port.
+    std::cout << "Remote ip: " << connectionInfo.remoteIp << std::endl;
+
+    if (msg->type == ix::WebSocketMessageType::Open)
+    {
+        std::cout << "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;
+
+        // The uri the client did connect to.
+        std::cout << "Uri: " << msg->openInfo.uri << std::endl;
+
+        std::cout << "Headers:" << std::endl;
+        for (auto it : msg->openInfo.headers)
+        {
+            std::cout << it.first << ": " << it.second << std::endl;
+        }
+    }
+    else if (msg->type == ix::WebSocketMessageType::Message)
+    {
+        // For an echo server, we just send back to the client whatever was received by the server
+        // All connected clients are available in an std::set. See the broadcast cpp example.
+        // Second parameter tells whether we are sending the message in binary or text mode.
+        // Here we send it in the same mode as it was received.
+        webSocket.send(msg->str, msg->binary);
+    }
+);
+
+auto res = server.listen();
+if (!res.first)
+{
+    // Error handling
+    return 1;
+}
+
+// Run the server in the background. Server can be stoped by calling server.stop()
+server.start();
+
+// Block until server.stop() is called.
+server.wait();
+
+```
+
 ## HTTP client API
 
 ```cpp
@@ -417,11 +500,14 @@ If you want to handle how requests are processed, implement the setOnConnectionC
 ```cpp
 setOnConnectionCallback(
     [this](HttpRequestPtr request,
-           std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr
+           std::shared_ptr<ConnectionState> /*connectionState*/,
+           std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr
     {
         // Build a string for the response
         std::stringstream ss;
-        ss << request->method
+        ss << connectionInfo->remoteIp
+           << " "
+           << request->method
            << " "
            << request->uri;
 

ixcobra/ixcobra/IXCobraConnection.cpp

@@ -111,6 +111,12 @@ namespace ix
 
     void CobraConnection::disconnect()
     {
+        auto subscriptionIds = getSubscriptionsIds();
+        for (auto&& subscriptionId : subscriptionIds)
+        {
+            unsubscribe(subscriptionId);
+        }
+
         _authenticated = false;
         _webSocket->stop();
     }
@@ -614,6 +620,18 @@ namespace ix
         _webSocket->send(pdu.toStyledString());
     }
 
+    std::vector<std::string> CobraConnection::getSubscriptionsIds()
+    {
+        std::vector<std::string> subscriptionIds;
+        std::lock_guard<std::mutex> lock(_cbsMutex);
+
+        for (auto&& it : _cbs)
+        {
+            subscriptionIds.push_back(it.first);
+        }
+        return subscriptionIds;
+    }
+
     //
     // Enqueue strategy drops old messages when we are at full capacity
     //

ixcobra/ixcobra/IXCobraConnection.h

@@ -163,6 +163,9 @@ namespace ix
         /// Tells whether the internal queue is empty or not
         bool isQueueEmpty();
 
+        /// Retrieve all subscriptions ids
+        std::vector<std::string> getSubscriptionsIds();
+
         ///
         /// Member variables
         ///

ixredis/ixredis/IXRedisServer.cpp

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

ixredis/ixredis/IXRedisServer.h

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

ixsnake/CMakeLists.txt

@@ -7,12 +7,14 @@ 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.cpp

@@ -26,6 +26,12 @@ namespace snake
         }
 
         auto roles = appConfig.apps[appkey]["roles"];
+        if (roles.count(role) == 0)
+        {
+            std::cerr << "Missing role " << role << std::endl;
+            return std::string();
+        }
+
         auto channel = roles[role]["secret"];
         return channel;
     }

ixsnake/ixsnake/IXAppConfig.h

@@ -33,6 +33,9 @@ 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

@@ -7,15 +7,21 @@
 #pragma once
 
 #include <ixredis/IXRedisClient.h>
-#include <future>
+#include <thread>
 #include <ixwebsocket/IXConnectionState.h>
 #include <string>
+#include "IXStreamSql.h"
 
 namespace snake
 {
     class SnakeConnectionState : public ix::ConnectionState
     {
     public:
+        virtual ~SnakeConnectionState()
+        {
+            stopSubScriptionThread();
+        }
+
         std::string getNonce()
         {
             return _nonce;
@@ -30,6 +36,7 @@ namespace snake
         {
             return _appkey;
         }
+
         void setAppkey(const std::string& appkey)
         {
             _appkey = appkey;
@@ -39,6 +46,7 @@ namespace snake
         {
             return _role;
         }
+
         void setRole(const std::string& role)
         {
             _role = role;
@@ -49,7 +57,24 @@ namespace snake
             return _redisClient;
         }
 
-        std::future<void> fut;
+        void stopSubScriptionThread()
+        {
+            if (subscriptionThread.joinable())
+            {
+                subscriptionRedisClient.stop();
+                subscriptionThread.join();
+            }
+        }
+
+        // We could make those accessible through methods
+        std::thread subscriptionThread;
+        std::string appChannel;
+        std::string subscriptionId;
+        uint64_t id;
+        std::unique_ptr<StreamSql> streamSql;
+        ix::RedisClient subscriptionRedisClient;
+        ix::RedisClient::OnRedisSubscribeResponseCallback onRedisSubscribeResponseCallback;
+        ix::RedisClient::OnRedisSubscribeCallback onRedisSubscribeCallback;
 
     private:
         std::string _nonce;

ixsnake/ixsnake/IXSnakeProtocol.cpp

@@ -18,21 +18,22 @@
 namespace snake
 {
     void handleError(const std::string& action,
-                     std::shared_ptr<ix::WebSocket> ws,
-                     nlohmann::json pdu,
+                     ix::WebSocket& ws,
+                     uint64_t pduId,
                      const std::string& errMsg)
     {
         std::string actionError(action);
         actionError += "/error";
 
         nlohmann::json response = {
-            {"action", actionError}, {"id", pdu.value("id", 1)}, {"body", {{"reason", errMsg}}}};
-        ws->sendText(response.dump());
+            {"action", actionError}, {"id", pduId}, {"body", {{"reason", errMsg}}}};
+        ws.sendText(response.dump());
     }
 
     void handleHandshake(std::shared_ptr<SnakeConnectionState> state,
-                         std::shared_ptr<ix::WebSocket> ws,
-                         const nlohmann::json& pdu)
+                         ix::WebSocket& ws,
+                         const nlohmann::json& pdu,
+                         uint64_t pduId)
     {
         std::string role = pdu["body"]["data"]["role"];
 
@@ -41,7 +42,7 @@ namespace snake
 
         nlohmann::json response = {
             {"action", "auth/handshake/ok"},
-            {"id", pdu.value("id", 1)},
+            {"id", pduId},
             {"body",
              {
                  {"data", {{"nonce", state->getNonce()}, {"connection_id", state->getId()}}},
@@ -49,13 +50,14 @@ namespace snake
 
         auto serializedResponse = response.dump();
 
-        ws->sendText(serializedResponse);
+        ws.sendText(serializedResponse);
     }
 
     void handleAuth(std::shared_ptr<SnakeConnectionState> state,
-                    std::shared_ptr<ix::WebSocket> ws,
+                    ix::WebSocket& ws,
                     const AppConfig& appConfig,
-                    const nlohmann::json& pdu)
+                    const nlohmann::json& pdu,
+                    uint64_t pduId)
     {
         auto secret = getRoleSecret(appConfig, state->appkey(), state->role());
 
@@ -63,9 +65,9 @@ namespace snake
         {
             nlohmann::json response = {
                 {"action", "auth/authenticate/error"},
-                {"id", pdu.value("id", 1)},
+                {"id", pduId},
                 {"body", {{"error", "authentication_failed"}, {"reason", "invalid secret"}}}};
-            ws->sendText(response.dump());
+            ws.sendText(response.dump());
             return;
         }
 
@@ -79,19 +81,21 @@ namespace snake
                 {"action", "auth/authenticate/error"},
                 {"id", pdu.value("id", 1)},
                 {"body", {{"error", "authentication_failed"}, {"reason", "invalid hash"}}}};
-            ws->sendText(response.dump());
+            ws.sendText(response.dump());
             return;
         }
 
         nlohmann::json response = {
             {"action", "auth/authenticate/ok"}, {"id", pdu.value("id", 1)}, {"body", {}}};
 
-        ws->sendText(response.dump());
+        ws.sendText(response.dump());
     }
 
     void handlePublish(std::shared_ptr<SnakeConnectionState> state,
-                       std::shared_ptr<ix::WebSocket> ws,
-                       const nlohmann::json& pdu)
+                       ix::WebSocket& ws,
+                       const AppConfig& appConfig,
+                       const nlohmann::json& pdu,
+                       uint64_t pduId)
     {
         std::vector<std::string> channels;
 
@@ -111,10 +115,16 @@ namespace snake
         {
             std::stringstream ss;
             ss << "Missing channels or channel field in publish data";
-            handleError("rtm/publish", ws, pdu, ss.str());
+            handleError("rtm/publish", ws, pduId, ss.str());
             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;
@@ -125,7 +135,7 @@ namespace snake
             {
                 std::stringstream ss;
                 ss << "Cannot publish to redis host " << errMsg;
-                handleError("rtm/publish", ws, pdu, ss.str());
+                handleError("rtm/publish", ws, pduId, ss.str());
                 return;
             }
         }
@@ -133,26 +143,27 @@ namespace snake
         nlohmann::json response = {
             {"action", "rtm/publish/ok"}, {"id", pdu.value("id", 1)}, {"body", {}}};
 
-        ws->sendText(response.dump());
+        ws.sendText(response.dump());
     }
 
     //
     // FIXME: this is not cancellable. We should be able to cancel the redis subscription
     //
-    void handleRedisSubscription(std::shared_ptr<SnakeConnectionState> state,
-                                 std::shared_ptr<ix::WebSocket> ws,
+    void handleSubscribe(std::shared_ptr<SnakeConnectionState> state,
+                         ix::WebSocket& ws,
                          const AppConfig& appConfig,
-                                 const nlohmann::json& pdu)
+                         const nlohmann::json& pdu,
+                         uint64_t pduId)
     {
         std::string channel = pdu["body"]["channel"];
-        std::string subscriptionId = channel;
+        state->subscriptionId = channel;
 
         std::stringstream ss;
         ss << state->appkey() << "::" << channel;
 
-        std::string appChannel(ss.str());
+        state->appChannel = ss.str();
 
-        ix::RedisClient redisClient;
+        ix::RedisClient& redisClient = state->subscriptionRedisClient;
         int port = appConfig.redisPort;
 
         auto urls = appConfig.redisHosts;
@@ -163,7 +174,7 @@ namespace snake
         {
             std::stringstream ss;
             ss << "Cannot connect to redis host " << hostname << ":" << port;
-            handleError("rtm/subscribe", ws, pdu, ss.str());
+            handleError("rtm/subscribe", ws, pduId, ss.str());
             return;
         }
 
@@ -175,80 +186,90 @@ namespace snake
             {
                 std::stringstream ss;
                 ss << "Cannot authenticated to redis";
-                handleError("rtm/subscribe", ws, pdu, ss.str());
+                handleError("rtm/subscribe", ws, pduId, ss.str());
                 return;
             }
         }
 
-        int id = 0;
-        auto callback = [ws, &id, &subscriptionId](const std::string& messageStr) {
+        std::string filterStr;
+        if (pdu["body"].find("filter") != pdu["body"].end())
+        {
+            std::string filterStr = pdu["body"]["filter"];
+        }
+        state->streamSql = std::make_unique<StreamSql>(filterStr);
+        state->id = 0;
+        state->onRedisSubscribeCallback = [&ws, state](const std::string& messageStr) {
             auto msg = nlohmann::json::parse(messageStr);
 
             msg = msg["body"]["message"];
 
+            if (state->streamSql->valid() && !state->streamSql->match(msg))
+            {
+                return;
+            }
+
             nlohmann::json response = {
                 {"action", "rtm/subscription/data"},
-                {"id", id++},
+                {"id", state->id++},
                 {"body",
-                 {{"subscription_id", subscriptionId}, {"position", "0-0"}, {"messages", {msg}}}}};
+                 {{"subscription_id", state->subscriptionId}, {"position", "0-0"}, {"messages", {msg}}}}};
 
-            ws->sendText(response.dump());
+            ws.sendText(response.dump());
         };
 
-        auto responseCallback = [ws, pdu, &subscriptionId](const std::string& redisResponse) {
+        state->onRedisSubscribeResponseCallback = [&ws, state, pduId](const std::string& redisResponse) {
             std::stringstream ss;
             ss << "Redis Response: " << redisResponse << "...";
             ix::CoreLogger::log(ss.str().c_str());
 
             // Success
             nlohmann::json response = {{"action", "rtm/subscribe/ok"},
-                                       {"id", pdu.value("id", 1)},
-                                       {"body", {{"subscription_id", subscriptionId}}}};
-            ws->sendText(response.dump());
+                                       {"id", pduId},
+                                       {"body", {{"subscription_id", state->subscriptionId}}}};
+            ws.sendText(response.dump());
         };
 
         {
             std::stringstream ss;
-            ss << "Subscribing to " << appChannel << "...";
+            ss << "Subscribing to " << state->appChannel << "...";
             ix::CoreLogger::log(ss.str().c_str());
         }
 
-        if (!redisClient.subscribe(appChannel, responseCallback, callback))
+        auto subscription = [&redisClient, state, &ws, pduId]
+        {
+            if (!redisClient.subscribe(state->appChannel, 
+                                       state->onRedisSubscribeResponseCallback,
+                                       state->onRedisSubscribeCallback))
             {
                 std::stringstream ss;
-            ss << "Error subscribing to channel " << appChannel;
-            handleError("rtm/subscribe", ws, pdu, ss.str());
+                ss << "Error subscribing to channel " << state->appChannel;
+                handleError("rtm/subscribe", ws, pduId, ss.str());
                 return;
             }
-    }
+        };
 
-    void handleSubscribe(std::shared_ptr<SnakeConnectionState> state,
-                         std::shared_ptr<ix::WebSocket> ws,
-                         const AppConfig& appConfig,
-                         const nlohmann::json& pdu)
-    {
-        state->fut =
-            std::async(std::launch::async, handleRedisSubscription, state, ws, appConfig, pdu);
+        state->subscriptionThread = std::thread(subscription);
     }
 
     void handleUnSubscribe(std::shared_ptr<SnakeConnectionState> state,
-                           std::shared_ptr<ix::WebSocket> ws,
-                           const nlohmann::json& pdu)
+                           ix::WebSocket& ws,
+                           const nlohmann::json& pdu,
+                           uint64_t pduId)
     {
         // extract subscription_id
         auto body = pdu["body"];
         auto subscriptionId = body["subscription_id"];
 
-        state->redisClient().stop();
+        state->stopSubScriptionThread();
 
         nlohmann::json response = {{"action", "rtm/unsubscribe/ok"},
-                                   {"id", pdu.value("id", 1)},
+                                   {"id", pduId},
                                    {"body", {{"subscription_id", subscriptionId}}}};
-        ws->sendText(response.dump());
+        ws.sendText(response.dump());
     }
 
     void processCobraMessage(std::shared_ptr<SnakeConnectionState> state,
-                             std::shared_ptr<ix::WebSocket> ws,
+                             ix::WebSocket& ws,
                              const AppConfig& appConfig,
                              const std::string& str)
     {
@@ -263,31 +284,32 @@ namespace snake
             ss << "malformed json pdu: " << e.what() << " -> " << str << "";
 
             nlohmann::json response = {{"body", {{"error", "invalid_json"}, {"reason", ss.str()}}}};
-            ws->sendText(response.dump());
+            ws.sendText(response.dump());
             return;
         }
 
         auto action = pdu["action"];
+        uint64_t pduId = pdu.value("id", 1);
 
         if (action == "auth/handshake")
         {
-            handleHandshake(state, ws, pdu);
+            handleHandshake(state, ws, pdu, pduId);
         }
         else if (action == "auth/authenticate")
         {
-            handleAuth(state, ws, appConfig, pdu);
+            handleAuth(state, ws, appConfig, pdu, pduId);
         }
         else if (action == "rtm/publish")
         {
-            handlePublish(state, ws, pdu);
+            handlePublish(state, ws, appConfig, pdu, pduId);
         }
         else if (action == "rtm/subscribe")
         {
-            handleSubscribe(state, ws, appConfig, pdu);
+            handleSubscribe(state, ws, appConfig, pdu, pduId);
         }
         else if (action == "rtm/unsubscribe")
         {
-            handleUnSubscribe(state, ws, pdu);
+            handleUnSubscribe(state, ws, pdu, pduId);
         }
         else
         {

ixsnake/ixsnake/IXSnakeProtocol.h

@@ -20,7 +20,7 @@ namespace snake
     struct AppConfig;
 
     void processCobraMessage(std::shared_ptr<SnakeConnectionState> state,
-                             std::shared_ptr<ix::WebSocket> ws,
+                             ix::WebSocket& ws,
                              const AppConfig& appConfig,
                              const std::string& str);
 } // namespace snake

ixsnake/ixsnake/IXSnakeServer.cpp

@@ -59,18 +59,22 @@ namespace snake
         };
         _server.setConnectionStateFactory(factory);
 
-        _server.setOnConnectionCallback(
-            [this](std::shared_ptr<ix::WebSocket> webSocket,
-                   std::shared_ptr<ix::ConnectionState> connectionState) {
+        _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;
             
-                webSocket->setOnMessageCallback(
-                    [this, webSocket, state](const ix::WebSocketMessagePtr& msg) {
                 std::stringstream ss;
+                ss << "[" << state->getId() << "] ";
+
                 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;
@@ -111,13 +115,12 @@ namespace snake
                 }
                 else if (msg->type == ix::WebSocketMessageType::Message)
                 {
-                            ss << "Received " << msg->wireSize << " bytes" << std::endl;
+                    ss << "Received " << msg->wireSize << " bytes" << " " << msg->str << std::endl;
                     processCobraMessage(state, webSocket, _appConfig, msg->str);
                 }
 
                 ix::CoreLogger::log(ss.str().c_str(), logLevel);
         });
-            });
 
         auto res = _server.listen();
         if (!res.first)

ixsnake/ixsnake/IXStreamSql.cpp

@@ -0,0 +1,63 @@
+/*
+ *  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

@@ -0,0 +1,29 @@
+/*
+ *  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

@@ -0,0 +1,25 @@
+/*
+ *  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.cpp

@@ -16,7 +16,10 @@
 #include <random>
 #include <sstream>
 #include <vector>
+
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
+#endif
 
 namespace ix
 {
@@ -174,11 +177,13 @@ namespace ix
         ss << verb << " " << path << " HTTP/1.1\r\n";
         ss << "Host: " << host << "\r\n";
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         if (args->compress)
         {
             ss << "Accept-Encoding: gzip"
                << "\r\n";
         }
+#endif
 
         // Append extra headers
         for (auto&& it : args->extraHeaders)
@@ -495,6 +500,7 @@ namespace ix
 
         downloadSize = payload.size();
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         // If the content was compressed with gzip, decode it
         if (headers["Content-Encoding"] == "gzip")
         {
@@ -513,6 +519,7 @@ namespace ix
             }
             payload = decompressedPayload;
         }
+#endif
 
         return std::make_shared<HttpResponse>(code,
                                               description,
@@ -672,6 +679,7 @@ namespace ix
         return ss.str();
     }
 
+#ifdef IXWEBSOCKET_USE_ZLIB
     bool HttpClient::gzipInflate(const std::string& in, std::string& out)
     {
         z_stream inflateState;
@@ -716,6 +724,7 @@ namespace ix
         inflateEnd(&inflateState);
         return true;
     }
+#endif
 
     void HttpClient::log(const std::string& msg, HttpRequestArgsPtr args)
     {

ixwebsocket/IXHttpClient.h

@@ -90,7 +90,9 @@ namespace ix
     private:
         void log(const std::string& msg, HttpRequestArgsPtr args);
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         bool gzipInflate(const std::string& in, std::string& out);
+#endif
 
         // Async API background thread runner
         void run();
@@ -103,9 +105,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,14 @@
 #include "IXNetSystem.h"
 #include "IXSocketConnect.h"
 #include "IXUserAgent.h"
+#include <cstring>
 #include <fstream>
 #include <sstream>
 #include <vector>
+
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
-#include <cstring>
+#endif
 
 namespace
 {
@@ -41,6 +44,7 @@ namespace
         return std::make_pair(res.first, std::string(vec.begin(), vec.end()));
     }
 
+#ifdef IXWEBSOCKET_USE_ZLIB
     std::string gzipCompress(const std::string& str)
     {
         z_stream zs; // z_stream is zlib's control structure
@@ -50,8 +54,11 @@ 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();
@@ -69,18 +76,18 @@ 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);
 
         return outstring;
     }
+#endif
 } // namespace
 
 namespace ix
@@ -113,7 +120,8 @@ namespace ix
     }
 
     void HttpServer::handleConnection(std::unique_ptr<Socket> socket,
-                                      std::shared_ptr<ConnectionState> connectionState)
+                                      std::shared_ptr<ConnectionState> connectionState,
+                                      std::unique_ptr<ConnectionInfo> connectionInfo)
     {
         _connectedClientsCount++;
 
@@ -122,7 +130,8 @@ namespace ix
 
         if (std::get<0>(ret))
         {
-            auto response = _onConnectionCallback(std::get<2>(ret), connectionState);
+            auto response =
+                _onConnectionCallback(std::get<2>(ret), connectionState, std::move(connectionInfo));
             if (!Http::sendResponse(response, socket))
             {
                 logError("Cannot send response");
@@ -142,7 +151,8 @@ namespace ix
     {
         setOnConnectionCallback(
             [this](HttpRequestPtr request,
-                   std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
+                   std::shared_ptr<ConnectionState> /*connectionState*/,
+                   std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
                 std::string uri(request->uri);
                 if (uri.empty() || uri == "/")
                 {
@@ -163,16 +173,19 @@ namespace ix
 
                 std::string content = res.second;
 
+#ifdef IXWEBSOCKET_USE_ZLIB
                 std::string acceptEncoding = request->headers["Accept-encoding"];
                 if (acceptEncoding == "*" || acceptEncoding.find("gzip") != std::string::npos)
                 {
                     content = gzipCompress(content);
                     headers["Content-Encoding"] = "gzip";
                 }
+#endif
 
                 // Log request
                 std::stringstream ss;
-                ss << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                   << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri << " " << content.size();
                 logInfo(ss.str());
 
@@ -196,15 +209,16 @@ namespace ix
         // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Redirections
         //
         setOnConnectionCallback(
-            [this,
-             redirectUrl](HttpRequestPtr request,
-                          std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
+            [this, redirectUrl](HttpRequestPtr request,
+                                std::shared_ptr<ConnectionState> /*connectionState*/,
+                                std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                   << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
                 logInfo(ss.str());
 

ixwebsocket/IXHttpServer.h

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

ixwebsocket/IXSelectInterruptPipe.cpp

@@ -5,8 +5,10 @@
  */
 
 //
-// On macOS we use UNIX pipes to wake up select.
+// 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.
 //
+#ifndef _WIN32
 
 #include "IXSelectInterruptPipe.h"
 
@@ -144,3 +146,5 @@ namespace ix
         return _fildes[kPipeReadIndex];
     }
 } // namespace ix
+
+#endif // !_WIN32

ixwebsocket/IXSetThreadName.cpp

@@ -0,0 +1,81 @@
+/*
+ *  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(32);
+    const size_t SocketServer::kDefaultMaxConnections(128);
     const int SocketServer::kDefaultAddressFamily(AF_INET);
 
     SocketServer::SocketServer(
@@ -276,6 +276,7 @@ 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);
@@ -307,6 +308,45 @@ 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)
             {
@@ -339,10 +379,13 @@ namespace ix
 
             // Launch the handleConnection work asynchronously in its own thread.
             std::lock_guard<std::mutex> lock(_connectionsThreadsMutex);
-            _connectionsThreads.push_back(std::make_pair(
+            _connectionsThreads.push_back(
+                std::make_pair(connectionState,
+                               std::thread(&SocketServer::handleConnection,
+                                           this,
+                                           std::move(socket),
                                            connectionState,
-                std::thread(
-                    &SocketServer::handleConnection, this, std::move(socket), connectionState)));
+                                           std::move(connectionInfo))));
         }
     }
 

ixwebsocket/IXSocketServer.h

@@ -6,6 +6,7 @@
 
 #pragma once
 
+#include "IXConnectionInfo.h"
 #include "IXConnectionState.h"
 #include "IXSocketTLSOptions.h"
 #include <atomic>
@@ -102,7 +103,8 @@ namespace ix
         ConnectionStateFactory _connectionStateFactory;
 
         virtual void handleConnection(std::unique_ptr<Socket>,
-                                      std::shared_ptr<ConnectionState> connectionState) = 0;
+                                      std::shared_ptr<ConnectionState> connectionState,
+                                      std::unique_ptr<ConnectionInfo> connectionInfo) = 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 <cstring>
 #include <cstdlib>
+#include <cstring>
 
 namespace
 {

ixwebsocket/IXUserAgent.cpp

@@ -8,7 +8,9 @@
 
 #include "IXWebSocketVersion.h"
 #include <sstream>
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
+#endif
 
 // Platform name
 #if defined(_WIN32)
@@ -77,8 +79,10 @@ namespace ix
         ss << " nossl";
 #endif
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         // Zlib version
         ss << " zlib " << ZLIB_VERSION;
+#endif
 
         return ss.str();
     }

ixwebsocket/IXWebSocket.cpp

@@ -46,6 +46,7 @@ namespace ix
     WebSocket::~WebSocket()
     {
         stop();
+        _ws.setOnCloseCallback(nullptr);
     }
 
     void WebSocket::setUrl(const std::string& url)

ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp

@@ -28,21 +28,26 @@ namespace ix
     WebSocketPerMessageDeflateCompressor::WebSocketPerMessageDeflateCompressor()
         : _compressBufferSize(kBufferSize)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_deflateState, 0, sizeof(_deflateState));
 
         _deflateState.zalloc = Z_NULL;
         _deflateState.zfree = Z_NULL;
         _deflateState.opaque = Z_NULL;
+#endif
     }
 
     WebSocketPerMessageDeflateCompressor::~WebSocketPerMessageDeflateCompressor()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         deflateEnd(&_deflateState);
+#endif
     }
 
     bool WebSocketPerMessageDeflateCompressor::init(uint8_t deflateBits,
                                                     bool clientNoContextTakeOver)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         int ret = deflateInit2(&_deflateState,
                                Z_DEFAULT_COMPRESSION,
                                Z_DEFLATED,
@@ -57,17 +62,49 @@ namespace ix
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
+#else
+        return false;
+#endif
     }
 
-    bool WebSocketPerMessageDeflateCompressor::endsWith(const std::string& value,
-                                                        const std::string& ending)
+    template<typename T>
+    bool WebSocketPerMessageDeflateCompressor::endsWithEmptyUnCompressedBlock(const T& value)
     {
-        if (ending.size() > value.size()) return false;
-        return std::equal(ending.rbegin(), ending.rend(), value.rbegin());
+        if (kEmptyUncompressedBlock.size() > value.size()) return false;
+        auto N = value.size();
+        return value[N - 1] == kEmptyUncompressedBlock[3] &&
+               value[N - 2] == kEmptyUncompressedBlock[2] &&
+               value[N - 3] == kEmptyUncompressedBlock[1] &&
+               value[N - 4] == kEmptyUncompressedBlock[0];
     }
 
     bool WebSocketPerMessageDeflateCompressor::compress(const std::string& in, std::string& out)
     {
+        return compressData(in, 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)
+    {
+        return compressData(in, 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)
+    {
+#ifdef IXWEBSOCKET_USE_ZLIB
         //
         // 7.2.1.  Compression
         //
@@ -96,7 +133,8 @@ namespace ix
             // The normal buffer size should be 6 but
             // we remove the 4 octets from the tail (#4)
             uint8_t buf[2] = {0x02, 0x00};
-            out.append((char*) (buf), 2);
+            out.push_back(buf[0]);
+            out.push_back(buf[1]);
 
             return true;
         }
@@ -114,15 +152,18 @@ namespace ix
 
             output = _compressBufferSize - _deflateState.avail_out;
 
-            out.append((char*) (_compressBuffer.get()), output);
+            out.insert(out.end(), _compressBuffer.get(), _compressBuffer.get() + output);
         } while (_deflateState.avail_out == 0);
 
-        if (endsWith(out, kEmptyUncompressedBlock))
+        if (endsWithEmptyUnCompressedBlock(out))
         {
             out.resize(out.size() - 4);
         }
 
         return true;
+#else
+        return false;
+#endif
     }
 
     //
@@ -131,6 +172,7 @@ namespace ix
     WebSocketPerMessageDeflateDecompressor::WebSocketPerMessageDeflateDecompressor()
         : _compressBufferSize(kBufferSize)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_inflateState, 0, sizeof(_inflateState));
 
         _inflateState.zalloc = Z_NULL;
@@ -138,16 +180,20 @@ namespace ix
         _inflateState.opaque = Z_NULL;
         _inflateState.avail_in = 0;
         _inflateState.next_in = Z_NULL;
+#endif
     }
 
     WebSocketPerMessageDeflateDecompressor::~WebSocketPerMessageDeflateDecompressor()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         inflateEnd(&_inflateState);
+#endif
     }
 
     bool WebSocketPerMessageDeflateDecompressor::init(uint8_t inflateBits,
                                                       bool clientNoContextTakeOver)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         int ret = inflateInit2(&_inflateState, -1 * inflateBits);
 
         if (ret != Z_OK) return false;
@@ -157,10 +203,14 @@ namespace ix
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
+#else
+        return false;
+#endif
     }
 
     bool WebSocketPerMessageDeflateDecompressor::decompress(const std::string& in, std::string& out)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         //
         // 7.2.2.  Decompression
         //
@@ -197,5 +247,8 @@ namespace ix
         } while (_inflateState.avail_out == 0);
 
         return true;
+#else
+        return false;
+#endif
     }
 } // namespace ix

ixwebsocket/IXWebSocketPerMessageDeflateCodec.h

@@ -6,9 +6,12 @@
 
 #pragma once
 
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include "zlib.h"
+#endif
 #include <memory>
 #include <string>
+#include <vector>
 
 namespace ix
 {
@@ -20,14 +23,23 @@ namespace ix
 
         bool init(uint8_t deflateBits, bool clientNoContextTakeOver);
         bool compress(const std::string& in, std::string& out);
+        bool compress(const std::string& in, std::vector<uint8_t>& out);
+        bool compress(const std::vector<uint8_t>& in, std::string& out);
+        bool compress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out);
 
     private:
-        static bool endsWith(const std::string& value, const std::string& ending);
+        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;
         std::unique_ptr<unsigned char[]> _compressBuffer;
+
+#ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _deflateState;
+#endif
     };
 
     class WebSocketPerMessageDeflateDecompressor
@@ -43,7 +55,10 @@ namespace ix
         int _flush;
         size_t _compressBufferSize;
         std::unique_ptr<unsigned char[]> _compressBuffer;
+
+#ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _inflateState;
+#endif
     };
 
 } // namespace ix

ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp

@@ -61,6 +61,7 @@ namespace ix
         _clientMaxWindowBits = kDefaultClientMaxWindowBits;
         _serverMaxWindowBits = kDefaultServerMaxWindowBits;
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         // Split by ;
         std::string token;
         std::stringstream tokenStream(extension);
@@ -112,6 +113,7 @@ namespace ix
                 sanitizeClientMaxWindowBits();
             }
         }
+#endif
     }
 
     void WebSocketPerMessageDeflateOptions::sanitizeClientMaxWindowBits()
@@ -126,6 +128,7 @@ namespace ix
 
     std::string WebSocketPerMessageDeflateOptions::generateHeader()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         std::stringstream ss;
         ss << "Sec-WebSocket-Extensions: permessage-deflate";
 
@@ -138,11 +141,18 @@ namespace ix
         ss << "\r\n";
 
         return ss.str();
+#else
+        return std::string();
+#endif
     }
 
     bool WebSocketPerMessageDeflateOptions::enabled() const
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         return _enabled;
+#else
+        return false;
+#endif
     }
 
     bool WebSocketPerMessageDeflateOptions::getClientNoContextTakeover() const

ixwebsocket/IXWebSocketProxyServer.cpp

@@ -0,0 +1,123 @@
+/*
+ *  IXWebSocketProxyServer.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXWebSocketProxyServer.h"
+
+#include "IXWebSocketServer.h"
+#include <sstream>
+
+namespace ix
+{
+    class ProxyConnectionState : public ix::ConnectionState
+    {
+    public:
+        ProxyConnectionState()
+            : _connected(false)
+        {
+        }
+
+        ix::WebSocket& webSocket()
+        {
+            return _serverWebSocket;
+        }
+
+        bool isConnected()
+        {
+            return _connected;
+        }
+
+        void setConnected()
+        {
+            _connected = true;
+        }
+
+    private:
+        ix::WebSocket _serverWebSocket;
+        bool _connected;
+    };
+
+    int websocket_proxy_server_main(int port,
+                                    const std::string& hostname,
+                                    const ix::SocketTLSOptions& tlsOptions,
+                                    const std::string& remoteUrl,
+                                    bool /*verbose*/)
+    {
+        ix::WebSocketServer server(port, hostname);
+        server.setTLSOptions(tlsOptions);
+
+        auto factory = []() -> std::shared_ptr<ix::ConnectionState> {
+            return std::make_shared<ProxyConnectionState>();
+        };
+        server.setConnectionStateFactory(factory);
+
+        server.setOnConnectionCallback([remoteUrl](std::weak_ptr<ix::WebSocket> webSocket,
+                                                   std::shared_ptr<ConnectionState> connectionState,
+                                                   std::unique_ptr<ConnectionInfo> connectionInfo) {
+            auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
+            auto remoteIp = connectionInfo->remoteIp;
+
+            // Server connection
+            state->webSocket().setOnMessageCallback(
+                [webSocket, state, remoteIp](const WebSocketMessagePtr& msg) {
+                    if (msg->type == ix::WebSocketMessageType::Close)
+                    {
+                        state->setTerminated();
+                    }
+                    else if (msg->type == ix::WebSocketMessageType::Message)
+                    {
+                        auto ws = webSocket.lock();
+                        if (ws)
+                        {
+                            ws->send(msg->str, msg->binary);
+                        }
+                    }
+                });
+
+            // Client connection
+            auto ws = webSocket.lock();
+            if (ws)
+            {
+                ws->setOnMessageCallback([state, remoteUrl](const WebSocketMessagePtr& msg) {
+                    if (msg->type == ix::WebSocketMessageType::Open)
+                    {
+                        // Connect to the 'real' server
+                        std::string url(remoteUrl);
+                        url += msg->openInfo.uri;
+                        state->webSocket().setUrl(url);
+                        state->webSocket().disableAutomaticReconnection();
+                        state->webSocket().start();
+
+                        // we should sleep here for a bit until we've established the
+                        // connection with the remote server
+                        while (state->webSocket().getReadyState() != ReadyState::Open)
+                        {
+                            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+                        }
+                    }
+                    else if (msg->type == ix::WebSocketMessageType::Close)
+                    {
+                        state->webSocket().close(msg->closeInfo.code, msg->closeInfo.reason);
+                    }
+                    else if (msg->type == ix::WebSocketMessageType::Message)
+                    {
+                        state->webSocket().send(msg->str, msg->binary);
+                    }
+                });
+            }
+        });
+
+        auto res = server.listen();
+        if (!res.first)
+        {
+            return 1;
+        }
+
+        server.start();
+        server.wait();
+
+        return 0;
+    }
+} // namespace ix

ixwebsocket/IXWebSocketProxyServer.h

@@ -0,0 +1,20 @@
+/*
+ *  IXWebSocketProxyServer.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019-2020 Machine Zone, Inc. All rights reserved.
+ */
+#pragma once
+
+#include "IXSocketTLSOptions.h"
+#include <cstdint>
+#include <stddef.h>
+#include <string>
+
+namespace ix
+{
+    int websocket_proxy_server_main(int port,
+                                    const std::string& hostname,
+                                    const ix::SocketTLSOptions& tlsOptions,
+                                    const std::string& remoteUrl,
+                                    bool verbose);
+} // namespace ix

ixwebsocket/IXWebSocketServer.cpp

@@ -71,13 +71,38 @@ namespace ix
         _onConnectionCallback = callback;
     }
 
+    void WebSocketServer::setOnClientMessageCallback(const OnClientMessageCallback& callback)
+    {
+        _onClientMessageCallback = callback;
+    }
+
     void WebSocketServer::handleConnection(std::unique_ptr<Socket> socket,
-                                           std::shared_ptr<ConnectionState> connectionState)
+                                           std::shared_ptr<ConnectionState> connectionState,
+                                           std::unique_ptr<ConnectionInfo> connectionInfo)
     {
         setThreadName("WebSocketServer::" + connectionState->getId());
 
         auto webSocket = std::make_shared<WebSocket>();
-        _onConnectionCallback(webSocket, connectionState);
+        if (_onConnectionCallback)
+        {
+            _onConnectionCallback(webSocket, connectionState, std::move(connectionInfo));
+        }
+        else if (_onClientMessageCallback)
+        {
+            webSocket->setOnMessageCallback(
+                [this, &ws = *webSocket.get(), connectionState, &ci = *connectionInfo.get()](
+                    const WebSocketMessagePtr& msg) {
+                    _onClientMessageCallback(connectionState, ci, ws, msg);
+                });
+        }
+        else
+        {
+            logError(
+                "WebSocketServer Application developer error: No server callback is registerered.");
+            logError("Missing call to setOnConnectionCallback or setOnClientMessageCallback.");
+            connectionState->setTerminated();
+            return;
+        }
 
         webSocket->disableAutomaticReconnection();
 
@@ -111,6 +136,8 @@ namespace ix
             logError(ss.str());
         }
 
+        webSocket->setOnMessageCallback(nullptr);
+
         // Remove this client from our client set
         {
             std::lock_guard<std::mutex> lock(_clientsMutex);

ixwebsocket/IXWebSocketServer.h

@@ -23,7 +23,14 @@ namespace ix
     {
     public:
         using OnConnectionCallback =
-            std::function<void(std::shared_ptr<WebSocket>, std::shared_ptr<ConnectionState>)>;
+            std::function<void(std::weak_ptr<WebSocket>,
+                               std::shared_ptr<ConnectionState>,
+                               std::unique_ptr<ConnectionInfo> connectionInfo)>;
+
+        using OnClientMessageCallback = std::function<void(std::shared_ptr<ConnectionState>,
+                                                           ConnectionInfo&,
+                                                           WebSocket&,
+                                                           const WebSocketMessagePtr&)>;
 
         WebSocketServer(int port = SocketServer::kDefaultPort,
                         const std::string& host = SocketServer::kDefaultHost,
@@ -39,6 +46,7 @@ namespace ix
         void disablePerMessageDeflate();
 
         void setOnConnectionCallback(const OnConnectionCallback& callback);
+        void setOnClientMessageCallback(const OnClientMessageCallback& callback);
 
         // Get all the connected clients
         std::set<std::shared_ptr<WebSocket>> getClients();
@@ -52,6 +60,7 @@ namespace ix
         bool _enablePerMessageDeflate;
 
         OnConnectionCallback _onConnectionCallback;
+        OnClientMessageCallback _onClientMessageCallback;
 
         std::mutex _clientsMutex;
         std::set<std::shared_ptr<WebSocket>> _clients;
@@ -60,7 +69,8 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket> socket,
-                                      std::shared_ptr<ConnectionState> connectionState) final;
+                                      std::shared_ptr<ConnectionState> connectionState,
+                                      std::unique_ptr<ConnectionInfo> connectionInfo);
         virtual size_t getConnectedClientsCount() final;
     };
 } // namespace ix

ixwebsocket/IXWebSocketTransport.cpp

@@ -65,7 +65,6 @@ namespace ix
         , _receivedMessageCompressed(false)
         , _readyState(ReadyState::CLOSED)
         , _closeCode(WebSocketCloseConstants::kInternalErrorCode)
-        , _closeReason(WebSocketCloseConstants::kInternalErrorMessage)
         , _closeWireSize(0)
         , _closeRemote(false)
         , _enablePerMessageDeflate(false)
@@ -77,6 +76,7 @@ namespace ix
         , _pingCount(0)
         , _lastSendPingTimePoint(std::chrono::steady_clock::now())
     {
+        setCloseReason(WebSocketCloseConstants::kInternalErrorMessage);
         _readbuf.resize(kChunkSize);
     }
 
@@ -179,10 +179,12 @@ namespace ix
 
         if (readyState == ReadyState::CLOSED)
         {
-            std::lock_guard<std::mutex> lock(_closeDataMutex);
-            _onCloseCallback(_closeCode, _closeReason, _closeWireSize, _closeRemote);
+            if (_onCloseCallback)
+            {
+                _onCloseCallback(_closeCode, getCloseReason(), _closeWireSize, _closeRemote);
+            }
+            setCloseReason(WebSocketCloseConstants::kInternalErrorMessage);
             _closeCode = WebSocketCloseConstants::kInternalErrorCode;
-            _closeReason = WebSocketCloseConstants::kInternalErrorMessage;
             _closeWireSize = 0;
             _closeRemote = false;
         }
@@ -261,9 +263,10 @@ namespace ix
         {
             // compute lasting delay to wait for next ping / timeout, if at least one set
             auto now = std::chrono::steady_clock::now();
-            lastingTimeoutDelayInMs = (int) std::chrono::duration_cast<std::chrono::milliseconds>(
+            int timeSinceLastPingMs = (int) std::chrono::duration_cast<std::chrono::milliseconds>(
                                           now - _lastSendPingTimePoint)
                                           .count();
+            lastingTimeoutDelayInMs = (1000 * _pingIntervalSecs) - timeSinceLastPingMs;
         }
 
 #ifdef _WIN32
@@ -326,9 +329,10 @@ namespace ix
         return _txbuf.empty();
     }
 
+    template<class Iterator>
     void WebSocketTransport::appendToSendBuffer(const std::vector<uint8_t>& header,
-                                                std::string::const_iterator begin,
-                                                std::string::const_iterator end,
+                                                Iterator begin,
+                                                Iterator end,
                                                 uint64_t message_size,
                                                 uint8_t masking_key[4])
     {
@@ -638,11 +642,7 @@ namespace ix
                 {
                     // we got the CLOSE frame answer from our close, so we can close the connection
                     // if the code/reason are the same
-                    bool identicalReason;
-                    {
-                        std::lock_guard<std::mutex> lock(_closeDataMutex);
-                        identicalReason = _closeCode == code && _closeReason == reason;
-                    }
+                    bool identicalReason = _closeCode == code && getCloseReason() == reason;
 
                     if (identicalReason)
                     {
@@ -750,8 +750,9 @@ namespace ix
         return static_cast<unsigned>(seconds);
     }
 
+    template<class T>
     WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
-                                                   const std::string& message,
+                                                   const T& message,
                                                    bool compress,
                                                    const OnProgressCallback& onProgressCallback)
     {
@@ -764,8 +765,8 @@ namespace ix
         size_t wireSize = message.size();
         bool compressionError = false;
 
-        std::string::const_iterator message_begin = message.begin();
-        std::string::const_iterator message_end = message.end();
+        auto message_begin = message.cbegin();
+        auto message_end = message.cend();
 
         if (compress)
         {
@@ -780,8 +781,8 @@ namespace ix
             compressionError = false;
             wireSize = _compressedMessage.size();
 
-            message_begin = _compressedMessage.begin();
-            message_end = _compressedMessage.end();
+            message_begin = _compressedMessage.cbegin();
+            message_end = _compressedMessage.cend();
         }
 
         {
@@ -795,6 +796,11 @@ namespace ix
         if (wireSize < kChunkSize)
         {
             success = sendFragment(type, true, message_begin, message_end, compress);
+
+            if (onProgressCallback)
+            {
+                onProgressCallback(0, 1);
+            }
         }
         else
         {
@@ -859,10 +865,11 @@ namespace ix
         return WebSocketSendInfo(success, compressionError, payloadSize, wireSize);
     }
 
+    template<class Iterator>
     bool WebSocketTransport::sendFragment(wsheader_type::opcode_type type,
                                           bool fin,
-                                          std::string::const_iterator message_begin,
-                                          std::string::const_iterator message_end,
+                                          Iterator message_begin,
+                                          Iterator message_end,
                                           bool compress)
     {
         uint64_t message_size = static_cast<uint64_t>(message_end - message_begin);
@@ -1055,7 +1062,7 @@ namespace ix
         else
         {
             // no close code/reason set
-            sendData(wsheader_type::CLOSE, "", compress);
+            sendData(wsheader_type::CLOSE, std::string(""), compress);
         }
     }
 
@@ -1078,13 +1085,10 @@ namespace ix
     {
         closeSocket();
 
-        {
-            std::lock_guard<std::mutex> lock(_closeDataMutex);
+        setCloseReason(reason);
         _closeCode = code;
-            _closeReason = reason;
         _closeWireSize = closeWireSize;
         _closeRemote = remote;
-        }
 
         setReadyState(ReadyState::CLOSED);
         _requestInitCancellation = false;
@@ -1104,13 +1108,11 @@ namespace ix
             closeWireSize = reason.size();
         }
 
-        {
-            std::lock_guard<std::mutex> lock(_closeDataMutex);
+        setCloseReason(reason);
         _closeCode = code;
-            _closeReason = reason;
         _closeWireSize = closeWireSize;
         _closeRemote = remote;
-        }
+
         {
             std::lock_guard<std::mutex> lock(_closingTimePointMutex);
             _closingTimePoint = std::chrono::steady_clock::now();
@@ -1155,4 +1157,15 @@ namespace ix
         return true;
     }
 
+    void WebSocketTransport::setCloseReason(const std::string& reason)
+    {
+        std::lock_guard<std::mutex> lock(_closeReasonMutex);
+        _closeReason = reason;
+    }
+
+    const std::string& WebSocketTransport::getCloseReason() const
+    {
+        std::lock_guard<std::mutex> lock(_closeReasonMutex);
+        return _closeReason;
+    }
 } // namespace ix

ixwebsocket/IXWebSocketTransport.h

@@ -178,11 +178,11 @@ namespace ix
         std::atomic<ReadyState> _readyState;
 
         OnCloseCallback _onCloseCallback;
-        uint16_t _closeCode;
         std::string _closeReason;
-        size_t _closeWireSize;
-        bool _closeRemote;
-        mutable std::mutex _closeDataMutex;
+        mutable std::mutex _closeReasonMutex;
+        std::atomic<uint16_t> _closeCode;
+        std::atomic<size_t> _closeWireSize;
+        std::atomic<bool> _closeRemote;
 
         // Data used for Per Message Deflate compression (with zlib)
         WebSocketPerMessageDeflatePtr _perMessageDeflate;
@@ -239,16 +239,15 @@ namespace ix
         bool sendOnSocket();
         bool receiveFromSocket();
 
+        template<class T>
         WebSocketSendInfo sendData(wsheader_type::opcode_type type,
-                                   const std::string& message,
+                                   const T& message,
                                    bool compress,
                                    const OnProgressCallback& onProgressCallback = nullptr);
 
-        bool sendFragment(wsheader_type::opcode_type type,
-                          bool fin,
-                          std::string::const_iterator begin,
-                          std::string::const_iterator end,
-                          bool compress);
+        template<class Iterator>
+        bool sendFragment(
+            wsheader_type::opcode_type type, bool fin, Iterator begin, Iterator end, bool compress);
 
         void emitMessage(MessageKind messageKind,
                          const std::string& message,
@@ -256,9 +255,11 @@ namespace ix
                          const OnMessageCallback& onMessageCallback);
 
         bool isSendBufferEmpty() const;
+
+        template<class Iterator>
         void appendToSendBuffer(const std::vector<uint8_t>& header,
-                                std::string::const_iterator begin,
-                                std::string::const_iterator end,
+                                Iterator begin,
+                                Iterator end,
                                 uint64_t message_size,
                                 uint8_t masking_key[4]);
 
@@ -266,5 +267,8 @@ namespace ix
         void unmaskReceiveBuffer(const wsheader_type& ws);
 
         std::string getMergedChunks() const;
+
+        void setCloseReason(const std::string& reason);
+        const std::string& getCloseReason() const;
     };
 } // namespace ix

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "9.8.5"
+#define IX_WEBSOCKET_VERSION "10.1.1"

ixwebsocket/apple/IXSetThreadName_apple.cpp

@@ -1,20 +0,0 @@
-/*
- *  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

@@ -1,16 +0,0 @@
-/*
- *  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

@@ -1,20 +0,0 @@
-/*
- *  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

@@ -1,46 +0,0 @@
-/*
- *  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

@@ -34,7 +34,7 @@ ws:
 	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && ninja install)
 
 ws_install:
-	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && make -j 4 install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_ZLIB=0 -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && ninja install)
 
 ws_openssl:
 	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_OPEN_SSL=1 .. ; make -j 4)
@@ -104,8 +104,10 @@ test_server:
 # env TEST=Websocket_server make test
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
-test:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; ninja install)
+build_test:
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. ; ninja install)
+
+test: build_test
 	(cd test ; python2.7 run.py -r)
 
 test_make:
@@ -172,7 +174,7 @@ autobahn_report:
 	cp -rvf ~/sandbox/reports/clients/* ../bsergean.github.io/IXWebSocket/autobahn/
 
 httpd:
-	clang++ --std=c++14 --stdlib=libc++ httpd.cpp \
+	clang++ --std=c++14 --stdlib=libc++ -o ixhttpd httpd.cpp \
 		ixwebsocket/IXSelectInterruptFactory.cpp \
 		ixwebsocket/IXCancellationRequest.cpp \
 		ixwebsocket/IXSocketTLSOptions.cpp \
@@ -191,11 +193,11 @@ httpd:
 		ixwebsocket/IXConnectionState.cpp \
 		ixwebsocket/IXUrlParser.cpp \
 		ixwebsocket/IXSelectInterrupt.cpp \
-		ixwebsocket/apple/IXSetThreadName_apple.cpp \
+		ixwebsocket/IXSetThreadName.cpp \
 		-lz
 
 httpd_linux:
-	g++ --std=c++11 -o ixhttpd httpd.cpp -Iixwebsocket \
+	g++ --std=c++14 -o ixhttpd httpd.cpp -Iixwebsocket \
 		ixwebsocket/IXSelectInterruptFactory.cpp \
 		ixwebsocket/IXCancellationRequest.cpp \
 		ixwebsocket/IXSocketTLSOptions.cpp \
@@ -214,7 +216,7 @@ httpd_linux:
 		ixwebsocket/IXConnectionState.cpp \
 		ixwebsocket/IXUrlParser.cpp \
 		ixwebsocket/IXSelectInterrupt.cpp \
-		ixwebsocket/linux/IXSetThreadName_linux.cpp \
+		ixwebsocket/IXSetThreadName.cpp \
 		-lz -lpthread
 	cp -f ixhttpd /usr/local/bin
 
@@ -236,9 +238,12 @@ install_cmake_for_linux:
 doc:
 	mkdocs gh-deploy
 
-change:
+change: format
 	vim ixwebsocket/IXWebSocketVersion.h docs/CHANGELOG.md
 
+commit:
+	git commit -am "`sh tools/extract_latest_change.sh`"
+
 .PHONY: test
 .PHONY: build
 .PHONY: ws

test/CMakeLists.txt

@@ -37,11 +37,11 @@ set (SOURCES
 
   test_runner.cpp
   IXTest.cpp
-  IXGetFreePort.cpp
   ../third_party/msgpack11/msgpack11.cpp
 
   IXSocketTest.cpp
   IXSocketConnectTest.cpp
+  # IXWebSocketLeakTest.cpp # commented until we have a fix for #224
   IXWebSocketServerTest.cpp
   IXWebSocketTestConnectionDisconnection.cpp
   IXUrlParserTest.cpp
@@ -55,6 +55,8 @@ set (SOURCES
   IXSentryClientTest.cpp
   IXWebSocketChatTest.cpp
   IXWebSocketBroadcastTest.cpp
+  IXWebSocketPerMessageDeflateCompressorTest.cpp
+  IXStreamSqlTest.cpp
 )
 
 # Some unittest don't work on windows yet

test/IXCobraMetricsPublisherTest.cpp

@@ -108,7 +108,7 @@ namespace
             }
             else if (event->type == ix::CobraEventType::UnSubscribed)
             {
-                TLogger() << "Subscriber: ununexpected from channel " << event->subscriptionId;
+                TLogger() << "Subscriber: unsubscribed from channel " << event->subscriptionId;
                 if (event->subscriptionId != channel)
                 {
                     TLogger() << "Subscriber: unexpected channel " << event->subscriptionId;

test/IXCobraToSentryBotTest.cpp

@@ -12,8 +12,8 @@
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcobra/IXCobraMetricsPublisher.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixsentry/IXSentryClient.h>
 #include <ixredis/IXRedisServer.h>
+#include <ixsentry/IXSentryClient.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>
@@ -95,13 +95,15 @@ TEST_CASE("Cobra_to_sentry_bot", "[cobra_bots]")
 
         sentryServer.setOnConnectionCallback(
             [](HttpRequestPtr request,
-               std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
+               std::shared_ptr<ConnectionState> /*connectionState*/,
+               std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << request->method << " " << request->headers["User-Agent"] << " "
+                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                   << 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 <ixsentry/IXSentryClient.h>
 #include <ixredis/IXRedisServer.h>
+#include <ixsentry/IXSentryClient.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 <ixsentry/IXSentryClient.h>
 #include <ixredis/IXRedisServer.h>
+#include <ixsentry/IXSentryClient.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>
@@ -92,6 +92,9 @@ 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

@@ -4,9 +4,9 @@
  *  Copyright (c) 2019 Machine Zone. All rights reserved.
  */
 
-#include "IXGetFreePort.h"
 #include "catch.hpp"
 #include <iostream>
+#include <ixwebsocket/IXGetFreePort.h>
 #include <ixwebsocket/IXHttpClient.h>
 #include <ixwebsocket/IXHttpServer.h>
 
@@ -67,7 +67,8 @@ 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

@@ -0,0 +1,42 @@
+/*
+ *  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

@@ -84,13 +84,16 @@ namespace ix
 
     bool startWebSocketEchoServer(ix::WebSocketServer& server)
     {
-        server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback(
-                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
+        server.setOnClientMessageCallback(
+            [&server](std::shared_ptr<ConnectionState> /*connectionState*/,
+                      ConnectionInfo& connectionInfo,
+                      WebSocket& webSocket,
+                      const ix::WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 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)
@@ -106,14 +109,13 @@ namespace ix
                 {
                     for (auto&& client : server.getClients())
                     {
-                            if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             client->send(msg->str, msg->binary);
                         }
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)

test/IXTest.h

@@ -6,9 +6,9 @@
 
 #pragma once
 
-#include "IXGetFreePort.h"
 #include <iostream>
 #include <ixsnake/IXAppConfig.h>
+#include <ixwebsocket/IXGetFreePort.h>
 #include <ixwebsocket/IXSocketTLSOptions.h>
 #include <ixwebsocket/IXWebSocketServer.h>
 #include <mutex>

test/IXWebSocketBroadcastTest.cpp

@@ -189,15 +189,19 @@ namespace
         bool preferTLS = true;
         server.setTLSOptions(makeServerTLSOptions(preferTLS));
 
-        server.setOnConnectionCallback([&server, &connectionId](
-                                           std::shared_ptr<ix::WebSocket> webSocket,
-                                           std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback([webSocket, connectionState, &connectionId, &server](
+        server.setOnClientMessageCallback(
+            [&server, &connectionId](std::shared_ptr<ConnectionState> connectionState,
+                                     ConnectionInfo& connectionInfo,
+                                     WebSocket& webSocket,
                                      const ix::WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
+
+
                 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:";
@@ -216,14 +220,13 @@ namespace
                 {
                     for (auto&& client : server.getClients())
                     {
-                        if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             client->send(msg->str, msg->binary);
                         }
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketChatTest.cpp

@@ -193,13 +193,16 @@ namespace
 
     bool startServer(ix::WebSocketServer& server)
     {
-        server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback(
-                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
+        server.setOnClientMessageCallback(
+            [&server](std::shared_ptr<ConnectionState> connectionState,
+                      ConnectionInfo& connectionInfo,
+                      WebSocket& webSocket,
+                      const ix::WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 if (msg->type == ix::WebSocketMessageType::Open)
                 {
                     TLogger() << "New connection";
+                    TLogger() << "remote ip: " << remoteIp;
                     TLogger() << "id: " << connectionState->getId();
                     TLogger() << "Uri: " << msg->openInfo.uri;
                     TLogger() << "Headers:";
@@ -216,14 +219,13 @@ namespace
                 {
                     for (auto&& client : server.getClients())
                     {
-                            if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             client->sendBinary(msg->str);
                         }
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketCloseTest.cpp

@@ -168,19 +168,17 @@ namespace
                      std::mutex& mutexWrite)
     {
         // A dev/null server
-        server.setOnConnectionCallback(
+        server.setOnClientMessageCallback(
             [&receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](
-                std::shared_ptr<ix::WebSocket> webSocket,
-                std::shared_ptr<ConnectionState> connectionState) {
-                webSocket->setOnMessageCallback([webSocket,
-                                                 connectionState,
-                                                 &receivedCloseCode,
-                                                 &receivedCloseReason,
-                                                 &receivedCloseRemote,
-                                                 &mutexWrite](const ix::WebSocketMessagePtr& msg) {
+                std::shared_ptr<ConnectionState> connectionState,
+                ConnectionInfo& connectionInfo,
+                WebSocket& /*webSocket*/,
+                const ix::WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 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:";
@@ -203,7 +201,6 @@ namespace
                     receivedCloseRemote = msg->closeInfo.remote;
                 }
             });
-            });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketLeakTest.cpp

@@ -0,0 +1,183 @@
+/*
+ *  IXWebSocketServer.cpp
+ *  Author: Benjamin Sergeant, @marcelkauf
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXTest.h"
+#include "catch.hpp"
+#include <ixwebsocket/IXWebSocket.h>
+#include <ixwebsocket/IXWebSocketServer.h>
+#include <memory>
+#include <sstream>
+
+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

@@ -0,0 +1,76 @@
+/*
+ *  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

@@ -33,15 +33,19 @@ namespace ix
         };
         server.setConnectionStateFactory(factory);
 
-        server.setOnConnectionCallback([&server, &connectionId](
-                                           std::shared_ptr<ix::WebSocket> webSocket,
-                                           std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback([webSocket, connectionState, &connectionId, &server](
+        server.setOnClientMessageCallback(
+            [&server, &connectionId](std::shared_ptr<ConnectionState> connectionState,
+                                     ConnectionInfo& connectionInfo,
+                                     WebSocket& webSocket,
                                      const ix::WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
+
+
                 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:";
@@ -60,14 +64,13 @@ namespace ix
                 {
                     for (auto&& client : server.getClients())
                     {
-                        if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             client->send(msg->str, msg->binary);
                         }
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketSubProtocolTest.cpp

@@ -16,14 +16,16 @@ using namespace ix;
 
 bool startServer(ix::WebSocketServer& server, std::string& subProtocols)
 {
-    server.setOnConnectionCallback(
-        [&server, &subProtocols](std::shared_ptr<ix::WebSocket> webSocket,
-                                 std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback([webSocket, connectionState, &server, &subProtocols](
+    server.setOnClientMessageCallback(
+        [&server, &subProtocols](std::shared_ptr<ConnectionState> connectionState,
+                                 ConnectionInfo& connectionInfo,
+                                 WebSocket& webSocket,
                                  const ix::WebSocketMessagePtr& msg) {
+            auto remoteIp = connectionInfo.remoteIp;
             if (msg->type == ix::WebSocketMessageType::Open)
             {
                 TLogger() << "New connection";
+                TLogger() << "remote ip: " << remoteIp;
                 TLogger() << "id: " << connectionState->getId();
                 TLogger() << "Uri: " << msg->openInfo.uri;
                 TLogger() << "Headers:";
@@ -42,14 +44,13 @@ bool startServer(ix::WebSocketServer& server, std::string& subProtocols)
             {
                 for (auto&& client : server.getClients())
                 {
-                        if (client != webSocket)
+                    if (client.get() != &webSocket)
                     {
                         client->sendBinary(msg->str);
                     }
                 }
             }
         });
-        });
 
     auto res = server.listen();
     if (!res.first)

test/test_runner.cpp

@@ -10,10 +10,18 @@
 #include <ixwebsocket/IXNetSystem.h>
 #include <spdlog/spdlog.h>
 
+#ifndef _WIN32
+#include <signal.h>
+#endif
+
 int main(int argc, char* argv[])
 {
     ix::initNetSystem();
 
+#ifndef _WIN32
+    signal(SIGPIPE, SIG_IGN);
+#endif
+
     ix::CoreLogger::LogFunc logFunc = [](const char* msg, ix::LogLevel level) {
         switch (level)
         {
@@ -49,6 +57,7 @@ int main(int argc, char* argv[])
         }
     };
     ix::CoreLogger::setLogFunction(logFunc);
+    spdlog::set_level(spdlog::level::debug);
 
     int result = Catch::Session().run(argc, argv);
 

third_party/cpp-linenoise/linenoise.hpp

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

tools/extract_latest_change.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+grep -A 3 '^##' docs/CHANGELOG.md | head -n 3 | tail -n 1

ws/CMakeLists.txt

@@ -20,7 +20,6 @@ option(USE_TLS "Add TLS support" ON)
 
 include_directories(ws .)
 include_directories(ws ..)
-include_directories(ws ../third_party)
 include_directories(ws ../third_party/spdlog/include)
 include_directories(ws ../third_party/cpp-linenoise)
 
@@ -66,7 +65,6 @@ add_executable(ws
   ws_cobra_publish.cpp
   ws_httpd.cpp
   ws_autobahn.cpp
-  ws_proxy_server.cpp
   ws_sentry_minidump_upload.cpp
   ws_dns_lookup.cpp
   ws.cpp)

ws/test_ws.sh

@@ -14,6 +14,7 @@ function cleanup_and_exit {
 }
 
 WITH_TLS=${WITH_TLS:-0}
+BLOCKS=${BLOCKS:-20000}
 
 rm -rf /tmp/ws_test
 mkdir -p /tmp/ws_test
@@ -57,7 +58,7 @@ ws receive "${protocol}127.0.0.1:8090" ${delay} --pidfile /tmp/ws_test/pidfile.r
 
 mkdir -p /tmp/ws_test/send
 cd /tmp/ws_test/send
-dd if=/dev/urandom of=/tmp/ws_test/send/20M_file count=20000 bs=1024
+dd if=/dev/urandom of=/tmp/ws_test/send/20M_file count=$BLOCKS bs=1024
 
 # Start the sender job
 ws send ${client_tls} --pidfile /tmp/ws_test/pidfile.send "${protocol}127.0.0.1:8090" /tmp/ws_test/send/20M_file

ws/ws.cpp

@@ -11,17 +11,18 @@
 
 #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 <ixbots/IXCobraMetricsToRedisBot.h>
-#include <ixredis/IXRedisClient.h>
 #include <ixcore/utils/IXCoreLogger.h>
+#include <ixredis/IXRedisClient.h>
 #include <ixsentry/IXSentryClient.h>
 #include <ixwebsocket/IXNetSystem.h>
 #include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXUserAgent.h>
+#include <ixwebsocket/IXWebSocketProxyServer.h>
 #include <spdlog/sinks/basic_file_sink.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
@@ -74,6 +75,7 @@ int main(int argc, char** argv)
         }
     };
     ix::CoreLogger::setLogFunction(logFunc);
+    spdlog::set_level(spdlog::level::debug);
 
 #ifndef _WIN32
     signal(SIGPIPE, SIG_IGN);
@@ -122,6 +124,7 @@ 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;
@@ -193,8 +196,7 @@ 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,7 +360,8 @@ 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);
@@ -391,6 +394,7 @@ 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);
@@ -482,7 +486,24 @@ int main(int argc, char** argv)
     cobraBotConfig.cobraConfig.socketTLSOptions = tlsOptions;
 
     int ret = 1;
-    if (app.got_subcommand("transfer"))
+    if (app.got_subcommand("connect"))
+    {
+        ret = ix::ws_connect_main(url,
+                                  headers,
+                                  disableAutomaticReconnection,
+                                  disablePerMessageDeflate,
+                                  binaryMode,
+                                  maxWaitBetweenReconnectionRetries,
+                                  tlsOptions,
+                                  subprotocol,
+                                  pingIntervalSecs);
+    }
+    else if (app.got_subcommand("echo_server"))
+    {
+        ret = ix::ws_echo_server_main(
+            port, greetings, hostname, tlsOptions, ipv6, disablePerMessageDeflate, disablePong);
+    }
+    else if (app.got_subcommand("transfer"))
     {
         ret = ix::ws_transfer_main(port, hostname, tlsOptions);
     }
@@ -495,27 +516,10 @@ int main(int argc, char** argv)
         bool enablePerMessageDeflate = false;
         ret = ix::ws_receive_main(url, enablePerMessageDeflate, delayMs, tlsOptions);
     }
-    else if (app.got_subcommand("connect"))
-    {
-        ret = ix::ws_connect_main(url,
-                                  headers,
-                                  disableAutomaticReconnection,
-                                  disablePerMessageDeflate,
-                                  binaryMode,
-                                  maxWaitBetweenReconnectionRetries,
-                                  tlsOptions,
-                                  subprotocol,
-                                  pingIntervalSecs);
-    }
     else if (app.got_subcommand("chat"))
     {
         ret = ix::ws_chat_main(url, user);
     }
-    else if (app.got_subcommand("echo_server"))
-    {
-        ret = ix::ws_echo_server_main(
-            port, greetings, hostname, tlsOptions, ipv6, disablePerMessageDeflate, disablePong);
-    }
     else if (app.got_subcommand("broadcast_server"))
     {
         ret = ix::ws_broadcast_server_main(port, hostname, tlsOptions);
@@ -604,8 +608,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, scriptPath);
         }
     }
     else if (app.got_subcommand("cobra_to_sentry"))
@@ -620,14 +623,12 @@ 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"))
@@ -640,7 +641,8 @@ int main(int argc, char** argv)
                                 verbose,
                                 appsConfigPath,
                                 tlsOptions,
-                                disablePong);
+                                disablePong,
+                                republishChannel);
     }
     else if (app.got_subcommand("httpd"))
     {
@@ -656,7 +658,7 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("proxy_server"))
     {
-        ret = ix::ws_proxy_server_main(port, hostname, tlsOptions, remoteHost, verbose);
+        ret = ix::websocket_proxy_server_main(port, hostname, tlsOptions, remoteHost, verbose);
     }
     else if (app.got_subcommand("upload_minidump"))
     {

ws/ws.h

@@ -64,9 +64,7 @@ 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,
@@ -105,7 +103,8 @@ namespace ix
                       bool verbose,
                       const std::string& appsConfigPath,
                       const ix::SocketTLSOptions& tlsOptions,
-                      bool disablePong);
+                      bool disablePong,
+                      const std::string& republishChannel);
 
     int ws_httpd_main(int port,
                       const std::string& hostname,
@@ -117,12 +116,6 @@ namespace ix
 
     int ws_redis_server_main(int port, const std::string& hostname);
 
-    int ws_proxy_server_main(int port,
-                             const std::string& hostname,
-                             const ix::SocketTLSOptions& tlsOptions,
-                             const std::string& remoteHost,
-                             bool verbose);
-
     int ws_sentry_minidump_upload(const std::string& metadataPath,
                                   const std::string& minidump,
                                   const std::string& project,

ws/ws_broadcast_server.cpp

@@ -20,13 +20,16 @@ namespace ix
         ix::WebSocketServer server(port, hostname);
         server.setTLSOptions(tlsOptions);
 
-        server.setOnConnectionCallback([&server](std::shared_ptr<WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback([webSocket, connectionState, &server](
+        server.setOnClientMessageCallback(
+            [&server](std::shared_ptr<ConnectionState> connectionState,
+                      ConnectionInfo& connectionInfo,
+                      WebSocket& webSocket,
                       const WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 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:");
@@ -60,7 +63,7 @@ namespace ix
 
                     for (auto&& client : server.getClients())
                     {
-                        if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             client->send(msg->str, msg->binary, [](int current, int total) -> bool {
                                 spdlog::info("Step {} out of {}", current, total);
@@ -79,7 +82,6 @@ namespace ix
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)

ws/ws_cobra_metrics_publish.cpp

@@ -8,7 +8,6 @@
 #include <chrono>
 #include <fstream>
 #include <ixcobra/IXCobraMetricsPublisher.h>
-#include <jsoncpp/json/json.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
 #include <thread>

ws/ws_cobra_publish.cpp

@@ -8,7 +8,6 @@
 #include <chrono>
 #include <fstream>
 #include <ixcobra/IXCobraMetricsPublisher.h>
-#include <jsoncpp/json/json.h>
 #include <mutex>
 #include <spdlog/spdlog.h>
 #include <sstream>

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
@@ -200,7 +200,7 @@ namespace ix
             }
             else if (msg->type == ix::WebSocketMessageType::Pong)
             {
-                spdlog::info("Received pong");
+                spdlog::info("Received pong {}", msg->str);
             }
             else
             {

ws/ws_echo_server.cpp

@@ -42,14 +42,16 @@ namespace ix
             server.disablePong();
         }
 
-        server.setOnConnectionCallback(
-            [greetings](std::shared_ptr<ix::WebSocket> webSocket,
-                        std::shared_ptr<ConnectionState> connectionState) {
-                webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, greetings](const WebSocketMessagePtr& msg) {
+        server.setOnClientMessageCallback(
+            [greetings](std::shared_ptr<ConnectionState> connectionState,
+                        ConnectionInfo& connectionInfo,
+                        WebSocket& webSocket,
+                        const WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 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:");
@@ -60,7 +62,7 @@ namespace ix
 
                     if (greetings)
                     {
-                                webSocket->sendText("Welcome !");
+                        webSocket.sendText("Welcome !");
                     }
                 }
                 else if (msg->type == ix::WebSocketMessageType::Close)
@@ -80,10 +82,9 @@ namespace ix
                 else if (msg->type == ix::WebSocketMessageType::Message)
                 {
                     spdlog::info("Received {} bytes", msg->wireSize);
-                            webSocket->send(msg->str, msg->binary);
+                    webSocket.send(msg->str, msg->binary);
                 }
             });
-            });
 
         auto res = server.listen();
         if (!res.first)

ws/ws_proxy_server.cpp

@@ -1,173 +0,0 @@
-/*
- *  ws_proxy_server.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
- */
-
-#include <ixwebsocket/IXWebSocketServer.h>
-#include <spdlog/spdlog.h>
-#include <sstream>
-
-namespace ix
-{
-    class ProxyConnectionState : public ix::ConnectionState
-    {
-    public:
-        ProxyConnectionState()
-            : _connected(false)
-        {
-        }
-
-        ix::WebSocket& webSocket()
-        {
-            return _serverWebSocket;
-        }
-
-        bool isConnected()
-        {
-            return _connected;
-        }
-
-        void setConnected()
-        {
-            _connected = true;
-        }
-
-    private:
-        ix::WebSocket _serverWebSocket;
-        bool _connected;
-    };
-
-    int ws_proxy_server_main(int port,
-                             const std::string& hostname,
-                             const ix::SocketTLSOptions& tlsOptions,
-                             const std::string& remoteUrl,
-                             bool verbose)
-    {
-        spdlog::info("Listening on {}:{}", hostname, port);
-
-        ix::WebSocketServer server(port, hostname);
-        server.setTLSOptions(tlsOptions);
-
-        auto factory = []() -> std::shared_ptr<ix::ConnectionState> {
-            return std::make_shared<ProxyConnectionState>();
-        };
-        server.setConnectionStateFactory(factory);
-
-        server.setOnConnectionCallback([remoteUrl,
-                                        verbose](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState) {
-            auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
-
-            // Server connection
-            state->webSocket().setOnMessageCallback([webSocket, state, verbose](
-                                                        const WebSocketMessagePtr& msg) {
-                if (msg->type == ix::WebSocketMessageType::Open)
-                {
-                    spdlog::info("New connection to remote server");
-                    spdlog::info("id: {}", state->getId());
-                    spdlog::info("Uri: {}", msg->openInfo.uri);
-                    spdlog::info("Headers:");
-                    for (auto it : msg->openInfo.headers)
-                    {
-                        spdlog::info("{}: {}", it.first, it.second);
-                    }
-                }
-                else if (msg->type == ix::WebSocketMessageType::Close)
-                {
-                    spdlog::info("Closed remote server connection: client id {} code {} reason {}",
-                                 state->getId(),
-                                 msg->closeInfo.code,
-                                 msg->closeInfo.reason);
-                    state->setTerminated();
-                }
-                else if (msg->type == ix::WebSocketMessageType::Error)
-                {
-                    spdlog::error("Connection error: {}", msg->errorInfo.reason);
-                    spdlog::error("#retries: {}", msg->errorInfo.retries);
-                    spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
-                    spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
-                }
-                else if (msg->type == ix::WebSocketMessageType::Message)
-                {
-                    spdlog::info("Received {} bytes from server", msg->wireSize);
-                    if (verbose)
-                    {
-                        spdlog::info("payload {}", msg->str);
-                    }
-
-                    webSocket->send(msg->str, msg->binary);
-                }
-            });
-
-            // Client connection
-            webSocket->setOnMessageCallback(
-                [state, remoteUrl, verbose](const WebSocketMessagePtr& msg) {
-                    if (msg->type == ix::WebSocketMessageType::Open)
-                    {
-                        spdlog::info("New connection from client");
-                        spdlog::info("id: {}", state->getId());
-                        spdlog::info("Uri: {}", msg->openInfo.uri);
-                        spdlog::info("Headers:");
-                        for (auto it : msg->openInfo.headers)
-                        {
-                            spdlog::info("{}: {}", it.first, it.second);
-                        }
-
-                        // Connect to the 'real' server
-                        std::string url(remoteUrl);
-                        url += msg->openInfo.uri;
-                        state->webSocket().setUrl(url);
-                        state->webSocket().disableAutomaticReconnection();
-                        state->webSocket().start();
-
-                        // we should sleep here for a bit until we've established the
-                        // connection with the remote server
-                        while (state->webSocket().getReadyState() != ReadyState::Open)
-                        {
-                            spdlog::info("waiting for server connection establishment");
-                            std::this_thread::sleep_for(std::chrono::milliseconds(10));
-                        }
-                        spdlog::info("server connection established");
-                    }
-                    else if (msg->type == ix::WebSocketMessageType::Close)
-                    {
-                        spdlog::info("Closed client connection: client id {} code {} reason {}",
-                                     state->getId(),
-                                     msg->closeInfo.code,
-                                     msg->closeInfo.reason);
-                        state->webSocket().close(msg->closeInfo.code, msg->closeInfo.reason);
-                    }
-                    else if (msg->type == ix::WebSocketMessageType::Error)
-                    {
-                        spdlog::error("Connection error: {}", msg->errorInfo.reason);
-                        spdlog::error("#retries: {}", msg->errorInfo.retries);
-                        spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
-                        spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
-                    }
-                    else if (msg->type == ix::WebSocketMessageType::Message)
-                    {
-                        spdlog::info("Received {} bytes from client", msg->wireSize);
-                        if (verbose)
-                        {
-                            spdlog::info("payload {}", msg->str);
-                        }
-
-                        state->webSocket().send(msg->str, msg->binary);
-                    }
-                });
-        });
-
-        auto res = server.listen();
-        if (!res.first)
-        {
-            spdlog::info(res.second);
-            return 1;
-        }
-
-        server.start();
-        server.wait();
-
-        return 0;
-    }
-} // namespace ix

ws/ws_redis_cli.cpp

@@ -4,17 +4,15 @@
  *  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))
@@ -71,9 +69,7 @@ namespace ix
             {
                 if (response.first != RespType::String)
                 {
-                    std::cout << "("
-                              << redisClient.getRespTypeDescription(response.first)
-                              << ")"
+                    std::cout << "(" << redisClient.getRespTypeDescription(response.first) << ")"
                               << " ";
                 }
 

ws/ws_sentry_minidump_upload.cpp

@@ -6,7 +6,6 @@
 
 #include <fstream>
 #include <ixsentry/IXSentryClient.h>
-#include <jsoncpp/json/json.h>
 #include <spdlog/spdlog.h>
 #include <sstream>
 

ws/ws_snake.cpp

@@ -45,7 +45,8 @@ namespace ix
                       bool verbose,
                       const std::string& appsConfigPath,
                       const SocketTLSOptions& socketTLSOptions,
-                      bool disablePong)
+                      bool disablePong,
+                      const std::string& republishChannel)
     {
         snake::AppConfig appConfig;
         appConfig.port = port;
@@ -55,6 +56,7 @@ 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

@@ -19,13 +19,16 @@ namespace ix
         ix::WebSocketServer server(port, hostname);
         server.setTLSOptions(tlsOptions);
 
-        server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-                                                 std::shared_ptr<ConnectionState> connectionState) {
-            webSocket->setOnMessageCallback([webSocket, connectionState, &server](
+        server.setOnClientMessageCallback(
+            [&server](std::shared_ptr<ConnectionState> connectionState,
+                      ConnectionInfo& connectionInfo,
+                      WebSocket& webSocket,
                       const WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
                 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:");
@@ -40,7 +43,7 @@ namespace ix
                                  connectionState->getId(),
                                  msg->closeInfo.code,
                                  msg->closeInfo.reason);
-                    auto remaining = server.getClients().erase(webSocket);
+                    auto remaining = server.getClients().size() - 1;
                     spdlog::info("ws_transfer: {} remaining clients", remaining);
                 }
                 else if (msg->type == ix::WebSocketMessageType::Error)
@@ -62,7 +65,7 @@ namespace ix
                     size_t receivers = 0;
                     for (auto&& client : server.getClients())
                     {
-                        if (client != webSocket)
+                        if (client.get() != &webSocket)
                         {
                             auto readyState = client->getReadyState();
                             auto id = connectionState->getId();
@@ -116,7 +119,6 @@ namespace ix
                     }
                 }
             });
-        });
 
         auto res = server.listen();
         if (!res.first)