embedded help

.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,38 +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: |
-        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 -DUSE_ZLIB=0 ..
-    - 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

@@ -1,20 +0,0 @@
-name: windows
-on:
-  push:
-    paths-ignore:
-    - 'docs/**'
-
-jobs:
-  windows:
-    runs-on: windows-latest
-    steps:
-    - uses: actions/checkout@v1
-    - uses: seanmiddleditch/gha-setup-vsdevenv@master
-    - run: |
-        mkdir build
-        cd build
-        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

.gitignore

@@ -5,4 +5,3 @@ ixsnake/ixsnake/.certs/
 site/
 ws/.certs/
 ws/.srl
-ixhttpd

CMake/FindMbedTLS.cmake

@@ -7,7 +7,7 @@ find_library(MBEDCRYPTO_LIBRARY mbedcrypto)
 set(MBEDTLS_LIBRARIES "${MBEDTLS_LIBRARY}" "${MBEDX509_LIBRARY}" "${MBEDCRYPTO_LIBRARY}")
 
 include(FindPackageHandleStandardArgs)
-find_package_handle_standard_args(MbedTLS DEFAULT_MSG
+find_package_handle_standard_args(MBEDTLS DEFAULT_MSG
     MBEDTLS_INCLUDE_DIRS MBEDTLS_LIBRARY MBEDX509_LIBRARY MBEDCRYPTO_LIBRARY)
 
 mark_as_advanced(MBEDTLS_INCLUDE_DIRS MBEDTLS_LIBRARY MBEDX509_LIBRARY MBEDCRYPTO_LIBRARY)

CMakeLists.txt

@@ -30,15 +30,12 @@ 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
@@ -54,7 +51,6 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXWebSocketPerMessageDeflate.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
-    ixwebsocket/IXWebSocketProxyServer.cpp
     ixwebsocket/IXWebSocketServer.cpp
     ixwebsocket/IXWebSocketTransport.cpp
 )
@@ -62,11 +58,9 @@ 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
@@ -74,7 +68,6 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXProgressCallback.h
     ixwebsocket/IXSelectInterrupt.h
     ixwebsocket/IXSelectInterruptFactory.h
-    ixwebsocket/IXSelectInterruptPipe.h
     ixwebsocket/IXSetThreadName.h
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketConnect.h
@@ -99,13 +92,29 @@ 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)
@@ -171,8 +180,10 @@ if (USE_TLS)
       # set(CMAKE_INCLUDE_PATH ${CMAKE_INCLUDE_PATH} /opt/local/include/openssl-1.0)
     endif()
 
-    # Use OPENSSL_ROOT_DIR CMake variable if you need to use your own openssl
+    # This OPENSSL_FOUND check is to help find a cmake manually configured OpenSSL
-    find_package(OpenSSL REQUIRED) 
+    if (NOT OPENSSL_FOUND)
+      find_package(OpenSSL REQUIRED)
+    endif()
     message(STATUS "OpenSSL: " ${OPENSSL_VERSION})
 
     add_definitions(${OPENSSL_DEFINITIONS})
@@ -190,15 +201,17 @@ if (USE_TLS)
   endif()
 endif()
 
-option(USE_ZLIB "Enable zlib support" TRUE)
+# This ZLIB_FOUND check is to help find a cmake manually configured zlib
-
+if (NOT ZLIB_FOUND)
-if (USE_ZLIB)
+  find_package(ZLIB)
-  # Use ZLIB_ROOT CMake variable if you need to use your own zlib
+endif()
-  find_package(ZLIB REQUIRED)
+if (ZLIB_FOUND)
   include_directories(${ZLIB_INCLUDE_DIRS})
   target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})
-
+else()
-  target_compile_definitions(ixwebsocket PUBLIC IXWEBSOCKET_USE_ZLIB)
+  include_directories(third_party/zlib ${CMAKE_CURRENT_BINARY_DIR}/third_party/zlib)
+  add_subdirectory(third_party/zlib)
+  target_link_libraries(ixwebsocket zlibstatic)
 endif()
 
 if (WIN32)
@@ -225,29 +238,19 @@ if (CMAKE_CXX_COMPILER_ID MATCHES "MSVC")
     target_compile_options(ixwebsocket PRIVATE /MP)
 endif()
 
-target_include_directories(ixwebsocket PUBLIC
+target_include_directories(ixwebsocket PUBLIC ${IXWEBSOCKET_INCLUDE_DIRS})
-  $<BUILD_INTERFACE:${IXWEBSOCKET_INCLUDE_DIRS}/>
-  $<INSTALL_INTERFACE:include/ixwebsocket>
-)
 
 set_target_properties(ixwebsocket PROPERTIES PUBLIC_HEADER "${IXWEBSOCKET_HEADERS}")
 
 install(TARGETS ixwebsocket
-        EXPORT ixwebsocket
+        ARCHIVE DESTINATION ${CMAKE_INSTALL_PREFIX}/lib
-        ARCHIVE DESTINATION lib
+        PUBLIC_HEADER DESTINATION ${CMAKE_INSTALL_PREFIX}/include/ixwebsocket/
-        PUBLIC_HEADER DESTINATION include/ixwebsocket/
 )
 
-install(EXPORT ixwebsocket
-        FILE ixwebsocket-config.cmake
-        NAMESPACE ixwebsocket::
-        DESTINATION lib/cmake/ixwebsocket)
-
 if (USE_WS OR USE_TEST)
   add_subdirectory(ixcore)
   add_subdirectory(ixcrypto)
   add_subdirectory(ixcobra)
-  add_subdirectory(ixredis)
   add_subdirectory(ixsnake)
   add_subdirectory(ixsentry)
   add_subdirectory(ixbots)

Dockerfile

@@ -1 +1 @@
-docker/Dockerfile.alpine
+docker/Dockerfile.centos7_httpd

README.md

@@ -81,7 +81,7 @@ If your company or project is using this library, feel free to open an issue or
 
 - [Machine Zone](https://www.mz.com)
 - [Tokio](https://gitlab.com/HCInk/tokio), a discord library focused on audio playback with node bindings.
-- [libDiscordBot](https://github.com/tostc/libDiscordBot/tree/master), an easy to use Discord-bot framework.
+- [libDiscordBot](https://github.com/tostc/libDiscordBot/tree/master), a work in progress discord library
 - [gwebsocket](https://github.com/norrbotten/gwebsocket), a websocket (lua) module for Garry's Mod
 - [DisCPP](https://github.com/DisCPP/DisCPP), a simple but feature rich Discord API wrapper
 

docker/Dockerfile.alpine

@@ -1,8 +1,8 @@
-FROM alpine:3.12 as build
+FROM alpine:3.11 as build
 
 RUN apk add --no-cache \
     gcc g++ musl-dev linux-headers \
-    cmake mbedtls-dev make zlib-dev python3-dev ninja
+    cmake mbedtls-dev make zlib-dev ninja
 
 RUN addgroup -S app && \
     adduser -S -G app app && \
@@ -18,9 +18,9 @@ USER app
 RUN make ws_mbedtls_install && \
     sh tools/trim_repo_for_docker.sh
 
-FROM alpine:3.12 as runtime
+FROM alpine:3.11 as runtime
 
-RUN apk add --no-cache libstdc++ mbedtls ca-certificates python3 && \
+RUN apk add --no-cache libstdc++ mbedtls ca-certificates && \
     addgroup -S app && \
     adduser -S -G app app
 

docker/Dockerfile.centos7_httpd

@@ -0,0 +1,33 @@
+FROM centos:7 as build
+
+RUN yum install -y gcc-c++ make zlib-devel redhat-rpm-config
+
+RUN groupadd app && useradd -g app app
+RUN chown -R app:app /opt
+RUN chown -R app:app /usr/local
+
+# There is a bug in CMake where we cannot build from the root top folder
+# So we build from /opt
+COPY --chown=app:app . /opt
+WORKDIR /opt
+
+USER app
+RUN [ "make", "httpd_linux" ]
+RUN [ "rm", "-rf", "build" ]
+
+FROM centos:8 as runtime
+
+RUN groupadd app && useradd -g app app
+COPY --chown=app:app --from=build /usr/local/bin/ixhttpd /usr/local/bin/ixhttpd
+RUN chmod +x /usr/local/bin/ixhttpd
+RUN ldd /usr/local/bin/ixhttpd
+
+# Copy source code for gcc
+COPY --chown=app:app --from=build /opt /opt
+
+# Now run in usermode
+USER app
+WORKDIR /home/app
+
+ENTRYPOINT ["ixhttpd"]
+EXPOSE 9999

docker/Dockerfile.ubuntu_groovy

@@ -1,23 +0,0 @@
-# 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,167 +1,6 @@
 # Changelog
-
 All changes to this project will be documented in this file.
 
-## [10.1.5] - 2020-08-02
-
-(ws) Add a new ws sub-command, push_server. This command runs a server which sends many messages in a loop to a websocket client. We can receive above 200,000 messages per second (cf #235).
-
-## [10.1.4] - 2020-08-02
-
-(ws) Add a new ws sub-command, echo_client. This command sends a message to an echo server, and send back to a server whatever message it does receive. When connecting to a local ws echo_server, on my MacBook Pro 2015 I can send/receive around 30,000 messages per second. (cf #235)
-
-## [10.1.3] - 2020-08-02
-
-(ws) ws echo_server. Add a -q option to only enable warning and error log levels. This is useful for bench-marking so that we do not print a lot of things on the console. (cf #235)
-
-## [10.1.2] - 2020-07-31
-
-(build) make using zlib optional, with the caveat that some http and websocket features are not available when zlib is absent
-
-## [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
-
-## [9.8.4] - 2020-06-26
-
-(cobra bots) remove bots which is not required now that we can use Python extensions
-
-## [9.8.3] - 2020-06-25
-
-(cmake) new python code is optional and enabled at cmake time with -DUSE_PYTHON=1
-
-## [9.8.2] - 2020-06-24
-
-(cobra bots) new cobra metrics bot to send data to statsd using Python for processing the message
-
-## [9.8.1] - 2020-06-19
-
-(cobra metrics to statsd bot) fps slow frame info : do not include os name
-
-## [9.8.0] - 2020-06-19
-
-(cobra metrics to statsd bot) send info about memory warnings
-
-## [9.7.9] - 2020-06-18
-
-(http client) fix deadlock when following redirects
-
-## [9.7.8] - 2020-06-18
-
-(cobra metrics to statsd bot) send info about net requests
-
-## [9.7.7] - 2020-06-17
-
-(cobra client and bots) add batch_size subscription option for retrieving multiple messages at once
-
-## [9.7.6] - 2020-06-15
-
-(websocket) WebSocketServer is not a final class, so that users can extend it (fix #215)
-
-## [9.7.5] - 2020-06-15
-
-(cobra bots) minor aesthetic change, in how we display http headers with a : then space as key value separator instead of :: with no space
-
-## [9.7.4] - 2020-06-11
-
-(cobra metrics to statsd bot) change from a statsd type of gauge to a timing one
-
-## [9.7.3] - 2020-06-11
-
-(redis cobra bots) capture most used devices in a zset
-
-## [9.7.2] - 2020-06-11
-
-(ws) add bare bone redis-cli like sub-command, with command line editing powered by libnoise
-
-## [9.7.1] - 2020-06-11
-
-(redis cobra bots) ws cobra metrics to redis / hostname invalid parsing
-
-## [9.7.0] - 2020-06-11
-
-(redis cobra bots) xadd with maxlen + fix bug in xadd client implementation and ws cobra metrics to redis command argument parsing
-
-## [9.6.9] - 2020-06-10
-
-(redis cobra bots) update the cobra to redis bot to use the bot framework, and change it to report fps metrics into redis streams.
-
-## [9.6.6] - 2020-06-04
-
-(statsd cobra bots) statsd improvement: prefix does not need a dot as a suffix, message size can be larger than 256 bytes, error handling was invalid, use core logger for logging instead of std::cerr
-
 ## [9.6.5] - 2020-05-29
 
 (http server) support gzip compression

docs/build.md

@@ -17,15 +17,13 @@ There is a unittest which can be executed by typing `make test`.
 
 Options for building:
 
-* `-DUSE_ZLIB=1` will enable zlib support, required for http client + server + websocket per message deflate extension
 * `-DUSE_TLS=1` will enable TLS support
 * `-DUSE_OPEN_SSL=1` will use [openssl](https://www.openssl.org/) for the TLS support (default on Linux and Windows)
 * `-DUSE_MBED_TLS=1` will use [mbedlts](https://tls.mbed.org/) for the TLS support
 * `-DUSE_WS=1` will build the ws interactive command line tool
 * `-DUSE_TEST=1` will build the unittest
-* `-DUSE_PYTHON=1` will use Python3 for cobra bots, require Python3 to be installed.
 
-If you are on Windows, look at the [appveyor](https://github.com/machinezone/IXWebSocket/blob/master/appveyor.yml) file (not maintained much though) or rather the [github actions](https://github.com/machinezone/IXWebSocket/blob/master/.github/workflows/unittest_windows.yml) which have instructions for building dependencies.
+If you are on Windows, look at the [appveyor](https://github.com/machinezone/IXWebSocket/blob/master/appveyor.yml) file (not maintained much though) or rather the [github actions](https://github.com/machinezone/IXWebSocket/blob/master/.github/workflows/ccpp.yml#L40) which have instructions for building dependencies.
 
 It is also possible to externally include the project, so that everything is fetched over the wire when you build like so:
 

docs/performance.md

@@ -1,37 +0,0 @@
-
-## WebSocket Client performance
-
-We will run a client and a server on the same machine, connecting to localhost. This bench is run on a MacBook Pro from 2015. We can receive over 200,000 (small) messages per second, another way to put it is that it takes 5 micro-second to receive and process one message. This is an indication about the minimal latency to receive messages.
-
-### Receiving messages
-
-By using the push_server ws sub-command, the server will send the same message in a loop to any connected client.
-
-```
-ws push_server -q --send_msg 'yo'
-```
-
-By using the echo_client ws sub-command, with the -m (mute or no_send), we will display statistics on how many messages we can receive per second.
-
-```
-$ ws echo_client -m ws://localhost:8008
-[2020-08-02 12:31:17.284] [info] ws_echo_client: connected
-[2020-08-02 12:31:17.284] [info] Uri: /
-[2020-08-02 12:31:17.284] [info] Headers:
-[2020-08-02 12:31:17.284] [info] Connection: Upgrade
-[2020-08-02 12:31:17.284] [info] Sec-WebSocket-Accept: byy/pMK2d0PtRwExaaiOnXJTQHo=
-[2020-08-02 12:31:17.284] [info] Server: ixwebsocket/10.1.4 macos ssl/SecureTransport zlib 1.2.11
-[2020-08-02 12:31:17.284] [info] Upgrade: websocket
-[2020-08-02 12:31:17.663] [info] messages received: 0 per second 2595307 total
-[2020-08-02 12:31:18.668] [info] messages received: 79679 per second 2674986 total
-[2020-08-02 12:31:19.668] [info] messages received: 207438 per second 2882424 total
-[2020-08-02 12:31:20.673] [info] messages received: 209207 per second 3091631 total
-[2020-08-02 12:31:21.676] [info] messages received: 216056 per second 3307687 total
-[2020-08-02 12:31:22.680] [info] messages received: 214927 per second 3522614 total
-[2020-08-02 12:31:23.684] [info] messages received: 216960 per second 3739574 total
-[2020-08-02 12:31:24.688] [info] messages received: 215232 per second 3954806 total
-[2020-08-02 12:31:25.691] [info] messages received: 212300 per second 4167106 total
-[2020-08-02 12:31:26.694] [info] messages received: 212501 per second 4379607 total
-[2020-08-02 12:31:27.699] [info] messages received: 212330 per second 4591937 total
-[2020-08-02 12:31:28.702] [info] messages received: 216511 per second 4808448 total
-```

docs/usage.md

@@ -246,10 +246,6 @@ 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>
 
@@ -260,50 +256,39 @@ This api was actually changed to take a weak_ptr<WebSocket> as the first argumen
 ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
-    [&server](std::weak_ptr<WebSocket> webSocket,
+    [&server](std::shared_ptr<WebSocket> webSocket,
-              std::shared_ptr<ConnectionState> connectionState,
+              std::shared_ptr<ConnectionState> connectionState)
-              std::unique_ptr<ConnectionInfo> connectionInfo)
     {
-        std::cout << "Remote ip: " << connectionInfo->remoteIp << std::endl;
+        webSocket->setOnMessageCallback(
-
+            [webSocket, connectionState, &server](const ix::WebSocketMessagePtr msg)
-        auto ws = webSocket.lock();
+            {
-        if (ws)
+                if (msg->type == ix::WebSocketMessageType::Open)
-        {
-            ws->setOnMessageCallback(
-                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr msg)
                 {
-                    if (msg->type == ix::WebSocketMessageType::Open)
+                    std::cerr << "New connection" << std::endl;
+
+                    // A connection state object is available, and has a default id
+                    // You can subclass ConnectionState and pass an alternate factory
+                    // to override it. It is useful if you want to store custom
+                    // attributes per connection (authenticated bool flag, attributes, etc...)
+                    std::cerr << "id: " << connectionState->getId() << std::endl;
+
+                    // The uri the client did connect to.
+                    std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
+
+                    std::cerr << "Headers:" << std::endl;
+                    for (auto it : msg->openInfo.headers)
                     {
-                        std::cout << "New connection" << std::endl;
+                        std::cerr << it.first << ": " << it.second << 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.
-                        auto ws = webSocket.lock();
-                        if (ws)
-                        {
-                            ws->send(msg->str, msg->binary);
-                        }
                     }
                 }
+                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);
+                }
             }
         );
     }
@@ -324,74 +309,6 @@ 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
@@ -500,14 +417,11 @@ If you want to handle how requests are processed, implement the setOnConnectionC
 ```cpp
 setOnConnectionCallback(
     [this](HttpRequestPtr request,
-           std::shared_ptr<ConnectionState> /*connectionState*/,
+           std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr
-           std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr
     {
         // Build a string for the response
         std::stringstream ss;
-        ss << connectionInfo->remoteIp
+        ss << request->method
-           << " "
-           << request->method
            << " "
            << request->uri;
 

httpd.cpp

@@ -6,7 +6,7 @@
  *  Buid with make httpd
  */
 
-#include <ixwebsocket/IXHttpServer.h>
+#include "IXHttpServer.h"
 #include <sstream>
 #include <iostream>
 

ixbots/CMakeLists.txt

@@ -8,8 +8,6 @@ set (IXBOTS_SOURCES
     ixbots/IXCobraToSentryBot.cpp
     ixbots/IXCobraToStatsdBot.cpp
     ixbots/IXCobraToStdoutBot.cpp
-    ixbots/IXCobraMetricsToRedisBot.cpp
-    ixbots/IXCobraToPythonBot.cpp
     ixbots/IXStatsdClient.cpp
 )
 
@@ -19,8 +17,6 @@ set (IXBOTS_HEADERS
     ixbots/IXCobraToSentryBot.h
     ixbots/IXCobraToStatsdBot.h
     ixbots/IXCobraToStdoutBot.h
-    ixbots/IXCobraMetricsToRedisBot.h
-    ixbots/IXCobraToPythonBot.h
     ixbots/IXStatsdClient.h
 )
 
@@ -34,24 +30,14 @@ if (NOT JSONCPP_FOUND)
   set(JSONCPP_INCLUDE_DIRS ../third_party/jsoncpp)
 endif()
 
-if (USE_PYTHON)
-  target_compile_definitions(ixbots PUBLIC IXBOTS_USE_PYTHON)
-  find_package(Python COMPONENTS Development)
-endif()
-
 set(IXBOTS_INCLUDE_DIRS
     .
     ..
     ../ixcore
     ../ixwebsocket
     ../ixcobra
-    ../ixredis
     ../ixsentry
     ${JSONCPP_INCLUDE_DIRS}
     ${SPDLOG_INCLUDE_DIRS})
 
-if (USE_PYTHON)
-  set(IXBOTS_INCLUDE_DIRS ${IXBOTS_INCLUDE_DIRS} ${Python_INCLUDE_DIRS})
-endif()
-
 target_include_directories( ixbots PUBLIC ${IXBOTS_INCLUDE_DIRS} )

ixbots/ixbots/IXCobraBot.cpp

@@ -8,7 +8,6 @@
 
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcore/utils/IXCoreLogger.h>
-#include <ixwebsocket/IXSetThreadName.h>
 
 #include <algorithm>
 #include <chrono>
@@ -29,7 +28,6 @@ namespace ix
         auto runtime = botConfig.runtime;
         auto maxEventsPerMinute = botConfig.maxEventsPerMinute;
         auto limitReceivedEvents = botConfig.limitReceivedEvents;
-        auto batchSize = botConfig.batchSize;
 
         ix::CobraConnection conn;
         conn.configure(config);
@@ -45,7 +43,6 @@ namespace ix
         std::atomic<bool> stop(false);
         std::atomic<bool> throttled(false);
         std::atomic<bool> fatalCobraError(false);
-        std::atomic<bool> stalledConnection(false);
         int minuteCounter = 0;
 
         auto timer = [&sentCount,
@@ -56,9 +53,7 @@ namespace ix
                       &receivedCountPerSecs,
                       &receivedCountPerMinutes,
                       &minuteCounter,
-                      &conn,
                       &stop] {
-            setThreadName("Bot progress");
             while (!stop)
             {
                 //
@@ -75,11 +70,7 @@ namespace ix
                    << sentCountPerSecs
                    << " "
                    << sentCountTotal;
-
+                CoreLogger::info(ss.str());
-                if (conn.isAuthenticated())
-                {
-                    CoreLogger::info(ss.str());
-                }
 
                 receivedCountPerSecs = receivedCount - receivedCountTotal;
                 sentCountPerSecs = sentCount - sentCountTotal;
@@ -102,14 +93,7 @@ namespace ix
 
         std::thread t1(timer);
 
-        auto heartbeat = [&sentCount,
+        auto heartbeat = [&sentCount, &receivedCount, &stop, &enableHeartbeat, &heartBeatTimeout, &fatalCobraError] {
-                          &receivedCount,
-                          &stop,
-                          &enableHeartbeat,
-                          &heartBeatTimeout,
-                          &stalledConnection]
-        {
-            setThreadName("Bot heartbeat");
             std::string state("na");
 
             if (!enableHeartbeat) return;
@@ -124,12 +108,9 @@ namespace ix
 
                 if (currentState == state)
                 {
-                    ss.str("");
+                    CoreLogger::error("no messages received or sent for 1 minute, exiting");
-                    ss << "no messages received or sent for "
+                    fatalCobraError = true;
-                       << heartBeatTimeout << " seconds, reconnecting";
+                    break;
-
-                    CoreLogger::error(ss.str());
-                    stalledConnection = true;
                 }
                 state = currentState;
 
@@ -154,7 +135,6 @@ namespace ix
                                &receivedCountPerMinutes,
                                maxEventsPerMinute,
                                limitReceivedEvents,
-                               batchSize,
                                &fatalCobraError,
                                &sentCount](const CobraEventPtr& event) {
             if (event->type == ix::CobraEventType::Open)
@@ -163,7 +143,7 @@ namespace ix
 
                 for (auto&& it : event->headers)
                 {
-                    CoreLogger::info(it.first + ": " + it.second);
+                    CoreLogger::info(it.first + "::" + it.second);
                 }
             }
             else if (event->type == ix::CobraEventType::Closed)
@@ -176,7 +156,7 @@ namespace ix
                 CoreLogger::info("Subscribing to " + channel);
                 CoreLogger::info("Subscribing at position " + subscriptionPosition);
                 CoreLogger::info("Subscribing with filter " + filter);
-                conn.subscribe(channel, filter, subscriptionPosition, batchSize,
+                conn.subscribe(channel, filter, subscriptionPosition,
                     [&sentCount, &receivedCountPerMinutes,
                      maxEventsPerMinute, limitReceivedEvents,
                      &throttled, &receivedCount,
@@ -251,13 +231,6 @@ namespace ix
                 std::this_thread::sleep_for(duration);
 
                 if (fatalCobraError) break;
-
-                if (stalledConnection)
-                {
-                    conn.disconnect();
-                    conn.connect();
-                    stalledConnection = false;
-                }
             }
         }
         // Run for a duration, used by unittesting now
@@ -269,13 +242,6 @@ namespace ix
                 std::this_thread::sleep_for(duration);
 
                 if (fatalCobraError) break;
-
-                if (stalledConnection)
-                {
-                    conn.disconnect();
-                    conn.connect();
-                    stalledConnection = false;
-                }
             }
         }
 
@@ -299,22 +265,4 @@ namespace ix
     {
         _onBotMessageCallback = callback;
     }
-
-    std::string CobraBot::getDeviceIdentifier(const Json::Value& msg)
-    {
-        std::string deviceId("na");
-
-        auto osName = msg["device"]["os_name"];
-        if (osName == "Android")
-        {
-            deviceId = msg["device"]["model"].asString();
-        }
-        else if (osName == "iOS")
-        {
-            deviceId = msg["device"]["hardware_model"].asString();
-        }
-
-        return deviceId;
-    }
-
 } // namespace ix

ixbots/ixbots/IXCobraBot.h

@@ -28,8 +28,6 @@ namespace ix
         int64_t run(const CobraBotConfig& botConfig);
         void setOnBotMessageCallback(const OnBotMessageCallback& callback);
 
-        std::string getDeviceIdentifier(const Json::Value& msg);
-
     private:
         OnBotMessageCallback _onBotMessageCallback;
     };

ixbots/ixbots/IXCobraBotConfig.h

@@ -27,6 +27,5 @@ namespace ix
         int runtime = -1;
         int maxEventsPerMinute = std::numeric_limits<int>::max();
         bool limitReceivedEvents = false;
-        int batchSize = 1;
     };
 } // namespace ix

ixbots/ixbots/IXCobraMetricsToRedisBot.cpp

@@ -1,149 +0,0 @@
-/*
- *  IXCobraMetricsToRedisBot.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-
-#include "IXCobraMetricsToRedisBot.h"
-
-#include "IXCobraBot.h"
-#include "IXStatsdClient.h"
-#include <chrono>
-#include <ixcobra/IXCobraConnection.h>
-#include <ixcore/utils/IXCoreLogger.h>
-#include <sstream>
-#include <vector>
-#include <algorithm>
-#include <map>
-#include <cctype>
-
-
-namespace
-{
-    std::string removeSpaces(const std::string& str)
-    {
-        std::string out(str);
-        out.erase(
-            std::remove_if(out.begin(), out.end(), [](unsigned char x) { return std::isspace(x); }),
-            out.end());
-
-        return out;
-    }
-}
-
-namespace ix
-{
-    bool processPerfMetricsEventSlowFrames(const Json::Value& msg,
-                                           RedisClient& redisClient,
-                                           const std::string& deviceId)
-    {
-        auto frameRateHistogramCounts = msg["data"]["FrameRateHistogramCounts"];
-
-        int slowFrames = 0;
-        slowFrames += frameRateHistogramCounts[4].asInt();
-        slowFrames += frameRateHistogramCounts[5].asInt();
-        slowFrames += frameRateHistogramCounts[6].asInt();
-        slowFrames += frameRateHistogramCounts[7].asInt();
-
-        //
-        // XADD without a device id
-        //
-        std::stringstream ss;
-        ss << msg["id"].asString() << "_slow_frames" << "."
-           << msg["device"]["game"].asString() << "."
-           << msg["device"]["os_name"].asString() << "."
-           << removeSpaces(msg["data"]["Tag"].asString());
-
-        int maxLen;
-        maxLen = 100000;
-        std::string id = ss.str();
-        std::string errMsg;
-        if (redisClient.xadd(id, std::to_string(slowFrames), maxLen, errMsg).empty())
-        {
-            CoreLogger::info(std::string("redis XADD error: ") + errMsg);
-        }
-
-        //
-        // XADD with a device id
-        //
-        ss.str(""); // reset the stringstream
-        ss << msg["id"].asString() << "_slow_frames_by_device" << "."
-           << deviceId << "."
-           << msg["device"]["game"].asString() << "."
-           << msg["device"]["os_name"].asString() << "."
-           << removeSpaces(msg["data"]["Tag"].asString());
-
-        id = ss.str();
-        maxLen = 1000;
-        if (redisClient.xadd(id, std::to_string(slowFrames), maxLen, errMsg).empty())
-        {
-            CoreLogger::info(std::string("redis XADD error: ") + errMsg);
-        }
-
-        //
-        // Add device to the device zset, and increment the score
-        // so that we know which devices are used more than others
-        // ZINCRBY myzset 1 one
-        //
-        ss.str(""); // reset the stringstream
-        ss << msg["id"].asString() << "_slow_frames_devices" << "."
-           << msg["device"]["game"].asString();
-
-        id = ss.str();
-        std::vector<std::string> args = {
-            "ZINCRBY", id, "1", deviceId
-        };
-        auto response = redisClient.send(args, errMsg);
-        if (response.first == RespType::Error)
-        {
-            CoreLogger::info(std::string("redis ZINCRBY error: ") + errMsg);
-        }
-
-        return true;
-    }
-
-    int64_t cobra_metrics_to_redis_bot(const ix::CobraBotConfig& config,
-                                       RedisClient& redisClient,
-                                       bool verbose)
-    {
-        CobraBot bot;
-
-        bot.setOnBotMessageCallback(
-            [&redisClient, &verbose, &bot]
-             (const Json::Value& msg,
-              const std::string& /*position*/,
-              std::atomic<bool>& /*throttled*/,
-              std::atomic<bool>& /*fatalCobraError*/,
-              std::atomic<uint64_t>& sentCount) -> void {
-            if (msg["device"].isNull())
-            {
-                CoreLogger::info("no device entry, skipping event");
-                return;
-            }
-
-            if (msg["id"].isNull())
-            {
-                CoreLogger::info("no id entry, skipping event");
-                return;
-            }
-
-            //
-            // Display full message with
-            if (verbose)
-            {
-                CoreLogger::info(msg.toStyledString());
-            }
-
-            bool success = false;
-            if (msg["id"].asString() == "engine_performance_metrics_id")
-            {
-                auto deviceId = bot.getDeviceIdentifier(msg);
-                success = processPerfMetricsEventSlowFrames(msg, redisClient, deviceId);
-            }
-
-            if (success) sentCount++;
-        });
-
-        return bot.run(config);
-    }
-} // namespace ix

ixbots/ixbots/IXCobraMetricsToRedisBot.h

@@ -1,20 +0,0 @@
-/*
- *  IXCobraMetricsToRedisBot.h
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-#pragma once
-
-#include <cstdint>
-#include <ixredis/IXRedisClient.h>
-#include "IXCobraBotConfig.h"
-#include <stddef.h>
-#include <string>
-
-namespace ix
-{
-    int64_t cobra_metrics_to_redis_bot(const ix::CobraBotConfig& config,
-                                       RedisClient& redisClient,
-                                       bool verbose);
-} // namespace ix
-

ixbots/ixbots/IXCobraToPythonBot.cpp

@@ -1,332 +0,0 @@
-/*
- *  IXCobraToPythonBot.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
- */
-
-#include "IXCobraToPythonBot.h"
-
-#include "IXCobraBot.h"
-#include "IXStatsdClient.h"
-#include <chrono>
-#include <ixcobra/IXCobraConnection.h>
-#include <ixcore/utils/IXCoreLogger.h>
-#include <sstream>
-#include <vector>
-#include <algorithm>
-#include <map>
-#include <cctype>
-
-//
-// I cannot get Windows to easily build on CI (github action) so support
-// is disabled for now. It should be a simple fix 
-// (linking error about missing debug build)
-//
-
-#ifdef IXBOTS_USE_PYTHON
-#define PY_SSIZE_T_CLEAN
-#include <Python.h>
-#endif
-
-#ifdef IXBOTS_USE_PYTHON
-namespace
-{
-    //
-    // This function is unused at this point. It produce a correct output,
-    // but triggers memory leaks when called repeateadly, as I cannot figure out how to
-    // make the reference counting Python functions to work properly (Py_DECREF and friends)
-    //
-    PyObject* jsonToPythonObject(const Json::Value& val)
-    {
-        switch(val.type())
-        {
-            case Json::nullValue:
-            {
-                return Py_None;
-            }
-
-            case Json::intValue:
-            {
-                return PyLong_FromLong(val.asInt64());
-            }
-
-            case Json::uintValue:
-            {
-                return PyLong_FromLong(val.asUInt64());
-            }
-
-            case Json::realValue:
-            {
-                return PyFloat_FromDouble(val.asDouble());
-            }
-
-            case Json::stringValue:
-            {
-                return PyUnicode_FromString(val.asCString());
-            }
-
-            case Json::booleanValue:
-            {
-                return val.asBool() ? Py_True : Py_False;
-            }
-
-            case Json::arrayValue:
-            {
-                PyObject* list = PyList_New(val.size());
-                Py_ssize_t i = 0;
-                for (auto&& it = val.begin(); it != val.end(); ++it)
-                {
-                    PyList_SetItem(list, i++, jsonToPythonObject(*it));
-                }
-                return list;
-            }
-
-            case Json::objectValue:
-            {
-                PyObject* dict = PyDict_New();
-                for (auto&& it = val.begin(); it != val.end(); ++it)
-                {
-                    PyObject* key = jsonToPythonObject(it.key());
-                    PyObject* value = jsonToPythonObject(*it);
-
-                    PyDict_SetItem(dict, key, value);
-                }
-                return dict;
-            }
-        }
-    }
-}
-#endif
-
-namespace ix
-{
-    int64_t cobra_to_python_bot(const ix::CobraBotConfig& config,
-                                StatsdClient& statsdClient,
-                                const std::string& scriptPath)
-    {
-#ifndef IXBOTS_USE_PYTHON
-        CoreLogger::error("Command is disabled. "
-                          "Needs to be configured with USE_PYTHON=1");
-        return -1;
-#else
-        CobraBot bot;
-        Py_InitializeEx(0); // 0 arg so that we do not install signal handlers 
-                            // which prevent us from using Ctrl-C
-
-        size_t lastIndex = scriptPath.find_last_of("."); 
-        std::string modulePath = scriptPath.substr(0, lastIndex);
-
-        PyObject* pyModuleName = PyUnicode_DecodeFSDefault(modulePath.c_str());
-
-        if (pyModuleName == nullptr)
-        {
-            CoreLogger::error("Python error: Cannot decode file system path");
-            PyErr_Print();
-            return false;
-        }
-
-        // Import module
-        PyObject* pyModule = PyImport_Import(pyModuleName);
-        Py_DECREF(pyModuleName);
-        if (pyModule == nullptr)
-        {
-            CoreLogger::error("Python error: Cannot import module.");
-            CoreLogger::error("Module name cannot countain dash characters.");
-            CoreLogger::error("Is PYTHONPATH set correctly ?");
-            PyErr_Print();
-            return false;
-        }
-
-        // module main funtion name is named 'run'
-        const std::string entryPoint("run");
-        PyObject* pyFunc = PyObject_GetAttrString(pyModule, entryPoint.c_str());
-
-        if (!pyFunc)
-        {
-            CoreLogger::error("run symbol is missing from module.");
-            PyErr_Print();
-            return false;
-        }
-
-        if (!PyCallable_Check(pyFunc))
-        {
-            CoreLogger::error("run symbol is not a function.");
-            PyErr_Print();
-            return false;
-        }
-
-        bot.setOnBotMessageCallback(
-            [&statsdClient, pyFunc]
-                (const Json::Value& msg,
-                 const std::string& /*position*/,
-                 std::atomic<bool>& /*throttled*/,
-                 std::atomic<bool>& fatalCobraError,
-                 std::atomic<uint64_t>& sentCount) -> void {
-            //
-            // Invoke python script here. First build function parameters, a tuple
-            //
-            const int kVersion = 1; // We can bump this and let the interface evolve
-
-            PyObject *pyArgs = PyTuple_New(2);
-            PyTuple_SetItem(pyArgs, 0, PyLong_FromLong(kVersion)); // First argument
-
-            //
-            // It would be better to create a Python object (a dictionary) 
-            // from the json msg, but it is simpler to serialize it to a string
-            // and decode it on the Python side of the fence
-            //
-            PyObject* pySerializedJson = PyUnicode_FromString(msg.toStyledString().c_str());
-            PyTuple_SetItem(pyArgs, 1, pySerializedJson); // Second argument
-
-            // Invoke the python routine
-            PyObject* pyList = PyObject_CallObject(pyFunc, pyArgs);
-
-            // Error calling the function
-            if (pyList == nullptr)
-            {
-                fatalCobraError = true;
-                CoreLogger::error("run() function call failed. Input msg: ");
-                auto serializedMsg = msg.toStyledString();
-                CoreLogger::error(serializedMsg);
-                PyErr_Print();
-                CoreLogger::error("================");
-                return;
-            }
-
-            // Invalid return type
-            if (!PyList_Check(pyList))
-            {
-                fatalCobraError = true;
-                CoreLogger::error("run() return type should be a list");
-                return;
-            }
-
-            // The result is a list of dict containing sufficient info 
-            // to send messages to statsd
-            auto listSize = PyList_Size(pyList);
-            
-            for (Py_ssize_t i = 0 ; i < listSize; ++i)
-            {
-                PyObject* dict = PyList_GetItem(pyList, i);
-
-                // Make sure this is a dict
-                if (!PyDict_Check(dict))
-                {
-                    fatalCobraError = true;
-                    CoreLogger::error("list element is not a dict");
-                    continue;
-                }
-
-                //
-                // Retrieve object kind
-                //
-                PyObject* pyKind = PyDict_GetItemString(dict, "kind");
-                if (!PyUnicode_Check(pyKind))
-                {
-                    fatalCobraError = true;
-                    CoreLogger::error("kind entry is not a string");
-                    continue;
-                }
-                std::string kind(PyUnicode_AsUTF8(pyKind));
-
-                bool counter = false;
-                bool gauge = false;
-                bool timing = false;
-
-                if (kind == "counter")
-                {
-                    counter = true;
-                }
-                else if (kind == "gauge")
-                {
-                    gauge = true;
-                }
-                else if (kind == "timing")
-                {
-                    timing = true;
-                }
-                else
-                {
-                    fatalCobraError = true;
-                    CoreLogger::error(std::string("invalid kind entry: ") + kind +
-                                      ". Supported ones are counter, gauge, timing");
-                    continue;
-                }
-
-                //
-                // Retrieve object key
-                //
-                PyObject* pyKey = PyDict_GetItemString(dict, "key");
-                if (!PyUnicode_Check(pyKey))
-                {
-                    fatalCobraError = true;
-                    CoreLogger::error("key entry is not a string");
-                    continue;
-                }
-                std::string key(PyUnicode_AsUTF8(pyKey));
-
-                //
-                // Retrieve object value and send data to statsd
-                //
-                PyObject* pyValue = PyDict_GetItemString(dict, "value");
-
-                // Send data to statsd
-                if (PyFloat_Check(pyValue))
-                {
-                    double value = PyFloat_AsDouble(pyValue);
-
-                    if (counter)
-                    {
-                        statsdClient.count(key, value);
-                    }
-                    else if (gauge)
-                    {
-                        statsdClient.gauge(key, value);
-                    }
-                    else if (timing)
-                    {
-                        statsdClient.timing(key, value);
-                    }
-                }
-                else if (PyLong_Check(pyValue))
-                {
-                    long value = PyLong_AsLong(pyValue);
-
-                    if (counter)
-                    {
-                        statsdClient.count(key, value);
-                    }
-                    else if (gauge)
-                    {
-                        statsdClient.gauge(key, value);
-                    }
-                    else if (timing)
-                    {
-                        statsdClient.timing(key, value);
-                    }
-                }
-                else
-                {
-                    fatalCobraError = true;
-                    CoreLogger::error("value entry is neither an int or a float");
-                    continue;
-                }
-
-                sentCount++; // should we update this for each statsd object sent ?
-            }
-
-            Py_DECREF(pyArgs);
-            Py_DECREF(pyList);
-        });
-
-        bool status = bot.run(config);
-
-        // Cleanup - we should do something similar in all exit case ...
-        Py_DECREF(pyFunc);
-        Py_DECREF(pyModule);
-        Py_FinalizeEx();
-
-        return status;
-#endif
-    }
-}

ixbots/ixbots/IXCobraToPythonBot.h

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

ixbots/ixbots/IXCobraToStatsdBot.cpp

@@ -70,17 +70,11 @@ namespace ix
                                                      std::atomic<bool>& fatalCobraError,
                                                      std::atomic<uint64_t>& sentCount) -> void {
                 std::string id;
-                size_t idx = 0;
                 for (auto&& attr : tokens)
                 {
+                    id += ".";
                     auto val = extractAttr(attr, msg);
                     id += val.asString();
-
-                    // We add a dot separator unless we are processing the last token
-                    if (idx++ != tokens.size() - 1)
-                    {
-                        id += ".";
-                    }
                 }
 
                 if (gauge.empty() && timer.empty())

ixbots/ixbots/IXStatsdClient.cpp

@@ -39,28 +39,21 @@
 
 #include "IXStatsdClient.h"
 
+#include <iostream>
 #include <ixwebsocket/IXNetSystem.h>
-#include <ixwebsocket/IXSetThreadName.h>
+#include <stdio.h>
-#include <ixcore/utils/IXCoreLogger.h>
-#include <sstream>
 #include <stdlib.h>
 #include <string.h>
 
 namespace ix
 {
-    StatsdClient::StatsdClient(const std::string& host,
+    StatsdClient::StatsdClient(const std::string& host, int port, const std::string& prefix)
-                               int port,
-                               const std::string& prefix,
-                               bool verbose)
         : _host(host)
         , _port(port)
         , _prefix(prefix)
         , _stop(false)
-        , _verbose(verbose)
     {
         _thread = std::thread([this] {
-            setThreadName("Statsd");
-
             while (!_stop)
             {
                 flushQueue();
@@ -122,15 +115,11 @@ namespace ix
     {
         cleanup(key);
 
-        std::stringstream ss;
+        char buf[256];
-        ss << _prefix << "." << key << ":" << value << "|" << type;
+        snprintf(
+            buf, sizeof(buf), "%s%s:%zd|%s\n", _prefix.c_str(), key.c_str(), value, type.c_str());
 
-        if (_verbose)
+        enqueue(buf);
-        {
-            CoreLogger::info(ss.str());
-        }
-
-        enqueue(ss.str() + "\n");
         return 0;
     }
 
@@ -148,13 +137,10 @@ namespace ix
         {
             auto message = _queue.front();
             auto ret = _socket.sendto(message);
-            if (ret == -1)
+            if (ret != 0)
             {
-                CoreLogger::error(std::string("statsd error: ") + strerror(UdpSocket::getErrno()));
+                std::cerr << "error: " << strerror(UdpSocket::getErrno()) << std::endl;
             }
-
-            // we always dequeue regardless of the ability to send the message
-            // so that we keep our queue size under control
             _queue.pop_front();
         }
     }

ixbots/ixbots/IXStatsdClient.h

@@ -20,8 +20,7 @@ namespace ix
     public:
         StatsdClient(const std::string& host = "127.0.0.1",
                      int port = 8125,
-                     const std::string& prefix = "",
+                     const std::string& prefix = "");
-                     bool verbose = false);
         ~StatsdClient();
 
         bool init(std::string& errMsg);
@@ -53,7 +52,6 @@ namespace ix
         std::mutex _mutex; // for the queue
 
         std::deque<std::string> _queue;
-        bool _verbose;
     };
 
 } // end namespace ix

ixcobra/ixcobra/IXCobraConnection.cpp

@@ -111,12 +111,6 @@ namespace ix
 
     void CobraConnection::disconnect()
     {
-        auto subscriptionIds = getSubscriptionsIds();
-        for (auto&& subscriptionId : subscriptionIds)
-        {
-            unsubscribe(subscriptionId);
-        }
-
         _authenticated = false;
         _webSocket->stop();
     }
@@ -568,13 +562,11 @@ namespace ix
     void CobraConnection::subscribe(const std::string& channel,
                                     const std::string& filter,
                                     const std::string& position,
-                                    int batchSize,
                                     SubscriptionCallback cb)
     {
         // Create and send a subscribe pdu
         Json::Value body;
         body["channel"] = channel;
-        body["batch_size"] = batchSize;
 
         if (!filter.empty())
         {
@@ -620,18 +612,6 @@ 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

@@ -88,7 +88,6 @@ namespace ix
         void subscribe(const std::string& channel,
                        const std::string& filter = std::string(),
                        const std::string& position = std::string(),
-                       int batchSize = 1,
                        SubscriptionCallback cb = nullptr);
 
         /// Unsubscribe from a channel
@@ -163,9 +162,6 @@ 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/CMakeLists.txt

@@ -1,27 +0,0 @@
-#
-# Author: Benjamin Sergeant
-# Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
-#
-
-set (IXREDIS_SOURCES
-    ixredis/IXRedisClient.cpp
-    ixredis/IXRedisServer.cpp
-)
-
-set (IXREDIS_HEADERS
-    ixredis/IXRedisClient.h
-    ixredis/IXRedisServer.h
-)
-
-add_library(ixredis STATIC
-    ${IXREDIS_SOURCES}
-    ${IXREDIS_HEADERS}
-)
-
-set(IXREDIS_INCLUDE_DIRS
-    .
-    ..
-    ../ixcore
-    ../ixwebsocket)
-
-target_include_directories( ixredis PUBLIC ${IXREDIS_INCLUDE_DIRS} )

ixsnake/CMakeLists.txt

@@ -7,14 +7,16 @@ set (IXSNAKE_SOURCES
     ixsnake/IXSnakeServer.cpp
     ixsnake/IXSnakeProtocol.cpp
     ixsnake/IXAppConfig.cpp
-    ixsnake/IXStreamSql.cpp
+    ixsnake/IXRedisClient.cpp
+    ixsnake/IXRedisServer.cpp
 )
 
 set (IXSNAKE_HEADERS
     ixsnake/IXSnakeServer.h
     ixsnake/IXSnakeProtocol.h
     ixsnake/IXAppConfig.h
-    ixsnake/IXStreamSql.h
+    ixsnake/IXRedisClient.h
+    ixsnake/IXRedisServer.h
 )
 
 add_library(ixsnake STATIC
@@ -28,7 +30,6 @@ set(IXSNAKE_INCLUDE_DIRS
     ../ixcore
     ../ixcrypto
     ../ixwebsocket
-    ../ixredis
     ../third_party)
 
 target_include_directories( ixsnake PUBLIC ${IXSNAKE_INCLUDE_DIRS} )

ixsnake/ixsnake/IXAppConfig.cpp

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

ixsnake/ixsnake/IXRedisClient.cpp

@@ -9,7 +9,6 @@
 #include <cstring>
 #include <iomanip>
 #include <iostream>
-#include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXSocketFactory.h>
 #include <ixwebsocket/IXSocketTLSOptions.h>
 #include <sstream>
@@ -251,16 +250,12 @@ namespace ix
     }
 
     std::string RedisClient::prepareXaddCommand(const std::string& stream,
-                                                const std::string& message,
+                                                const std::string& message)
-                                                int maxLen)
     {
         std::stringstream ss;
-        ss << "*8\r\n";
+        ss << "*5\r\n";
         ss << writeString("XADD");
         ss << writeString(stream);
-        ss << writeString("MAXLEN");
-        ss << writeString("~");
-        ss << writeString(std::to_string(maxLen));
         ss << writeString("*");
         ss << writeString("field");
         ss << writeString(message);
@@ -270,7 +265,6 @@ namespace ix
 
     std::string RedisClient::xadd(const std::string& stream,
                                   const std::string& message,
-                                  int maxLen,
                                   std::string& errMsg)
     {
         errMsg.clear();
@@ -281,7 +275,7 @@ namespace ix
             return std::string();
         }
 
-        std::string command = prepareXaddCommand(stream, message, maxLen);
+        std::string command = prepareXaddCommand(stream, message);
 
         bool sent = _socket->writeBytes(command, nullptr);
         if (!sent)
@@ -354,104 +348,4 @@ namespace ix
 
         return success;
     }
-
-    std::pair<RespType, std::string> RedisClient::send(
-        const std::vector<std::string>& args,
-        std::string& errMsg)
-    {
-        std::stringstream ss;
-        ss << "*";
-        ss << std::to_string(args.size());
-        ss << "\r\n";
-
-        for (auto&& arg : args)
-        {
-            ss << writeString(arg);
-        }
-
-        bool sent = _socket->writeBytes(ss.str(), nullptr);
-        if (!sent)
-        {
-            errMsg = "Cannot write bytes to socket";
-            return std::make_pair(RespType::Error, "");
-        }
-
-        return readResponse(errMsg);
-    }
-
-    std::pair<RespType, std::string> RedisClient::readResponse(std::string& errMsg)
-    {
-        // Read result
-        auto pollResult = _socket->isReadyToRead(-1);
-        if (pollResult == PollResultType::Error)
-        {
-            errMsg = "Error while polling for result";
-            return std::make_pair(RespType::Error, "");
-        }
-
-        // First line is the string length
-        auto lineResult = _socket->readLine(nullptr);
-        auto lineValid = lineResult.first;
-        auto line = lineResult.second;
-
-        if (!lineValid)
-        {
-            errMsg = "Error while polling for result";
-            return std::make_pair(RespType::Error, "");
-        }
-
-        std::string response;
-
-        if (line[0] == '+') // Simple string
-        {
-            std::stringstream ss;
-            response = line.substr(1, line.size() - 3);
-            return std::make_pair(RespType::String, response);
-        }
-        else if (line[0] == '-') // Errors
-        {
-            std::stringstream ss;
-            response = line.substr(1, line.size() - 3);
-            return std::make_pair(RespType::Error, response);
-        }
-        else if (line[0] == ':') // Integers
-        {
-            std::stringstream ss;
-            response = line.substr(1, line.size() - 3);
-            return std::make_pair(RespType::Integer, response);
-        }
-        else if (line[0] == '$') // Bulk strings
-        {
-            int stringSize;
-            {
-                std::stringstream ss;
-                ss << line.substr(1, line.size() - 1);
-                ss >> stringSize;
-            }
-
-            // Read the result, which is the stream id computed by the redis server
-            lineResult = _socket->readLine(nullptr);
-            lineValid = lineResult.first;
-            line = lineResult.second;
-
-            std::string str = line.substr(0, stringSize);
-            return std::make_pair(RespType::String, str);
-        }
-        else
-        {
-            errMsg = "Unhandled response type";
-            return std::make_pair(RespType::Unknown, std::string());
-        }
-    }
-
-    std::string RedisClient::getRespTypeDescription(RespType respType)
-    {
-        switch (respType)
-        {
-            case RespType::Integer: return "integer";
-            case RespType::Error: return "error";
-            case RespType::String: return "string";
-            default: return "unknown";
-        }
-    }
 } // namespace ix

ixsnake/ixsnake/IXRedisClient.h

@@ -8,20 +8,12 @@
 
 #include <atomic>
 #include <functional>
+#include <ixwebsocket/IXSocket.h>
 #include <memory>
 #include <string>
-#include <ixwebsocket/IXSocket.h>
 
 namespace ix
 {
-    enum class RespType : int
-    {
-        String = 0,
-        Error = 1,
-        Integer = 2,
-        Unknown = 3
-    };
-
     class RedisClient
     {
     public:
@@ -48,22 +40,13 @@ namespace ix
         // XADD
         std::string xadd(const std::string& channel,
                          const std::string& message,
-                         int maxLen,
                          std::string& errMsg);
-        std::string prepareXaddCommand(const std::string& stream,
+
-                                       const std::string& message,
+        std::string prepareXaddCommand(const std::string& stream, const std::string& message);
-                                       int maxLen);
+
         std::string readXaddReply(std::string& errMsg);
-        bool sendCommand(
-            const std::string& commands, int commandsCount, std::string& errMsg);
 
-        // Arbitrary commands
+        bool sendCommand(const std::string& commands, int commandsCount, std::string& errMsg);
-        std::pair<RespType, std::string> send(
-            const std::vector<std::string>& args,
-            std::string& errMsg);
-        std::pair<RespType, std::string> readResponse(std::string& errMsg);
-
-        std::string getRespTypeDescription(RespType respType);
 
         void stop();
 

ixsnake/ixsnake/IXRedisServer.cpp

@@ -45,11 +45,8 @@ 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)
@@ -115,7 +112,7 @@ namespace ix
             it.second.erase(socket.get());
         }
 
-        for (auto&& it : _subscribers)
+        for (auto it : _subscribers)
         {
             std::stringstream ss;
             ss << "Subscription id: " << it.first << " #subscribers: " << it.second.size();

ixsnake/ixsnake/IXRedisServer.h

@@ -6,8 +6,8 @@
 
 #pragma once
 
-#include <ixwebsocket/IXSocket.h>
+#include "IXSocket.h"
-#include <ixwebsocket/IXSocketServer.h>
+#include "IXSocketServer.h"
 #include <functional>
 #include <map>
 #include <memory>
@@ -44,8 +44,7 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket>,
-                                      std::shared_ptr<ConnectionState> connectionState,
+                                      std::shared_ptr<ConnectionState> connectionState) final;
-                                      std::unique_ptr<ConnectionInfo> connectionInfo) final;
         virtual size_t getConnectedClientsCount() final;
 
         bool startsWith(const std::string& str, const std::string& start);

ixsnake/ixsnake/IXSnakeConnectionState.h

@@ -6,22 +6,16 @@
 
 #pragma once
 
-#include <ixredis/IXRedisClient.h>
+#include "IXRedisClient.h"
-#include <thread>
+#include <future>
 #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;
@@ -36,7 +30,6 @@ namespace snake
         {
             return _appkey;
         }
-
         void setAppkey(const std::string& appkey)
         {
             _appkey = appkey;
@@ -46,7 +39,6 @@ namespace snake
         {
             return _role;
         }
-
         void setRole(const std::string& role)
         {
             _role = role;
@@ -57,24 +49,7 @@ namespace snake
             return _redisClient;
         }
 
-        void stopSubScriptionThread()
+        std::future<void> fut;
-        {
-            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,22 +18,21 @@
 namespace snake
 {
     void handleError(const std::string& action,
-                     ix::WebSocket& ws,
+                     std::shared_ptr<ix::WebSocket> ws,
-                     uint64_t pduId,
+                     nlohmann::json pdu,
                      const std::string& errMsg)
     {
         std::string actionError(action);
         actionError += "/error";
 
         nlohmann::json response = {
-            {"action", actionError}, {"id", pduId}, {"body", {{"reason", errMsg}}}};
+            {"action", actionError}, {"id", pdu.value("id", 1)}, {"body", {{"reason", errMsg}}}};
-        ws.sendText(response.dump());
+        ws->sendText(response.dump());
     }
 
     void handleHandshake(std::shared_ptr<SnakeConnectionState> state,
-                         ix::WebSocket& ws,
+                         std::shared_ptr<ix::WebSocket> ws,
-                         const nlohmann::json& pdu,
+                         const nlohmann::json& pdu)
-                         uint64_t pduId)
     {
         std::string role = pdu["body"]["data"]["role"];
 
@@ -42,7 +41,7 @@ namespace snake
 
         nlohmann::json response = {
             {"action", "auth/handshake/ok"},
-            {"id", pduId},
+            {"id", pdu.value("id", 1)},
             {"body",
              {
                  {"data", {{"nonce", state->getNonce()}, {"connection_id", state->getId()}}},
@@ -50,14 +49,13 @@ namespace snake
 
         auto serializedResponse = response.dump();
 
-        ws.sendText(serializedResponse);
+        ws->sendText(serializedResponse);
     }
 
     void handleAuth(std::shared_ptr<SnakeConnectionState> state,
-                    ix::WebSocket& ws,
+                    std::shared_ptr<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());
 
@@ -65,9 +63,9 @@ namespace snake
         {
             nlohmann::json response = {
                 {"action", "auth/authenticate/error"},
-                {"id", pduId},
+                {"id", pdu.value("id", 1)},
                 {"body", {{"error", "authentication_failed"}, {"reason", "invalid secret"}}}};
-            ws.sendText(response.dump());
+            ws->sendText(response.dump());
             return;
         }
 
@@ -81,21 +79,19 @@ 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,
-                       ix::WebSocket& ws,
+                       std::shared_ptr<ix::WebSocket> ws,
-                       const AppConfig& appConfig,
+                       const nlohmann::json& pdu)
-                       const nlohmann::json& pdu,
-                       uint64_t pduId)
     {
         std::vector<std::string> channels;
 
@@ -115,16 +111,10 @@ namespace snake
         {
             std::stringstream ss;
             ss << "Missing channels or channel field in publish data";
-            handleError("rtm/publish", ws, pduId, ss.str());
+            handleError("rtm/publish", ws, pdu, 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;
@@ -135,7 +125,7 @@ namespace snake
             {
                 std::stringstream ss;
                 ss << "Cannot publish to redis host " << errMsg;
-                handleError("rtm/publish", ws, pduId, ss.str());
+                handleError("rtm/publish", ws, pdu, ss.str());
                 return;
             }
         }
@@ -143,27 +133,26 @@ 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 handleSubscribe(std::shared_ptr<SnakeConnectionState> state,
+    void handleRedisSubscription(std::shared_ptr<SnakeConnectionState> state,
-                         ix::WebSocket& ws,
+                                 std::shared_ptr<ix::WebSocket> ws,
-                         const AppConfig& appConfig,
+                                 const AppConfig& appConfig,
-                         const nlohmann::json& pdu,
+                                 const nlohmann::json& pdu)
-                         uint64_t pduId)
     {
         std::string channel = pdu["body"]["channel"];
-        state->subscriptionId = channel;
+        std::string subscriptionId = channel;
 
         std::stringstream ss;
         ss << state->appkey() << "::" << channel;
 
-        state->appChannel = ss.str();
+        std::string appChannel(ss.str());
 
-        ix::RedisClient& redisClient = state->subscriptionRedisClient;
+        ix::RedisClient redisClient;
         int port = appConfig.redisPort;
 
         auto urls = appConfig.redisHosts;
@@ -174,7 +163,7 @@ namespace snake
         {
             std::stringstream ss;
             ss << "Cannot connect to redis host " << hostname << ":" << port;
-            handleError("rtm/subscribe", ws, pduId, ss.str());
+            handleError("rtm/subscribe", ws, pdu, ss.str());
             return;
         }
 
@@ -186,90 +175,80 @@ namespace snake
             {
                 std::stringstream ss;
                 ss << "Cannot authenticated to redis";
-                handleError("rtm/subscribe", ws, pduId, ss.str());
+                handleError("rtm/subscribe", ws, pdu, ss.str());
                 return;
             }
         }
 
-        std::string filterStr;
+        int id = 0;
-        if (pdu["body"].find("filter") != pdu["body"].end())
+        auto callback = [ws, &id, &subscriptionId](const std::string& messageStr) {
-        {
-            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", state->id++},
+                {"id", id++},
                 {"body",
-                 {{"subscription_id", state->subscriptionId}, {"position", "0-0"}, {"messages", {msg}}}}};
+                 {{"subscription_id", subscriptionId}, {"position", "0-0"}, {"messages", {msg}}}}};
 
-            ws.sendText(response.dump());
+            ws->sendText(response.dump());
         };
 
-        state->onRedisSubscribeResponseCallback = [&ws, state, pduId](const std::string& redisResponse) {
+        auto responseCallback = [ws, pdu, &subscriptionId](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", pduId},
+                                       {"id", pdu.value("id", 1)},
-                                       {"body", {{"subscription_id", state->subscriptionId}}}};
+                                       {"body", {{"subscription_id", subscriptionId}}}};
-            ws.sendText(response.dump());
+            ws->sendText(response.dump());
         };
 
         {
             std::stringstream ss;
-            ss << "Subscribing to " << state->appChannel << "...";
+            ss << "Subscribing to " << appChannel << "...";
             ix::CoreLogger::log(ss.str().c_str());
         }
 
-        auto subscription = [&redisClient, state, &ws, pduId]
+        if (!redisClient.subscribe(appChannel, responseCallback, callback))
         {
-            if (!redisClient.subscribe(state->appChannel, 
+            std::stringstream ss;
-                                       state->onRedisSubscribeResponseCallback,
+            ss << "Error subscribing to channel " << appChannel;
-                                       state->onRedisSubscribeCallback))
+            handleError("rtm/subscribe", ws, pdu, ss.str());
-            {
+            return;
-                std::stringstream ss;
+        }
-                ss << "Error subscribing to channel " << state->appChannel;
+    }
-                handleError("rtm/subscribe", ws, pduId, ss.str());
-                return;
-            }
-        };
 
-        state->subscriptionThread = std::thread(subscription);
+    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);
     }
 
     void handleUnSubscribe(std::shared_ptr<SnakeConnectionState> state,
-                           ix::WebSocket& ws,
+                           std::shared_ptr<ix::WebSocket> ws,
-                           const nlohmann::json& pdu,
+                           const nlohmann::json& pdu)
-                           uint64_t pduId)
     {
         // extract subscription_id
         auto body = pdu["body"];
         auto subscriptionId = body["subscription_id"];
 
-        state->stopSubScriptionThread();
+        state->redisClient().stop();
 
         nlohmann::json response = {{"action", "rtm/unsubscribe/ok"},
-                                   {"id", pduId},
+                                   {"id", pdu.value("id", 1)},
                                    {"body", {{"subscription_id", subscriptionId}}}};
-        ws.sendText(response.dump());
+        ws->sendText(response.dump());
     }
 
     void processCobraMessage(std::shared_ptr<SnakeConnectionState> state,
-                             ix::WebSocket& ws,
+                             std::shared_ptr<ix::WebSocket> ws,
                              const AppConfig& appConfig,
                              const std::string& str)
     {
@@ -284,32 +263,31 @@ 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, pduId);
+            handleHandshake(state, ws, pdu);
         }
         else if (action == "auth/authenticate")
         {
-            handleAuth(state, ws, appConfig, pdu, pduId);
+            handleAuth(state, ws, appConfig, pdu);
         }
         else if (action == "rtm/publish")
         {
-            handlePublish(state, ws, appConfig, pdu, pduId);
+            handlePublish(state, ws, pdu);
         }
         else if (action == "rtm/subscribe")
         {
-            handleSubscribe(state, ws, appConfig, pdu, pduId);
+            handleSubscribe(state, ws, appConfig, pdu);
         }
         else if (action == "rtm/unsubscribe")
         {
-            handleUnSubscribe(state, ws, pdu, pduId);
+            handleUnSubscribe(state, ws, pdu);
         }
         else
         {

ixsnake/ixsnake/IXSnakeProtocol.h

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

ixsnake/ixsnake/IXSnakeServer.cpp

@@ -59,68 +59,65 @@ namespace snake
         };
         _server.setConnectionStateFactory(factory);
 
-        _server.setOnClientMessageCallback(
+        _server.setOnConnectionCallback(
-            [this](std::shared_ptr<ix::ConnectionState> connectionState,
+            [this](std::shared_ptr<ix::WebSocket> webSocket,
-                   ix::ConnectionInfo& connectionInfo,
+                   std::shared_ptr<ix::ConnectionState> connectionState) {
-                   ix::WebSocket& webSocket,
-                   const ix::WebSocketMessagePtr& msg) {
                 auto state = std::dynamic_pointer_cast<SnakeConnectionState>(connectionState);
-                auto remoteIp = connectionInfo.remoteIp;
-            
-                std::stringstream ss;
-                ss << "[" << state->getId() << "] ";
 
-                ix::LogLevel logLevel = ix::LogLevel::Debug;
+                webSocket->setOnMessageCallback(
-                if (msg->type == ix::WebSocketMessageType::Open)
+                    [this, webSocket, state](const ix::WebSocketMessagePtr& msg) {
-                {
+                        std::stringstream ss;
-                    ss << "New connection" << std::endl;
+                        ix::LogLevel logLevel = ix::LogLevel::Debug;
-                    ss << "remote ip: " << remoteIp << std::endl;
+                        if (msg->type == ix::WebSocketMessageType::Open)
-                    ss << "id: " << state->getId() << std::endl;
+                        {
-                    ss << "Uri: " << msg->openInfo.uri << std::endl;
+                            ss << "New connection" << std::endl;
-                    ss << "Headers:" << std::endl;
+                            ss << "id: " << state->getId() << std::endl;
-                    for (auto it : msg->openInfo.headers)
+                            ss << "Uri: " << msg->openInfo.uri << std::endl;
-                    {
+                            ss << "Headers:" << std::endl;
-                        ss << it.first << ": " << it.second << std::endl;
+                            for (auto it : msg->openInfo.headers)
-                    }
+                            {
+                                ss << it.first << ": " << it.second << std::endl;
+                            }
 
-                    std::string appkey = parseAppKey(msg->openInfo.uri);
+                            std::string appkey = parseAppKey(msg->openInfo.uri);
-                    state->setAppkey(appkey);
+                            state->setAppkey(appkey);
 
-                    // Connect to redis first
+                            // Connect to redis first
-                    if (!state->redisClient().connect(_appConfig.redisHosts[0],
+                            if (!state->redisClient().connect(_appConfig.redisHosts[0],
-                                                      _appConfig.redisPort))
+                                                              _appConfig.redisPort))
-                    {
+                            {
-                        ss << "Cannot connect to redis host" << std::endl;
+                                ss << "Cannot connect to redis host" << std::endl;
-                        logLevel = ix::LogLevel::Error;
+                                logLevel = ix::LogLevel::Error;
-                    }
+                            }
-                }
+                        }
-                else if (msg->type == ix::WebSocketMessageType::Close)
+                        else if (msg->type == ix::WebSocketMessageType::Close)
-                {
+                        {
-                    ss << "Closed connection"
+                            ss << "Closed connection"
-                       << " code " << msg->closeInfo.code << " reason "
+                               << " code " << msg->closeInfo.code << " reason "
-                       << msg->closeInfo.reason << std::endl;
+                               << msg->closeInfo.reason << std::endl;
-                }
+                        }
-                else if (msg->type == ix::WebSocketMessageType::Error)
+                        else if (msg->type == ix::WebSocketMessageType::Error)
-                {
+                        {
-                    std::stringstream ss;
+                            std::stringstream ss;
-                    ss << "Connection error: " << msg->errorInfo.reason << std::endl;
+                            ss << "Connection error: " << msg->errorInfo.reason << std::endl;
-                    ss << "#retries: " << msg->errorInfo.retries << std::endl;
+                            ss << "#retries: " << msg->errorInfo.retries << std::endl;
-                    ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
+                            ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
-                    ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
+                            ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
-                    logLevel = ix::LogLevel::Error;
+                            logLevel = ix::LogLevel::Error;
-                }
+                        }
-                else if (msg->type == ix::WebSocketMessageType::Fragment)
+                        else if (msg->type == ix::WebSocketMessageType::Fragment)
-                {
+                        {
-                    ss << "Received message fragment" << std::endl;
+                            ss << "Received message fragment" << std::endl;
-                }
+                        }
-                else if (msg->type == ix::WebSocketMessageType::Message)
+                        else if (msg->type == ix::WebSocketMessageType::Message)
-                {
+                        {
-                    ss << "Received " << msg->wireSize << " bytes" << " " << msg->str << std::endl;
+                            ss << "Received " << msg->wireSize << " bytes" << std::endl;
-                    processCobraMessage(state, webSocket, _appConfig, msg->str);
+                            processCobraMessage(state, webSocket, _appConfig, msg->str);
-                }
+                        }
 
-                ix::CoreLogger::log(ss.str().c_str(), logLevel);
+                        ix::CoreLogger::log(ss.str().c_str(), logLevel);
-        });
+                    });
+            });
 
         auto res = _server.listen();
         if (!res.first)

ixsnake/ixsnake/IXStreamSql.cpp

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

ixsnake/ixsnake/IXStreamSql.h

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

ixwebsocket/IXConnectionInfo.h

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

ixwebsocket/IXHttpClient.cpp

@@ -16,10 +16,7 @@
 #include <random>
 #include <sstream>
 #include <vector>
-
-#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
-#endif
 
 namespace ix
 {
@@ -130,7 +127,7 @@ namespace ix
     {
         // We only have one socket connection, so we cannot
         // make multiple requests concurrently.
-        std::lock_guard<std::recursive_mutex> lock(_mutex);
+        std::lock_guard<std::mutex> lock(_mutex);
 
         uint64_t uploadSize = 0;
         uint64_t downloadSize = 0;
@@ -177,13 +174,11 @@ 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)
@@ -500,7 +495,6 @@ namespace ix
 
         downloadSize = payload.size();
 
-#ifdef IXWEBSOCKET_USE_ZLIB
         // If the content was compressed with gzip, decode it
         if (headers["Content-Encoding"] == "gzip")
         {
@@ -519,7 +513,6 @@ namespace ix
             }
             payload = decompressedPayload;
         }
-#endif
 
         return std::make_shared<HttpResponse>(code,
                                               description,
@@ -679,7 +672,6 @@ namespace ix
         return ss.str();
     }
 
-#ifdef IXWEBSOCKET_USE_ZLIB
     bool HttpClient::gzipInflate(const std::string& in, std::string& out)
     {
         z_stream inflateState;
@@ -724,7 +716,6 @@ namespace ix
         inflateEnd(&inflateState);
         return true;
     }
-#endif
 
     void HttpClient::log(const std::string& msg, HttpRequestArgsPtr args)
     {

ixwebsocket/IXHttpClient.h

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

ixwebsocket/IXHttpServer.cpp

@@ -9,14 +9,11 @@
 #include "IXNetSystem.h"
 #include "IXSocketConnect.h"
 #include "IXUserAgent.h"
-#include <cstring>
 #include <fstream>
 #include <sstream>
 #include <vector>
-
-#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
-#endif
+#include <cstring>
 
 namespace
 {
@@ -44,7 +41,6 @@ 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
@@ -54,11 +50,8 @@ namespace
         const int windowBits = 15;
         const int GZIP_ENCODING = 16;
 
-        deflateInit2(&zs,
+        deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
-                     Z_DEFAULT_COMPRESSION,
+                     windowBits | GZIP_ENCODING, 8,
-                     Z_DEFLATED,
-                     windowBits | GZIP_ENCODING,
-                     8,
                      Z_DEFAULT_STRATEGY);
 
         zs.next_in = (Bytef*) str.data();
@@ -76,18 +69,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
@@ -120,8 +113,7 @@ 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++;
 
@@ -130,8 +122,7 @@ namespace ix
 
         if (std::get<0>(ret))
         {
-            auto response =
+            auto response = _onConnectionCallback(std::get<2>(ret), connectionState);
-                _onConnectionCallback(std::get<2>(ret), connectionState, std::move(connectionInfo));
             if (!Http::sendResponse(response, socket))
             {
                 logError("Cannot send response");
@@ -151,8 +142,7 @@ namespace ix
     {
         setOnConnectionCallback(
             [this](HttpRequestPtr request,
-                   std::shared_ptr<ConnectionState> /*connectionState*/,
+                   std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
-                   std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
                 std::string uri(request->uri);
                 if (uri.empty() || uri == "/")
                 {
@@ -173,19 +163,15 @@ 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)
+                if (acceptEncoding == "gzip" || acceptEncoding == "*")
                 {
                     content = gzipCompress(content);
-                    headers["Content-Encoding"] = "gzip";
                 }
-#endif
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                ss << request->method << " " << request->headers["User-Agent"] << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri << " " << content.size();
                 logInfo(ss.str());
 
@@ -209,16 +195,15 @@ namespace ix
         // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Redirections
         //
         setOnConnectionCallback(
-            [this, redirectUrl](HttpRequestPtr request,
+            [this,
-                                std::shared_ptr<ConnectionState> /*connectionState*/,
+             redirectUrl](HttpRequestPtr request,
-                                std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
+                          std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                ss << request->method << " " << request->headers["User-Agent"] << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
                 logInfo(ss.str());
 

ixwebsocket/IXHttpServer.h

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

ixwebsocket/IXSelectInterruptFactory.cpp

@@ -6,6 +6,7 @@
 
 #include "IXSelectInterruptFactory.h"
 
+#include "IXUniquePtr.h"
 #if defined(__linux__) || defined(__APPLE__)
 #include "IXSelectInterruptPipe.h"
 #else
@@ -17,9 +18,9 @@ namespace ix
     SelectInterruptPtr createSelectInterrupt()
     {
 #if defined(__linux__) || defined(__APPLE__)
-        return std::make_unique<SelectInterruptPipe>();
+        return ix::make_unique<SelectInterruptPipe>();
 #else
-        return std::make_unique<SelectInterrupt>();
+        return ix::make_unique<SelectInterrupt>();
 #endif
     }
 } // namespace ix

ixwebsocket/IXSelectInterruptPipe.cpp

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

ixwebsocket/IXSetThreadName.cpp

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

ixwebsocket/IXSocket.cpp

@@ -35,122 +35,19 @@ namespace ix
         : _sockfd(fd)
         , _selectInterrupt(createSelectInterrupt())
     {
-#if defined(__APPLE__)
+        ;
-        _kqueuefd = kqueue();
-#endif
     }
 
     Socket::~Socket()
     {
         close();
-
-#if defined(__APPLE__)
-        ::close(_kqueuefd);
-#endif
     }
 
     PollResultType Socket::poll(bool readyToRead,
                                 int timeoutMs,
                                 int sockfd,
-                                const SelectInterruptPtr& selectInterrupt,
+                                const SelectInterruptPtr& selectInterrupt)
-                                int kqueuefd)
     {
-#if defined(__APPLE__)
-        // FIXME int kqueuefd = kqueue();
-
-        struct kevent ke;
-        EV_SET(&ke, sockfd, (readyToRead) ? EVFILT_READ : EVFILT_WRITE, EV_ADD, 0, 0, NULL);
-        if (kevent(kqueuefd, &ke, 1, NULL, 0, NULL) == -1) return PollResultType::Error;
-
-        int retval, numevents = 0;
-
-        int nfds = 1;
-#if 0
-        if (selectInterrupt) 
-        {
-            nfds = 2;
-            int interruptFd = selectInterrupt->getFd();
-
-            struct kevent ke;
-            EV_SET(&ke, interruptFd, EVFILT_READ, EV_ADD, 0, 0, NULL);
-            if (kevent(kqueuefd, &ke, 1, NULL, 0, NULL) == -1) return PollResultType::Error;
-        }
-#endif
-
-        struct kevent *events;
-        events = (struct kevent*) malloc(sizeof(struct kevent) * nfds);
-
-        if (timeoutMs != 0)
-        {
-            struct timespec timeout;
-            timeout.tv_sec = timeoutMs / 1000;
-            timeout.tv_nsec = (timeoutMs % 1000) * 1000 * 1000;
-            retval = kevent(kqueuefd, NULL, 0, events, nfds, &timeout);
-        }
-        else
-        {
-            retval = kevent(kqueuefd, NULL, 0, events, nfds, NULL);
-        }
-
-#if 0
-        if (retval > 0) {
-            int j;
-
-            numevents = retval;
-            for(j = 0; j < numevents; j++) {
-                int mask = 0;
-                struct kevent *e = events+j;
-
-                if (e->filter == EVFILT_READ) mask |= AE_READABLE;
-                if (e->filter == EVFILT_WRITE) mask |= AE_WRITABLE;
-                eventLoop->fired[j].fd = e->ident;
-                eventLoop->fired[j].mask = mask;
-            }
-        }
-#else
-        PollResultType pollResult = PollResultType::ReadyForRead;
-        if (retval < 0)
-        {
-            pollResult = PollResultType::Error;
-        }
-
-        if (retval > 0) {
-            struct kevent *e = events;
-            if (e->filter == EVFILT_READ)
-            {
-                pollResult = PollResultType::ReadyForRead;
-            }
-            else if (e->filter == EVFILT_WRITE) 
-            {
-                pollResult = PollResultType::ReadyForWrite;
-
-                int optval = -1;
-                socklen_t optlen = sizeof(optval);
-
-                // getsockopt() puts the errno value for connect into optval so 0
-                // means no-error.
-                if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &optval, &optlen) == -1 || optval != 0)
-                {
-                    pollResult = PollResultType::Error;
-
-                    // set errno to optval so that external callers can have an
-                    // appropriate error description when calling strerror
-                    errno = optval;
-                }
-            }
-        }
-        else
-        {
-            pollResult = PollResultType::Timeout;
-        }
-#endif
-
-        free(events);
-
-        // ::close(kqueuefd); //FMXE
-
-        return pollResult;
-#else
         //
         // We used to use ::select to poll but on Android 9 we get large fds out of
         // ::connect which crash in FD_SET as they are larger than FD_SETSIZE. Switching
@@ -245,7 +142,6 @@ namespace ix
         }
 
         return pollResult;
-#endif
     }
 
     PollResultType Socket::isReadyToRead(int timeoutMs)
@@ -256,7 +152,7 @@ namespace ix
         }
 
         bool readyToRead = true;
-        return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt, _kqueuefd);
+        return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt);
     }
 
     PollResultType Socket::isReadyToWrite(int timeoutMs)
@@ -267,7 +163,7 @@ namespace ix
         }
 
         bool readyToRead = false;
-        return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt, _kqueuefd);
+        return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt);
     }
 
     // Wake up from poll/select by writing to the pipe which is watched by select

ixwebsocket/IXSocket.h

@@ -13,13 +13,6 @@
 #include <string>
 #include <vector>
 
-// For kqueue
-#if defined(__APPLE__)
-#include <sys/types.h>
-#include <sys/event.h>
-#include <sys/time.h>
-#endif
-
 #ifdef _WIN32
 #include <BaseTsd.h>
 typedef SSIZE_T ssize_t;
@@ -101,8 +94,7 @@ namespace ix
         static PollResultType poll(bool readyToRead,
                                    int timeoutMs,
                                    int sockfd,
-                                   const SelectInterruptPtr& selectInterrupt,
+                                   const SelectInterruptPtr& selectInterrupt);
-                                   int kqueuefd);
 
 
         // Used as special codes for pipe communication
@@ -122,9 +114,5 @@ namespace ix
         static constexpr size_t kChunkSize = 1 << 15;
 
         SelectInterruptPtr _selectInterrupt;
-
-#if defined(__APPLE__)
-        int _kqueuefd;
-#endif
     };
 } // namespace ix

ixwebsocket/IXSocketConnect.cpp

@@ -10,6 +10,7 @@
 #include "IXNetSystem.h"
 #include "IXSelectInterrupt.h"
 #include "IXSocket.h"
+#include "IXUniquePtr.h"
 #include <fcntl.h>
 #include <string.h>
 #include <sys/types.h>
@@ -65,11 +66,8 @@ namespace ix
 
             int timeoutMs = 10;
             bool readyToRead = false;
-            auto selectInterrupt = std::make_unique<SelectInterrupt>();
+            auto selectInterrupt = ix::make_unique<SelectInterrupt>();
-
+            PollResultType pollResult = Socket::poll(readyToRead, timeoutMs, fd, selectInterrupt);
-            int kqueuefd = kqueue();
-            PollResultType pollResult = Socket::poll(readyToRead, timeoutMs, fd, selectInterrupt, kqueuefd);
-            ::close(kqueuefd);
 
             if (pollResult == PollResultType::Timeout)
             {

ixwebsocket/IXSocketFactory.cpp

@@ -5,6 +5,7 @@
  */
 
 #include "IXSocketFactory.h"
+#include "IXUniquePtr.h"
 
 #ifdef IXWEBSOCKET_USE_TLS
 
@@ -35,17 +36,17 @@ namespace ix
 
         if (!tls)
         {
-            socket = std::make_unique<Socket>(fd);
+            socket = ix::make_unique<Socket>(fd);
         }
         else
         {
 #ifdef IXWEBSOCKET_USE_TLS
 #if defined(IXWEBSOCKET_USE_MBED_TLS)
-            socket = std::make_unique<SocketMbedTLS>(tlsOptions, fd);
+            socket = ix::make_unique<SocketMbedTLS>(tlsOptions, fd);
 #elif defined(IXWEBSOCKET_USE_OPEN_SSL)
-            socket = std::make_unique<SocketOpenSSL>(tlsOptions, fd);
+            socket = ix::make_unique<SocketOpenSSL>(tlsOptions, fd);
 #elif defined(__APPLE__)
-            socket = std::make_unique<SocketAppleSSL>(tlsOptions, fd);
+            socket = ix::make_unique<SocketAppleSSL>(tlsOptions, fd);
 #endif
 #else
             errorMsg = "TLS support is not enabled on this platform.";

ixwebsocket/IXSocketServer.cpp

@@ -12,6 +12,7 @@
 #include "IXSocket.h"
 #include "IXSocketConnect.h"
 #include "IXSocketFactory.h"
+#include "IXUniquePtr.h"
 #include <assert.h>
 #include <sstream>
 #include <stdio.h>
@@ -22,7 +23,7 @@ namespace ix
     const int SocketServer::kDefaultPort(8080);
     const std::string SocketServer::kDefaultHost("127.0.0.1");
     const int SocketServer::kDefaultTcpBacklog(5);
-    const size_t SocketServer::kDefaultMaxConnections(128);
+    const size_t SocketServer::kDefaultMaxConnections(32);
     const int SocketServer::kDefaultAddressFamily(AF_INET);
 
     SocketServer::SocketServer(
@@ -258,12 +259,9 @@ namespace ix
             // Use poll to check whether a new connection is in progress
             int timeoutMs = 10;
             bool readyToRead = true;
-            auto selectInterrupt = std::make_unique<SelectInterrupt>();
+            auto selectInterrupt = ix::make_unique<SelectInterrupt>();
-
-            int kqueuefd = kqueue();
             PollResultType pollResult =
-                Socket::poll(readyToRead, timeoutMs, _serverFd, selectInterrupt, kqueuefd);
+                Socket::poll(readyToRead, timeoutMs, _serverFd, selectInterrupt);
-            ::close(kqueuefd);
 
             if (pollResult == PollResultType::Error)
             {
@@ -279,7 +277,6 @@ namespace ix
             }
 
             // Accept a connection.
-            // FIXME: Is this working for ipv6 ?
             struct sockaddr_in client; // client address information
             int clientFd;              // socket connected to client
             socklen_t addressLen = sizeof(client);
@@ -311,45 +308,6 @@ namespace ix
                 continue;
             }
 
-            std::unique_ptr<ConnectionInfo> connectionInfo;
-
-            if (_addressFamily == AF_INET)
-            {
-                char remoteIp[INET_ADDRSTRLEN];
-                if (inet_ntop(AF_INET, &client.sin_addr, remoteIp, INET_ADDRSTRLEN) == nullptr)
-                {
-                    int err = Socket::getErrno();
-                    std::stringstream ss;
-                    ss << "SocketServer::run() error calling inet_ntop (ipv4): " << err << ", "
-                       << strerror(err);
-                    logError(ss.str());
-
-                    Socket::closeSocket(clientFd);
-
-                    continue;
-                }
-
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
-            }
-            else // AF_INET6
-            {
-                char remoteIp[INET6_ADDRSTRLEN];
-                if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp, INET6_ADDRSTRLEN) == nullptr)
-                {
-                    int err = Socket::getErrno();
-                    std::stringstream ss;
-                    ss << "SocketServer::run() error calling inet_ntop (ipv6): " << err << ", "
-                       << strerror(err);
-                    logError(ss.str());
-
-                    Socket::closeSocket(clientFd);
-
-                    continue;
-                }
-
-                connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
-            }
-
             std::shared_ptr<ConnectionState> connectionState;
             if (_connectionStateFactory)
             {
@@ -382,13 +340,10 @@ namespace ix
 
             // Launch the handleConnection work asynchronously in its own thread.
             std::lock_guard<std::mutex> lock(_connectionsThreadsMutex);
-            _connectionsThreads.push_back(
+            _connectionsThreads.push_back(std::make_pair(
-                std::make_pair(connectionState,
+                connectionState,
-                               std::thread(&SocketServer::handleConnection,
+                std::thread(
-                                           this,
+                    &SocketServer::handleConnection, this, std::move(socket), connectionState)));
-                                           std::move(socket),
-                                           connectionState,
-                                           std::move(connectionInfo))));
         }
     }
 

ixwebsocket/IXSocketServer.h

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

ixwebsocket/IXUniquePtr.h

@@ -0,0 +1,18 @@
+/*
+ *  IXUniquePtr.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace ix
+{
+    template<typename T, typename... Args>
+    std::unique_ptr<T> make_unique(Args&&... args)
+    {
+        return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
+    }
+}

ixwebsocket/IXUrlParser.cpp

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

ixwebsocket/IXUserAgent.cpp

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

ixwebsocket/IXWebSocket.cpp

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

ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp

@@ -28,26 +28,21 @@ 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,
@@ -62,49 +57,17 @@ namespace ix
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
-#else
-        return false;
-#endif
     }
 
-    template<typename T>
+    bool WebSocketPerMessageDeflateCompressor::endsWith(const std::string& value,
-    bool WebSocketPerMessageDeflateCompressor::endsWithEmptyUnCompressedBlock(const T& value)
+                                                        const std::string& ending)
     {
-        if (kEmptyUncompressedBlock.size() > value.size()) return false;
+        if (ending.size() > value.size()) return false;
-        auto N = value.size();
+        return std::equal(ending.rbegin(), ending.rend(), value.rbegin());
-        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
         //
@@ -133,8 +96,7 @@ 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.push_back(buf[0]);
+            out.append((char*) (buf), 2);
-            out.push_back(buf[1]);
 
             return true;
         }
@@ -152,18 +114,15 @@ namespace ix
 
             output = _compressBufferSize - _deflateState.avail_out;
 
-            out.insert(out.end(), _compressBuffer.get(), _compressBuffer.get() + output);
+            out.append((char*) (_compressBuffer.get()), output);
         } while (_deflateState.avail_out == 0);
 
-        if (endsWithEmptyUnCompressedBlock(out))
+        if (endsWith(out, kEmptyUncompressedBlock))
         {
             out.resize(out.size() - 4);
         }
 
         return true;
-#else
-        return false;
-#endif
     }
 
     //
@@ -172,7 +131,6 @@ namespace ix
     WebSocketPerMessageDeflateDecompressor::WebSocketPerMessageDeflateDecompressor()
         : _compressBufferSize(kBufferSize)
     {
-#ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_inflateState, 0, sizeof(_inflateState));
 
         _inflateState.zalloc = Z_NULL;
@@ -180,20 +138,16 @@ 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;
@@ -203,14 +157,10 @@ 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
         //
@@ -247,8 +197,5 @@ namespace ix
         } while (_inflateState.avail_out == 0);
 
         return true;
-#else
-        return false;
-#endif
     }
 } // namespace ix

ixwebsocket/IXWebSocketPerMessageDeflateCodec.h

@@ -6,12 +6,9 @@
 
 #pragma once
 
-#ifdef IXWEBSOCKET_USE_ZLIB
 #include "zlib.h"
-#endif
 #include <memory>
 #include <string>
-#include <vector>
 
 namespace ix
 {
@@ -23,23 +20,14 @@ 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:
-        template<typename T, typename S>
+        static bool endsWith(const std::string& value, const std::string& ending);
-        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
@@ -55,10 +43,7 @@ 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,7 +61,6 @@ namespace ix
         _clientMaxWindowBits = kDefaultClientMaxWindowBits;
         _serverMaxWindowBits = kDefaultServerMaxWindowBits;
 
-#ifdef IXWEBSOCKET_USE_ZLIB
         // Split by ;
         std::string token;
         std::stringstream tokenStream(extension);
@@ -113,7 +112,6 @@ namespace ix
                 sanitizeClientMaxWindowBits();
             }
         }
-#endif
     }
 
     void WebSocketPerMessageDeflateOptions::sanitizeClientMaxWindowBits()
@@ -128,7 +126,6 @@ namespace ix
 
     std::string WebSocketPerMessageDeflateOptions::generateHeader()
     {
-#ifdef IXWEBSOCKET_USE_ZLIB
         std::stringstream ss;
         ss << "Sec-WebSocket-Extensions: permessage-deflate";
 
@@ -141,18 +138,11 @@ 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

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

@@ -1,20 +0,0 @@
-/*
- *  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,38 +71,13 @@ 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>();
-        if (_onConnectionCallback)
+        _onConnectionCallback(webSocket, connectionState);
-        {
-            _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();
 
@@ -136,8 +111,6 @@ 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

@@ -19,18 +19,11 @@
 
 namespace ix
 {
-    class WebSocketServer : public SocketServer
+    class WebSocketServer final : public SocketServer
     {
     public:
         using OnConnectionCallback =
-            std::function<void(std::weak_ptr<WebSocket>,
+            std::function<void(std::shared_ptr<WebSocket>, std::shared_ptr<ConnectionState>)>;
-                               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,
@@ -46,7 +39,6 @@ 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();
@@ -60,7 +52,6 @@ namespace ix
         bool _enablePerMessageDeflate;
 
         OnConnectionCallback _onConnectionCallback;
-        OnClientMessageCallback _onClientMessageCallback;
 
         std::mutex _clientsMutex;
         std::set<std::shared_ptr<WebSocket>> _clients;
@@ -69,8 +60,7 @@ namespace ix
 
         // Methods
         virtual void handleConnection(std::unique_ptr<Socket> socket,
-                                      std::shared_ptr<ConnectionState> connectionState,
+                                      std::shared_ptr<ConnectionState> connectionState) final;
-                                      std::unique_ptr<ConnectionInfo> connectionInfo);
         virtual size_t getConnectedClientsCount() final;
     };
 } // namespace ix

ixwebsocket/IXWebSocketTransport.cpp

@@ -65,6 +65,7 @@ namespace ix
         , _receivedMessageCompressed(false)
         , _readyState(ReadyState::CLOSED)
         , _closeCode(WebSocketCloseConstants::kInternalErrorCode)
+        , _closeReason(WebSocketCloseConstants::kInternalErrorMessage)
         , _closeWireSize(0)
         , _closeRemote(false)
         , _enablePerMessageDeflate(false)
@@ -76,7 +77,6 @@ namespace ix
         , _pingCount(0)
         , _lastSendPingTimePoint(std::chrono::steady_clock::now())
     {
-        setCloseReason(WebSocketCloseConstants::kInternalErrorMessage);
         _readbuf.resize(kChunkSize);
     }
 
@@ -179,12 +179,10 @@ namespace ix
 
         if (readyState == ReadyState::CLOSED)
         {
-            if (_onCloseCallback)
+            std::lock_guard<std::mutex> lock(_closeDataMutex);
-            {
+            _onCloseCallback(_closeCode, _closeReason, _closeWireSize, _closeRemote);
-                _onCloseCallback(_closeCode, getCloseReason(), _closeWireSize, _closeRemote);
-            }
-            setCloseReason(WebSocketCloseConstants::kInternalErrorMessage);
             _closeCode = WebSocketCloseConstants::kInternalErrorCode;
+            _closeReason = WebSocketCloseConstants::kInternalErrorMessage;
             _closeWireSize = 0;
             _closeRemote = false;
         }
@@ -263,10 +261,9 @@ namespace ix
         {
             // compute lasting delay to wait for next ping / timeout, if at least one set
             auto now = std::chrono::steady_clock::now();
-            int timeSinceLastPingMs = (int) std::chrono::duration_cast<std::chrono::milliseconds>(
+            lastingTimeoutDelayInMs = (int) std::chrono::duration_cast<std::chrono::milliseconds>(
                                           now - _lastSendPingTimePoint)
                                           .count();
-            lastingTimeoutDelayInMs = (1000 * _pingIntervalSecs) - timeSinceLastPingMs;
         }
 
 #ifdef _WIN32
@@ -329,10 +326,9 @@ namespace ix
         return _txbuf.empty();
     }
 
-    template<class Iterator>
     void WebSocketTransport::appendToSendBuffer(const std::vector<uint8_t>& header,
-                                                Iterator begin,
+                                                std::string::const_iterator begin,
-                                                Iterator end,
+                                                std::string::const_iterator end,
                                                 uint64_t message_size,
                                                 uint8_t masking_key[4])
     {
@@ -642,7 +638,11 @@ 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 = _closeCode == code && getCloseReason() == reason;
+                    bool identicalReason;
+                    {
+                        std::lock_guard<std::mutex> lock(_closeDataMutex);
+                        identicalReason = _closeCode == code && _closeReason == reason;
+                    }
 
                     if (identicalReason)
                     {
@@ -750,9 +750,8 @@ namespace ix
         return static_cast<unsigned>(seconds);
     }
 
-    template<class T>
     WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
-                                                   const T& message,
+                                                   const std::string& message,
                                                    bool compress,
                                                    const OnProgressCallback& onProgressCallback)
     {
@@ -765,8 +764,8 @@ namespace ix
         size_t wireSize = message.size();
         bool compressionError = false;
 
-        auto message_begin = message.cbegin();
+        std::string::const_iterator message_begin = message.begin();
-        auto message_end = message.cend();
+        std::string::const_iterator message_end = message.end();
 
         if (compress)
         {
@@ -781,8 +780,8 @@ namespace ix
             compressionError = false;
             wireSize = _compressedMessage.size();
 
-            message_begin = _compressedMessage.cbegin();
+            message_begin = _compressedMessage.begin();
-            message_end = _compressedMessage.cend();
+            message_end = _compressedMessage.end();
         }
 
         {
@@ -796,11 +795,6 @@ namespace ix
         if (wireSize < kChunkSize)
         {
             success = sendFragment(type, true, message_begin, message_end, compress);
-
-            if (onProgressCallback)
-            {
-                onProgressCallback(0, 1);
-            }
         }
         else
         {
@@ -865,11 +859,10 @@ namespace ix
         return WebSocketSendInfo(success, compressionError, payloadSize, wireSize);
     }
 
-    template<class Iterator>
     bool WebSocketTransport::sendFragment(wsheader_type::opcode_type type,
                                           bool fin,
-                                          Iterator message_begin,
+                                          std::string::const_iterator message_begin,
-                                          Iterator message_end,
+                                          std::string::const_iterator message_end,
                                           bool compress)
     {
         uint64_t message_size = static_cast<uint64_t>(message_end - message_begin);
@@ -1062,7 +1055,7 @@ namespace ix
         else
         {
             // no close code/reason set
-            sendData(wsheader_type::CLOSE, std::string(""), compress);
+            sendData(wsheader_type::CLOSE, "", compress);
         }
     }
 
@@ -1085,10 +1078,13 @@ namespace ix
     {
         closeSocket();
 
-        setCloseReason(reason);
+        {
-        _closeCode = code;
+            std::lock_guard<std::mutex> lock(_closeDataMutex);
-        _closeWireSize = closeWireSize;
+            _closeCode = code;
-        _closeRemote = remote;
+            _closeReason = reason;
+            _closeWireSize = closeWireSize;
+            _closeRemote = remote;
+        }
 
         setReadyState(ReadyState::CLOSED);
         _requestInitCancellation = false;
@@ -1108,11 +1104,13 @@ namespace ix
             closeWireSize = reason.size();
         }
 
-        setCloseReason(reason);
+        {
-        _closeCode = code;
+            std::lock_guard<std::mutex> lock(_closeDataMutex);
-        _closeWireSize = closeWireSize;
+            _closeCode = code;
-        _closeRemote = remote;
+            _closeReason = reason;
-
+            _closeWireSize = closeWireSize;
+            _closeRemote = remote;
+        }
         {
             std::lock_guard<std::mutex> lock(_closingTimePointMutex);
             _closingTimePoint = std::chrono::steady_clock::now();
@@ -1157,15 +1155,4 @@ 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;
-        mutable std::mutex _closeReasonMutex;
+        size_t _closeWireSize;
-        std::atomic<uint16_t> _closeCode;
+        bool _closeRemote;
-        std::atomic<size_t> _closeWireSize;
+        mutable std::mutex _closeDataMutex;
-        std::atomic<bool> _closeRemote;
 
         // Data used for Per Message Deflate compression (with zlib)
         WebSocketPerMessageDeflatePtr _perMessageDeflate;
@@ -239,15 +239,16 @@ namespace ix
         bool sendOnSocket();
         bool receiveFromSocket();
 
-        template<class T>
         WebSocketSendInfo sendData(wsheader_type::opcode_type type,
-                                   const T& message,
+                                   const std::string& message,
                                    bool compress,
                                    const OnProgressCallback& onProgressCallback = nullptr);
 
-        template<class Iterator>
+        bool sendFragment(wsheader_type::opcode_type type,
-        bool sendFragment(
+                          bool fin,
-            wsheader_type::opcode_type type, bool fin, Iterator begin, Iterator end, bool compress);
+                          std::string::const_iterator begin,
+                          std::string::const_iterator end,
+                          bool compress);
 
         void emitMessage(MessageKind messageKind,
                          const std::string& message,
@@ -255,11 +256,9 @@ namespace ix
                          const OnMessageCallback& onMessageCallback);
 
         bool isSendBufferEmpty() const;
-
-        template<class Iterator>
         void appendToSendBuffer(const std::vector<uint8_t>& header,
-                                Iterator begin,
+                                std::string::const_iterator begin,
-                                Iterator end,
+                                std::string::const_iterator end,
                                 uint64_t message_size,
                                 uint8_t masking_key[4]);
 
@@ -267,8 +266,5 @@ 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 "10.1.5"
+#define IX_WEBSOCKET_VERSION "9.6.5"

ixwebsocket/apple/IXSetThreadName_apple.cpp

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

ixwebsocket/freebsd/IXSetThreadName_freebsd.cpp

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

ixwebsocket/linux/IXSetThreadName_linux.cpp

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

ixwebsocket/windows/IXSetThreadName_windows.cpp

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

makefile

@@ -19,31 +19,26 @@ install: brew
 #
 # Release, Debug, MinSizeRel, RelWithDebInfo are the build types
 #
-# Default rule does not use python as that requires first time users to have Python3 installed
-#
 brew:
 	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_EXPORT_COMPILE_COMMANDS=ON -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; ninja install)
 
-# Docker default target. We've had problems with OpenSSL and TLS 1.3 (on the
+# Docker default target. We've add problem with OpenSSL and TLS 1.3 (on the
 # server side ?) and I can't work-around it easily, so we're using mbedtls on
 # Linux for the SSL backend, which works great.
 ws_mbedtls_install:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_MBED_TLS=1 .. ; ninja install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_TLS=1 -DUSE_WS=1 -DUSE_MBED_TLS=1 .. ; ninja install)
 
 ws:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && ninja install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 .. ; ninja install)
 
 ws_install:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. -DUSE_TEST=0 && ninja install)
+	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_TLS=1 -DUSE_WS=1 .. ; make -j 4 install)
-
-ws_install_release:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. -DUSE_TEST=0 && 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)
+	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_OPEN_SSL=1 .. ; make -j 4)
 
 ws_mbedtls:
-	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_MBED_TLS=1 .. ; make -j 4)
+	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_MBED_TLS=1 .. ; make -j 4)
 
 ws_no_ssl:
 	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_WS=1 .. ; make -j 4)
@@ -107,23 +102,21 @@ test_server:
 # env TEST=Websocket_server make test
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
-build_test:
+test:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. ; ninja install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; ninja install)
-
-test: build_test
 	(cd test ; python2.7 run.py -r)
 
 test_make:
-	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; make -j 4)
+	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_WS=1 -DUSE_TEST=1 .. ; make -j 4)
 	(cd test ; python2.7 run.py -r)
 
 test_tsan:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
 test_ubsan:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableUndefinedBehaviorSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableUndefinedBehaviorSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
@@ -131,37 +124,37 @@ test_asan: build_test_asan
 	(cd test ; python2.7 run.py -r)
 
 build_test_asan:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableAddressSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableAddressSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 
 test_tsan_openssl:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
 test_ubsan_openssl:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableUndefinedBehaviorSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableUndefinedBehaviorSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
 test_tsan_openssl_release:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Release -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -configuration Release -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Release -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_OPEN_SSL=1 .. && xcodebuild -project ixwebsocket.xcodeproj -configuration Release -target ixwebsocket_unittest -enableThreadSanitizer YES)
 	(cd build/test ; ln -sf Release/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
 test_tsan_mbedtls:
-	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_MBED_TLS=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
+	mkdir -p build && (cd build && cmake -GXcode -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_TEST=1 -DUSE_MBED_TLS=1 .. && xcodebuild -project ixwebsocket.xcodeproj -target ixwebsocket_unittest -enableThreadSanitizer YES)
 	(cd build/test ; ln -sf Debug/ixwebsocket_unittest)
 	(cd test ; python2.7 run.py -r)
 
 build_test_openssl:
-	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_OPEN_SSL=1 -DUSE_TEST=1 .. ; ninja install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_OPEN_SSL=1 -DUSE_TEST=1 .. ; ninja install)
 
 test_openssl: build_test_openssl
 	(cd test ; python2.7 run.py -r)
 
 build_test_mbedtls:
-	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_MBED_TLS=1 -DUSE_TEST=1 .. ; make -j 4)
+	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_TLS=1 -DUSE_MBED_TLS=1 -DUSE_TEST=1 .. ; make -j 4)
 
 test_mbedtls: build_test_mbedtls
 	(cd test ; python2.7 run.py -r)
@@ -177,7 +170,7 @@ autobahn_report:
 	cp -rvf ~/sandbox/reports/clients/* ../bsergean.github.io/IXWebSocket/autobahn/
 
 httpd:
-	clang++ --std=c++14 --stdlib=libc++ -o ixhttpd httpd.cpp \
+	clang++ --std=c++11 --stdlib=libc++ -Iixwebsocket httpd.cpp \
 		ixwebsocket/IXSelectInterruptFactory.cpp \
 		ixwebsocket/IXCancellationRequest.cpp \
 		ixwebsocket/IXSocketTLSOptions.cpp \
@@ -196,11 +189,11 @@ httpd:
 		ixwebsocket/IXConnectionState.cpp \
 		ixwebsocket/IXUrlParser.cpp \
 		ixwebsocket/IXSelectInterrupt.cpp \
-		ixwebsocket/IXSetThreadName.cpp \
+		ixwebsocket/apple/IXSetThreadName_apple.cpp \
 		-lz
 
 httpd_linux:
-	g++ --std=c++14 -o ixhttpd httpd.cpp -Iixwebsocket \
+	g++ --std=c++11 -o ixhttpd httpd.cpp -Iixwebsocket \
 		ixwebsocket/IXSelectInterruptFactory.cpp \
 		ixwebsocket/IXCancellationRequest.cpp \
 		ixwebsocket/IXSocketTLSOptions.cpp \
@@ -219,7 +212,7 @@ httpd_linux:
 		ixwebsocket/IXConnectionState.cpp \
 		ixwebsocket/IXUrlParser.cpp \
 		ixwebsocket/IXSelectInterrupt.cpp \
-		ixwebsocket/IXSetThreadName.cpp \
+		ixwebsocket/linux/IXSetThreadName_linux.cpp \
 		-lz -lpthread
 	cp -f ixhttpd /usr/local/bin
 
@@ -230,7 +223,6 @@ rebase_upstream:
 	git reset --hard upstream/master
 	git push origin master --force
 
-# Legacy target
 install_cmake_for_linux:
 	mkdir -p /tmp/cmake
 	(cd /tmp/cmake ; curl -L -O https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz ; tar zxf cmake-3.14.0-Linux-x86_64.tar.gz)
@@ -241,12 +233,6 @@ install_cmake_for_linux:
 doc:
 	mkdocs gh-deploy
 
-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,8 +55,6 @@ set (SOURCES
   IXSentryClientTest.cpp
   IXWebSocketChatTest.cpp
   IXWebSocketBroadcastTest.cpp
-  IXWebSocketPerMessageDeflateCompressorTest.cpp
-  IXStreamSqlTest.cpp
 )
 
 # Some unittest don't work on windows yet
@@ -93,30 +91,15 @@ if (JSONCPP_FOUND)
   target_link_libraries(ixwebsocket_unittest ${JSONCPP_LIBRARIES})
 endif()
 
-if (USE_PYTHON)
-  find_package(Python COMPONENTS Development)
-  if (NOT Python_FOUND)
-    message(FATAL_ERROR "Python3 not found")
-  endif()
-  message("Python_FOUND:${Python_FOUND}")
-  message("Python_VERSION:${Python_VERSION}")
-  message("Python_Development_FOUND:${Python_Development_FOUND}")
-  message("Python_LIBRARIES:${Python_LIBRARIES}")
-endif()
-
 # library with the most dependencies come first
 target_link_libraries(ixwebsocket_unittest ixbots)
 target_link_libraries(ixwebsocket_unittest ixsnake)
 target_link_libraries(ixwebsocket_unittest ixcobra)
 target_link_libraries(ixwebsocket_unittest ixsentry)
-target_link_libraries(ixwebsocket_unittest ixredis)
 target_link_libraries(ixwebsocket_unittest ixwebsocket)
 target_link_libraries(ixwebsocket_unittest ixcrypto)
 target_link_libraries(ixwebsocket_unittest ixcore)
 
 target_link_libraries(ixwebsocket_unittest spdlog)
-if (USE_PYTHON)
-  target_link_libraries(ixwebsocket_unittest ${Python_LIBRARIES})
-endif()
 
 install(TARGETS ixwebsocket_unittest DESTINATION bin)

test/IXCobraChatTest.cpp

@@ -10,7 +10,7 @@
 #include <iostream>
 #include <ixcobra/IXCobraConnection.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixredis/IXRedisServer.h>
+#include <ixsnake/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 
 using namespace ix;
@@ -125,12 +125,10 @@ namespace
     {
         std::string filter;
         std::string position("$");
-        int batchSize = 1;
 
         _conn.subscribe(channel,
                         filter,
                         position,
-                        batchSize,
                         [this](const Json::Value& msg, const std::string& /*position*/) {
                             spdlog::info("receive {}", msg.toStyledString());
 

test/IXCobraMetricsPublisherTest.cpp

@@ -8,7 +8,7 @@
 #include <iostream>
 #include <ixcobra/IXCobraMetricsPublisher.h>
 #include <ixcrypto/IXUuid.h>
-#include <ixredis/IXRedisServer.h>
+#include <ixsnake/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <set>
 
@@ -76,12 +76,10 @@ namespace
                 log("Subscriber authenticated");
                 std::string filter;
                 std::string position("$");
-                int batchSize = 1;
 
                 conn.subscribe(channel,
                                filter,
                                position,
-                               batchSize,
                                [](const Json::Value& msg, const std::string& /*position*/) {
                                    log(msg.toStyledString());
 
@@ -108,7 +106,7 @@ namespace
             }
             else if (event->type == ix::CobraEventType::UnSubscribed)
             {
-                TLogger() << "Subscriber: unsubscribed from channel " << event->subscriptionId;
+                TLogger() << "Subscriber: ununexpected 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 <ixredis/IXRedisServer.h>
 #include <ixsentry/IXSentryClient.h>
+#include <ixsnake/IXRedisServer.h>
 #include <ixsnake/IXSnakeServer.h>
 #include <ixwebsocket/IXHttpServer.h>
 #include <ixwebsocket/IXUserAgent.h>
@@ -95,15 +95,13 @@ TEST_CASE("Cobra_to_sentry_bot", "[cobra_bots]")
 
         sentryServer.setOnConnectionCallback(
             [](HttpRequestPtr request,
-               std::shared_ptr<ConnectionState> /*connectionState*/,
+               std::shared_ptr<ConnectionState> /*connectionState*/) -> HttpResponsePtr {
-               std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
                 WebSocketHttpHeaders headers;
                 headers["Server"] = userAgent();
 
                 // Log request
                 std::stringstream ss;
-                ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
+                ss << request->method << " " << request->headers["User-Agent"] << " "
-                   << request->method << " " << request->headers["User-Agent"] << " "
                    << request->uri;
 
                 if (request->method == "POST")

test/IXCobraToStatsdBotTest.cpp

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

test/IXCobraToStdoutBotTest.cpp

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

test/IXHttpServerTest.cpp

@@ -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,8 +67,7 @@ TEST_CASE("http server", "[httpd]")
 
 TEST_CASE("http server redirection", "[httpd_redirect]")
 {
-    SECTION(
+    SECTION("Connect to a local HTTP server, with redirection enabled")
-        "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");
@@ -118,54 +117,4 @@ TEST_CASE("http server redirection", "[httpd_redirect]")
 
         server.stop();
     }
-
-    SECTION("Connect to a local HTTP server, with redirection enabled, but we do follow redirects")
-    {
-        int port = getFreePort();
-        ix::HttpServer server(port, "127.0.0.1");
-        server.makeRedirectServer("http://www.google.com");
-
-        auto res = server.listen();
-        REQUIRE(res.first);
-        server.start();
-
-        HttpClient httpClient;
-        WebSocketHttpHeaders headers;
-
-        std::string url("http://127.0.0.1:");
-        url += std::to_string(port);
-        url += "/data/foo.txt";
-        auto args = httpClient.createRequest(url);
-
-        args->extraHeaders = headers;
-        args->connectTimeout = 60;
-        args->transferTimeout = 60;
-        args->followRedirects = true;
-        args->maxRedirects = 10;
-        args->verbose = true;
-        args->compress = true;
-        args->logger = [](const std::string& msg) { std::cout << msg; };
-        args->onProgressCallback = [](int current, int total) -> bool {
-            std::cerr << "\r"
-                      << "Downloaded " << current << " bytes out of " << total;
-            return true;
-        };
-
-        auto response = httpClient.get(url, args);
-
-        for (auto it : response->headers)
-        {
-            std::cerr << it.first << ": " << it.second << std::endl;
-        }
-
-        std::cerr << "Upload size: " << response->uploadSize << std::endl;
-        std::cerr << "Download size: " << response->downloadSize << std::endl;
-        std::cerr << "Status: " << response->statusCode << std::endl;
-        std::cerr << "Error message: " << response->errorMsg << std::endl;
-
-        REQUIRE(response->errorCode == HttpErrorCode::Ok);
-        REQUIRE(response->statusCode == 200);
-
-        server.stop();
-    }
 }

test/IXStreamSqlTest.cpp

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

test/IXTest.cpp

@@ -84,38 +84,36 @@ namespace ix
 
     bool startWebSocketEchoServer(ix::WebSocketServer& server)
     {
-        server.setOnClientMessageCallback(
+        server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-            [&server](std::shared_ptr<ConnectionState> /*connectionState*/,
+                                                 std::shared_ptr<ConnectionState> connectionState) {
-                      ConnectionInfo& connectionInfo,
+            webSocket->setOnMessageCallback(
-                      WebSocket& webSocket,
+                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
-                      const ix::WebSocketMessagePtr& msg) {
+                    if (msg->type == ix::WebSocketMessageType::Open)
-                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)
                     {
-                        TLogger() << it.first << ": " << it.second;
+                        TLogger() << "New connection";
-                    }
+                        TLogger() << "Uri: " << msg->openInfo.uri;
-                }
+                        TLogger() << "Headers:";
-                else if (msg->type == ix::WebSocketMessageType::Close)
+                        for (auto it : msg->openInfo.headers)
-                {
-                    TLogger() << "Closed connection";
-                }
-                else if (msg->type == ix::WebSocketMessageType::Message)
-                {
-                    for (auto&& client : server.getClients())
-                    {
-                        if (client.get() != &webSocket)
                         {
-                            client->send(msg->str, msg->binary);
+                            TLogger() << it.first << ": " << it.second;
                         }
                     }
-                }
+                    else if (msg->type == ix::WebSocketMessageType::Close)
-            });
+                    {
+                        TLogger() << "Closed connection";
+                    }
+                    else if (msg->type == ix::WebSocketMessageType::Message)
+                    {
+                        for (auto&& client : server.getClients())
+                        {
+                            if (client != 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,19 +189,15 @@ namespace
         bool preferTLS = true;
         server.setTLSOptions(makeServerTLSOptions(preferTLS));
 
-        server.setOnClientMessageCallback(
+        server.setOnConnectionCallback([&server, &connectionId](
-            [&server, &connectionId](std::shared_ptr<ConnectionState> connectionState,
+                                           std::shared_ptr<ix::WebSocket> webSocket,
-                                     ConnectionInfo& connectionInfo,
+                                           std::shared_ptr<ConnectionState> connectionState) {
-                                     WebSocket& webSocket,
+            webSocket->setOnMessageCallback([webSocket, connectionState, &connectionId, &server](
-                                     const ix::WebSocketMessagePtr& msg) {
+                                                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:";
@@ -220,13 +216,14 @@ namespace
                 {
                     for (auto&& client : server.getClients())
                     {
-                        if (client.get() != &webSocket)
+                        if (client != webSocket)
                         {
                             client->send(msg->str, msg->binary);
                         }
                     }
                 }
             });
+        });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketChatTest.cpp

@@ -193,39 +193,37 @@ namespace
 
     bool startServer(ix::WebSocketServer& server)
     {
-        server.setOnClientMessageCallback(
+        server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
-            [&server](std::shared_ptr<ConnectionState> connectionState,
+                                                 std::shared_ptr<ConnectionState> connectionState) {
-                      ConnectionInfo& connectionInfo,
+            webSocket->setOnMessageCallback(
-                      WebSocket& webSocket,
+                [webSocket, connectionState, &server](const ix::WebSocketMessagePtr& msg) {
-                      const ix::WebSocketMessagePtr& msg) {
+                    if (msg->type == ix::WebSocketMessageType::Open)
-                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:";
-                    for (auto it : msg->openInfo.headers)
                     {
-                        TLogger() << it.first << ": " << it.second;
+                        TLogger() << "New connection";
-                    }
+                        TLogger() << "id: " << connectionState->getId();
-                }
+                        TLogger() << "Uri: " << msg->openInfo.uri;
-                else if (msg->type == ix::WebSocketMessageType::Close)
+                        TLogger() << "Headers:";
-                {
+                        for (auto it : msg->openInfo.headers)
-                    log("Closed connection");
-                }
-                else if (msg->type == ix::WebSocketMessageType::Message)
-                {
-                    for (auto&& client : server.getClients())
-                    {
-                        if (client.get() != &webSocket)
                         {
-                            client->sendBinary(msg->str);
+                            TLogger() << it.first << ": " << it.second;
                         }
                     }
-                }
+                    else if (msg->type == ix::WebSocketMessageType::Close)
-            });
+                    {
+                        log("Closed connection");
+                    }
+                    else if (msg->type == ix::WebSocketMessageType::Message)
+                    {
+                        for (auto&& client : server.getClients())
+                        {
+                            if (client != webSocket)
+                            {
+                                client->sendBinary(msg->str);
+                            }
+                        }
+                    }
+                });
+        });
 
         auto res = server.listen();
         if (!res.first)

test/IXWebSocketCloseTest.cpp

@@ -168,38 +168,41 @@ namespace
                      std::mutex& mutexWrite)
     {
         // A dev/null server
-        server.setOnClientMessageCallback(
+        server.setOnConnectionCallback(
             [&receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](
-                std::shared_ptr<ConnectionState> connectionState,
+                std::shared_ptr<ix::WebSocket> webSocket,
-                ConnectionInfo& connectionInfo,
+                std::shared_ptr<ConnectionState> connectionState) {
-                WebSocket& /*webSocket*/,
+                webSocket->setOnMessageCallback([webSocket,
-                const ix::WebSocketMessagePtr& msg) {
+                                                 connectionState,
-                auto remoteIp = connectionInfo.remoteIp;
+                                                 &receivedCloseCode,
-                if (msg->type == ix::WebSocketMessageType::Open)
+                                                 &receivedCloseReason,
-                {
+                                                 &receivedCloseRemote,
-                    TLogger() << "New server connection";
+                                                 &mutexWrite](const ix::WebSocketMessagePtr& msg) {
-                    TLogger() << "remote ip: " << remoteIp;
+                    if (msg->type == ix::WebSocketMessageType::Open)
-                    TLogger() << "id: " << connectionState->getId();
-                    TLogger() << "Uri: " << msg->openInfo.uri;
-                    TLogger() << "Headers:";
-                    for (auto it : msg->openInfo.headers)
                     {
-                        TLogger() << it.first << ": " << it.second;
+                        TLogger() << "New server connection";
+                        TLogger() << "id: " << connectionState->getId();
+                        TLogger() << "Uri: " << msg->openInfo.uri;
+                        TLogger() << "Headers:";
+                        for (auto it : msg->openInfo.headers)
+                        {
+                            TLogger() << it.first << ": " << it.second;
+                        }
                     }
-                }
+                    else if (msg->type == ix::WebSocketMessageType::Close)
-                else if (msg->type == ix::WebSocketMessageType::Close)
+                    {
-                {
+                        std::stringstream ss;
-                    std::stringstream ss;
+                        ss << "Server closed connection(" << msg->closeInfo.code << ","
-                    ss << "Server closed connection(" << msg->closeInfo.code << ","
+                           << msg->closeInfo.reason << ")";
-                       << msg->closeInfo.reason << ")";
+                        log(ss.str());
-                    log(ss.str());
 
-                    std::lock_guard<std::mutex> lck(mutexWrite);
+                        std::lock_guard<std::mutex> lck(mutexWrite);
 
-                    receivedCloseCode = msg->closeInfo.code;
+                        receivedCloseCode = msg->closeInfo.code;
-                    receivedCloseReason = std::string(msg->closeInfo.reason);
+                        receivedCloseReason = std::string(msg->closeInfo.reason);
-                    receivedCloseRemote = msg->closeInfo.remote;
+                        receivedCloseRemote = msg->closeInfo.remote;
-                }
+                    }
+                });
             });
 
         auto res = server.listen();

test/IXWebSocketLeakTest.cpp

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

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

test/IXWebSocketPingTest.cpp

@@ -19,7 +19,7 @@ namespace
     class WebSocketClient
     {
     public:
-        WebSocketClient(int port);
+        WebSocketClient(int port, bool useHeartBeatMethod);
 
         void start();
         void stop();
@@ -29,10 +29,12 @@ namespace
     private:
         ix::WebSocket _webSocket;
         int _port;
+        bool _useHeartBeatMethod;
     };
 
-    WebSocketClient::WebSocketClient(int port)
+    WebSocketClient::WebSocketClient(int port, bool useHeartBeatMethod)
         : _port(port)
+        , _useHeartBeatMethod(useHeartBeatMethod)
     {
         ;
     }
@@ -61,37 +63,49 @@ namespace
 
         // The important bit for this test.
         // Set a 1 second heartbeat with the setter method to test
-        _webSocket.setPingInterval(1);
+        if (_useHeartBeatMethod)
+        {
+            _webSocket.setPingInterval(1);
+        }
+        else
+        {
+            _webSocket.setPingInterval(1);
+        }
 
         std::stringstream ss;
         log(std::string("Connecting to url: ") + url);
 
-        _webSocket.setOnMessageCallback([](const ix::WebSocketMessagePtr& msg) {
+        _webSocket.setOnMessageCallback([](ix::WebSocketMessageType messageType,
+                                           const std::string& str,
+                                           size_t wireSize,
+                                           const ix::WebSocketErrorInfo& error,
+                                           const ix::WebSocketOpenInfo& openInfo,
+                                           const ix::WebSocketCloseInfo& closeInfo) {
             std::stringstream ss;
-            if (msg->type == ix::WebSocketMessageType::Open)
+            if (messageType == ix::WebSocketMessageType::Open)
             {
                 log("client connected");
             }
-            else if (msg->type == ix::WebSocketMessageType::Close)
+            else if (messageType == ix::WebSocketMessageType::Close)
             {
                 log("client disconnected");
             }
-            else if (msg->type == ix::WebSocketMessageType::Error)
+            else if (messageType == ix::WebSocketMessageType::Error)
             {
-                ss << "Error ! " << msg->errorInfo.reason;
+                ss << "Error ! " << error.reason;
                 log(ss.str());
             }
-            else if (msg->type == ix::WebSocketMessageType::Pong)
+            else if (messageType == ix::WebSocketMessageType::Pong)
             {
-                ss << "Received pong message " << msg->str;
+                ss << "Received pong message " << str;
                 log(ss.str());
             }
-            else if (msg->type == ix::WebSocketMessageType::Ping)
+            else if (messageType == ix::WebSocketMessageType::Ping)
             {
-                ss << "Received ping message " << msg->str;
+                ss << "Received ping message " << str;
                 log(ss.str());
             }
-            else if (msg->type == ix::WebSocketMessageType::Message)
+            else if (messageType == ix::WebSocketMessageType::Message)
             {
                 // too many messages to log
             }
@@ -118,28 +132,33 @@ namespace
                                              std::shared_ptr<ConnectionState> connectionState) {
                 webSocket->setOnMessageCallback(
                     [webSocket, connectionState, &server, &receivedPingMessages](
-                        const ix::WebSocketMessagePtr& msg) {
+                        ix::WebSocketMessageType messageType,
-                        if (msg->type == ix::WebSocketMessageType::Open)
+                        const std::string& str,
+                        size_t wireSize,
+                        const ix::WebSocketErrorInfo& error,
+                        const ix::WebSocketOpenInfo& openInfo,
+                        const ix::WebSocketCloseInfo& closeInfo) {
+                        if (messageType == ix::WebSocketMessageType::Open)
                         {
                             TLogger() << "New server connection";
                             TLogger() << "id: " << connectionState->getId();
-                            TLogger() << "Uri: " << msg->openInfo.uri;
+                            TLogger() << "Uri: " << openInfo.uri;
                             TLogger() << "Headers:";
-                            for (auto it : msg->openInfo.headers)
+                            for (auto it : openInfo.headers)
                             {
                                 TLogger() << it.first << ": " << it.second;
                             }
                         }
-                        else if (msg->type == ix::WebSocketMessageType::Close)
+                        else if (messageType == ix::WebSocketMessageType::Close)
                         {
                             log("Server closed connection");
                         }
-                        else if (msg->type == ix::WebSocketMessageType::Ping)
+                        else if (messageType == ix::WebSocketMessageType::Ping)
                         {
                             log("Server received a ping");
                             receivedPingMessages++;
                         }
-                        else if (msg->type == ix::WebSocketMessageType::Message)
+                        else if (messageType == ix::WebSocketMessageType::Message)
                         {
                             // to many messages to log
                             for (auto client : server.getClients())
@@ -174,7 +193,8 @@ TEST_CASE("Websocket_ping_no_data_sent_setPingInterval", "[setPingInterval]")
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = false; // so use setPingInterval
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -216,7 +236,8 @@ TEST_CASE("Websocket_ping_data_sent_setPingInterval", "[setPingInterval]")
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = false; // so use setPingInterval
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -240,7 +261,7 @@ TEST_CASE("Websocket_ping_data_sent_setPingInterval", "[setPingInterval]")
         // Here we test ping interval
         // client has sent data, but ping should have been sent no matter what
         // -> expected ping messages == 3 as 900+900+1300 = 3100 seconds, 1 ping sent every second
-        REQUIRE(serverReceivedPingMessages >= 2);
+        REQUIRE(serverReceivedPingMessages == 3);
 
         // Give us 1000ms for the server to notice that clients went away
         ix::msleep(1000);
@@ -263,7 +284,8 @@ TEST_CASE("Websocket_ping_data_sent_setPingInterval_half_full", "[setPingInterva
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = false; // so use setPingInterval
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -316,7 +338,8 @@ TEST_CASE("Websocket_ping_data_sent_setPingInterval_full", "[setPingInterval]")
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = false; // so use setPingInterval
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -340,9 +363,8 @@ TEST_CASE("Websocket_ping_data_sent_setPingInterval_full", "[setPingInterval]")
 
         // Here we test ping interval
         // client has sent data, but ping should have been sent no matter what
-        // -> expected ping messages == 2, 1 ping sent every second
+        // -> expected ping messages == 1, 1 ping sent every second
-        // The first ping is sent right away on connect
+        REQUIRE(serverReceivedPingMessages == 1);
-        REQUIRE(serverReceivedPingMessages == 2);
 
         ix::msleep(100);
 
@@ -370,7 +392,8 @@ TEST_CASE("Websocket_ping_no_data_sent_setHeartBeatPeriod", "[setPingInterval]")
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = true;
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -383,13 +406,14 @@ TEST_CASE("Websocket_ping_no_data_sent_setHeartBeatPeriod", "[setPingInterval]")
 
         REQUIRE(server.getClients().size() == 1);
 
-        ix::msleep(2100);
+        ix::msleep(1900);
 
         webSocketClient.stop();
 
+
         // Here we test ping interval
-        // -> expected ping messages == 2 as 2100 seconds, 1 ping sent every second
+        // -> expected ping messages == 1 as 1900 seconds, 1 ping sent every second
-        REQUIRE(serverReceivedPingMessages == 2);
+        REQUIRE(serverReceivedPingMessages == 1);
 
         // Give us 1000ms for the server to notice that clients went away
         ix::msleep(1000);
@@ -412,7 +436,8 @@ TEST_CASE("Websocket_ping_data_sent_setHeartBeatPeriod", "[setPingInterval]")
         REQUIRE(startServer(server, serverReceivedPingMessages));
 
         std::string session = ix::generateSessionId();
-        WebSocketClient webSocketClient(port);
+        bool useSetHeartBeatPeriodMethod = true;
+        WebSocketClient webSocketClient(port, useSetHeartBeatPeriodMethod);
 
         webSocketClient.start();
 
@@ -439,7 +464,7 @@ TEST_CASE("Websocket_ping_data_sent_setHeartBeatPeriod", "[setPingInterval]")
         // Here we test ping interval
         // client has sent data, but ping should have been sent no matter what
         // -> expected ping messages == 2 as 900+900+1100 = 2900 seconds, 1 ping sent every second
-        REQUIRE(serverReceivedPingMessages >= 2);
+        REQUIRE(serverReceivedPingMessages == 2);
 
         // Give us 1000ms for the server to notice that clients went away
         ix::msleep(1000);

test/IXWebSocketServerTest.cpp

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

test/IXWebSocketSubProtocolTest.cpp

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

test/compatibility/cpp/libwebsockets/devnull_client.cpp

@@ -1,189 +0,0 @@
-/*
- * lws-minimal-ws-client
- *
- * Written in 2010-2019 by Andy Green <andy@warmcat.com>
- *
- * This file is made available under the Creative Commons CC0 1.0
- * Universal Public Domain Dedication.
- *
- * This demonstrates the a minimal ws client using lws.
- *
- * Original programs connects to https://libwebsockets.org/ and makes a
- * wss connection to the dumb-increment protocol there.  While
- * connected, it prints the numbers it is being sent by
- * dumb-increment protocol.
- *
- * This is modified to make a test client which counts how much messages
- * per second can be received.
- *
- * libwebsockets$ make && ./a.out
- * g++ --std=c++14 -I/usr/local/opt/openssl/include devnull_client.cpp -lwebsockets
- * messages received: 0 per second 0 total
- * [2020/08/02 19:22:21:4774] U: LWS minimal ws client rx [-d <logs>] [--h2]
- * [2020/08/02 19:22:21:4814] U: callback_dumb_increment: established
- * messages received: 0 per second 0 total
- * messages received: 180015 per second 180015 total
- * messages received: 172866 per second 352881 total
- * messages received: 176177 per second 529058 total
- * messages received: 174191 per second 703249 total
- * messages received: 193397 per second 896646 total
- * messages received: 196385 per second 1093031 total
- * messages received: 194593 per second 1287624 total
- * messages received: 189484 per second 1477108 total
- * messages received: 200825 per second 1677933 total
- * messages received: 183542 per second 1861475 total
- * ^C[2020/08/02 19:22:33:4450] U: Completed OK
- * 
- */
-
-#include <libwebsockets.h>
-#include <string.h>
-#include <signal.h>
-
-#include <atomic>
-#include <thread>
-#include <iostream>
-
-static int interrupted;
-static struct lws *client_wsi;
-
-std::atomic<uint64_t> receivedCount(0);
-
-static int
-callback_dumb_increment(struct lws *wsi, enum lws_callback_reasons reason,
-              void *user, void *in, size_t len)
-{
-        switch (reason) {
-
-        /* because we are protocols[0] ... */
-        case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
-                lwsl_err("CLIENT_CONNECTION_ERROR: %s\n",
-                         in ? (char *)in : "(null)");
-                client_wsi = NULL;
-                break;
-
-        case LWS_CALLBACK_CLIENT_ESTABLISHED:
-                lwsl_user("%s: established\n", __func__);
-                break;
-
-        case LWS_CALLBACK_CLIENT_RECEIVE:
-                receivedCount++;
-                break;
-
-        case LWS_CALLBACK_CLIENT_CLOSED:
-                client_wsi = NULL;
-                break;
-
-        default:
-                break;
-        }
-
-        return lws_callback_http_dummy(wsi, reason, user, in, len);
-}
-
-static const struct lws_protocols protocols[] = {
-        {
-                "dumb-increment-protocol",
-                callback_dumb_increment,
-                0,
-                0,
-        },
-        { NULL, NULL, 0, 0 }
-};
-
-static void
-sigint_handler(int sig)
-{
-        interrupted = 1;
-}
-
-int main(int argc, const char **argv)
-{
-        uint64_t receivedCountTotal(0);
-        uint64_t receivedCountPerSecs(0);
-
-        auto timer = [&receivedCountTotal, &receivedCountPerSecs] {
-            while (!interrupted)
-            {
-                std::cerr << "messages received: "
-                          << receivedCountPerSecs
-                          << " per second "
-                          << receivedCountTotal 
-                          << " total"
-                          << std::endl;
-
-                receivedCountPerSecs = receivedCount - receivedCountTotal;
-                receivedCountTotal += receivedCountPerSecs;
-
-                auto duration = std::chrono::seconds(1);
-                std::this_thread::sleep_for(duration);
-            }
-        };
-
-        std::thread t1(timer);
-
-        struct lws_context_creation_info info;
-        struct lws_client_connect_info i;
-        struct lws_context *context;
-        const char *p;
-        int n = 0, logs = LLL_USER | LLL_ERR | LLL_WARN | LLL_NOTICE
-                /* for LLL_ verbosity above NOTICE to be built into lws, lws
-                 * must have been configured with -DCMAKE_BUILD_TYPE=DEBUG
-                 * instead of =RELEASE */
-                /* | LLL_INFO */ /* | LLL_PARSER */ /* | LLL_HEADER */
-                /* | LLL_EXT */ /* | LLL_CLIENT */ /* | LLL_LATENCY */
-                /* | LLL_DEBUG */;
-
-        signal(SIGINT, sigint_handler);
-        if ((p = lws_cmdline_option(argc, argv, "-d")))
-                logs = atoi(p);
-
-        lws_set_log_level(logs, NULL);
-        lwsl_user("LWS minimal ws client rx [-d <logs>] [--h2]\n");
-
-        memset(&info, 0, sizeof info); /* otherwise uninitialized garbage */
-        info.port = CONTEXT_PORT_NO_LISTEN; /* we do not run any server */
-        info.protocols = protocols;
-        info.timeout_secs = 10;
-
-        /*
-         * since we know this lws context is only ever going to be used with
-         * one client wsis / fds / sockets at a time, let lws know it doesn't
-         * have to use the default allocations for fd tables up to ulimit -n.
-         * It will just allocate for 1 internal and 1 (+ 1 http2 nwsi) that we
-         * will use.
-         */
-        info.fd_limit_per_thread = 1 + 1 + 1;
-
-        context = lws_create_context(&info);
-        if (!context) {
-                lwsl_err("lws init failed\n");
-                return 1;
-        }
-
-        memset(&i, 0, sizeof i); /* otherwise uninitialized garbage */
-        i.context = context;
-        i.port = 8008;
-        i.address = "127.0.0.1";
-        i.path = "/";
-        i.host = i.address;
-        i.origin = i.address;
-        i.protocol = protocols[0].name; /* "dumb-increment-protocol" */
-        i.pwsi = &client_wsi;
-
-        if (lws_cmdline_option(argc, argv, "--h2"))
-                i.alpn = "h2";
-
-        lws_client_connect_via_info(&i);
-
-        while (n >= 0 && client_wsi && !interrupted)
-                n = lws_service(context, 0);
-
-        lws_context_destroy(context);
-
-        lwsl_user("Completed %s\n", receivedCount > 10 ? "OK" : "Failed");
-
-        t1.join();
-
-        return receivedCount > 10;
-}