websocket server: closed connection threads are joined properly

* close method change and fix code

* missing mutex

CMakeLists.txt

@@ -15,11 +15,15 @@ if (NOT WIN32)
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
 endif()
 
+if ("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang")
+  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wshorten-64-to-32")
+endif()
+
 set( IXWEBSOCKET_SOURCES
-    ixwebsocket/IXEventFd.cpp
     ixwebsocket/IXSocket.cpp
     ixwebsocket/IXSocketServer.cpp
     ixwebsocket/IXSocketConnect.cpp
+    ixwebsocket/IXSocketFactory.cpp
     ixwebsocket/IXDNSLookup.cpp
     ixwebsocket/IXCancellationRequest.cpp
     ixwebsocket/IXWebSocket.cpp
@@ -29,13 +33,19 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXWebSocketPerMessageDeflate.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
+    ixwebsocket/IXWebSocketHttpHeaders.cpp
+    ixwebsocket/IXHttpClient.cpp
+    ixwebsocket/IXUrlParser.cpp
+    ixwebsocket/IXSelectInterrupt.cpp
+    ixwebsocket/IXSelectInterruptFactory.cpp
+    ixwebsocket/IXConnectionState.cpp
 )
 
 set( IXWEBSOCKET_HEADERS
-    ixwebsocket/IXEventFd.h
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketServer.h
     ixwebsocket/IXSocketConnect.h
+    ixwebsocket/IXSocketFactory.h
     ixwebsocket/IXSetThreadName.h
     ixwebsocket/IXDNSLookup.h
     ixwebsocket/IXCancellationRequest.h
@@ -51,8 +61,19 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.h
     ixwebsocket/IXWebSocketHttpHeaders.h
     ixwebsocket/libwshandshake.hpp
+    ixwebsocket/IXHttpClient.h
+    ixwebsocket/IXUrlParser.h
+    ixwebsocket/IXSelectInterrupt.h
+    ixwebsocket/IXSelectInterruptFactory.h
+    ixwebsocket/IXConnectionState.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)
@@ -60,8 +81,11 @@ elseif (WIN32)
     list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/windows/IXSetThreadName_windows.cpp)
 else()
     list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/linux/IXSetThreadName_linux.cpp)
+    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptEventFd.cpp)
+    list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSelectInterruptEventFd.h)
 endif()
 
+set(USE_OPEN_SSL FALSE)
 if (USE_TLS)
     add_definitions(-DIXWEBSOCKET_USE_TLS)
 
@@ -72,6 +96,7 @@ if (USE_TLS)
         list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSocketSChannel.h)
         list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSocketSChannel.cpp)
     else()
+        set(USE_OPEN_SSL TRUE)
         list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSocketOpenSSL.h)
         list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSocketOpenSSL.cpp)
     endif()
@@ -82,14 +107,16 @@ add_library( ixwebsocket STATIC
     ${IXWEBSOCKET_HEADERS}
 )
 
-# gcc/Linux needs -pthread
-find_package(Threads)
+if (APPLE AND USE_TLS)
+  target_link_libraries(ixwebsocket "-framework foundation" "-framework security")
+endif()
 
-if(UNIX AND NOT APPLE)
+if(USE_OPEN_SSL)
   find_package(OpenSSL REQUIRED)
   add_definitions(${OPENSSL_DEFINITIONS})
   message(STATUS "OpenSSL: " ${OPENSSL_VERSION})
   include_directories(${OPENSSL_INCLUDE_DIR})
+  target_link_libraries(ixwebsocket ${OPENSSL_LIBRARIES})
 endif()
 
 if (WIN32)
@@ -106,13 +133,19 @@ if (WIN32)
   add_definitions(-D_CRT_SECURE_NO_WARNINGS)
 
 else()
+  # gcc/Linux needs -pthread
+  find_package(Threads)
+
   target_link_libraries(ixwebsocket 
-    z ${OPENSSL_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
+    z ${CMAKE_THREAD_LIBS_INIT})
 endif()
 
 set( IXWEBSOCKET_INCLUDE_DIRS
     .
-    ../../shared/OpenSSL/include)
+)
+
 target_include_directories( ixwebsocket PUBLIC ${IXWEBSOCKET_INCLUDE_DIRS} )
 
+if (NOT WIN32)
     add_subdirectory(ws)
+endif()

DOCKER_VERSION

@@ -0,0 +1 @@
+1.4.0

Dockerfile

@@ -1 +0,0 @@
-docker/Dockerfile.debian

Dockerfile

@@ -0,0 +1,47 @@
+# Build time
+FROM debian:buster as build
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update 
+RUN apt-get -y install wget 
+RUN mkdir -p /tmp/cmake
+WORKDIR /tmp/cmake
+RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
+RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
+
+RUN apt-get -y install g++ 
+RUN apt-get -y install libssl-dev
+RUN apt-get -y install libz-dev
+RUN apt-get -y install make
+
+COPY . .
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+RUN ["make"]
+
+# Runtime
+FROM debian:buster as runtime
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update 
+# Runtime 
+RUN apt-get install -y libssl1.1 
+
+# Debugging
+RUN apt-get install -y strace
+RUN apt-get install -y gdb
+RUN apt-get install -y procps
+RUN apt-get install -y htop
+
+RUN adduser --disabled-password --gecos '' app
+COPY --chown=app:app --from=build /usr/local/bin/ws /usr/local/bin/ws
+RUN chmod +x /usr/local/bin/ws
+RUN ldd /usr/local/bin/ws
+
+# Now run in usermode
+USER app
+WORKDIR /home/app
+
+CMD ["ws"]

README.md

@@ -4,18 +4,18 @@
 
 ## Introduction
 
-[*WebSocket*](https://en.wikipedia.org/wiki/WebSocket) is a computer communications protocol, providing full-duplex
-communication channels over a single TCP connection. *IXWebSocket* is a C++ library for client and server Websocket communication. The code is derived from [easywsclient](https://github.com/dhbaird/easywsclient) and from the [Satori C SDK](https://github.com/satori-com/satori-rtm-sdk-c). It has been tested on the following platforms.
+[*WebSocket*](https://en.wikipedia.org/wiki/WebSocket) is a computer communications protocol, providing full-duplex and bi-directionnal communication channels over a single TCP connection. *IXWebSocket* is a C++ library for client and server Websocket communication, and for client HTTP communication. The code is derived from [easywsclient](https://github.com/dhbaird/easywsclient) and from the [Satori C SDK](https://github.com/satori-com/satori-rtm-sdk-c). It has been tested on the following platforms.
 
 * macOS
 * iOS
 * Linux
 * Android
-* Windows (no TLS support yet)
+
+The code was made to compile once on Windows but support is currently broken on this platform.
 
 ## Examples
 
-The ws folder countains many interactive programs for chat and file transfers demonstrating client and server usage.
+The [*ws*](https://github.com/machinezone/IXWebSocket/tree/master/ws) folder countains many interactive programs for chat, [file transfers](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_send.cpp), [curl like](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_http_client.cpp) http clients, demonstrating client and server usage.
 
 Here is what the client API looks like.
 
@@ -25,7 +25,7 @@ ix::WebSocket webSocket;
 std::string url("ws://localhost:8080/");
 webSocket.setUrl(url);
 
-// Optional heart beat, sent every 45 seconds when there isn't any traffic
+// Optional heart beat, sent every 45 seconds when there is not any traffic
 // to make sure that load balancers do not kill an idle connection.
 webSocket.setHeartBeatPeriod(45);
 
@@ -47,9 +47,12 @@ webSocket.setOnMessageCallback(
 // Now that our callback is setup, we can start our background thread and receive messages
 webSocket.start();
 
-// Send a message to the server
+// Send a message to the server (default to BINARY mode)
 webSocket.send("hello world");
 
+// The message can be sent in TEXT mode
+webSocket.sendText("hello again");
+
 // ... finally ...
 
 // Stop the connection
@@ -64,10 +67,11 @@ Here is what the server API looks like. Note that server support is very recent
 ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
-    [&server](std::shared_ptr<ix::WebSocket> webSocket)
+    [&server](std::shared_ptr<WebSocket> webSocket,
+              std::shared_ptr<ConnectionState> connectionState)
     {
         webSocket->setOnMessageCallback(
-            [webSocket, &server](ix::WebSocketMessageType messageType,
+            [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
                const std::string& str,
                size_t wireSize,
                const ix::WebSocketErrorInfo& error,
@@ -77,7 +81,16 @@ server.setOnConnectionCallback(
                 if (messageType == ix::WebSocket_MessageType_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: " << openInfo.uri << std::endl;
+
                     std::cerr << "Headers:" << std::endl;
                     for (auto it : openInfo.headers)
                     {
@@ -110,12 +123,81 @@ server.wait();
 
 ```
 
+Here is what the HTTP client API looks like. Note that HTTP client support is very recent and subject to changes.
+
+```
+//
+// Preparation
+//
+HttpClient httpClient;
+HttpRequestArgs args;
+
+// Custom headers can be set
+WebSocketHttpHeaders headers;
+headers["Foo"] = "bar";
+args.extraHeaders = headers;
+
+// Timeout options
+args.connectTimeout = connectTimeout;
+args.transferTimeout = transferTimeout;
+
+// Redirect options
+args.followRedirects = followRedirects;
+args.maxRedirects = maxRedirects;
+
+// Misc
+args.compress = compress; // Enable gzip compression
+args.verbose = verbose;
+args.logger = [](const std::string& msg)
+{
+    std::cout << msg;
+};
+
+//
+// Request
+//
+HttpResponse out;
+std::string url = "https://www.google.com";
+
+// HEAD request
+out = httpClient.head(url, args);
+
+// GET request
+out = httpClient.get(url, args);
+
+// POST request with parameters
+HttpParameters httpParameters;
+httpParameters["foo"] = "bar";
+out = httpClient.post(url, httpParameters, args);
+
+// POST request with a body
+out = httpClient.post(url, std::string("foo=bar"), args);
+
+//
+// Result
+//
+auto statusCode = std::get<0>(out);
+auto errorCode = std::get<1>(out);
+auto responseHeaders = std::get<2>(out);
+auto payload = std::get<3>(out);
+auto errorMsg = std::get<4>(out);
+auto uploadSize = std::get<5>(out);
+auto downloadSize = std::get<6>(out);
+```
+
 ## Build
 
 CMakefiles for the library and the examples are available. This library has few dependencies, so it is possible to just add the source files into your project.
 
 There is a Dockerfile for running some code on Linux, and a unittest which can be executed by typing `make test`.
 
+You can build and install the ws command line tool with Homebrew.
+
+```
+brew tap bsergean/IXWebSocket
+brew install IXWebSocket
+```
+
 ## Implementation details
 
 ### Per Message Deflate compression.
@@ -136,17 +218,17 @@ If the remote end (server) breaks the connection, the code will try to perpetual
 
 ### Large messages
 
-Large frames are broken up into smaller chunks or messages to avoid filling up the os tcp buffers, which is permitted thanks to WebSocket [fragmentation](https://tools.ietf.org/html/rfc6455#section-5.4). Messages up to 500M were sent and received succesfully.
+Large frames are broken up into smaller chunks or messages to avoid filling up the os tcp buffers, which is permitted thanks to WebSocket [fragmentation](https://tools.ietf.org/html/rfc6455#section-5.4). Messages up to 1G were sent and received succesfully.
 
 ## Limitations
 
-* There is no text support for sending data, only the binary protocol is supported. Sending json or text over the binary protocol works well.
+* No utf-8 validation is made when sending TEXT message with sendText()
 * Automatic reconnection works at the TCP socket level, and will detect remote end disconnects. However, if the device/computer network become unreachable (by turning off wifi), it is quite hard to reliably and timely detect it at the socket level using `recv` and `send` error codes. [Here](https://stackoverflow.com/questions/14782143/linux-socket-how-to-detect-disconnected-network-in-a-client-program) is a good discussion on the subject. This behavior is consistent with other runtimes such as node.js. One way to detect a disconnected device with low level C code is to do a name resolution with DNS but this can be expensive. Mobile devices have good and reliable API to do that.
-* The server code is using select to detect incoming data, and creates one OS thread per connection. This isn't as scalable as strategies using epoll or kqueue.
+* The server code is using select to detect incoming data, and creates one OS thread per connection. This is not as scalable as strategies using epoll or kqueue.
 
 ## C++ code organization
 
-Here's a simplistic diagram which explains how the code is structured in term of class/modules.
+Here is a simplistic diagram which explains how the code is structured in term of class/modules.
 
 ```
 +-----------------------+ --- Public
@@ -198,7 +280,7 @@ If the connection was closed and sending failed, the return value will be set to
 1. WebSocket_ReadyState_Connecting - The connection is not yet open.
 2. WebSocket_ReadyState_Open       - The connection is open and ready to communicate.
 3. WebSocket_ReadyState_Closing    - The connection is in the process of closing.
-4. WebSocket_MessageType_Close     - The connection is closed or couldn't be opened.
+4. WebSocket_MessageType_Close     - The connection is closed or could not be opened.
 
 ### Open and Close notifications
 
@@ -308,7 +390,7 @@ websocket.ping("ping data, optional (empty string is ok): limited to 125 bytes l
 ### Heartbeat.
 
 You can configure an optional heart beat / keep-alive, sent every 45 seconds
-when there isn't any traffic to make sure that load balancers do not kill an
+when there is no any traffic to make sure that load balancers do not kill an
 idle connection.
 
 ```

docker-compose.yml

@@ -0,0 +1,21 @@
+version: "3"
+services:
+  ws:
+    stdin_open: true
+    tty: true
+    image: bsergean/ws:build
+    ports:
+      - "8765:8765"
+    entrypoint: bash
+    networks:
+      - ws-net
+    depends_on:
+      - redis1
+
+  redis1:
+    image: redis:alpine
+    networks:
+      - ws-net
+
+networks:
+  ws-net:

docker/Dockerfile

@@ -1,16 +0,0 @@
-FROM debian:stretch
-
-# RUN yum install -y gcc-c++ make cmake openssl-devel gdb
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-RUN apt-get -y install g++ 
-RUN apt-get -y install libssl-dev
-RUN apt-get -y install gdb
-RUN apt-get -y install screen
-RUN apt-get -y install procps
-RUN apt-get -y install lsof
-
-COPY . .
-
-WORKDIR examples/ws_connect
-RUN ["sh", "build_linux.sh"]

docker/Dockerfile.alpine

@@ -1,11 +0,0 @@
-FROM alpine:3.8
-
-RUN apk add --no-cache g++ musl-dev make cmake openssl-dev
-
-COPY . .
-
-WORKDIR examples/ws_connect
-RUN ["sh", "build_linux.sh"]
-
-EXPOSE 8765
-CMD ["ws_connect"]

docker/Dockerfile.centos

@@ -1,11 +0,0 @@
-FROM alpine:3.8
-
-RUN apk add --no-cache g++ musl-dev make cmake openssl-dev
-
-COPY . .
-
-WORKDIR examples/ws_connect
-RUN ["sh", "build_linux.sh"]
-
-EXPOSE 8765
-CMD ["ws_connect"]

docker/Dockerfile.debian

@@ -1,22 +0,0 @@
-FROM debian:stretch
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-RUN apt-get -y install g++ 
-RUN apt-get -y install libssl-dev
-RUN apt-get -y install gdb
-RUN apt-get -y install screen
-RUN apt-get -y install procps
-RUN apt-get -y install lsof
-RUN apt-get -y install libz-dev
-RUN apt-get -y install vim
-RUN apt-get -y install make
-RUN apt-get -y install cmake
-
-COPY . .
-
-WORKDIR ws
-RUN ["sh", "docker_build.sh"]
-
-EXPOSE 8765
-CMD ["/ws/ws", "transfer", "8765"]

docker/Dockerfile.gcc

@@ -1,8 +0,0 @@
-FROM gcc:8
-
-# RUN yum install -y gcc-c++ make cmake openssl-devel gdb
-
-COPY . .
-
-WORKDIR examples/ws_connect
-RUN ["sh", "build_linux.sh"]

ixwebsocket/IXConnectionState.cpp

@@ -0,0 +1,43 @@
+/*
+ *  IXConnectionState.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXConnectionState.h"
+
+namespace ix
+{
+    std::atomic<uint64_t> ConnectionState::_globalId(0);
+
+    ConnectionState::ConnectionState() : _terminated(false)
+    {
+        computeId();
+    }
+
+    void ConnectionState::computeId()
+    {
+        _id = std::to_string(_globalId++);
+    }
+
+    const std::string& ConnectionState::getId() const
+    {
+        return _id;
+    }
+
+    std::shared_ptr<ConnectionState> ConnectionState::createConnectionState()
+    {
+        return std::make_shared<ConnectionState>();
+    }
+
+    bool ConnectionState::isTerminated() const
+    {
+        return _terminated;
+    }
+
+    bool ConnectionState::setTerminated()
+    {
+        _terminated = true;
+    }
+}
+

ixwebsocket/IXConnectionState.h

@@ -0,0 +1,37 @@
+/*
+ *  IXConnectionState.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <stdint.h>
+#include <string>
+#include <atomic>
+#include <memory>
+
+namespace ix
+{
+    class ConnectionState {
+    public:
+        ConnectionState();
+        virtual ~ConnectionState() = default;
+
+        virtual void computeId();
+        virtual const std::string& getId() const;
+
+        bool setTerminated();
+        bool isTerminated() const;
+
+        static std::shared_ptr<ConnectionState> createConnectionState();
+
+    protected:
+        std::atomic<bool> _terminated;
+        std::string _id;
+
+        static std::atomic<uint64_t> _globalId;
+    };
+}
+
+

ixwebsocket/IXDNSLookup.cpp

@@ -73,7 +73,7 @@ namespace ix
         errMsg = "no error";
 
         // Maybe a cancellation request got in before the background thread terminated ?
-        if (isCancellationRequested())
+        if (isCancellationRequested && isCancellationRequested())
         {
             errMsg = "cancellation requested";
             return nullptr;
@@ -121,7 +121,7 @@ namespace ix
             }
 
             // Were we cancelled ?
-            if (isCancellationRequested())
+            if (isCancellationRequested && isCancellationRequested())
             {
                 errMsg = "cancellation requested";
                 return nullptr;
@@ -129,7 +129,7 @@ namespace ix
         }
 
         // Maybe a cancellation request got in before the bg terminated ?
-        if (isCancellationRequested())
+        if (isCancellationRequested && isCancellationRequested())
         {
             errMsg = "cancellation requested";
             return nullptr;

ixwebsocket/IXEventFd.cpp

@@ -1,82 +0,0 @@
-/*
- *  IXEventFd.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
- */
-
-//
-// Linux/Android has a special type of virtual files. select(2) will react
-// when reading/writing to those files, unlike closing sockets.
-//
-// https://linux.die.net/man/2/eventfd
-// http://www.sourcexr.com/articles/2013/10/26/lightweight-inter-process-signaling-with-eventfd
-//
-// eventfd was added in Linux kernel 2.x, and our oldest Android (Kitkat 4.4)
-// is on Kernel 3.x
-//
-// cf Android/Kernel table here
-// https://android.stackexchange.com/questions/51651/which-android-runs-which-linux-kernel
-//
-
-#include "IXEventFd.h"
-
-#ifdef __linux__
-# include <sys/eventfd.h>
-#endif
-
-#ifndef _WIN32
-#include <unistd.h> // for write
-#endif
-
-namespace ix
-{
-    EventFd::EventFd() :
-        _eventfd(-1)
-    {
-#ifdef __linux__
-        _eventfd = eventfd(0, 0);
-#endif
-    }
-
-    EventFd::~EventFd()
-    {
-#ifdef __linux__
-        ::close(_eventfd);
-#endif
-    }
-
-    bool EventFd::notify()
-    {
-#if defined(__linux__)
-        if (_eventfd == -1) return false;
-
-        // select will wake up when a non-zero value is written to our eventfd
-        uint64_t value = 1;
-
-        // we should write 8 bytes for an uint64_t
-        return write(_eventfd, &value, sizeof(value)) == 8;
-#else
-        return true;
-#endif
-    }
-
-    bool EventFd::clear()
-    {
-#if defined(__linux__)
-        if (_eventfd == -1) return false;
-
-        // 0 is a special value ; select will not wake up
-        uint64_t value = 0;
-
-        // we should write 8 bytes for an uint64_t
-        return write(_eventfd, &value, sizeof(value)) == 8;
-#else
-        return true;
-#endif
-    }
-
-    int EventFd::getFd()
-    {
-        return _eventfd;
-    }
-}

ixwebsocket/IXEventFd.h

@@ -1,23 +0,0 @@
-/*
- *  IXEventFd.h
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
- */
-
-#pragma once
-
-namespace ix
-{
-    class EventFd {
-    public:
-        EventFd();
-        virtual ~EventFd();
-
-        bool notify();
-        bool clear();
-        int getFd();
-
-    private:
-        int _eventfd;
-    };
-}

ixwebsocket/IXHttpClient.cpp

@@ -0,0 +1,467 @@
+/*
+ *  IXHttpClient.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXHttpClient.h"
+#include "IXUrlParser.h"
+#include "IXWebSocketHttpHeaders.h"
+#include "IXSocketFactory.h"
+
+#include <sstream>
+#include <iomanip>
+#include <vector>
+#include <cstring>
+
+#include <zlib.h>
+
+namespace ix
+{
+    const std::string HttpClient::kPost = "POST";
+    const std::string HttpClient::kGet = "GET";
+    const std::string HttpClient::kHead = "HEAD";
+
+    HttpClient::HttpClient()
+    {
+
+    }
+
+    HttpClient::~HttpClient()
+    {
+
+    }
+
+    HttpResponse HttpClient::request(
+        const std::string& url,
+        const std::string& verb,
+        const std::string& body,
+        const HttpRequestArgs& args,
+        int redirects)
+    {
+        uint64_t uploadSize = 0;
+        uint64_t downloadSize = 0;
+        int code = 0;
+        WebSocketHttpHeaders headers;
+        std::string payload;
+
+        std::string protocol, host, path, query;
+        int port;
+        bool websocket = false;
+
+        if (!UrlParser::parse(url, protocol, host, path, query, port, websocket))
+        {
+            std::stringstream ss;
+            ss << "Cannot parse url: " << url;
+            return std::make_tuple(code, HttpErrorCode_UrlMalformed,
+                                   headers, payload, ss.str(),
+                                   uploadSize, downloadSize);
+        }
+
+        bool tls = protocol == "https";
+        std::string errorMsg;
+        _socket = createSocket(tls, errorMsg);
+
+        if (!_socket)
+        {
+            return std::make_tuple(code, HttpErrorCode_CannotCreateSocket,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        // Build request string
+        std::stringstream ss;
+        ss << verb << " " << path << " HTTP/1.1\r\n";
+        ss << "Host: " << host << "\r\n";
+        ss << "User-Agent: ixwebsocket/1.0.0" << "\r\n";
+        ss << "Accept: */*" << "\r\n";
+
+        if (args.compress)
+        {
+            ss << "Accept-Encoding: gzip" << "\r\n";
+        }
+
+        // Append extra headers
+        for (auto&& it : args.extraHeaders)
+        {
+            ss << it.first << ": " << it.second << "\r\n";
+        }
+
+        if (verb == kPost)
+        {
+            ss << "Content-Length: " << body.size() << "\r\n";
+
+            // Set default Content-Type if unspecified
+            if (args.extraHeaders.find("Content-Type") == args.extraHeaders.end())
+            {
+                ss << "Content-Type: application/x-www-form-urlencoded" << "\r\n";
+            }
+            ss << "\r\n";
+            ss << body;
+        }
+        else
+        {
+            ss << "\r\n";
+        }
+
+        std::string req(ss.str());
+        std::string errMsg;
+        std::atomic<bool> requestInitCancellation(false);
+
+        // Make a cancellation object dealing with connection timeout
+        auto isCancellationRequested =
+            makeCancellationRequestWithTimeout(args.connectTimeout, requestInitCancellation);
+
+        bool success = _socket->connect(host, port, errMsg, isCancellationRequested);
+        if (!success)
+        {
+            std::stringstream ss;
+            ss << "Cannot connect to url: " << url;
+            return std::make_tuple(code, HttpErrorCode_CannotConnect,
+                                   headers, payload, ss.str(),
+                                   uploadSize, downloadSize);
+        }
+
+        // Make a new cancellation object dealing with transfer timeout
+        isCancellationRequested =
+            makeCancellationRequestWithTimeout(args.transferTimeout, requestInitCancellation);
+
+        if (args.verbose)
+        {
+            std::stringstream ss;
+            ss << "Sending " << verb << " request "
+               << "to " << host << ":" << port << std::endl
+               << "request size: " << req.size() << " bytes" << std::endl
+               << "=============" << std::endl
+               << req
+               << "=============" << std::endl
+               << std::endl;
+
+            log(ss.str(), args);
+        }
+
+        if (!_socket->writeBytes(req, isCancellationRequested))
+        {
+            std::string errorMsg("Cannot send request");
+            return std::make_tuple(code, HttpErrorCode_SendError,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        uploadSize = req.size();
+
+        auto lineResult = _socket->readLine(isCancellationRequested);
+        auto lineValid = lineResult.first;
+        auto line = lineResult.second;
+
+        if (!lineValid)
+        {
+            std::string errorMsg("Cannot retrieve status line");
+            return std::make_tuple(code, HttpErrorCode_CannotReadStatusLine,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        if (args.verbose)
+        {
+            std::stringstream ss;
+            ss << "Status line " << line;
+            log(ss.str(), args);
+        }
+
+        if (sscanf(line.c_str(), "HTTP/1.1 %d", &code) != 1)
+        {
+            std::string errorMsg("Cannot parse response code from status line");
+            return std::make_tuple(code, HttpErrorCode_MissingStatus,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        auto result = parseHttpHeaders(_socket, isCancellationRequested);
+        auto headersValid = result.first;
+        headers = result.second;
+
+        if (!headersValid)
+        {
+            std::string errorMsg("Cannot parse http headers");
+            return std::make_tuple(code, HttpErrorCode_HeaderParsingError,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        // Redirect ?
+        if ((code >= 301 && code <= 308) && args.followRedirects)
+        {
+            if (headers.find("Location") == headers.end())
+            {
+                std::string errorMsg("Missing location header for redirect");
+                return std::make_tuple(code, HttpErrorCode_MissingLocation,
+                                       headers, payload, errorMsg,
+                                       uploadSize, downloadSize);
+            }
+
+            if (redirects >= args.maxRedirects)
+            {
+                std::stringstream ss;
+                ss << "Too many redirects: " << redirects;
+                return std::make_tuple(code, HttpErrorCode_TooManyRedirects,
+                                       headers, payload, ss.str(),
+                                       uploadSize, downloadSize);
+            }
+
+            // Recurse
+            std::string location = headers["Location"];
+            return request(location, verb, body, args, redirects+1);
+        }
+
+        if (verb == "HEAD")
+        {
+            return std::make_tuple(code, HttpErrorCode_Ok,
+                                   headers, payload, std::string(),
+                                   uploadSize, downloadSize);
+        }
+
+        // Parse response:
+        if (headers.find("Content-Length") != headers.end())
+        {
+            ssize_t contentLength = -1;
+            ss.str("");
+            ss << headers["Content-Length"];
+            ss >> contentLength;
+
+            payload.reserve(contentLength);
+
+            auto chunkResult = _socket->readBytes(contentLength,
+                                                  args.onProgressCallback,
+                                                  isCancellationRequested);
+            if (!chunkResult.first)
+            {
+                errorMsg = "Cannot read chunk";
+                return std::make_tuple(code, HttpErrorCode_ChunkReadError,
+                                       headers, payload, errorMsg,
+                                       uploadSize, downloadSize);
+            }
+            payload += chunkResult.second;
+        }
+        else if (headers.find("Transfer-Encoding") != headers.end() &&
+                 headers["Transfer-Encoding"] == "chunked")
+        {
+            std::stringstream ss;
+
+            while (true)
+            {
+                lineResult = _socket->readLine(isCancellationRequested);
+                line = lineResult.second;
+
+                if (!lineResult.first)
+                {
+                    return std::make_tuple(code, HttpErrorCode_ChunkReadError,
+                                           headers, payload, errorMsg,
+                                           uploadSize, downloadSize);
+                }
+
+                uint64_t chunkSize;
+                ss.str("");
+                ss << std::hex << line;
+                ss >> chunkSize;
+
+                if (args.verbose)
+                {
+                    std::stringstream oss;
+                    oss << "Reading " << chunkSize << " bytes"
+                        << std::endl;
+                    log(oss.str(), args);
+                }
+
+                payload.reserve(payload.size() + chunkSize);
+
+                // Read a chunk
+                auto chunkResult = _socket->readBytes(chunkSize,
+                                                      args.onProgressCallback,
+                                                      isCancellationRequested);
+                if (!chunkResult.first)
+                {
+                    errorMsg = "Cannot read chunk";
+                    return std::make_tuple(code, HttpErrorCode_ChunkReadError,
+                                           headers, payload, errorMsg,
+                                           uploadSize, downloadSize);
+                }
+                payload += chunkResult.second;
+
+                // Read the line that terminates the chunk (\r\n)
+                lineResult = _socket->readLine(isCancellationRequested);
+
+                if (!lineResult.first)
+                {
+                    return std::make_tuple(code, HttpErrorCode_ChunkReadError,
+                                           headers, payload, errorMsg,
+                                           uploadSize, downloadSize);
+                }
+
+                if (chunkSize == 0) break;
+            }
+        }
+        else if (code == 204)
+        {
+            ; // 204 is NoContent response code
+        }
+        else
+        {
+            std::string errorMsg("Cannot read http body");
+            return std::make_tuple(code, HttpErrorCode_CannotReadBody,
+                                   headers, payload, errorMsg,
+                                   uploadSize, downloadSize);
+        }
+
+        downloadSize = payload.size();
+
+        // If the content was compressed with gzip, decode it
+        if (headers["Content-Encoding"] == "gzip")
+        {
+            std::string decompressedPayload;
+            if (!gzipInflate(payload, decompressedPayload))
+            {
+                std::string errorMsg("Error decompressing payload");
+                return std::make_tuple(code, HttpErrorCode_Gzip,
+                                       headers, payload, errorMsg,
+                                       uploadSize, downloadSize);
+            }
+            payload = decompressedPayload;
+        }
+
+        return std::make_tuple(code, HttpErrorCode_Ok,
+                               headers, payload, std::string(),
+                               uploadSize, downloadSize);
+    }
+
+    HttpResponse HttpClient::get(const std::string& url,
+                                 const HttpRequestArgs& args)
+    {
+        return request(url, kGet, std::string(), args);
+    }
+
+    HttpResponse HttpClient::head(const std::string& url,
+                                  const HttpRequestArgs& args)
+    {
+        return request(url, kHead, std::string(), args);
+    }
+
+    HttpResponse HttpClient::post(const std::string& url,
+                                  const HttpParameters& httpParameters,
+                                  const HttpRequestArgs& args)
+    {
+        return request(url, kPost, serializeHttpParameters(httpParameters), args);
+    }
+
+    HttpResponse HttpClient::post(const std::string& url,
+                                  const std::string& body,
+                                  const HttpRequestArgs& args)
+    {
+        return request(url, kPost, body, args);
+    }
+
+    std::string HttpClient::urlEncode(const std::string& value)
+    {
+        std::ostringstream escaped;
+        escaped.fill('0');
+        escaped << std::hex;
+
+        for (std::string::const_iterator i = value.begin(), n = value.end();
+             i != n; ++i)
+        {
+            std::string::value_type c = (*i);
+
+            // Keep alphanumeric and other accepted characters intact
+            if (isalnum(c) || c == '-' || c == '_' || c == '.' || c == '~')
+            {
+                escaped << c;
+                continue;
+            }
+
+            // Any other characters are percent-encoded
+            escaped << std::uppercase;
+            escaped << '%' << std::setw(2) << int((unsigned char) c);
+            escaped << std::nouppercase;
+        }
+
+        return escaped.str();
+    }
+
+    std::string HttpClient::serializeHttpParameters(const HttpParameters& httpParameters)
+    {
+        std::stringstream ss;
+        size_t count = httpParameters.size();
+        size_t i = 0;
+
+        for (auto&& it : httpParameters)
+        {
+            ss << urlEncode(it.first)
+               << "="
+               << urlEncode(it.second);
+
+            if (i++ < (count-1))
+            {
+               ss << "&";
+            }
+        }
+        return ss.str();
+    }
+
+    bool HttpClient::gzipInflate(
+        const std::string& in,
+        std::string& out)
+    {
+        z_stream inflateState;
+        std::memset(&inflateState, 0, sizeof(inflateState));
+
+        inflateState.zalloc = Z_NULL;
+        inflateState.zfree = Z_NULL;
+        inflateState.opaque = Z_NULL;
+        inflateState.avail_in = 0;
+        inflateState.next_in = Z_NULL;
+
+        if (inflateInit2(&inflateState, 16+MAX_WBITS) != Z_OK)
+        {
+            return false;
+        }
+
+        inflateState.avail_in = (uInt) in.size();
+        inflateState.next_in = (unsigned char *)(const_cast<char *>(in.data()));
+
+        const int kBufferSize = 1 << 14;
+
+        std::unique_ptr<unsigned char[]> compressBuffer =
+            std::make_unique<unsigned char[]>(kBufferSize);
+
+        do
+        {
+            inflateState.avail_out = (uInt) kBufferSize;
+            inflateState.next_out = compressBuffer.get();
+
+            int ret = inflate(&inflateState, Z_SYNC_FLUSH);
+
+            if (ret == Z_NEED_DICT || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR)
+            {
+                inflateEnd(&inflateState);
+                return false;
+            }
+
+            out.append(
+                reinterpret_cast<char *>(compressBuffer.get()),
+                kBufferSize - inflateState.avail_out
+            );
+        } while (inflateState.avail_out == 0);
+
+        inflateEnd(&inflateState);
+        return true;
+    }
+
+    void HttpClient::log(const std::string& msg,
+                         const HttpRequestArgs& args)
+    {
+        if (args.logger)
+        {
+            args.logger(msg);
+        }
+    }
+}

ixwebsocket/IXHttpClient.h

@@ -0,0 +1,107 @@
+/*
+ *  IXHttpClient.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <functional>
+#include <mutex>
+#include <atomic>
+#include <tuple>
+#include <memory>
+#include <map>
+
+#include "IXSocket.h"
+#include "IXWebSocketHttpHeaders.h"
+
+namespace ix
+{
+    enum HttpErrorCode
+    {
+        HttpErrorCode_Ok = 0,
+        HttpErrorCode_CannotConnect = 1,
+        HttpErrorCode_Timeout = 2,
+        HttpErrorCode_Gzip = 3,
+        HttpErrorCode_UrlMalformed = 4,
+        HttpErrorCode_CannotCreateSocket = 5,
+        HttpErrorCode_SendError = 6,
+        HttpErrorCode_ReadError = 7,
+        HttpErrorCode_CannotReadStatusLine = 8,
+        HttpErrorCode_MissingStatus = 9,
+        HttpErrorCode_HeaderParsingError = 10,
+        HttpErrorCode_MissingLocation = 11,
+        HttpErrorCode_TooManyRedirects = 12,
+        HttpErrorCode_ChunkReadError = 13,
+        HttpErrorCode_CannotReadBody = 14
+    };
+
+    using HttpResponse = std::tuple<int, // status
+                                    HttpErrorCode, // error code
+                                    WebSocketHttpHeaders,
+                                    std::string, // payload
+                                    std::string, // error msg
+                                    uint64_t,    // upload size
+                                    uint64_t>;   // download size
+
+    using HttpParameters = std::map<std::string, std::string>;
+    using Logger = std::function<void(const std::string&)>;
+
+    struct HttpRequestArgs
+    {
+        std::string url;
+        WebSocketHttpHeaders extraHeaders;
+        std::string body;
+        int connectTimeout;
+        int transferTimeout;
+        bool followRedirects;
+        int maxRedirects;
+        bool verbose;
+        bool compress;
+        Logger logger;
+        OnProgressCallback onProgressCallback;
+    };
+
+    class HttpClient {
+    public:
+        HttpClient();
+        ~HttpClient();
+
+        HttpResponse get(const std::string& url,
+                         const HttpRequestArgs& args);
+        HttpResponse head(const std::string& url,
+                          const HttpRequestArgs& args);
+
+        HttpResponse post(const std::string& url,
+                          const HttpParameters& httpParameters,
+                          const HttpRequestArgs& args);
+        HttpResponse post(const std::string& url,
+                          const std::string& body,
+                          const HttpRequestArgs& args);
+
+    private:
+        HttpResponse request(const std::string& url,
+                             const std::string& verb,
+                             const std::string& body,
+                             const HttpRequestArgs& args,
+                             int redirects = 0);
+
+        std::string serializeHttpParameters(const HttpParameters& httpParameters);
+
+        std::string urlEncode(const std::string& value);
+
+        void log(const std::string& msg, const HttpRequestArgs& args);
+
+        bool gzipInflate(
+            const std::string& in,
+            std::string& out);
+
+        std::shared_ptr<Socket> _socket;
+
+        const static std::string kPost;
+        const static std::string kGet;
+        const static std::string kHead;
+    };
+}

ixwebsocket/IXSelectInterrupt.cpp

@@ -0,0 +1,46 @@
+/*
+ *  IXSelectInterrupt.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXSelectInterrupt.h"
+
+namespace ix
+{
+    SelectInterrupt::SelectInterrupt()
+    {
+        ;
+    }
+
+    SelectInterrupt::~SelectInterrupt()
+    {
+        ;
+    }
+
+    bool SelectInterrupt::init(std::string& /*errorMsg*/)
+    {
+        return true;
+    }
+
+    bool SelectInterrupt::notify(uint64_t /*value*/)
+    {
+        return true;
+    }
+
+    uint64_t SelectInterrupt::read()
+    {
+        return 0;
+    }
+
+    bool SelectInterrupt::clear()
+    {
+        return true;
+    }
+
+    int SelectInterrupt::getFd() const
+    {
+        return -1;
+    }
+}
+

ixwebsocket/IXSelectInterrupt.h

@@ -0,0 +1,28 @@
+/*
+ *  IXSelectInterrupt.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <stdint.h>
+#include <string>
+
+namespace ix
+{
+    class SelectInterrupt {
+    public:
+        SelectInterrupt();
+        virtual ~SelectInterrupt();
+
+        virtual bool init(std::string& errorMsg);
+
+        virtual bool notify(uint64_t value);
+        virtual bool clear();
+        virtual uint64_t read();
+        virtual int getFd() const;
+    };
+}
+
+

ixwebsocket/IXSelectInterruptEventFd.cpp

@@ -0,0 +1,116 @@
+/*
+ *  IXSelectInterruptEventFd.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018-2019 Machine Zone, Inc. All rights reserved.
+ */
+
+//
+// On Linux we use eventd to wake up select.
+//
+
+//
+// Linux/Android has a special type of virtual files. select(2) will react
+// when reading/writing to those files, unlike closing sockets.
+//
+// https://linux.die.net/man/2/eventfd
+// http://www.sourcexr.com/articles/2013/10/26/lightweight-inter-process-signaling-with-eventfd
+//
+// eventfd was added in Linux kernel 2.x, and our oldest Android (Kitkat 4.4)
+// is on Kernel 3.x
+//
+// cf Android/Kernel table here
+// https://android.stackexchange.com/questions/51651/which-android-runs-which-linux-kernel
+//
+// On macOS we use UNIX pipes to wake up select.
+//
+
+#include "IXSelectInterruptEventFd.h"
+
+#include <sys/eventfd.h>
+
+#include <unistd.h> // for write
+#include <string.h> // for strerror
+#include <fcntl.h>
+#include <errno.h>
+#include <assert.h>
+#include <sstream>
+
+namespace ix
+{
+    SelectInterruptEventFd::SelectInterruptEventFd()
+    {
+        _eventfd = -1;
+    }
+
+    SelectInterruptEventFd::~SelectInterruptEventFd()
+    {
+        ::close(_eventfd);
+    }
+
+    bool SelectInterruptEventFd::init(std::string& errorMsg)
+    {
+        // calling init twice is a programming error
+        assert(_eventfd == -1);
+
+        _eventfd = eventfd(0, 0);
+        if (_eventfd < 0)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptEventFd::init() failed in eventfd()"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _eventfd = -1;
+            return false;
+        }
+
+        if (fcntl(_eventfd, F_SETFL, O_NONBLOCK) == -1)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptEventFd::init() failed in fcntl() call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _eventfd = -1;
+            return false;
+        }
+
+        return true;
+    }
+
+    bool SelectInterruptEventFd::notify(uint64_t value)
+    {
+        int fd = _eventfd;
+
+        if (fd == -1) return false;
+
+        // we should write 8 bytes for an uint64_t
+        return write(fd, &value, sizeof(value)) == 8;
+    }
+
+    // TODO: return max uint64_t for errors ?
+    uint64_t SelectInterruptEventFd::read()
+    {
+        int fd = _eventfd;
+
+        uint64_t value = 0;
+        ::read(fd, &value, sizeof(value));
+        return value;
+    }
+
+    bool SelectInterruptEventFd::clear()
+    {
+        if (_eventfd == -1) return false;
+
+        // 0 is a special value ; select will not wake up
+        uint64_t value = 0;
+
+        // we should write 8 bytes for an uint64_t
+        return write(_eventfd, &value, sizeof(value)) == 8;
+    }
+
+    int SelectInterruptEventFd::getFd() const
+    {
+        return _eventfd;
+    }
+}

ixwebsocket/IXSelectInterruptEventFd.h

@@ -0,0 +1,32 @@
+/*
+ *  IXSelectInterruptEventFd.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018-2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include "IXSelectInterrupt.h"
+
+#include <stdint.h>
+#include <string>
+
+namespace ix
+{
+    class SelectInterruptEventFd : public SelectInterrupt {
+    public:
+        SelectInterruptEventFd();
+        virtual ~SelectInterruptEventFd();
+
+        bool init(std::string& errorMsg) final;
+
+        bool notify(uint64_t value) final;
+        bool clear() final;
+        uint64_t read() final;
+        int getFd() const final;
+
+    private:
+        int _eventfd;
+    };
+}
+

ixwebsocket/IXSelectInterruptFactory.cpp

@@ -0,0 +1,25 @@
+/*
+ *  IXSelectInterruptFactory.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXSelectInterruptFactory.h"
+
+#if defined(__linux__) || defined(__APPLE__)
+# include <ixwebsocket/IXSelectInterruptPipe.h>
+#else
+# include <ixwebsocket/IXSelectInterrupt.h>
+#endif
+
+namespace ix
+{
+    std::shared_ptr<SelectInterrupt> createSelectInterrupt()
+    {
+#if defined(__linux__) || defined(__APPLE__)
+        return std::make_shared<SelectInterruptPipe>();
+#else
+        return std::make_shared<SelectInterrupt>();
+#endif
+    }
+}

ixwebsocket/IXSelectInterruptFactory.h

@@ -0,0 +1,15 @@
+/*
+ *  IXSelectInterruptFactory.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace ix
+{
+    class SelectInterrupt;
+    std::shared_ptr<SelectInterrupt> createSelectInterrupt();
+}

ixwebsocket/IXSelectInterruptPipe.cpp

@@ -0,0 +1,138 @@
+/*
+ *  IXSelectInterruptPipe.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018-2019 Machine Zone, Inc. All rights reserved.
+ */
+
+//
+// On macOS we use UNIX pipes to wake up select.
+//
+
+#include "IXSelectInterruptPipe.h"
+
+#include <unistd.h> // for write
+#include <string.h> // for strerror
+#include <fcntl.h>
+#include <errno.h>
+#include <assert.h>
+#include <sstream>
+
+namespace ix
+{
+    // File descriptor at index 0 in _fildes is the read end of the pipe
+    // File descriptor at index 1 in _fildes is the write end of the pipe
+    const int SelectInterruptPipe::kPipeReadIndex = 0;
+    const int SelectInterruptPipe::kPipeWriteIndex = 1;
+
+    SelectInterruptPipe::SelectInterruptPipe()
+    {
+        _fildes[kPipeReadIndex] = -1;
+        _fildes[kPipeWriteIndex] = -1;
+    }
+
+    SelectInterruptPipe::~SelectInterruptPipe()
+    {
+        ::close(_fildes[kPipeReadIndex]);
+        ::close(_fildes[kPipeWriteIndex]);
+        _fildes[kPipeReadIndex] = -1;
+        _fildes[kPipeWriteIndex] = -1;
+    }
+
+    bool SelectInterruptPipe::init(std::string& errorMsg)
+    {
+        // calling init twice is a programming error
+        assert(_fildes[kPipeReadIndex] == -1);
+        assert(_fildes[kPipeWriteIndex] == -1);
+
+        if (pipe(_fildes) < 0)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptPipe::init() failed in pipe() call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+            return false;
+        }
+
+        if (fcntl(_fildes[kPipeReadIndex], F_SETFL, O_NONBLOCK) == -1)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptPipe::init() failed in fcntl(..., O_NONBLOCK) call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _fildes[kPipeReadIndex] = -1;
+            _fildes[kPipeWriteIndex] = -1;
+            return false;
+        }
+
+        if (fcntl(_fildes[kPipeWriteIndex], F_SETFL, O_NONBLOCK) == -1)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptPipe::init() failed in fcntl(..., O_NONBLOCK) call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _fildes[kPipeReadIndex] = -1;
+            _fildes[kPipeWriteIndex] = -1;
+            return false;
+        }
+
+#ifdef F_SETNOSIGPIPE
+        if (fcntl(_fildes[kPipeWriteIndex], F_SETNOSIGPIPE, 1) == -1)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptPipe::init() failed in fcntl(.... F_SETNOSIGPIPE) call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _fildes[kPipeReadIndex] = -1;
+            _fildes[kPipeWriteIndex] = -1;
+            return false;
+        }
+
+        if (fcntl(_fildes[kPipeWriteIndex], F_SETNOSIGPIPE, 1) == -1)
+        {
+            std::stringstream ss;
+            ss << "SelectInterruptPipe::init() failed in fcntl(..., F_SETNOSIGPIPE) call"
+               << " : " << strerror(errno);
+            errorMsg = ss.str();
+
+            _fildes[kPipeReadIndex] = -1;
+            _fildes[kPipeWriteIndex] = -1;
+            return false;
+        }
+#endif
+
+        return true;
+    }
+
+    bool SelectInterruptPipe::notify(uint64_t value)
+    {
+        int fd = _fildes[kPipeWriteIndex];
+        if (fd == -1) return false;
+
+        // we should write 8 bytes for an uint64_t
+        return write(fd, &value, sizeof(value)) == 8;
+    }
+
+    // TODO: return max uint64_t for errors ?
+    uint64_t SelectInterruptPipe::read()
+    {
+        int fd = _fildes[kPipeReadIndex];
+
+        uint64_t value = 0;
+        ::read(fd, &value, sizeof(value));
+
+        return value;
+    }
+
+    bool SelectInterruptPipe::clear()
+    {
+        return true;
+    }
+
+    int SelectInterruptPipe::getFd() const
+    {
+        return _fildes[kPipeReadIndex];
+    }
+}

ixwebsocket/IXSelectInterruptPipe.h

@@ -0,0 +1,39 @@
+/*
+ *  IXSelectInterruptPipe.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018-2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include "IXSelectInterrupt.h"
+
+#include <stdint.h>
+#include <string>
+
+namespace ix
+{
+    class SelectInterruptPipe : public SelectInterrupt {
+    public:
+        SelectInterruptPipe();
+        virtual ~SelectInterruptPipe();
+
+        bool init(std::string& errorMsg) final;
+
+        bool notify(uint64_t value) final;
+        bool clear() final;
+        uint64_t read() final;
+        int getFd() const final;
+
+    private:
+        // Store file descriptors used by the communication pipe. Communication
+        // happens between a control thread and a background thread, which is
+        // blocked on select.
+        int _fildes[2];
+
+        // Used to identify the read/write idx
+        static const int kPipeReadIndex;
+        static const int kPipeWriteIndex;
+    };
+}
+

ixwebsocket/IXSocket.cpp

@@ -7,6 +7,8 @@
 #include "IXSocket.h"
 #include "IXSocketConnect.h"
 #include "IXNetSystem.h"
+#include "IXSelectInterrupt.h"
+#include "IXSelectInterruptFactory.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -19,15 +21,23 @@
 #include <algorithm>
 #include <iostream>
 
+#ifdef min
+#undef min
+#endif
+
 namespace ix
 {
     const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
     const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
+    const uint64_t Socket::kSendRequest = 1;
+    const uint64_t Socket::kCloseRequest = 2;
+    constexpr size_t Socket::kChunkSize;
 
     Socket::Socket(int fd) :
-        _sockfd(fd)
+        _sockfd(fd),
+        _selectInterrupt(createSelectInterrupt())
     {
-
+        ;
     }
 
     Socket::~Socket()
@@ -39,44 +49,93 @@ namespace ix
     {
         if (_sockfd == -1)
         {
-            onPollCallback(PollResultType_Error);
+            if (onPollCallback) onPollCallback(PollResultType::Error);
             return;
         }
 
-        fd_set rfds;
-        FD_ZERO(&rfds);
-        FD_SET(_sockfd, &rfds);
+        PollResultType pollResult = isReadyToRead(1000 * timeoutSecs);
 
-#ifdef __linux__
-        FD_SET(_eventfd.getFd(), &rfds);
-#endif
+        if (onPollCallback) onPollCallback(pollResult);
+    }
+
+    PollResultType Socket::select(bool readyToRead, int timeoutMs)
+    {
+        fd_set rfds;
+        fd_set wfds;
+        FD_ZERO(&rfds);
+        FD_ZERO(&wfds);
+
+        fd_set* fds = (readyToRead) ? &rfds : & wfds;
+        FD_SET(_sockfd, fds);
+
+        // File descriptor used to interrupt select when needed
+        int interruptFd = _selectInterrupt->getFd();
+        if (interruptFd != -1)
+        {
+            FD_SET(interruptFd, fds);
+        }
 
         struct timeval timeout;
-        timeout.tv_sec = timeoutSecs;
-        timeout.tv_usec = 0;
+        timeout.tv_sec = timeoutMs / 1000;
+        timeout.tv_usec = (timeoutMs < 1000) ? 0 : 1000 * (timeoutMs % 1000);
 
+        // Compute the highest fd.
         int sockfd = _sockfd;
-        int nfds = (std::max)(sockfd, _eventfd.getFd());
-        int ret = select(nfds + 1, &rfds, nullptr, nullptr,
-                         (timeoutSecs < 0) ? nullptr : &timeout);
+        int nfds = (std::max)(sockfd, interruptFd);
 
-        PollResultType pollResult = PollResultType_ReadyForRead;
+        int ret = ::select(nfds + 1, &rfds, &wfds, nullptr,
+                           (timeoutMs < 0) ? nullptr : &timeout);
+
+        PollResultType pollResult = PollResultType::ReadyForRead;
         if (ret < 0)
         {
-            pollResult = PollResultType_Error;
+            pollResult = PollResultType::Error;
         }
         else if (ret == 0)
         {
-            pollResult = PollResultType_Timeout;
+            pollResult = PollResultType::Timeout;
         }
-
-        onPollCallback(pollResult);
-    }
-
-    void Socket::wakeUpFromPoll()
+        else if (interruptFd != -1 && FD_ISSET(interruptFd, &rfds))
         {
-        // this will wake up the thread blocked on select, only needed on Linux
-        _eventfd.notify();
+            uint64_t value = _selectInterrupt->read();
+
+            if (value == kSendRequest)
+            {
+                pollResult = PollResultType::SendRequest;
+            }
+            else if (value == kCloseRequest)
+            {
+                pollResult = PollResultType::CloseRequest;
+            }
+        }
+        else if (sockfd != -1 && readyToRead && FD_ISSET(sockfd, &rfds))
+        {
+            pollResult = PollResultType::ReadyForRead;
+        }
+        else if (sockfd != -1 && !readyToRead && FD_ISSET(sockfd, &wfds))
+        {
+            pollResult = PollResultType::ReadyForWrite;
+        }
+
+        return pollResult;
+    }
+
+    PollResultType Socket::isReadyToRead(int timeoutMs)
+    {
+        bool readyToRead = true;
+        return select(readyToRead, timeoutMs);
+    }
+
+    PollResultType Socket::isReadyToWrite(int timeoutMs)
+    {
+        bool readyToRead = false;
+        return select(readyToRead, timeoutMs);
+    }
+
+    // Wake up from poll/select by writing to the pipe which is watched by select
+    bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
+    {
+        return _selectInterrupt->notify(wakeUpCode);
     }
 
     bool Socket::connect(const std::string& host,
@@ -86,7 +145,7 @@ namespace ix
     {
         std::lock_guard<std::mutex> lock(_socketMutex);
 
-        if (!_eventfd.clear()) return false;
+        if (!_selectInterrupt->clear()) return false;
 
         _sockfd = SocketConnect::connect(host, port, errMsg, isCancellationRequested);
         return _sockfd != -1;
@@ -145,69 +204,9 @@ namespace ix
 #endif
     }
 
-    bool Socket::init()
+    bool Socket::init(std::string& errorMsg)
     {
-#ifdef _WIN32
-        INT rc;
-        WSADATA wsaData;
-
-        rc = WSAStartup(MAKEWORD(2, 2), &wsaData);
-        return rc != 0;
-#else
-        return true;
-#endif
-    }
-
-    void Socket::cleanup()
-    {
-#ifdef _WIN32
-        WSACleanup();
-#endif
-    }
-
-    bool Socket::readByte(void* buffer,
-                          const CancellationRequest& isCancellationRequested)
-    {
-        while (true)
-        {
-            if (isCancellationRequested()) return false;
-
-            ssize_t ret;
-            ret = recv(buffer, 1);
-
-            // We read one byte, as needed, all good.
-            if (ret == 1)
-            {
-                return true;
-            }
-            // There is possibly something to be read, try again
-            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
-                                 getErrno() == EAGAIN))
-            {
-                // Wait with a timeout until something is written.
-                // This way we are not busy looping
-                fd_set rfds;
-                struct timeval timeout;
-                timeout.tv_sec = 0;
-                timeout.tv_usec = 1 * 1000; // 1ms timeout
-
-                FD_ZERO(&rfds);
-                FD_SET(_sockfd, &rfds);
-
-                if (select(_sockfd + 1, &rfds, nullptr, nullptr, &timeout) < 0 &&
-                    (errno == EBADF || errno == EINVAL))
-                {
-                    return false;
-                }
-
-                continue;
-            }
-            // There was an error during the read, abort
-            else
-            {
-                return false;
-            }
-        }
+        return _selectInterrupt->init(errorMsg);
     }
 
     bool Socket::writeBytes(const std::string& str,
@@ -215,7 +214,7 @@ namespace ix
     {
         while (true)
         {
-            if (isCancellationRequested()) return false;
+            if (isCancellationRequested && isCancellationRequested()) return false;
 
             char* buffer = const_cast<char*>(str.c_str());
             int len = (int) str.size();
@@ -227,7 +226,7 @@ namespace ix
             {
                 return ret == len;
             }
-            // There is possibly something to be write, try again
+            // There is possibly something to be writen, try again
             else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
                                  getErrno() == EAGAIN))
             {
@@ -241,7 +240,42 @@ namespace ix
         }
     }
 
-    std::pair<bool, std::string> Socket::readLine(const CancellationRequest& isCancellationRequested)
+    bool Socket::readByte(void* buffer,
+                          const CancellationRequest& isCancellationRequested)
+    {
+        while (true)
+        {
+            if (isCancellationRequested && isCancellationRequested()) return false;
+
+            ssize_t ret;
+            ret = recv(buffer, 1);
+
+            // We read one byte, as needed, all good.
+            if (ret == 1)
+            {
+                return true;
+            }
+            // There is possibly something to be read, try again
+            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
+                                 getErrno() == EAGAIN))
+            {
+                // Wait with a 1ms timeout until the socket is ready to read.
+                // This way we are not busy looping
+                if (isReadyToRead(1) == PollResultType::Error)
+                {
+                    return false;
+                }
+            }
+            // There was an error during the read, abort
+            else
+            {
+                return false;
+            }
+        }
+    }
+
+    std::pair<bool, std::string> Socket::readLine(
+        const CancellationRequest& isCancellationRequested)
     {
         char c;
         std::string line;
@@ -251,7 +285,8 @@ namespace ix
         {
             if (!readByte(&c, isCancellationRequested))
             {
-                return std::make_pair(false, std::string());
+                // Return what we were able to read
+                return std::make_pair(false, line);
             }
 
             line += c;
@@ -259,4 +294,52 @@ namespace ix
 
         return std::make_pair(true, line);
     }
+
+    std::pair<bool, std::string> Socket::readBytes(
+        size_t length,
+        const OnProgressCallback& onProgressCallback,
+        const CancellationRequest& isCancellationRequested)
+    {
+        if (_readBuffer.empty())
+        {
+            _readBuffer.resize(kChunkSize);
+        }
+
+        std::vector<uint8_t> output;
+        while (output.size() != length)
+        {
+            if (isCancellationRequested && isCancellationRequested())
+            {
+                return std::make_pair(false, std::string());
+            }
+
+            size_t size = std::min(kChunkSize, length - output.size());
+            ssize_t ret = recv((char*)&_readBuffer[0], size);
+
+            if (ret <= 0 && (getErrno() != EWOULDBLOCK &&
+                             getErrno() != EAGAIN))
+            {
+                // Error
+                return std::make_pair(false, std::string());
+            }
+            else if (ret > 0)
+            {
+                output.insert(output.end(),
+                              _readBuffer.begin(),
+                              _readBuffer.begin() + ret);
+            }
+
+            if (onProgressCallback) onProgressCallback((int) output.size(), (int) length);
+
+            // Wait with a 1ms timeout until the socket is ready to read.
+            // This way we are not busy looping
+            if (isReadyToRead(1) == PollResultType::Error)
+            {
+                return std::make_pair(false, std::string());
+            }
+        }
+
+        return std::make_pair(true, std::string(output.begin(),
+                                                output.end()));
+    }
 }

ixwebsocket/IXSocket.h

@@ -10,22 +10,29 @@
 #include <functional>
 #include <mutex>
 #include <atomic>
+#include <vector>
+#include <memory>
 
 #ifdef _WIN32
 #include <BaseTsd.h>
 typedef SSIZE_T ssize_t;
 #endif
 
-#include "IXEventFd.h"
 #include "IXCancellationRequest.h"
+#include "IXProgressCallback.h"
 
 namespace ix
 {
-    enum PollResultType
+    class SelectInterrupt;
+
+    enum class PollResultType
     {
-        PollResultType_ReadyForRead = 0,
-        PollResultType_Timeout = 1,
-        PollResultType_Error = 2
+        ReadyForRead = 0,
+        ReadyForWrite = 1,
+        Timeout = 2,
+        Error = 3,
+        SendRequest = 4,
+        CloseRequest = 5
     };
 
     class Socket {
@@ -34,12 +41,17 @@ namespace ix
 
         Socket(int fd = -1);
         virtual ~Socket();
+        bool init(std::string& errorMsg);
 
         void configure();
 
-        virtual void poll(const OnPollCallback& onPollCallback,
+        // Functions to check whether there is activity on the socket
+        void poll(const OnPollCallback& onPollCallback,
                   int timeoutSecs = kDefaultPollTimeout);
-        virtual void wakeUpFromPoll();
+        bool wakeUpFromPoll(uint8_t wakeUpCode);
+
+        PollResultType isReadyToWrite(int timeoutMs);
+        PollResultType isReadyToRead(int timeoutMs);
 
         // Virtual methods
         virtual bool connect(const std::string& url,
@@ -58,21 +70,36 @@ namespace ix
                       const CancellationRequest& isCancellationRequested);
         bool writeBytes(const std::string& str,
                         const CancellationRequest& isCancellationRequested);
-        std::pair<bool, std::string> readLine(const CancellationRequest& isCancellationRequested);
+
+        std::pair<bool, std::string> readLine(
+            const CancellationRequest& isCancellationRequested);
+        std::pair<bool, std::string> readBytes(
+            size_t length,
+            const OnProgressCallback& onProgressCallback,
+            const CancellationRequest& isCancellationRequested);
 
         static int getErrno();
-        static bool init(); // Required on Windows to initialize WinSocket
-        static void cleanup(); // Required on Windows to cleanup WinSocket
+
+        // Used as special codes for pipe communication
+        static const uint64_t kSendRequest;
+        static const uint64_t kCloseRequest;
 
     protected:
         void closeSocket(int fd);
 
         std::atomic<int> _sockfd;
         std::mutex _socketMutex;
-        EventFd _eventfd;
 
     private:
+        PollResultType select(bool readyToRead, int timeoutMs);
+
         static const int kDefaultPollTimeout;
         static const int kDefaultPollNoTimeout;
+
+        // Buffer for reading from our socket. That buffer is never resized.
+        std::vector<uint8_t> _readBuffer;
+        static constexpr size_t kChunkSize = 1 << 15;
+
+        std::shared_ptr<SelectInterrupt> _selectInterrupt;
     };
 }

ixwebsocket/IXSocketConnect.cpp

@@ -66,7 +66,7 @@ namespace ix
 
         for (;;)
         {
-            if (isCancellationRequested()) // Must handle timeout as well
+            if (isCancellationRequested && isCancellationRequested()) // Must handle timeout as well
             {
                 closeSocket(fd);
                 errMsg = "Cancelled";

ixwebsocket/IXSocketFactory.cpp

@@ -0,0 +1,74 @@
+/*
+ *  IXSocketFactory.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXSocketFactory.h"
+
+#ifdef IXWEBSOCKET_USE_TLS
+
+# ifdef __APPLE__
+#  include <ixwebsocket/IXSocketAppleSSL.h>
+# elif defined(_WIN32)
+#  include <ixwebsocket/IXSocketSChannel.h>
+# else
+#  include <ixwebsocket/IXSocketOpenSSL.h>
+# endif
+
+#else
+
+#include <ixwebsocket/IXSocket.h>
+
+#endif
+
+namespace ix
+{
+    std::shared_ptr<Socket> createSocket(bool tls,
+                                         std::string& errorMsg)
+    {
+        errorMsg.clear();
+        std::shared_ptr<Socket> socket;
+
+        if (!tls)
+        {
+            socket = std::make_shared<Socket>();
+        }
+        else
+        {
+#ifdef IXWEBSOCKET_USE_TLS
+# ifdef __APPLE__
+            socket = std::make_shared<SocketAppleSSL>();
+# elif defined(_WIN32)
+            socket = std::make_shared<SocketSChannel>();
+# else
+            socket = std::make_shared<SocketOpenSSL>();
+# endif
+#else
+            errorMsg = "TLS support is not enabled on this platform.";
+            return nullptr;
+#endif
+        }
+
+        if (!socket->init(errorMsg))
+        {
+            socket.reset();
+        }
+
+        return socket;
+    }
+
+    std::shared_ptr<Socket> createSocket(int fd,
+                                         std::string& errorMsg)
+    {
+        errorMsg.clear();
+
+        std::shared_ptr<Socket> socket = std::make_shared<Socket>(fd);
+        if (!socket->init(errorMsg))
+        {
+            socket.reset();
+        }
+
+        return socket;
+    }
+}

ixwebsocket/IXSocketFactory.h

@@ -0,0 +1,21 @@
+
+/*
+ *  IXSocketFactory.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+namespace ix
+{
+    class Socket;
+    std::shared_ptr<Socket> createSocket(bool tls,
+                                         std::string& errorMsg);
+
+    std::shared_ptr<Socket> createSocket(int fd,
+                                         std::string& errorMsg);
+}

ixwebsocket/IXSocketOpenSSL.cpp

@@ -21,6 +21,7 @@
 namespace ix
 {
     std::atomic<bool> SocketOpenSSL::_openSSLInitializationSuccessful(false);
+    std::once_flag SocketOpenSSL::_openSSLInitFlag;
 
     SocketOpenSSL::SocketOpenSSL(int fd) : Socket(fd),
         _ssl_connection(nullptr),

ixwebsocket/IXSocketOpenSSL.h

@@ -50,7 +50,7 @@ namespace ix
         const SSL_METHOD* _ssl_method;
         mutable std::mutex _mutex;  // OpenSSL routines are not thread-safe
 
-        std::once_flag _openSSLInitFlag;
+        static std::once_flag _openSSLInitFlag;
         static std::atomic<bool> _openSSLInitializationSuccessful;
     };
 

ixwebsocket/IXSocketSChannel.cpp

@@ -18,7 +18,7 @@
 # include <ws2def.h>
 # include <WS2tcpip.h>
 # include <schannel.h>
-# include <sslsock.h>
+//# include <sslsock.h>
 # include <io.h>
 
 #define WIN32_LEAN_AND_MEAN
@@ -75,7 +75,7 @@ namespace ix
                                  int port,
                                  std::string& errMsg)
     {
-        return Socket::connect(host, port, errMsg);
+        return Socket::connect(host, port, errMsg, nullptr);
     }
 
 
@@ -89,17 +89,17 @@ namespace ix
         Socket::close();
     }
 
-    int SocketSChannel::send(char* buf, size_t nbyte)
+    ssize_t SocketSChannel::send(char* buf, size_t nbyte)
     {
         return Socket::send(buf, nbyte);
     }
 
-    int SocketSChannel::send(const std::string& buffer)
+    ssize_t SocketSChannel::send(const std::string& buffer)
     {
         return Socket::send(buffer);
     }
 
-    int SocketSChannel::recv(void* buf, size_t nbyte)
+    ssize_t SocketSChannel::recv(void* buf, size_t nbyte)
     {
         return Socket::recv(buf, nbyte);
     }

ixwebsocket/IXSocketSChannel.h

@@ -24,9 +24,9 @@ namespace ix
         // The important override
         virtual void secureSocket() final;
 
-        virtual int send(char* buffer, size_t length) final;
-        virtual int send(const std::string& buffer) final;
-        virtual int recv(void* buffer, size_t length) final;
+        virtual ssize_t send(char* buffer, size_t length) final;
+        virtual ssize_t send(const std::string& buffer) final;
+        virtual ssize_t recv(void* buffer, size_t length) final;
 
     private:
     };

ixwebsocket/IXSocketServer.cpp

@@ -29,7 +29,8 @@ namespace ix
         _host(host),
         _backlog(backlog),
         _maxConnections(maxConnections),
-        _stop(false)
+        _stop(false),
+        _connectionStateFactory(&ConnectionState::createConnectionState)
     {
 
     }
@@ -135,6 +136,9 @@ namespace ix
 
     void SocketServer::stop()
     {
+        closeTerminatedThreads();
+        assert(_connectionsThreads.empty());
+
         if (!_thread.joinable()) return; // nothing to do
 
         _stop = true;
@@ -145,18 +149,50 @@ namespace ix
         ::close(_serverFd);
     }
 
+    void SocketServer::setConnectionStateFactory(
+        const ConnectionStateFactory& connectionStateFactory)
+    {
+        _connectionStateFactory = connectionStateFactory;
+    }
+
+    // join the threads for connections that have been closed
+    void SocketServer::closeTerminatedThreads()
+    {
+        auto it = _connectionsThreads.begin();
+        auto itEnd  = _connectionsThreads.end();
+
+        while (it != itEnd)
+        {
+            auto& connectionState = it->first;
+            auto& thread = it->second;
+
+            if (!connectionState->isTerminated() ||
+                !thread.joinable())
+            {
+                ++it;
+                continue;
+            }
+
+            thread.join();
+            it = _connectionsThreads.erase(it);
+        }
+    }
+
     void SocketServer::run()
     {
         // Set the socket to non blocking mode, so that accept calls are not blocking
         SocketConnect::configure(_serverFd);
 
-        // Return value of std::async, ignored
-        std::future<void> f;
-
         for (;;)
         {
             if (_stop) return;
 
+            // Garbage collection to shutdown/join threads for closed connections.
+            // We could run this in its own thread, so that we dont need to accept 
+            // a new connection to close a thread.
+            // We could also use a condition variable to be notify when we need to do this
+            closeTerminatedThreads();
+
             // Use select to check whether a new connection is in progress
             fd_set rfds;
             struct timeval timeout;
@@ -214,14 +250,19 @@ namespace ix
                 continue;
             }
 
+            std::shared_ptr<ConnectionState> connectionState;
+            if (_connectionStateFactory)
+            {
+                connectionState = _connectionStateFactory();
+            }
+
             // Launch the handleConnection work asynchronously in its own thread.
-            //
-            // the destructor of a future returned by std::async blocks,
-            // so we need to declare it outside of this loop
-            f = std::async(std::launch::async,
-                           &SocketServer::handleConnection,
+            _connectionsThreads.push_back(std::make_pair(
+                    connectionState,
+                    std::thread(&SocketServer::handleConnection,
                                 this,
-                           clientFd);
+                                clientFd,
+                                connectionState)));
         }
     }
 }

ixwebsocket/IXSocketServer.h

@@ -6,10 +6,13 @@
 
 #pragma once
 
+#include "IXConnectionState.h"
+
 #include <utility> // pair
 #include <string>
 #include <set>
 #include <thread>
+#include <list>
 #include <mutex>
 #include <functional>
 #include <memory>
@@ -20,6 +23,12 @@ namespace ix
 {
     class SocketServer {
     public:
+        using ConnectionStateFactory = std::function<std::shared_ptr<ConnectionState>()>;
+
+        // We use a list as we only care about remove and append operations.
+        using ConnectionThreads = std::list<std::pair<std::shared_ptr<ConnectionState>,
+                                                      std::thread>>;
+
         SocketServer(int port = SocketServer::kDefaultPort,
                      const std::string& host = SocketServer::kDefaultHost,
                      int backlog = SocketServer::kDefaultTcpBacklog,
@@ -27,6 +36,8 @@ namespace ix
         virtual ~SocketServer();
         virtual void stop();
 
+        void setConnectionStateFactory(const ConnectionStateFactory& connectionStateFactory);
+
         const static int kDefaultPort;
         const static std::string kDefaultHost;
         const static int kDefaultTcpBacklog;
@@ -57,12 +68,20 @@ namespace ix
         std::atomic<bool> _stop;
         std::thread _thread;
 
+        ConnectionThreads _connectionsThreads;
+
         std::condition_variable _conditionVariable;
         std::mutex _conditionVariableMutex;
 
+        //
+        ConnectionStateFactory _connectionStateFactory;
+
         // Methods
         void run();
-        virtual void handleConnection(int fd) = 0;
+        virtual void handleConnection(int fd,
+                                      std::shared_ptr<ConnectionState> connectionState) = 0;
         virtual size_t getConnectedClientsCount() = 0;
+
+        void closeTerminatedThreads();
     };
 }

ixwebsocket/IXUrlParser.cpp

@@ -0,0 +1,104 @@
+/*
+ *  IXUrlParser.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXUrlParser.h"
+
+#include <iostream>
+#include <sstream>
+
+
+namespace ix
+{
+    //
+    // The only difference between those 2 regex is the protocol
+    //
+    std::regex UrlParser::_httpRegex("(http|https)://([^/ :]+):?([^/ ]*)(/?[^ #?]*)\\x3f?([^ #]*)#?([^ ]*)");
+    std::regex UrlParser::_webSocketRegex("(ws|wss)://([^/ :]+):?([^/ ]*)(/?[^ #?]*)\\x3f?([^ #]*)#?([^ ]*)");
+
+    bool UrlParser::parse(const std::string& url,
+                          std::string& protocol,
+                          std::string& host,
+                          std::string& path,
+                          std::string& query,
+                          int& port,
+                          bool websocket)
+    {
+        std::cmatch what;
+        if (!regex_match(url.c_str(), what,
+                         websocket ? _webSocketRegex : _httpRegex))
+        {
+            return false;
+        }
+
+        std::string portStr;
+
+        protocol = std::string(what[1].first, what[1].second);
+        host     = std::string(what[2].first, what[2].second);
+        portStr  = std::string(what[3].first, what[3].second);
+        path     = std::string(what[4].first, what[4].second);
+        query    = std::string(what[5].first, what[5].second);
+
+        if (portStr.empty())
+        {
+            if (protocol == "ws" || protocol == "http")
+            {
+                port = 80;
+            }
+            else if (protocol == "wss" || protocol == "https")
+            {
+                port = 443;
+            }
+            else
+            {
+                // Invalid protocol. Should be caught by regex check
+                // but this missing branch trigger cpplint linter.
+                return false;
+            }
+        }
+        else
+        {
+            std::stringstream ss;
+            ss << portStr;
+            ss >> port;
+        }
+
+        if (path.empty())
+        {
+            path = "/";
+        }
+        else if (path[0] != '/')
+        {
+            path = '/' + path;
+        }
+
+        if (!query.empty())
+        {
+            path += "?";
+            path += query;
+        }
+
+        return true;
+    }
+
+    void UrlParser::printUrl(const std::string& url, bool websocket)
+    {
+        std::string protocol, host, path, query;
+        int port {0};
+
+        if (!parse(url, protocol, host, path, query, port, websocket))
+        {
+            return;
+        }
+
+        std::cout << "[" << url << "]" << std::endl;
+        std::cout << protocol << std::endl;
+        std::cout << host << std::endl;
+        std::cout << port << std::endl;
+        std::cout << path << std::endl;
+        std::cout << query << std::endl;
+        std::cout << "-------------------------------" << std::endl;
+    }
+}

ixwebsocket/IXUrlParser.h

@@ -0,0 +1,31 @@
+/*
+ *  IXUrlParser.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <string>
+#include <regex>
+
+namespace ix
+{
+    class UrlParser
+    {
+    public:
+        static bool parse(const std::string& url,
+                          std::string& protocol,
+                          std::string& host,
+                          std::string& path,
+                          std::string& query,
+                          int& port,
+                          bool websocket);
+
+        static void printUrl(const std::string& url, bool websocket);
+
+    private:
+        static std::regex _httpRegex;
+        static std::regex _webSocketRegex;
+    };
+}

ixwebsocket/IXWebSocket.cpp

@@ -79,10 +79,10 @@ namespace ix
         return _perMessageDeflateOptions;
     }
 
-    void WebSocket::setHeartBeatPeriod(int hearBeatPeriod)
+    void WebSocket::setHeartBeatPeriod(int heartBeatPeriod)
     {
         std::lock_guard<std::mutex> lock(_configMutex);
-        _heartBeatPeriod = hearBeatPeriod;
+        _heartBeatPeriod = heartBeatPeriod;
     }
 
     int WebSocket::getHeartBeatPeriod() const
@@ -252,6 +252,11 @@ namespace ix
                         {
                             webSocketMessageType = WebSocket_MessageType_Pong;
                         } break;
+
+                        case WebSocketTransport::FRAGMENT:
+                        {
+                            webSocketMessageType = WebSocket_MessageType_Fragment;
+                        } break;
                     }
 
                     WebSocketErrorInfo webSocketErrorInfo;
@@ -297,7 +302,13 @@ namespace ix
     WebSocketSendInfo WebSocket::send(const std::string& text,
                                       const OnProgressCallback& onProgressCallback)
     {
-        return sendMessage(text, false, onProgressCallback);
+        return sendMessage(text, SendMessageKind::Binary, onProgressCallback);
+    }
+
+    WebSocketSendInfo WebSocket::sendText(const std::string& text,
+                                          const OnProgressCallback& onProgressCallback)
+    {
+        return sendMessage(text, SendMessageKind::Text, onProgressCallback);
     }
 
     WebSocketSendInfo WebSocket::ping(const std::string& text)
@@ -306,11 +317,11 @@ namespace ix
         constexpr size_t pingMaxPayloadSize = 125;
         if (text.size() > pingMaxPayloadSize) return WebSocketSendInfo(false);
 
-        return sendMessage(text, true);
+        return sendMessage(text, SendMessageKind::Ping);
     }
 
     WebSocketSendInfo WebSocket::sendMessage(const std::string& text,
-                                             bool ping,
+                                             SendMessageKind sendMessageKind,
                                              const OnProgressCallback& onProgressCallback)
     {
         if (!isConnected()) return WebSocketSendInfo(false);
@@ -327,13 +338,22 @@ namespace ix
         std::lock_guard<std::mutex> lock(_writeMutex);
         WebSocketSendInfo webSocketSendInfo;
 
-        if (ping)
+        switch (sendMessageKind)
         {
-            webSocketSendInfo = _ws.sendPing(text);
-        }
-        else
+            case SendMessageKind::Text:
+            {
+                webSocketSendInfo = _ws.sendText(text, onProgressCallback);
+            } break;
+
+            case SendMessageKind::Binary:
             {
                 webSocketSendInfo = _ws.sendBinary(text, onProgressCallback);
+            } break;
+
+            case SendMessageKind::Ping:
+            {
+                webSocketSendInfo = _ws.sendPing(text);
+            } break;
         }
 
         WebSocket::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize, false);
@@ -374,4 +394,9 @@ namespace ix
     {
         _automaticReconnection = false;
     }
+
+    size_t WebSocket::bufferedAmount() const
+    {
+        return _ws.bufferedAmount();
+    }
 }

ixwebsocket/IXWebSocket.h

@@ -39,7 +39,8 @@ namespace ix
         WebSocket_MessageType_Close = 2,
         WebSocket_MessageType_Error = 3,
         WebSocket_MessageType_Ping = 4,
-        WebSocket_MessageType_Pong = 5
+        WebSocket_MessageType_Pong = 5,
+        WebSocket_MessageType_Fragment = 6
     };
 
     struct WebSocketOpenInfo
@@ -88,7 +89,7 @@ namespace ix
         void setUrl(const std::string& url);
         void setPerMessageDeflateOptions(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
         void setHandshakeTimeout(int handshakeTimeoutSecs);
-        void setHeartBeatPeriod(int hearBeatPeriod);
+        void setHeartBeatPeriod(int heartBeatPeriod);
 
         // Run asynchronously, by calling start and stop.
         void start();
@@ -100,6 +101,8 @@ namespace ix
 
         WebSocketSendInfo send(const std::string& text,
                                const OnProgressCallback& onProgressCallback = nullptr);
+        WebSocketSendInfo sendText(const std::string& text,
+                                   const OnProgressCallback& onProgressCallback = nullptr);
         WebSocketSendInfo ping(const std::string& text);
         void close();
 
@@ -111,6 +114,7 @@ namespace ix
         const std::string& getUrl() const;
         const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
         int getHeartBeatPeriod() const;
+        size_t bufferedAmount() const;
 
         void enableAutomaticReconnection();
         void disableAutomaticReconnection();
@@ -118,7 +122,7 @@ namespace ix
     private:
 
         WebSocketSendInfo sendMessage(const std::string& text,
-                                      bool ping,
+                                      SendMessageKind sendMessageKind,
                                       const OnProgressCallback& callback = nullptr);
 
         bool isConnected() const;

ixwebsocket/IXWebSocketHandshake.cpp

@@ -6,6 +6,7 @@
 
 #include "IXWebSocketHandshake.h"
 #include "IXSocketConnect.h"
+#include "IXUrlParser.h"
 
 #include "libwshandshake.hpp"
 
@@ -32,90 +33,6 @@ namespace ix
 
     }
 
-    bool WebSocketHandshake::parseUrl(const std::string& url,
-                                      std::string& protocol,
-                                      std::string& host,
-                                      std::string& path,
-                                      std::string& query,
-                                      int& port)
-    {
-        std::regex ex("(ws|wss)://([^/ :]+):?([^/ ]*)(/?[^ #?]*)\\x3f?([^ #]*)#?([^ ]*)");
-        std::cmatch what;
-        if (!regex_match(url.c_str(), what, ex))
-        {
-            return false;
-        }
-
-        std::string portStr;
-
-        protocol = std::string(what[1].first, what[1].second);
-        host     = std::string(what[2].first, what[2].second);
-        portStr  = std::string(what[3].first, what[3].second);
-        path     = std::string(what[4].first, what[4].second);
-        query    = std::string(what[5].first, what[5].second);
-
-        if (portStr.empty())
-        {
-            if (protocol == "ws")
-            {
-                port = 80;
-            }
-            else if (protocol == "wss")
-            {
-                port = 443;
-            }
-            else
-            {
-                // Invalid protocol. Should be caught by regex check
-                // but this missing branch trigger cpplint linter.
-                return false;
-            }
-        }
-        else
-        {
-            std::stringstream ss;
-            ss << portStr;
-            ss >> port;
-        }
-
-        if (path.empty())
-        {
-            path = "/";
-        }
-        else if (path[0] != '/')
-        {
-            path = '/' + path;
-        }
-
-        if (!query.empty())
-        {
-            path += "?";
-            path += query;
-        }
-
-        return true;
-    }
-
-    void WebSocketHandshake::printUrl(const std::string& url)
-    {
-        std::string protocol, host, path, query;
-        int port {0};
-
-        if (!WebSocketHandshake::parseUrl(url, protocol, host,
-                                          path, query, port))
-        {
-            return;
-        }
-
-        std::cout << "[" << url << "]" << std::endl;
-        std::cout << protocol << std::endl;
-        std::cout << host << std::endl;
-        std::cout << port << std::endl;
-        std::cout << path << std::endl;
-        std::cout << query << std::endl;
-        std::cout << "-------------------------------" << std::endl;
-    }
-
     std::string WebSocketHandshake::trim(const std::string& str)
     {
         std::string out(str);
@@ -192,67 +109,12 @@ namespace ix
         return s;
     }
 
-
-    std::pair<bool, WebSocketHttpHeaders> WebSocketHandshake::parseHttpHeaders(
-        const CancellationRequest& isCancellationRequested)
-    {
-        WebSocketHttpHeaders headers;
-
-        char line[256];
-        int i;
-
-        while (true)
-        {
-            int colon = 0;
-
-            for (i = 0;
-                 i < 2 || (i < 255 && line[i-2] != '\r' && line[i-1] != '\n');
-                 ++i)
-            {
-                if (!_socket->readByte(line+i, isCancellationRequested))
-                {
-                    return std::make_pair(false, headers);
-                }
-
-                if (line[i] == ':' && colon == 0)
-                {
-                    colon = i;
-                }
-            }
-            if (line[0] == '\r' && line[1] == '\n')
-            {
-                break;
-            }
-
-            // line is a single header entry. split by ':', and add it to our
-            // header map. ignore lines with no colon.
-            if (colon > 0)
-            {
-                line[i] = '\0';
-                std::string lineStr(line);
-                // colon is ':', colon+1 is ' ', colon+2 is the start of the value.
-                // i is end of string (\0), i-colon is length of string minus key;
-                // subtract 1 for '\0', 1 for '\n', 1 for '\r',
-                // 1 for the ' ' after the ':', and total is -4
-                std::string name(lineStr.substr(0, colon));
-                std::string value(lineStr.substr(colon + 2, i - colon - 4));
-
-                // Make the name lower case.
-                std::transform(name.begin(), name.end(), name.begin(), ::tolower);
-
-                headers[name] = value;
-            }
-        }
-
-        return std::make_pair(true, headers);
-    }
-
     WebSocketInitResult WebSocketHandshake::sendErrorResponse(int code, const std::string& reason)
     {
         std::stringstream ss;
         ss << "HTTP/1.1 ";
         ss << code;
-        ss << "\r\n";
+        ss << " ";
         ss << reason;
         ss << "\r\n";
 
@@ -355,7 +217,7 @@ namespace ix
             return WebSocketInitResult(false, status, ss.str());
         }
 
-        auto result = parseHttpHeaders(isCancellationRequested);
+        auto result = parseHttpHeaders(_socket, isCancellationRequested);
         auto headersValid = result.first;
         auto headers = result.second;
 
@@ -450,7 +312,7 @@ namespace ix
         }
 
         // Retrieve and validate HTTP headers
-        auto result = parseHttpHeaders(isCancellationRequested);
+        auto result = parseHttpHeaders(_socket, isCancellationRequested);
         auto headersValid = result.first;
         auto headers = result.second;
 
@@ -491,7 +353,7 @@ namespace ix
         WebSocketHandshakeKeyGen::generate(headers["sec-websocket-key"].c_str(), output);
 
         std::stringstream ss;
-        ss << "HTTP/1.1 101\r\n";
+        ss << "HTTP/1.1 101 Switching Protocols\r\n";
         ss << "Sec-WebSocket-Accept: " << std::string(output) << "\r\n";
         ss << "Upgrade: websocket\r\n";
         ss << "Connection: Upgrade\r\n";

ixwebsocket/IXWebSocketHandshake.h

@@ -59,19 +59,10 @@ namespace ix
         WebSocketInitResult serverHandshake(int fd,
                                             int timeoutSecs);
 
-        static bool parseUrl(const std::string& url,
-                             std::string& protocol,
-                             std::string& host,
-                             std::string& path,
-                             std::string& query,
-                             int& port);
-
     private:
-        static void printUrl(const std::string& url);
         std::string genRandomString(const int len);
 
         // Parse HTTP headers
-        std::pair<bool, WebSocketHttpHeaders> parseHttpHeaders(const CancellationRequest& isCancellationRequested);
         WebSocketInitResult sendErrorResponse(int code, const std::string& reason);
 
         std::tuple<std::string, std::string, std::string> parseRequestLine(const std::string& line);

ixwebsocket/IXWebSocketHttpHeaders.cpp

@@ -0,0 +1,82 @@
+/*
+ *  IXWebSocketHttpHeaders.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXWebSocketHttpHeaders.h"
+#include "IXSocket.h"
+#include <algorithm>
+#include <locale>
+
+namespace ix
+{
+    bool CaseInsensitiveLess::NocaseCompare::operator()(const unsigned char & c1, const unsigned char & c2) const
+    {
+#ifdef _WIN32
+        return std::tolower(c1, std::locale()) < std::tolower(c2, std::locale());
+#else
+        return std::tolower(c1) < std::tolower(c2);
+#endif
+    }
+
+    bool CaseInsensitiveLess::operator()(const std::string & s1, const std::string & s2) const
+    {
+        return std::lexicographical_compare
+                (s1.begin(), s1.end(),   // source range
+                 s2.begin(), s2.end(),   // dest range
+                 NocaseCompare());       // comparison
+    }
+
+    std::pair<bool, WebSocketHttpHeaders> parseHttpHeaders(
+        std::shared_ptr<Socket> socket,
+        const CancellationRequest& isCancellationRequested)
+    {
+        WebSocketHttpHeaders headers;
+
+        char line[1024];
+        int i;
+
+        while (true)
+        {
+            int colon = 0;
+
+            for (i = 0;
+                 i < 2 || (i < 1023 && line[i-2] != '\r' && line[i-1] != '\n');
+                 ++i)
+            {
+                if (!socket->readByte(line+i, isCancellationRequested))
+                {
+                    return std::make_pair(false, headers);
+                }
+
+                if (line[i] == ':' && colon == 0)
+                {
+                    colon = i;
+                }
+            }
+            if (line[0] == '\r' && line[1] == '\n')
+            {
+                break;
+            }
+
+            // line is a single header entry. split by ':', and add it to our
+            // header map. ignore lines with no colon.
+            if (colon > 0)
+            {
+                line[i] = '\0';
+                std::string lineStr(line);
+                // colon is ':', colon+1 is ' ', colon+2 is the start of the value.
+                // i is end of string (\0), i-colon is length of string minus key;
+                // subtract 1 for '\0', 1 for '\n', 1 for '\r',
+                // 1 for the ' ' after the ':', and total is -4
+                std::string name(lineStr.substr(0, colon));
+                std::string value(lineStr.substr(colon + 2, i - colon - 4));
+
+                headers[name] = value;
+            }
+        }
+
+        return std::make_pair(true, headers);
+    }
+}

ixwebsocket/IXWebSocketHttpHeaders.h

@@ -6,10 +6,30 @@
 
 #pragma once
 
+#include "IXCancellationRequest.h"
+
 #include <string>
-#include <unordered_map>
+#include <map>
+#include <memory>
 
 namespace ix
 {
-    using WebSocketHttpHeaders = std::unordered_map<std::string, std::string>;
+    class Socket;
+
+    struct CaseInsensitiveLess
+    {
+        // Case Insensitive compare_less binary function
+        struct NocaseCompare
+        {
+            bool operator() (const unsigned char& c1, const unsigned char& c2) const;
+        };
+
+        bool operator() (const std::string & s1, const std::string & s2) const;
+    };
+
+    using WebSocketHttpHeaders = std::map<std::string, std::string, CaseInsensitiveLess>;
+
+    std::pair<bool, WebSocketHttpHeaders> parseHttpHeaders(
+        std::shared_ptr<Socket> socket,
+        const CancellationRequest& isCancellationRequested);
 }

ixwebsocket/IXWebSocketServer.cpp

@@ -49,10 +49,12 @@ namespace ix
         _onConnectionCallback = callback;
     }
 
-    void WebSocketServer::handleConnection(int fd)
+    void WebSocketServer::handleConnection(
+        int fd,
+        std::shared_ptr<ConnectionState> connectionState)
     {
         auto webSocket = std::make_shared<WebSocket>();
-        _onConnectionCallback(webSocket);
+        _onConnectionCallback(webSocket, connectionState);
 
         webSocket->disableAutomaticReconnection();
 
@@ -89,6 +91,7 @@ namespace ix
         }
 
         logInfo("WebSocketServer::handleConnection() done");
+        connectionState->setTerminated();
     }
 
     std::set<std::shared_ptr<WebSocket>> WebSocketServer::getClients()

ixwebsocket/IXWebSocketServer.h

@@ -20,7 +20,8 @@
 
 namespace ix
 {
-    using OnConnectionCallback = std::function<void(std::shared_ptr<WebSocket>)>;
+    using OnConnectionCallback = std::function<void(std::shared_ptr<WebSocket>,
+                                                    std::shared_ptr<ConnectionState>)>;
 
     class WebSocketServer : public SocketServer {
     public:
@@ -49,7 +50,8 @@ namespace ix
         const static int kDefaultHandShakeTimeoutSecs;
 
         // Methods
-        virtual void handleConnection(int fd) final;
+        virtual void handleConnection(int fd,
+                                      std::shared_ptr<ConnectionState> connectionState) final;
         virtual size_t getConnectedClientsCount() final;
     };
 }

ixwebsocket/IXWebSocketTransport.cpp

@@ -1,7 +1,31 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2012, 2013 <dhbaird@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
 /*
  *  IXWebSocketTransport.cpp
  *  Author: Benjamin Sergeant
- *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
  */
 
 //
@@ -11,14 +35,8 @@
 #include "IXWebSocketTransport.h"
 #include "IXWebSocketHandshake.h"
 #include "IXWebSocketHttpHeaders.h"
-
-#ifdef IXWEBSOCKET_USE_TLS
-# ifdef __APPLE__
-#  include "IXSocketAppleSSL.h"
-# else
-#  include "IXSocketOpenSSL.h"
-# endif
-#endif
+#include "IXUrlParser.h"
+#include "IXSocketFactory.h"
 
 #include <string.h>
 #include <stdlib.h>
@@ -35,11 +53,12 @@
 
 namespace ix
 {
-    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::hearbeat");
+    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::heartbeat");
     const int WebSocketTransport::kDefaultHeartBeatPeriod(-1);
     constexpr size_t WebSocketTransport::kChunkSize;
 
     WebSocketTransport::WebSocketTransport() :
+        _useMask(true),
         _readyState(CLOSED),
         _closeCode(0),
         _closeWireSize(0),
@@ -57,11 +76,11 @@ namespace ix
     }
 
     void WebSocketTransport::configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                                       int hearBeatPeriod)
+                                       int heartBeatPeriod)
     {
         _perMessageDeflateOptions = perMessageDeflateOptions;
         _enablePerMessageDeflate = _perMessageDeflateOptions.enabled();
-        _heartBeatPeriod = hearBeatPeriod;
+        _heartBeatPeriod = heartBeatPeriod;
     }
 
     // Client
@@ -70,31 +89,21 @@ namespace ix
     {
         std::string protocol, host, path, query;
         int port;
+        bool websocket = true;
 
-        if (!WebSocketHandshake::parseUrl(url, protocol, host,
-                                          path, query, port))
+        if (!UrlParser::parse(url, protocol, host, path, query, port, websocket))
         {
             return WebSocketInitResult(false, 0,
                                        std::string("Could not parse URL ") + url);
         }
 
-        if (protocol == "wss")
+        bool tls = protocol == "wss";
+        std::string errorMsg;
+        _socket = createSocket(tls, errorMsg);
+
+        if (!_socket)
         {
-            _socket.reset();
-#ifdef IXWEBSOCKET_USE_TLS
-# ifdef __APPLE__
-             _socket = std::make_shared<SocketAppleSSL>();
-# else
-             _socket = std::make_shared<SocketOpenSSL>();
-# endif
-#else
-            return WebSocketInitResult(false, 0, "TLS is not supported.");
-#endif
-        }
-        else
-        {
-            _socket.reset();
-            _socket = std::make_shared<Socket>();
+            return WebSocketInitResult(false, 0, errorMsg);
         }
 
         WebSocketHandshake webSocketHandshake(_requestInitCancellation,
@@ -115,8 +124,16 @@ namespace ix
     // Server
     WebSocketInitResult WebSocketTransport::connectToSocket(int fd, int timeoutSecs)
     {
-        _socket.reset();
-        _socket = std::make_shared<Socket>(fd);
+        // Server should not mask the data it sends to the client
+        _useMask = false;
+
+        std::string errorMsg;
+        _socket = createSocket(fd, errorMsg);
+
+        if (!_socket)
+        {
+            return WebSocketInitResult(false, 0, errorMsg);
+        }
 
         WebSocketHandshake webSocketHandshake(_requestInitCancellation,
                                               _socket,
@@ -176,15 +193,37 @@ namespace ix
                 // If (1) heartbeat is enabled, and (2) no data was received or
                 // send for a duration exceeding our heart-beat period, send a
                 // ping to the server.
-                if (pollResult == PollResultType_Timeout &&
+                if (pollResult == PollResultType::Timeout &&
                     heartBeatPeriodExceeded())
                 {
                     std::stringstream ss;
                     ss << kHeartBeatPingMessage << "::" << _heartBeatPeriod << "s";
                     sendPing(ss.str());
-                    return;
                 }
+                // Make sure we send all the buffered data
+                // there can be a lot of it for large messages.
+                else if (pollResult == PollResultType::SendRequest)
+                {
+                    while (!isSendBufferEmpty() && !_requestInitCancellation)
+                    {
+                        // Wait with a 10ms timeout until the socket is ready to write.
+                        // This way we are not busy looping
+                        PollResultType result = _socket->isReadyToWrite(10);
 
+                        if (result == PollResultType::Error)
+                        {
+                            _socket->close();
+                            setReadyState(CLOSED);
+                            break;
+                        }
+                        else if (result == PollResultType::ReadyForWrite)
+                        {
+                            sendOnSocket();
+                        }
+                    }
+                }
+                else if (pollResult == PollResultType::ReadyForRead)
+                {
                     while (true)
                     {
                         ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
@@ -208,11 +247,21 @@ namespace ix
                                           _readbuf.begin() + ret);
                         }
                     }
-
-                if (isSendBufferEmpty() && _readyState == CLOSING)
+                }
+                else if (pollResult == PollResultType::Error)
+                {
+                    _socket->close();
+                }
+                else if (pollResult == PollResultType::CloseRequest)
+                {
+                    _socket->close();
+                }
+
+                // Avoid a race condition where we get stuck in select
+                // while closing.
+                if (_readyState == CLOSING)
                 {
                     _socket->close();
-                    setReadyState(CLOSED);
                 }
             },
             _heartBeatPeriod);
@@ -235,17 +284,13 @@ namespace ix
         _txbuf.insert(_txbuf.end(), header.begin(), header.end());
         _txbuf.insert(_txbuf.end(), begin, end);
 
-        // Masking
+        if (_useMask)
+        {
             for (size_t i = 0; i != (size_t) message_size; ++i)
             {
                 *(_txbuf.end() - (size_t) message_size + i) ^= masking_key[i&0x3];
             }
         }
-
-    void WebSocketTransport::appendToSendBuffer(const std::vector<uint8_t>& buffer)
-    {
-        std::lock_guard<std::mutex> lock(_txbufMutex);
-        _txbuf.insert(_txbuf.end(), buffer.begin(), buffer.end());
     }
 
     void WebSocketTransport::unmaskReceiveBuffer(const wsheader_type& ws)
@@ -390,6 +435,10 @@ namespace ix
                         emitMessage(MSG, getMergedChunks(), ws, onMessageCallback);
                         _chunks.clear();
                     }
+                    else
+                    {
+                        emitMessage(FRAGMENT, std::string(), ws, onMessageCallback);
+                    }
                 }
             }
             else if (ws.opcode == wsheader_type::PING)
@@ -425,14 +474,8 @@ namespace ix
                 std::string reason(_rxbuf.begin()+ws.header_size + 2,
                                    _rxbuf.begin()+ws.header_size + 2 + (size_t) ws.N);
 
-                {
-                    std::lock_guard<std::mutex> lock(_closeDataMutex);
-                    _closeCode = code;
-                    _closeReason = reason;
-                    _closeWireSize = _rxbuf.size();
-                }
 
-                close();
+                close(code, reason, _rxbuf.size());
             }
             else
             {
@@ -473,7 +516,7 @@ namespace ix
         size_t wireSize = message.size();
 
         // When the RSV1 bit is 1 it means the message is compressed
-        if (_enablePerMessageDeflate && ws.rsv1)
+        if (_enablePerMessageDeflate && ws.rsv1 && messageKind != FRAGMENT)
         {
             std::string decompressedMessage;
             bool success = _perMessageDeflate.decompress(message, decompressedMessage);
@@ -571,7 +614,7 @@ namespace ix
                 // Send message
                 sendFragment(opcodeType, fin, begin, end, compress);
 
-                if (onProgressCallback && !onProgressCallback(i, steps))
+                if (onProgressCallback && !onProgressCallback((int)i, (int) steps))
                 {
                     break;
                 }
@@ -580,6 +623,12 @@ namespace ix
             }
         }
 
+        // Request to flush the send buffer on the background thread if it isn't empty
+        if (!isSendBufferEmpty())
+        {
+            _socket->wakeUpFromPoll(Socket::kSendRequest);
+        }
+
         return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
     }
 
@@ -589,7 +638,7 @@ namespace ix
                                           std::string::const_iterator message_end,
                                           bool compress)
     {
-        auto message_size = message_end - message_begin;
+        uint64_t message_size = static_cast<uint64_t>(message_end - message_begin);
 
         unsigned x = getRandomUnsigned();
         uint8_t masking_key[4] = {};
@@ -601,7 +650,8 @@ namespace ix
         std::vector<uint8_t> header;
         header.assign(2 +
                       (message_size >= 126 ? 2 : 0) +
-                      (message_size >= 65536 ? 6 : 0) + 4, 0);
+                      (message_size >= 65536 ? 6 : 0) +
+                      (_useMask ? 4 : 0), 0);
         header[0] = type;
 
         // The fin bit indicate that this is the last fragment. Fin is French for end.
@@ -618,27 +668,33 @@ namespace ix
 
         if (message_size < 126)
         {
-            header[1] = (message_size & 0xff) | 0x80;
+            header[1] = (message_size & 0xff) | (_useMask ? 0x80 : 0);
 
+            if (_useMask)
+            {
                 header[2] = masking_key[0];
                 header[3] = masking_key[1];
                 header[4] = masking_key[2];
                 header[5] = masking_key[3];
             }
+        }
         else if (message_size < 65536)
         {
-            header[1] = 126 | 0x80;
+            header[1] = 126 | (_useMask ? 0x80 : 0);
             header[2] = (message_size >> 8) & 0xff;
             header[3] = (message_size >> 0) & 0xff;
 
+            if (_useMask)
+            {
                 header[4] = masking_key[0];
                 header[5] = masking_key[1];
                 header[6] = masking_key[2];
                 header[7] = masking_key[3];
             }
+        }
         else
         { // TODO: run coverage testing here
-            header[1] = 127 | 0x80;
+            header[1] = 127 | (_useMask ? 0x80 : 0);
             header[2] = (message_size >> 56) & 0xff;
             header[3] = (message_size >> 48) & 0xff;
             header[4] = (message_size >> 40) & 0xff;
@@ -648,11 +704,14 @@ namespace ix
             header[8] = (message_size >>  8) & 0xff;
             header[9] = (message_size >>  0) & 0xff;
 
+            if (_useMask)
+            {
                 header[10] = masking_key[0];
                 header[11] = masking_key[1];
                 header[12] = masking_key[2];
                 header[13] = masking_key[3];
             }
+        }
 
         // _txbuf will keep growing until it can be transmitted over the socket:
         appendToSendBuffer(header, message_begin, message_end,
@@ -677,6 +736,15 @@ namespace ix
                         _enablePerMessageDeflate, onProgressCallback);
     }
 
+    WebSocketSendInfo WebSocketTransport::sendText(
+        const std::string& message,
+        const OnProgressCallback& onProgressCallback)
+
+    {
+        return sendData(wsheader_type::TEXT_FRAME, message,
+                        _enablePerMessageDeflate, onProgressCallback);
+    }
+
     void WebSocketTransport::sendOnSocket()
     {
         std::lock_guard<std::mutex> lock(_txbufMutex);
@@ -707,7 +775,7 @@ namespace ix
         _lastSendTimePoint = std::chrono::steady_clock::now();
     }
 
-    void WebSocketTransport::close()
+    void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)
     {
         _requestInitCancellation = true;
 
@@ -715,18 +783,29 @@ namespace ix
 
         // See list of close events here:
         // https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
-        // We use 1000: normal closure.
-        //
-        // >>> struct.pack('!H', 1000)
-        // b'\x03\xe8'
-        //
-        const std::string normalClosure = std::string("\x03\xe8");
+        const std::string closure{(char)(code >> 8), (char)(code & 0xff)};
+
         bool compress = false;
-        sendData(wsheader_type::CLOSE, normalClosure, compress);
+        sendData(wsheader_type::CLOSE, closure, compress);
         setReadyState(CLOSING);
 
-        _socket->wakeUpFromPoll();
+        _socket->wakeUpFromPoll(Socket::kCloseRequest);
         _socket->close();
+
+        {
+            std::lock_guard<std::mutex> lock(_closeDataMutex);
+            _closeCode = code;
+            _closeReason = reason;
+            _closeWireSize = closeWireSize;
+        }
+        
+        setReadyState(CLOSED);
+    }
+
+    size_t WebSocketTransport::bufferedAmount() const
+    {
+        std::lock_guard<std::mutex> lock(_txbufMutex);
+        return _txbuf.size();
     }
 
 } // namespace ix

ixwebsocket/IXWebSocketTransport.h

@@ -30,6 +30,13 @@ namespace ix
 {
     class Socket;
 
+    enum class SendMessageKind
+    {
+        Text,
+        Binary,
+        Ping
+    };
+
     class WebSocketTransport
     {
     public:
@@ -45,7 +52,8 @@ namespace ix
         {
             MSG,
             PING,
-            PONG
+            PONG,
+            FRAGMENT
         };
 
         using OnMessageCallback = std::function<void(const std::string&,
@@ -60,7 +68,7 @@ namespace ix
         ~WebSocketTransport();
 
         void configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                       int hearBeatPeriod);
+                       int heartBeatPeriod);
 
         WebSocketInitResult connectToUrl(const std::string& url, // Client
                                          int timeoutSecs);
@@ -70,12 +78,19 @@ namespace ix
         void poll();
         WebSocketSendInfo sendBinary(const std::string& message,
                                      const OnProgressCallback& onProgressCallback);
+        WebSocketSendInfo sendText(const std::string& message,
+                                   const OnProgressCallback& onProgressCallback);
         WebSocketSendInfo sendPing(const std::string& message);
-        void close();
+
+        void close(uint16_t code = 1000,
+                   const std::string& reason = "Normal closure",
+                   size_t closeWireSize = 0);
+
         ReadyStateValues getReadyState() const;
         void setReadyState(ReadyStateValues readyStateValue);
         void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
         void dispatch(const OnMessageCallback& onMessageCallback);
+        size_t bufferedAmount() const;
 
     private:
         std::string _url;
@@ -98,6 +113,10 @@ namespace ix
             uint8_t masking_key[4];
         };
 
+        // Tells whether we should mask the data we send.
+        // client should mask but server should not
+        bool _useMask;
+
         // Buffer for reading from our socket. That buffer is never resized.
         std::vector<uint8_t> _readbuf;
 
@@ -146,7 +165,7 @@ namespace ix
         mutable std::mutex _lastSendTimePointMutex;
         std::chrono::time_point<std::chrono::steady_clock> _lastSendTimePoint;
 
-        // No data was send through the socket for longer that the hearbeat period
+        // No data was send through the socket for longer than the heartbeat period
         bool heartBeatPeriodExceeded();
 
         void sendOnSocket();
@@ -172,7 +191,6 @@ namespace ix
                                 std::string::const_iterator end,
                                 uint64_t message_size,
                                 uint8_t masking_key[4]);
-        void appendToSendBuffer(const std::vector<uint8_t>& buffer);
 
         unsigned getRandomUnsigned();
         void unmaskReceiveBuffer(const wsheader_type& ws);

makefile

@@ -3,19 +3,33 @@
 #
 all: brew
 
+install: brew
+
 brew:
 	mkdir -p build && (cd build ; cmake .. ; make -j install)
 
 .PHONY: docker
+
+NAME   := bsergean/ws
+TAG    := $(shell cat DOCKER_VERSION)
+IMG    := ${NAME}:${TAG}
+LATEST := ${NAME}:latest
+BUILD  := ${NAME}:build
+
 docker:
-	docker build -t broadcast_server:latest .
+	docker build -t ${IMG} .
+	docker tag ${IMG} ${BUILD}
+
+docker_push:
+	docker tag ${IMG} ${LATEST}
+	docker push ${LATEST}
 
 run:
-	docker run --cap-add sys_ptrace -it broadcast_server:latest bash
+	docker run --cap-add sys_ptrace -it ws:latest
 
 # this is helpful to remove trailing whitespaces
 trail:
-	sh third_party/remove_trailing_whitespaces.sh
+	sh third_party/remote_trailing_whitespaces.sh
 
 build:
 	(cd examples/satori_publisher ; mkdir -p build ; cd build ; cmake .. ; make)
@@ -34,7 +48,10 @@ test_server:
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
 test:
-	python test/run.py
+	(cd test ; python2.7 run.py)
+
+ws_test: all
+	(cd ws ; bash test_ws.sh)
 
 # For the fork that is configured with appveyor
 rebase_upstream:
@@ -43,5 +60,9 @@ rebase_upstream:
 	git reset --hard upstream/master
 	git push origin master --force
 
+install_cmake_for_linux:
+	mkdir -p /tmp/cmake
+	(cd /tmp/cmake ; curl -L -O https://github.com/Kitware/CMake/releases/download/v3.14.0-rc4/cmake-3.14.0-rc4-Linux-x86_64.tar.gz ; tar zxf cmake-3.14.0-rc4-Linux-x86_64.tar.gz)
+
 .PHONY: test
 .PHONY: build

test/CMakeLists.txt

@@ -29,6 +29,7 @@ set (SOURCES
 
   IXDNSLookupTest.cpp
   IXSocketTest.cpp
+  IXSocketConnectTest.cpp
 )
 
 # Some unittest don't work on windows yet

test/IXSocketConnectTest.cpp

@@ -0,0 +1,43 @@
+/*
+ *  IXSocketConnectTest.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone. All rights reserved.
+ */
+
+#include "catch.hpp"
+
+#include "IXTest.h"
+#include <ixwebsocket/IXSocketConnect.h>
+#include <iostream>
+
+using namespace ix;
+
+
+TEST_CASE("socket_connect", "[net]")
+{
+    SECTION("Test connecting to a known hostname")
+    {
+        std::string errMsg;
+        int fd = SocketConnect::connect("www.google.com", 80, errMsg, [] { return false; });
+        std::cerr << "Error message: " << errMsg << std::endl;
+        REQUIRE(fd != -1);
+    }
+
+    SECTION("Test connecting to a non-existing hostname")
+    {
+        std::string errMsg;
+        std::string hostname("12313lhjlkjhopiupoijlkasdckljqwehrlkqjwehraospidcuaposidcasdc");
+        int fd = SocketConnect::connect(hostname, 80, errMsg, [] { return false; });
+        std::cerr << "Error message: " << errMsg << std::endl;
+        REQUIRE(fd == -1);
+    }
+
+    SECTION("Test connecting to a good hostname, with cancellation")
+    {
+        std::string errMsg;
+        // The callback returning true means we are requesting cancellation
+        int fd = SocketConnect::connect("www.google.com", 80, errMsg, [] { return true; });
+        std::cerr << "Error message: " << errMsg << std::endl;
+        REQUIRE(fd == -1);
+    }
+}

test/IXSocketTest.cpp

@@ -5,19 +5,13 @@
  */
 
 #include <iostream>
+#include <ixwebsocket/IXSocketFactory.h>
 #include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXCancellationRequest.h>
 
-#if defined(__APPLE__) or defined(__linux__)
-# ifdef __APPLE__
-#  include <ixwebsocket/IXSocketAppleSSL.h>
-# else
-#  include <ixwebsocket/IXSocketOpenSSL.h>
-# endif
-#endif
-
 #include "IXTest.h"
 #include "catch.hpp"
+#include <string.h>
 
 using namespace ix;
 
@@ -39,16 +33,15 @@ namespace ix
         Logger() << "errMsg: " << errMsg;
         REQUIRE(success);
 
-        std::cout << "Sending request: " << request
-                  << "to " << host << ":" << port
-                  << std::endl;
+        Logger() << "Sending request: " << request
+                 << "to " << host << ":" << port;
         REQUIRE(socket->writeBytes(request, isCancellationRequested));
 
         auto lineResult = socket->readLine(isCancellationRequested);
         auto lineValid = lineResult.first;
         auto line = lineResult.second;
 
-        std::cout << "read error: " << strerror(Socket::getErrno()) << std::endl;
+        Logger() << "read error: " << strerror(Socket::getErrno());
 
         REQUIRE(lineValid);
 
@@ -62,24 +55,30 @@ TEST_CASE("socket", "[socket]")
 {
     SECTION("Connect to google HTTP server. Send GET request without header. Should return 200")
     {
-        std::shared_ptr<Socket> socket(new Socket);
+        std::string errMsg;
+        bool tls = false;
+        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
         std::string host("www.google.com");
         int port = 80;
-        std::string request("GET / HTTP/1.1\r\n\r\n");
+
+        std::stringstream ss;
+        ss << "GET / HTTP/1.1\r\n";
+        ss << "Host: " << host << "\r\n";
+        ss << "\r\n";
+        std::string request(ss.str());
+
         int expectedStatus = 200;
         int timeoutSecs = 3;
 
         testSocket(host, port, request, socket, expectedStatus, timeoutSecs);
     }
 
-#if defined(__APPLE__) or defined(__linux__)
+#if defined(__APPLE__) || defined(__linux__)
     SECTION("Connect to google HTTPS server. Send GET request without header. Should return 200")
     {
-# ifdef __APPLE__
-        std::shared_ptr<Socket> socket = std::make_shared<SocketAppleSSL>();
-# else
-        std::shared_ptr<Socket> socket = std::make_shared<SocketOpenSSL>();
-# endif
+        std::string errMsg;
+        bool tls = true;
+        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
         std::string host("www.google.com");
         int port = 443;
         std::string request("GET / HTTP/1.1\r\n\r\n");

test/IXTest.cpp

@@ -16,6 +16,7 @@
 #include <iostream>
 #include <stdlib.h>
 #include <stack>
+#include <iomanip>
 
 namespace ix
 {
@@ -69,10 +70,15 @@ namespace ix
         Logger() << msg;
     }
 
+    int getAnyFreePortSimple()
+    {
+        static int defaultPort = 8090;
+        return defaultPort++;
+    }
+
     int getAnyFreePort()
     {
         int defaultPort = 8090;
-
         int sockfd;
         if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
         {
@@ -103,7 +109,7 @@ namespace ix
         }
 
         struct sockaddr_in sa; // server address information
-        unsigned int len;
+        socklen_t len;
         if (getsockname(sockfd, (struct sockaddr *) &sa, &len) < 0)
         {
             log("Cannot compute a free port. getsockname error.");
@@ -122,8 +128,15 @@ namespace ix
     {
         while (true)
         {
+#if defined(__has_feature)
+# if __has_feature(address_sanitizer)
+            int port = getAnyFreePortSimple();
+# else
             int port = getAnyFreePort();
-
+# endif
+#else
+            int port = getAnyFreePort();
+#endif
             //
             // Only port above 1024 can be used by non root users, but for some
             // reason I got port 7 returned with macOS when binding on port 0...
@@ -136,4 +149,21 @@ namespace ix
 
         return -1;
     }
+
+    void hexDump(const std::string& prefix,
+                 const std::string& s)
+    {
+        std::ostringstream ss;
+        bool upper_case = false;
+
+        for (std::string::size_type i = 0; i < s.length(); ++i)
+        {
+            ss << std::hex
+               << std::setfill('0')
+               << std::setw(2)
+               << (upper_case ? std::uppercase : std::nouppercase) << (int)s[i];
+        }
+
+        std::cout << prefix << ": " << s << " => " << ss.str() << std::endl;
+    }
 }

test/IXWebSocketHeartBeatTest.cpp

@@ -65,7 +65,7 @@ namespace
         _webSocket.setUrl(url);
 
         // The important bit for this test.
-        // Set a 1 second hearbeat ; if no traffic is present on the connection for 1 second
+        // Set a 1 second heartbeat ; if no traffic is present on the connection for 1 second
         // a ping message will be sent by the client.
         _webSocket.setHeartBeatPeriod(1);
 
@@ -128,10 +128,11 @@ namespace
     {
         // A dev/null server
         server.setOnConnectionCallback(
-            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket)
+            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket,
+                                             std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
+                    [webSocket, connectionState, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -141,6 +142,7 @@ namespace
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             Logger() << "New server connection";
+                            Logger() << "id: " << connectionState->getId();
                             Logger() << "Uri: " << openInfo.uri;
                             Logger() << "Headers:";
                             for (auto it : openInfo.headers)
@@ -210,6 +212,10 @@ TEST_CASE("Websocket_heartbeat", "[heartbeat]")
         webSocketClientA.stop();
         webSocketClientB.stop();
 
+
+        // Here we test heart beat period exceeded for clientA
+        // but it should not be exceeded for clientB which has sent data.
+        // -> expected ping messages == 2, but add a small buffer to make this more reliable.
         REQUIRE(serverReceivedPingMessages >= 2);
         REQUIRE(serverReceivedPingMessages <= 4);
 

test/IXWebSocketServerTest.cpp

@@ -8,6 +8,7 @@
 #include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXWebSocket.h>
 #include <ixwebsocket/IXWebSocketServer.h>
+#include <ixwebsocket/IXSocketFactory.h>
 
 #include "IXTest.h"
 
@@ -17,13 +18,32 @@ using namespace ix;
 
 namespace ix
 {
-    bool startServer(ix::WebSocketServer& server)
+    // Test that we can override the connectionState impl to provide our own
+    class ConnectionStateCustom : public ConnectionState
     {
+        void computeId()
+        {
+            // a very boring invariant id that we can test against in the unittest
+            _id = "foobarConnectionId";
+        }
+    };
+
+    bool startServer(ix::WebSocketServer& server,
+                     std::string& connectionId)
+    {
+        auto factory = []() -> std::shared_ptr<ConnectionState>
+        {
+            return std::make_shared<ConnectionStateCustom>();
+        };
+        server.setConnectionStateFactory(factory);
+
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket)
+            [&server, &connectionId](std::shared_ptr<ix::WebSocket> webSocket,
+                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, connectionState,
+                     &connectionId, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -33,12 +53,16 @@ namespace ix
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             Logger() << "New connection";
+                            connectionState->computeId();
+                            Logger() << "id: " << connectionState->getId();
                             Logger() << "Uri: " << openInfo.uri;
                             Logger() << "Headers:";
                             for (auto it : openInfo.headers)
                             {
                                 Logger() << it.first << ": " << it.second;
                             }
+
+                            connectionId = connectionState->getId();
                         }
                         else if (messageType == ix::WebSocket_MessageType_Close)
                         {
@@ -77,19 +101,21 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        REQUIRE(startServer(server));
+        std::string connectionId;
+        REQUIRE(startServer(server, connectionId));
 
-        Socket socket;
-        std::string host("localhost");
         std::string errMsg;
+        bool tls = false;
+        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        std::string host("localhost");
         auto isCancellationRequested = []() -> bool
         {
             return false;
         };
-        bool success = socket.connect(host, port, errMsg, isCancellationRequested);
+        bool success = socket->connect(host, port, errMsg, isCancellationRequested);
         REQUIRE(success);
 
-        auto lineResult = socket.readLine(isCancellationRequested);
+        auto lineResult = socket->readLine(isCancellationRequested);
         auto lineValid = lineResult.first;
         auto line = lineResult.second;
 
@@ -109,22 +135,24 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        REQUIRE(startServer(server));
+        std::string connectionId;
+        REQUIRE(startServer(server, connectionId));
 
-        Socket socket;
-        std::string host("localhost");
         std::string errMsg;
+        bool tls = false;
+        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        std::string host("localhost");
         auto isCancellationRequested = []() -> bool
         {
             return false;
         };
-        bool success = socket.connect(host, port, errMsg, isCancellationRequested);
+        bool success = socket->connect(host, port, errMsg, isCancellationRequested);
         REQUIRE(success);
 
         Logger() << "writeBytes";
-        socket.writeBytes("GET /\r\n", isCancellationRequested);
+        socket->writeBytes("GET /\r\n", isCancellationRequested);
 
-        auto lineResult = socket.readLine(isCancellationRequested);
+        auto lineResult = socket->readLine(isCancellationRequested);
         auto lineValid = lineResult.first;
         auto line = lineResult.second;
 
@@ -144,26 +172,28 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        REQUIRE(startServer(server));
+        std::string connectionId;
+        REQUIRE(startServer(server, connectionId));
 
-        Socket socket;
-        std::string host("localhost");
         std::string errMsg;
+        bool tls = false;
+        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        std::string host("localhost");
         auto isCancellationRequested = []() -> bool
         {
             return false;
         };
-        bool success = socket.connect(host, port, errMsg, isCancellationRequested);
+        bool success = socket->connect(host, port, errMsg, isCancellationRequested);
         REQUIRE(success);
 
-        socket.writeBytes("GET / HTTP/1.1\r\n"
+        socket->writeBytes("GET / HTTP/1.1\r\n"
                            "Upgrade: websocket\r\n"
                            "Sec-WebSocket-Version: 13\r\n"
                            "Sec-WebSocket-Key: foobar\r\n"
                            "\r\n",
                            isCancellationRequested);
 
-        auto lineResult = socket.readLine(isCancellationRequested);
+        auto lineResult = socket->readLine(isCancellationRequested);
         auto lineValid = lineResult.first;
         auto line = lineResult.second;
 
@@ -174,6 +204,8 @@ TEST_CASE("Websocket_server", "[websocket_server]")
         // Give us 500ms for the server to notice that clients went away
         ix::msleep(500);
 
+        REQUIRE(connectionId == "foobarConnectionId");
+
         server.stop();
         REQUIRE(server.getClients().size() == 0);
     }

test/cmd_websocket_chat.cpp

@@ -164,10 +164,21 @@ namespace
                     ss << "cmd_websocket_chat: Error ! " << error.reason;
                     log(ss.str());
                 }
+                else if (messageType == ix::WebSocket_MessageType_Ping)
+                {
+                    log("cmd_websocket_chat: received ping message");
+                }
+                else if (messageType == ix::WebSocket_MessageType_Pong)
+                {
+                    log("cmd_websocket_chat: received pong message");
+                }
+                else if (messageType == ix::WebSocket_MessageType_Fragment)
+                {
+                    log("cmd_websocket_chat: received message fragment");
+                }
                 else
                 {
-                    // FIXME: missing ping/pong messages
-                    ss << "Invalid ix::WebSocketMessageType";
+                    ss << "Unexpected ix::WebSocketMessageType";
                     log(ss.str());
                 }
             });
@@ -206,10 +217,11 @@ namespace
     bool startServer(ix::WebSocketServer& server)
     {
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket)
+            [&server](std::shared_ptr<ix::WebSocket> webSocket,
+                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -219,6 +231,7 @@ namespace
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             Logger() << "New connection";
+                            Logger() << "id: " << connectionState->getId();
                             Logger() << "Uri: " << openInfo.uri;
                             Logger() << "Headers:";
                             for (auto it : openInfo.headers)

test/run.py

@@ -1,26 +1,97 @@
+#!/usr/bin/env python2.7
+'''
+'''
+
+from __future__ import print_function
+
 import os
+import sys
 import platform
-import shutil
+import argparse
+import multiprocessing
+import tempfile
+import time
+import datetime
+import threading
+import subprocess
+import re
+import xml.etree.ElementTree as ET
+from xml.dom import minidom
 
-osName = platform.system()
-print('os name = {}'.format(osName))
+hasClick = True
+try:
+    import click
+except ImportError:
+    hasClick = False
 
-root = os.path.dirname(os.path.realpath(__file__))
-buildDir = os.path.join(root, 'build')
 
-if not os.path.exists(buildDir):
-    os.mkdir(buildDir)
+DEFAULT_EXE = 'ixwebsocket_unittest'
 
-os.chdir(buildDir)
 
-if osName == 'Windows':
-    generator = '-G"NMake Makefiles"'
-    make = 'nmake'
-    testBinary ='ixwebsocket_unittest.exe'
+class Command(object):
+    """Run system commands with timeout
+    
+    From http://www.bo-yang.net/2016/12/01/python-run-command-with-timeout
+    Python3 might have a builtin way to do that.
+    """
+    def __init__(self, cmd):
+        self.cmd = cmd
+        self.process = None
+
+    def run_command(self):
+        self.process = subprocess.Popen(self.cmd, shell=True)
+        self.process.communicate()
+
+    def run(self, timeout=None):
+        '''5 minutes default timeout'''
+        
+        if timeout is None:
+            timeout = 5 * 60
+
+        thread = threading.Thread(target=self.run_command, args=())
+        thread.start()
+        thread.join(timeout)
+
+        if thread.is_alive():
+            print('Command timeout, kill it: ' + self.cmd)
+            self.process.terminate()
+            thread.join()
+            return False, 255
         else:
-    generator = ''
-    make = 'make -j6'
-    testBinary ='./ixwebsocket_unittest'
+            return True, self.process.returncode
+
+
+def runCommand(cmd, assertOnFailure=True, timeout=None):
+    '''Small wrapper to run a command and make sure it succeed'''
+
+    if timeout is None:
+        timeout = 30 * 60 # 30 minute default timeout
+
+    print('\nRunning', cmd)
+    command = Command(cmd)
+    timedout, ret = command.run(timeout)
+
+    if timedout:
+        print('Unittest timed out')
+
+    msg = 'cmd {} failed with error code {}'.format(cmd, ret)
+    if ret != 0:
+        print(msg)
+        if assertOnFailure:
+            assert False
+
+
+def runCMake(sanitizer, buildDir):
+    '''Generate a makefile from CMake.
+    We do an out of dir build, so that cleaning up is easy
+    (remove build sub-folder).
+    '''
+
+    # CMake installed via Self Service ends up here.
+    cmake_executable = '/Applications/CMake.app/Contents/bin/cmake'
+
+    if not os.path.exists(cmake_executable):
+        cmake_executable = 'cmake'
 
     sanitizersFlags = {
         'asan': '-DSANITIZE_ADDRESS=On',
@@ -28,55 +99,386 @@ sanitizersFlags = {
         'tsan': '-DSANITIZE_THREAD=On',
         'none': ''
     }
-sanitizer = 'tsan'
-if osName == 'Linux':
-    sanitizer = 'none'
+    sanitizerFlag = sanitizersFlags[sanitizer]
 
-sanitizerFlags = sanitizersFlags[sanitizer]
+    # CMake installed via Self Service ends up here.
+    cmakeExecutable = '/Applications/CMake.app/Contents/bin/cmake'
+    if not os.path.exists(cmakeExecutable):
+        cmakeExecutable = 'cmake'
 
-# if osName == 'Windows':
-#     os.environ['CC'] = 'clang-cl'
-#     os.environ['CXX'] = 'clang-cl'
+    fmt = '''
+{cmakeExecutable} -H. \
+    {sanitizerFlag} \
+    -B{buildDir} \
+    -DCMAKE_BUILD_TYPE=Debug \
+    -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
+'''
+    cmakeCmd = fmt.format(**locals())
+    runCommand(cmakeCmd)
 
-cmakeCmd = 'cmake -DCMAKE_BUILD_TYPE=Debug {} {} ..'.format(generator, sanitizerFlags)
-print(cmakeCmd)
-ret = os.system(cmakeCmd)
-assert ret == 0, 'CMake failed, exiting'
 
-ret = os.system(make)
-assert ret == 0, 'Make failed, exiting'
+def runTest(args, buildDir, xmlOutput, testRunName):
+    '''Execute the unittest.
+    '''
+    if args is None:
+        args = ''
 
-def findFiles(prefix):
-    '''Find all files under a given directory'''
+    fmt = '{buildDir}/{DEFAULT_EXE} -o {xmlOutput} -n "{testRunName}" -r junit "{args}"'
+    testCommand = fmt.format(**locals())
+    runCommand(testCommand,
+               assertOnFailure=False)
 
-    paths = []
 
-    for root, _, files in os.walk(prefix):
-        for path in files:
-            fullPath = os.path.join(root, path)
+def validateTestSuite(xmlOutput):
+    '''
+    Parse the output XML file to validate that all tests passed.
 
-            if os.path.islink(fullPath):
+    Assume that the XML file contains only one testsuite.
+    (which is true when generate by catch2)
+    '''
+    tree = ET.parse(xmlOutput)
+    root = tree.getroot()
+    testSuite = root[0]
+    testSuiteAttributes = testSuite.attrib
+
+    tests = testSuiteAttributes['tests']
+
+    success = True
+
+    for testcase in testSuite:
+        if testcase.tag != 'testcase':
             continue
 
-            paths.append(fullPath)
+        testName = testcase.attrib['name']
+        systemOutput = None
 
-    return paths
+        for child in testcase:
+            if child.tag == 'system-out':
+                systemOutput = child.text
 
-#for path in findFiles('.'):
-#    print(path)
+            if child.tag == 'failure':
+                success = False
 
-# We need to copy the zlib DLL in the current work directory
-shutil.copy(os.path.join(
-    '..',
-    '..',
-    'third_party',
-    'ZLIB-Windows',
-    'zlib-1.2.11_deploy_v140',
-    'release_dynamic',
-    'x64',
-    'bin',
-    'zlib.dll'), '.')
+                print("Testcase '{}' failed".format(testName))
+                print(' ', systemOutput)
 
-testCommand = '{} {}'.format(testBinary, os.getenv('TEST', ''))
-ret = os.system(testCommand)
-assert ret == 0, 'Test command failed'
+    return success, tests
+
+
+def log(msg, color):
+    if hasClick:
+        click.secho(msg, fg=color)
+    else:
+        print(msg)
+
+
+def isSuccessFullRun(output):
+    '''When being run from lldb, we cannot capture the exit code 
+    so we have to parse the output which is produced in a 
+    consistent way. Whenever we'll be on a recent enough version of lldb we 
+    won't have to do this.
+    '''
+    pid = None
+    matchingPids = False
+    exitCode = -1
+
+    # 'Process 279 exited with status = 1 (0x00000001) ',
+    exitPattern = re.compile('^Process (?P<pid>[0-9]+) exited with status = (?P<exitCode>[0-9]+)')
+
+    # "Process 99232 launched: '/Users/bse...
+    launchedPattern = re.compile('^Process (?P<pid>[0-9]+) launched: ')
+
+    for line in output:
+        match = exitPattern.match(line)
+        if match:
+            exitCode = int(match.group('exitCode'))
+            pid = match.group('pid')
+
+        match = launchedPattern.match(line)
+        if match:
+            matchingPids = (pid == match.group('pid'))
+
+    return exitCode == 0 and matchingPids
+
+
+def testLLDBOutput():
+    failedOutputWithCrashLines = [
+        '    frame #15: 0x00007fff73f4d305 libsystem_pthread.dylib`_pthread_body + 126',
+        '    frame #16: 0x00007fff73f5026f libsystem_pthread.dylib`_pthread_start + 70',
+        '    frame #17: 0x00007fff73f4c415 libsystem_pthread.dylib`thread_start + 13',
+        '(lldb) quit 1'
+    ]
+
+    failedOutputWithFailedUnittest = [
+        '===============================================================================', 
+        'test cases:  1 |  0 passed | 1 failed', 'assertions: 15 | 14 passed | 1 failed', 
+        '', 
+        'Process 279 exited with status = 1 (0x00000001) ',
+        '',
+        "Process 279 launched: '/Users/bsergeant/src/foss/ixwebsocket/test/build/Darwin/ixwebsocket_unittest' (x86_64)"
+    ]
+
+    successLines = [
+        '...',
+        '...',
+        'All tests passed (16 assertions in 1 test case)',
+        '',
+        'Process 99232 exited with status = 0 (0x00000000) ',
+        '',
+        "Process 99232 launched: '/Users/bsergeant/src/foss/ixwebsocket/test/build/Darwin/ixwebsocket_unittest' (x86_64)"
+    ]
+
+    assert not isSuccessFullRun(failedOutputWithCrashLines)
+    assert not isSuccessFullRun(failedOutputWithFailedUnittest)
+    assert isSuccessFullRun(successLines)
+
+
+def executeJob(job):
+    '''Execute a unittest and capture info about it (runtime, success, etc...)'''
+
+    start = time.time()
+
+    sys.stderr.write('.')
+
+    # 2 minutes of timeout for a single test
+    timeout = 2 * 60
+    command = Command(job['cmd'])
+    timedout, ret = command.run(timeout)
+
+    job['exit_code'] = ret
+    job['success'] = ret == 0
+    job['runtime'] = time.time() - start
+
+    # Record unittest console output
+    job['output'] = ''
+    path = job['output_path']
+
+    if os.path.exists(path):
+        with open(path) as f:
+            output = f.read()
+            job['output'] = output
+
+        outputLines = output.splitlines()
+
+        if job['use_lldb']:
+            job['success'] = isSuccessFullRun(outputLines)
+
+        # Cleanup tmp file now that its content was read
+        os.unlink(path)
+
+    return job
+
+
+def executeJobs(jobs):
+    '''Execute a list of job concurrently on multiple CPU/cores'''
+
+    poolSize = multiprocessing.cpu_count()
+
+    pool = multiprocessing.Pool(poolSize)
+    results = pool.map(executeJob, jobs)
+    pool.close()
+    pool.join()
+
+    return results
+
+
+def computeAllTestNames(buildDir):
+    '''Compute all test case names, by executing the unittest in a custom mode'''
+
+    executable = os.path.join(buildDir, DEFAULT_EXE)
+    cmd = '"{}" --list-test-names-only'.format(executable)
+    names = os.popen(cmd).read().splitlines()
+    names.sort()  # Sort test names for execution determinism
+    return names
+
+
+def prettyPrintXML(root):
+    '''Pretty print an XML file. Default writer write it on a single line
+    which makes it hard for human to inspect.'''
+
+    serializedXml = ET.tostring(root, encoding='utf-8')
+    reparsed = minidom.parseString(serializedXml)
+    prettyPrinted = reparsed.toprettyxml(indent="  ")
+    return prettyPrinted
+
+
+def generateXmlOutput(results, xmlOutput, testRunName, runTime):
+    '''Generate a junit compatible XML file
+    
+    We prefer doing this ourself instead of letting Catch2 do it.
+    When the test is crashing (as has happened on Jenkins), an invalid file 
+    with no trailer can be created which trigger an XML reading error in validateTestSuite.
+
+    Something like that:
+    ```
+    <testsuite>
+      <foo>
+    ```
+    '''
+
+    root = ET.Element('testsuites')
+    testSuite = ET.Element('testsuite', {
+        'name': testRunName,
+        'tests': str(len(results)),
+        'failures': str(sum(1 for result in results if not result['success'])),
+        'time': str(runTime),
+        'timestamp': datetime.datetime.utcnow().isoformat(),
+    })
+    root.append(testSuite)
+
+    for result in results:
+        testCase = ET.Element('testcase', {
+            'name': result['name'],
+            'time': str(result['runtime'])
+        })
+
+        systemOut = ET.Element('system-out')
+        systemOut.text = result['output'].decode('utf-8')
+        testCase.append(systemOut)
+
+        if not result['success']:
+            failure = ET.Element('failure')
+            testCase.append(failure)
+
+        testSuite.append(testCase)
+
+    with open(xmlOutput, 'w') as f:
+        content = prettyPrintXML(root)
+        f.write(content.encode('utf-8'))
+
+
+def run(testName, buildDir, sanitizer, xmlOutput, testRunName, buildOnly, useLLDB):
+    '''Main driver. Run cmake, compiles, execute and validate the testsuite.'''
+
+    runCMake(sanitizer, buildDir)
+    runCommand('make -C {} -j8'.format(buildDir))
+
+    if buildOnly:
+        return
+
+    # A specific test case can be provided on the command line
+    if testName:
+        testNames = [testName]
+    else:
+        # Default case
+        testNames = computeAllTestNames(buildDir)
+
+    # This should be empty. It is useful to have a blacklist during transitions
+    # We could add something for asan as well.
+    blackLists = {
+        'ubsan': []
+    }
+    blackList = blackLists.get(sanitizer, [])
+
+    # Run through LLDB to capture crashes
+    lldb = ''
+    if useLLDB:
+        lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
+
+    # Jobs is a list of python dicts
+    jobs = []
+
+    for testName in testNames:
+        outputPath = tempfile.mktemp(suffix=testName + '.log')
+
+        if testName in blackList:
+            log('Skipping blacklisted test {}'.format(testName), 'yellow')
+            continue
+
+        # testName can contains spaces, so we enclose them in double quotes
+        executable = os.path.join(buildDir, DEFAULT_EXE)
+
+        cmd = '{} "{}" "{}" >& "{}"'.format(lldb, executable, testName, outputPath)
+
+        jobs.append({
+            'name': testName,
+            'cmd': cmd,
+            'output_path': outputPath,
+            'use_lldb': useLLDB
+        })
+
+    start = time.time()
+    results = executeJobs(jobs)
+    runTime = time.time() - start
+    generateXmlOutput(results, xmlOutput, testRunName, runTime)
+
+    # Validate and report results
+    print('\nParsing junit test result file: {}'.format(xmlOutput))
+    log('## Results', 'blue')
+    success, tests = validateTestSuite(xmlOutput)
+
+    if success:
+        label = 'tests' if int(tests) > 1 else 'test'
+        msg = 'All test passed (#{} {})'.format(tests, label)
+        color = 'green'
+    else:
+        msg = 'unittest failed'
+        color = 'red'
+
+    log(msg, color)
+    log('Execution time: %.2fs' % (runTime), 'blue')
+    sys.exit(0 if success else 1)
+
+
+def main():
+    buildDir = 'build/' + platform.system()
+    if not os.path.exists(buildDir):
+        os.makedirs(buildDir)
+
+    defaultOutput = DEFAULT_EXE + '.xml'
+
+    parser = argparse.ArgumentParser(description='Build and Run the engine unittest')
+
+    sanitizers = ['tsan', 'asan', 'ubsan', 'none']
+
+    parser.add_argument('--sanitizer', choices=sanitizers,
+                        help='Run a clang sanitizer.')
+    parser.add_argument('--test', '-t', help='Test name.')
+    parser.add_argument('--list', '-l', action='store_true',
+                        help='Print test names and exit.')
+    parser.add_argument('--no_sanitizer', action='store_true',
+                        help='Do not execute a clang sanitizer.')
+    parser.add_argument('--validate', action='store_true',
+                        help='Validate XML output.')
+    parser.add_argument('--build_only', '-b', action='store_true',
+                        help='Stop after building. Do not run the unittest.')
+    parser.add_argument('--output', '-o', help='Output XML file.')
+    parser.add_argument('--lldb', action='store_true',
+                        help='Run the test through lldb.')
+    parser.add_argument('--run_name', '-n',
+                        help='Name of the test run.')
+
+    args = parser.parse_args()
+
+    # Default sanitizer is tsan
+    sanitizer = args.sanitizer
+    if args.sanitizer is None:
+        sanitizer = 'tsan'
+
+    defaultRunName = 'ixengine_{}_{}'.format(platform.system(), sanitizer)
+
+    xmlOutput = args.output or defaultOutput
+    testRunName = args.run_name or os.getenv('IXENGINE_TEST_RUN_NAME') or defaultRunName
+
+    if args.list:
+        # catch2 exit with a different error code when requesting the list of files
+        try:
+            runTest('--list-test-names-only', buildDir, xmlOutput, testRunName)
+        except AssertionError:
+            pass
+        return
+
+    if args.validate:
+        validateTestSuite(xmlOutput)
+        return
+
+    if platform.system() != 'Darwin' and args.lldb:
+        print('LLDB is only supported on Apple at this point')
+        args.lldb = False
+
+    return run(args.test, buildDir, sanitizer, xmlOutput, 
+               testRunName, args.build_only, args.lldb)
+
+
+if __name__ == '__main__':
+    main()

test/test_runner.cpp

@@ -7,14 +7,8 @@
 #define CATCH_CONFIG_RUNNER
 #include "catch.hpp"
 
-#include <ixwebsocket/IXSocket.h>
-
 int main(int argc, char* argv[])
 {
-    ix::Socket::init(); // for Windows
-
     int result = Catch::Session().run(argc, argv);
-
-    ix::Socket::cleanup(); // for Windows
     return result;
 }

third_party/README.md

@@ -0,0 +1 @@
+Except ZLIB on Windows (whose port is currently broken...) all dependencies here are for the ws command line tools, not for the IXWebSockets library which is standalone.

third_party/remote_trailing_whitespaces.sh

@@ -1,2 +1,3 @@
 find . -type f -name '*.cpp' -exec sed -i '' 's/[[:space:]]*$//' {} \+
 find . -type f -name '*.h' -exec sed -i '' 's/[[:space:]]*$//' {} \+
+find . -type f -name '*.md' -exec sed -i '' 's/[[:space:]]*$//' {} \+

third_party/statsd-client-cpp/.gitignore

@@ -0,0 +1,13 @@
+# Compiled Object files
+*.slo
+*.lo
+*.o
+
+# Compiled Dynamic libraries
+*.so
+*.dylib
+
+# Compiled Static libraries
+*.lai
+*.la
+*.a

third_party/statsd-client-cpp/CMakeLists.txt

@@ -0,0 +1,18 @@
+cmake_minimum_required(VERSION 3.1)
+project(helloCLion)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+
+include_directories(
+    src
+)
+
+add_library(statsdcppclient STATIC src/statsd_client.cpp)
+add_definitions("-fPIC")
+target_link_libraries(statsdcppclient pthread)
+
+add_executable(system_monitor demo/system_monitor.cpp)
+target_link_libraries(system_monitor statsdcppclient)
+
+add_executable(test_client demo/test_client.cpp)
+target_link_libraries(test_client statsdcppclient)

third_party/statsd-client-cpp/LICENSE

@@ -0,0 +1,27 @@
+Copyright (c) 2014, Rex
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* Neither the name of the {organization} nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

third_party/statsd-client-cpp/README.md

@@ -0,0 +1,34 @@
+# a client sdk for StatsD, written in C++
+
+## API
+See [header file](src/statsd_client.h) for more api detail.
+
+** Notice: this client is not thread-safe **
+
+## Demo
+### test\_client
+This simple demo shows how the use this client.
+
+### system\_monitor
+This is a daemon for monitoring a Linux system.
+It'll wake up every minute and monitor the following:
+
+* load
+* cpu
+* free memory
+* free swap (disabled)
+* received bytes
+* transmitted bytes
+* procs
+* uptime
+
+The stats sent to statsd will be in "host.MACAddress" namespace.
+
+Usage:
+
+    system_monitor statsd-host interface-to-monitor
+
+e.g.
+
+    `system_monitor 172.16.42.1 eth0`
+

third_party/statsd-client-cpp/demo/system_monitor.cpp

@@ -0,0 +1,164 @@
+
+#include <sys/types.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <signal.h>
+#include <unistd.h>
+#include <string.h>
+#include <netdb.h>
+#include <sys/sysinfo.h>
+
+#include <sys/ioctl.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <string>
+#include <vector>
+
+#include "statsd_client.h"
+
+using namespace std;
+
+static int running = 1;
+
+void sigterm(int sig)
+{
+    running = 0;
+}
+
+string localhost() {
+    struct addrinfo hints, *info, *p;
+    string hostname(1024, '\0');
+    gethostname((char*)hostname.data(), hostname.capacity());
+
+    memset(&hints, 0, sizeof hints);
+    hints.ai_family = AF_UNSPEC; /*either IPV4 or IPV6*/
+    hints.ai_socktype = SOCK_STREAM;
+    hints.ai_flags = AI_CANONNAME;
+
+    if ( getaddrinfo(hostname.c_str(), "http", &hints, &info) == 0) {
+        for(p = info; p != NULL; p = p->ai_next) {
+            hostname = p->ai_canonname;
+        }
+        freeaddrinfo(info);
+    }
+
+    string::size_type pos = hostname.find(".");
+    while ( pos != string::npos )
+    {
+        hostname[pos] = '_';
+        pos = hostname.find(".", pos);
+    }
+    return hostname;
+}
+
+vector<string>& StringSplitTrim(const string& sData,
+        const string& sDelim, vector<string>& vItems)
+{
+    vItems.clear();
+
+    string::size_type bpos = 0;
+    string::size_type epos = 0;
+    string::size_type nlen = sDelim.size();
+
+    while(sData.substr(epos,nlen) == sDelim)
+    {
+        epos += nlen;
+    }
+    bpos = epos;
+
+    while ((epos=sData.find(sDelim, epos)) != string::npos)
+    {
+        vItems.push_back(sData.substr(bpos, epos-bpos));
+        epos += nlen;
+        while(sData.substr(epos,nlen) == sDelim)
+        {
+            epos += nlen;
+        }
+        bpos = epos;
+    }
+
+    if(bpos != sData.size())
+    {
+        vItems.push_back(sData.substr(bpos, sData.size()-bpos));
+    }
+    return vItems;
+}
+
+int main(int argc, char *argv[])
+{
+    FILE *net, *stat;
+    struct sysinfo si;
+    char line[256];
+    unsigned int user, nice, sys, idle, total, busy, old_total=0, old_busy=0;
+
+    if (argc != 3) {
+        printf( "Usage: %s host port\n"
+                "Eg:    %s 127.0.0.1 8125\n",
+                argv[0], argv[0]);
+        exit(1);
+    }
+
+    signal(SIGHUP, SIG_IGN);
+    signal(SIGPIPE, SIG_IGN);
+    signal(SIGCHLD, SIG_IGN); /* will save one syscall per sleep */
+    signal(SIGTERM, sigterm);
+
+    if ( (net = fopen("/proc/net/dev", "r")) == NULL) {
+        perror("fopen");
+        exit(-1);
+    }
+
+    if ( (stat = fopen("/proc/stat", "r")) == NULL) {
+        perror("fopen");
+        exit(-1);
+    }
+
+    string ns = string("host.") + localhost().c_str() + ".";
+    statsd::StatsdClient client(argv[1], atoi(argv[2]), ns);
+
+    daemon(0,0);
+    printf("running in background.\n");
+
+    while(running) {
+        rewind(net);
+        vector<string> items;
+        while(!feof(net)) {
+            fgets(line, sizeof(line), net);
+            StringSplitTrim(line, " ", items);
+
+            if ( items.size() < 17 ) continue;
+            if ( items[0].find(":") == string::npos ) continue;
+            if ( items[1] == "0" and items[9] == "0" ) continue;
+
+            string netface = "network."+items[0].erase( items[0].find(":") );
+            client.count( netface+".receive.bytes", atoll(items[1].c_str()) );
+            client.count( netface+".receive.packets", atoll(items[2].c_str()) );
+            client.count( netface+".transmit.bytes", atoll(items[9].c_str()) );
+            client.count( netface+".transmit.packets", atoll(items[10].c_str()) );
+        }
+
+        sysinfo(&si);
+        client.gauge("system.load", 100*si.loads[0]/0x10000);
+        client.gauge("system.freemem", si.freeram/1024);
+        client.gauge("system.procs", si.procs);
+        client.count("system.uptime", si.uptime);
+
+        /* rewind doesn't do the trick for /proc/stat */
+        freopen("/proc/stat", "r", stat);
+        fgets(line, sizeof(line), stat);
+        sscanf(line, "cpu  %u %u %u %u", &user, &nice, &sys, &idle);
+        total = user + sys + idle;
+        busy = user + sys;
+
+        client.send("system.cpu", 100 * (busy - old_busy)/(total - old_total), "g", 1.0);
+
+        old_total = total;
+        old_busy = busy;
+        sleep(6);
+    }
+
+    fclose(net);
+    fclose(stat);
+
+    exit(0);
+}

third_party/statsd-client-cpp/demo/test_client.cpp

@@ -0,0 +1,28 @@
+
+#include <iostream>
+#include <unistd.h>
+#include "statsd_client.h"
+
+int main(void)
+{
+    std::cout << "running..." << std::endl;
+
+    statsd::StatsdClient client;
+    statsd::StatsdClient client2("127.0.0.1", 8125, "myproject.abx.", true);
+
+    client.count("count1", 123, 1.0);
+    client.count("count2", 125, 1.0);
+    client.gauge("speed", 10);
+    int i;
+    for (i=0; i<1000; i++)
+        client2.timing("request", i);
+    sleep(1);
+    client.inc("count1", 1.0);
+    client2.dec("count2", 1.0);
+//    for(i=0; i<1000; i++) {
+//        client2.count("count3", i, 0.8);
+//    }
+
+    std::cout << "done" << std::endl;
+    return 0;
+}

third_party/statsd-client-cpp/src/statsd_client.cpp

@@ -0,0 +1,246 @@
+#include <math.h>
+#include <netdb.h>
+#include <time.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <netinet/in.h>
+#include "statsd_client.h"
+
+using namespace std;
+namespace statsd {
+
+inline bool fequal(float a, float b)
+{
+    const float epsilon = 0.0001;
+    return ( fabs(a - b) < epsilon );
+}
+
+inline bool should_send(float sample_rate)
+{
+    if ( fequal(sample_rate, 1.0) )
+    {
+        return true;
+    }
+
+    float p = ((float)random() / RAND_MAX);
+    return sample_rate > p;
+}
+
+struct _StatsdClientData {
+    int     sock;
+    struct  sockaddr_in server;
+
+    string  ns;
+    string  host;
+    short   port;
+    bool    init;
+
+    char    errmsg[1024];
+};
+
+StatsdClient::StatsdClient(const string& host,
+                           int port,
+                           const string& ns,
+                           const bool batching)
+: batching_(batching), exit_(false)
+{
+    d = new _StatsdClientData;
+    d->sock = -1;
+    config(host, port, ns);
+    srandom(time(NULL));
+
+    if (batching_) {
+        pthread_mutex_init(&batching_mutex_lock_, nullptr);
+        batching_thread_ = std::thread([this] {
+          while (!exit_) {
+              std::deque<std::string> staged_message_queue;
+
+              pthread_mutex_lock(&batching_mutex_lock_);
+              batching_message_queue_.swap(staged_message_queue);
+              pthread_mutex_unlock(&batching_mutex_lock_);
+
+              while(!staged_message_queue.empty()) {
+                  send_to_daemon(staged_message_queue.front());
+                  staged_message_queue.pop_front();
+              }
+
+              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+          }
+        });
+    }
+}
+
+StatsdClient::~StatsdClient()
+{
+    if (batching_) {
+        exit_ = true;
+        batching_thread_.join();
+        pthread_mutex_destroy(&batching_mutex_lock_);
+    }
+
+
+    // close socket
+    if (d->sock >= 0) {
+        close(d->sock);
+        d->sock = -1;
+        delete d;
+        d = NULL;
+    }
+}
+
+void StatsdClient::config(const string& host, int port, const string& ns)
+{
+    d->ns = ns;
+    d->host = host;
+    d->port = port;
+    d->init = false;
+    if ( d->sock >= 0 ) {
+        close(d->sock);
+    }
+    d->sock = -1;
+}
+
+int StatsdClient::init()
+{
+    if ( d->init ) return 0;
+
+    d->sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if ( d->sock == -1 ) {
+        snprintf(d->errmsg, sizeof(d->errmsg), "could not create socket, err=%m");
+        return -1;
+    }
+
+    memset(&d->server, 0, sizeof(d->server));
+    d->server.sin_family = AF_INET;
+    d->server.sin_port = htons(d->port);
+
+    int ret = inet_aton(d->host.c_str(), &d->server.sin_addr);
+    if ( ret == 0 )
+    {
+        // host must be a domain, get it from internet
+        struct addrinfo hints, *result = NULL;
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_INET;
+        hints.ai_socktype = SOCK_DGRAM;
+
+        ret = getaddrinfo(d->host.c_str(), NULL, &hints, &result);
+        if ( ret ) {
+            close(d->sock);
+            d->sock = -1;
+            snprintf(d->errmsg, sizeof(d->errmsg),
+                     "getaddrinfo fail, error=%d, msg=%s", ret, gai_strerror(ret) );
+            return -2;
+        }
+        struct sockaddr_in* host_addr = (struct sockaddr_in*)result->ai_addr;
+        memcpy(&d->server.sin_addr, &host_addr->sin_addr, sizeof(struct in_addr));
+        freeaddrinfo(result);
+    }
+
+    d->init = true;
+    return 0;
+}
+
+/* will change the original string */
+void StatsdClient::cleanup(string& key)
+{
+    size_t pos = key.find_first_of(":|@");
+    while ( pos != string::npos )
+    {
+        key[pos] = '_';
+        pos = key.find_first_of(":|@");
+    }
+}
+
+int StatsdClient::dec(const string& key, float sample_rate)
+{
+    return count(key, -1, sample_rate);
+}
+
+int StatsdClient::inc(const string& key, float sample_rate)
+{
+    return count(key, 1, sample_rate);
+}
+
+int StatsdClient::count(const string& key, size_t value, float sample_rate)
+{
+    return send(key, value, "c", sample_rate);
+}
+
+int StatsdClient::gauge(const string& key, size_t value, float sample_rate)
+{
+    return send(key, value, "g", sample_rate);
+}
+
+int StatsdClient::timing(const string& key, size_t ms, float sample_rate)
+{
+    return send(key, ms, "ms", sample_rate);
+}
+
+int StatsdClient::send(string key, size_t value, const string &type, float sample_rate)
+{
+    if (!should_send(sample_rate)) {
+        return 0;
+    }
+
+    cleanup(key);
+
+    char buf[256];
+    if ( fequal( sample_rate, 1.0 ) )
+    {
+        snprintf(buf, sizeof(buf), "%s%s:%zd|%s",
+                 d->ns.c_str(), key.c_str(), value, type.c_str());
+    }
+    else
+    {
+        snprintf(buf, sizeof(buf), "%s%s:%zd|%s|@%.2f",
+                 d->ns.c_str(), key.c_str(), value, type.c_str(), sample_rate);
+    }
+
+    return send(buf);
+}
+
+int StatsdClient::send(const string &message)
+{
+    if (batching_) {
+        pthread_mutex_lock(&batching_mutex_lock_);
+        if (batching_message_queue_.empty() ||
+            batching_message_queue_.back().length() > max_batching_size) {
+            batching_message_queue_.push_back(message);
+        } else {
+            (*batching_message_queue_.rbegin()).append("\n").append(message);
+        }
+        pthread_mutex_unlock(&batching_mutex_lock_);
+
+        return 0;
+    } else {
+        return send_to_daemon(message);
+    }
+}
+
+
+int StatsdClient::send_to_daemon(const string &message) {
+    std::cout << "send_to_daemon: " << message.length() << " B" << std::endl;
+    int ret = init();
+    if ( ret )
+    {
+        return ret;
+    }
+    ret = sendto(d->sock, message.data(), message.size(), 0, (struct sockaddr *) &d->server, sizeof(d->server));
+    if ( ret == -1) {
+        snprintf(d->errmsg, sizeof(d->errmsg),
+                 "sendto server fail, host=%s:%d, err=%m", d->host.c_str(), d->port);
+        return -1;
+    }
+
+    return 0;
+}
+
+const char* StatsdClient::errmsg()
+{
+    return d->errmsg;
+}
+
+}
+

third_party/statsd-client-cpp/src/statsd_client.h

@@ -0,0 +1,66 @@
+
+#ifndef STATSD_CLIENT_H
+#define STATSD_CLIENT_H
+
+#include <sys/types.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <pthread.h>
+#include <string>
+#include <thread>
+#include <deque>
+#include <iostream>
+
+namespace statsd {
+
+struct _StatsdClientData;
+
+class StatsdClient {
+public:
+    StatsdClient(const std::string& host="127.0.0.1", int port=8125, const std::string& ns = "", const bool batching = false);
+    ~StatsdClient();
+
+public:
+    // you can config at anytime; client will use new address (useful for Singleton)
+    void config(const std::string& host, int port, const std::string& ns = "");
+    const char* errmsg();
+    int send_to_daemon(const std::string &);
+
+public:
+    int inc(const std::string& key, float sample_rate = 1.0);
+    int dec(const std::string& key, float sample_rate = 1.0);
+    int count(const std::string& key, size_t value, float sample_rate = 1.0);
+    int gauge(const std::string& key, size_t value, float sample_rate = 1.0);
+    int timing(const std::string& key, size_t ms, float sample_rate = 1.0);
+
+public:
+    /**
+     * (Low Level Api) manually send a message
+     * which might be composed of several lines.
+     */
+    int send(const std::string& message);
+
+    /* (Low Level Api) manually send a message
+     * type = "c", "g" or "ms"
+     */
+    int send(std::string key, size_t value,
+             const std::string& type, float sample_rate);
+
+protected:
+    int init();
+    void cleanup(std::string& key);
+
+protected:
+    struct _StatsdClientData* d;
+
+    bool batching_;
+    bool exit_;
+    pthread_mutex_t batching_mutex_lock_;
+    std::thread batching_thread_;
+    std::deque<std::string> batching_message_queue_;
+    const uint64_t max_batching_size = 32768;
+};
+
+}; // end namespace
+
+#endif

ws/.gitignore

@@ -1 +1,2 @@
 build
+node_modules

ws/CMakeLists.txt

@@ -7,7 +7,12 @@ cmake_minimum_required (VERSION 3.4.1)
 project (ws)
 
 # There's -Weverything too for clang
+if (NOT WIN32)
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
+endif()
+
+#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
+#set(CMAKE_LD_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
 
 set (CMAKE_CXX_STANDARD 14)
 
@@ -16,13 +21,22 @@ option(USE_TLS "Add TLS support" ON)
 include_directories(ws .)
 include_directories(ws ..)
 include_directories(ws ../third_party)
+include_directories(ws ../third_party/statsd-client-cpp/src)
 
 add_executable(ws 
   ../third_party/msgpack11/msgpack11.cpp
+  ../third_party/jsoncpp/jsoncpp.cpp
+  ../third_party/statsd-client-cpp/src/statsd_client.cpp
   ixcrypto/IXBase64.cpp
   ixcrypto/IXHash.cpp
   ixcrypto/IXUuid.cpp
+  ixcrypto/IXHMac.cpp
 
+  IXRedisClient.cpp
+  IXSentryClient.cpp
+  IXCobraConnection.cpp
+
+  ws_http_client.cpp
   ws_ping_pong.cpp
   ws_broadcast_server.cpp
   ws_echo_server.cpp
@@ -31,11 +45,21 @@ add_executable(ws
   ws_transfer.cpp
   ws_send.cpp
   ws_receive.cpp
+  ws_redis_publish.cpp
+  ws_redis_subscribe.cpp
+  ws_cobra_subscribe.cpp
+  ws_cobra_to_statsd.cpp
+  ws_cobra_to_sentry.cpp
   ws.cpp)
 
-if (APPLE AND USE_TLS)
-    target_link_libraries(ws "-framework foundation" "-framework security")
+target_link_libraries(ws ixwebsocket)
+
+if(NOT APPLE)
+  find_package(OpenSSL REQUIRED)
+  add_definitions(${OPENSSL_DEFINITIONS})
+  message(STATUS "OpenSSL: " ${OPENSSL_VERSION})
+  include_directories(${OPENSSL_INCLUDE_DIR})
+  target_link_libraries(ws ${OPENSSL_LIBRARIES})
 endif()
 
-target_link_libraries(ws ixwebsocket)
 install(TARGETS ws RUNTIME DESTINATION bin)

ws/IXCobraConnection.cpp

@@ -6,6 +6,7 @@
 
 #include "IXCobraConnection.h"
 #include <ixcrypto/IXHMac.h>
+#include <ixwebsocket/IXWebSocket.h>
 
 #include <algorithm>
 #include <stdexcept>
@@ -20,9 +21,10 @@ namespace ix
     constexpr size_t CobraConnection::kQueueMaxSize;
 
     CobraConnection::CobraConnection() :
+        _webSocket(new WebSocket()),
+        _publishMode(CobraConnection_PublishMode_Immediate),
         _authenticated(false),
-        _eventCallback(nullptr),
-        _publishMode(CobraConnection_PublishMode_Immediate)
+        _eventCallback(nullptr)
     {
         _pdu["action"] = "rtm/publish";
 
@@ -32,6 +34,7 @@ namespace ix
     CobraConnection::~CobraConnection()
     {
         disconnect();
+        setEventCallback(nullptr);
     }
 
     void CobraConnection::setTrafficTrackerCallback(const TrafficTrackerCallback& callback)
@@ -60,35 +63,39 @@ namespace ix
 
     void CobraConnection::invokeEventCallback(ix::CobraConnectionEventType eventType,
                                                const std::string& errorMsg,
-                                              const WebSocketHttpHeaders& headers)
+                                               const WebSocketHttpHeaders& headers,
+                                               const std::string& subscriptionId)
     {
         std::lock_guard<std::mutex> lock(_eventCallbackMutex);
         if (_eventCallback)
         {
-            _eventCallback(eventType, errorMsg, headers);
+            _eventCallback(eventType, errorMsg, headers, subscriptionId);
         }
     }
 
-    void CobraConnection::invokeErrorCallback(const std::string& errorMsg)
+    void CobraConnection::invokeErrorCallback(const std::string& errorMsg,
+                                               const std::string& serializedPdu)
     {
-        invokeEventCallback(ix::CobraConnection_EventType_Error, errorMsg);
+        std::stringstream ss;
+        ss << errorMsg << " : received pdu => " << serializedPdu;
+        invokeEventCallback(ix::CobraConnection_EventType_Error, ss.str());
     }
 
     void CobraConnection::disconnect()
     {
         _authenticated = false;
-        _webSocket.stop();
+        _webSocket->stop();
     }
 
     void CobraConnection::initWebSocketOnMessageCallback()
     {
-        _webSocket.setOnMessageCallback(
+        _webSocket->setOnMessageCallback(
             [this](ix::WebSocketMessageType messageType,
                    const std::string& str,
                    size_t wireSize,
                    const ix::WebSocketErrorInfo& error,
-                   const ix::WebSocketCloseInfo& closeInfo,
-                   const ix::WebSocketHttpHeaders& headers)
+                   const ix::WebSocketOpenInfo& openInfo,
+                   const ix::WebSocketCloseInfo& closeInfo)
             {
                 CobraConnection::invokeTrafficTrackerCallback(wireSize, true);
 
@@ -97,7 +104,7 @@ namespace ix
                 {
                     invokeEventCallback(ix::CobraConnection_EventType_Open,
                                         std::string(),
-                                        headers);
+                                        openInfo.headers);
                     sendHandshakeMessage();
                 }
                 else if (messageType == ix::WebSocket_MessageType_Close)
@@ -116,13 +123,13 @@ namespace ix
                     Json::Reader reader;
                     if (!reader.parse(str, data))
                     {
-                        invokeErrorCallback(std::string("Invalid json: ") + str);
+                        invokeErrorCallback("Invalid json", str);
                         return;
                     }
 
                     if (!data.isMember("action"))
                     {
-                        invokeErrorCallback("Missing action");
+                        invokeErrorCallback("Missing action", str);
                         return;
                     }
 
@@ -132,12 +139,12 @@ namespace ix
                     {
                         if (!handleHandshakeResponse(data))
                         {
-                            invokeErrorCallback("Error extracting nonce from handshake response");
+                            invokeErrorCallback("Error extracting nonce from handshake response", str);
                         }
                     }
                     else if (action == "auth/handshake/error")
                     {
-                        invokeErrorCallback("Handshake error."); // print full message ?
+                        invokeErrorCallback("Handshake error", str);
                     }
                     else if (action == "auth/authenticate/ok")
                     {
@@ -147,15 +154,37 @@ namespace ix
                     }
                     else if (action == "auth/authenticate/error")
                     {
-                        invokeErrorCallback("Authentication error."); // print full message ?
+                        invokeErrorCallback("Authentication error", str);
                     }
                     else if (action == "rtm/subscription/data")
                     {
                         handleSubscriptionData(data);
                     }
+                    else if (action == "rtm/subscribe/ok")
+                    {
+                        if (!handleSubscriptionResponse(data))
+                        {
+                            invokeErrorCallback("Error processing subscribe response", str);
+                        }
+                    }
+                    else if (action == "rtm/subscribe/error")
+                    {
+                        invokeErrorCallback("Subscription error", str);
+                    }
+                    else if (action == "rtm/unsubscribe/ok")
+                    {
+                        if (!handleUnsubscriptionResponse(data))
+                        {
+                            invokeErrorCallback("Error processing subscribe response", str);
+                        }
+                    }
+                    else if (action == "rtm/unsubscribe/error")
+                    {
+                        invokeErrorCallback("Unsubscription error", str);
+                    }
                     else
                     {
-                        invokeErrorCallback(std::string("Un-handled message type: ") + action);
+                        invokeErrorCallback("Un-handled message type", str);
                     }
                 }
                 else if (messageType == ix::WebSocket_MessageType_Error)
@@ -165,7 +194,7 @@ namespace ix
                     ss << "#retries: "         << error.retries     << std::endl;
                     ss << "Wait time(ms): "    << error.wait_time   << std::endl;
                     ss << "HTTP Status: "      << error.http_status << std::endl;
-                    invokeErrorCallback(ss.str());
+                    invokeErrorCallback(ss.str(), std::string());
                 }
         });
     }
@@ -192,8 +221,8 @@ namespace ix
         ss << _appkey;
 
         std::string url = ss.str();
-        _webSocket.setUrl(url);
-        _webSocket.setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
+        _webSocket->setUrl(url);
+        _webSocket->setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
     }
 
     //
@@ -226,7 +255,7 @@ namespace ix
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
 
-        return _webSocket.send(serializedJson).success;
+        return _webSocket->send(serializedJson).success;
     }
 
     // 
@@ -288,9 +317,40 @@ namespace ix
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
 
-        return _webSocket.send(serializedJson).success;
+        return _webSocket->send(serializedJson).success;
     }
 
+    bool CobraConnection::handleSubscriptionResponse(const Json::Value& pdu)
+    {
+        if (!pdu.isMember("body")) return false;
+        Json::Value body = pdu["body"];
+
+        if (!body.isMember("subscription_id")) return false;
+        Json::Value subscriptionId = body["subscription_id"];
+
+        if (!subscriptionId.isString()) return false;
+
+        invokeEventCallback(ix::CobraConnection_EventType_Subscribed,
+                            std::string(), WebSocketHttpHeaders(),
+                            subscriptionId.asString());
+        return true;
+    }
+
+    bool CobraConnection::handleUnsubscriptionResponse(const Json::Value& pdu)
+    {
+        if (!pdu.isMember("body")) return false;
+        Json::Value body = pdu["body"];
+
+        if (!body.isMember("subscription_id")) return false;
+        Json::Value subscriptionId = body["subscription_id"];
+
+        if (!subscriptionId.isString()) return false;
+
+        invokeEventCallback(ix::CobraConnection_EventType_UnSubscribed,
+                            std::string(), WebSocketHttpHeaders(),
+                            subscriptionId.asString());
+        return true;
+    }
 
     bool CobraConnection::handleSubscriptionData(const Json::Value& pdu)
     {
@@ -320,13 +380,13 @@ namespace ix
 
     bool CobraConnection::connect()
     {
-        _webSocket.start();
+        _webSocket->start();
         return true;
     }
 
     bool CobraConnection::isConnected() const
     {
-        return _webSocket.getReadyState() == ix::WebSocket_ReadyState_Open;
+        return _webSocket->getReadyState() == ix::WebSocket_ReadyState_Open;
     }
 
     std::string CobraConnection::serializeJson(const Json::Value& value)
@@ -381,7 +441,7 @@ namespace ix
         pdu["action"] = "rtm/subscribe";
         pdu["body"] = body;
 
-        _webSocket.send(pdu.toStyledString());
+        _webSocket->send(pdu.toStyledString());
 
         // Set the callback
         std::lock_guard<std::mutex> lock(_cbsMutex);
@@ -400,13 +460,13 @@ namespace ix
 
         // Create and send an unsubscribe pdu
         Json::Value body;
-        body["channel"] = channel;
+        body["subscription_id"] = channel;
 
         Json::Value pdu;
         pdu["action"] = "rtm/unsubscribe";
         pdu["body"] = body;
 
-        _webSocket.send(pdu.toStyledString());
+        _webSocket->send(pdu.toStyledString());
     }
 
     //
@@ -456,7 +516,7 @@ namespace ix
 
     bool CobraConnection::publishMessage(const std::string& serializedJson)
     {
-        auto webSocketSendInfo = _webSocket.send(serializedJson);
+        auto webSocketSendInfo = _webSocket->send(serializedJson);
         CobraConnection::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize,
                                                        false);
         return webSocketSendInfo.success;

ws/IXCobraConnection.h

@@ -11,19 +11,24 @@
 #include <string>
 #include <thread>
 #include <unordered_map>
+#include <memory>
 
 #include <jsoncpp/json/json.h>
-#include <ixwebsocket/IXWebSocket.h>
+#include <ixwebsocket/IXWebSocketHttpHeaders.h>
 #include <ixwebsocket/IXWebSocketPerMessageDeflateOptions.h>
 
 namespace ix
 {
+    class WebSocket;
+
     enum CobraConnectionEventType
     {
         CobraConnection_EventType_Authenticated = 0,
         CobraConnection_EventType_Error = 1,
         CobraConnection_EventType_Open = 2,
-        CobraConnection_EventType_Closed = 3
+        CobraConnection_EventType_Closed = 3,
+        CobraConnection_EventType_Subscribed = 4,
+        CobraConnection_EventType_UnSubscribed = 5
     };
 
     enum CobraConnectionPublishMode
@@ -35,7 +40,8 @@ namespace ix
     using SubscriptionCallback = std::function<void(const Json::Value&)>;
     using EventCallback = std::function<void(CobraConnectionEventType,
                                              const std::string&,
-                                             const WebSocketHttpHeaders&)>;
+                                             const WebSocketHttpHeaders&,
+                                             const std::string&)>;
     using TrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
 
     class CobraConnection
@@ -100,6 +106,8 @@ namespace ix
         bool handleHandshakeResponse(const Json::Value& data);
         bool sendAuthMessage(const std::string& nonce);
         bool handleSubscriptionData(const Json::Value& pdu);
+        bool handleSubscriptionResponse(const Json::Value& pdu);
+        bool handleUnsubscriptionResponse(const Json::Value& pdu);
 
         void initWebSocketOnMessageCallback();
 
@@ -113,13 +121,15 @@ namespace ix
         /// Invoke event callbacks
         void invokeEventCallback(CobraConnectionEventType eventType,
                                  const std::string& errorMsg = std::string(),
-                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders());
-        void invokeErrorCallback(const std::string& errorMsg);
+                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders(),
+                                 const std::string& subscriptionId = std::string());
+        void invokeErrorCallback(const std::string& errorMsg,
+                                 const std::string& serializedPdu);
 
         ///
         /// Member variables
         /// 
-        WebSocket _webSocket;
+        std::unique_ptr<WebSocket> _webSocket;
 
         /// Configuration data
         std::string _appkey;

ws/IXRedisClient.cpp

@@ -0,0 +1,207 @@
+/*
+ *  IXRedisClient.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include "IXRedisClient.h"
+#include <ixwebsocket/IXSocketFactory.h>
+#include <ixwebsocket/IXSocket.h>
+
+#include <sstream>
+#include <iomanip>
+#include <vector>
+#include <cstring>
+
+namespace ix
+{
+    bool RedisClient::connect(const std::string& hostname, int port)
+    {
+        bool tls = false;
+        std::string errorMsg;
+        _socket = createSocket(tls, errorMsg);
+
+        if (!_socket)
+        {
+            return false;
+        }
+
+        std::string errMsg;
+        return _socket->connect(hostname, port, errMsg, nullptr);
+    }
+
+    bool RedisClient::auth(const std::string& password,
+                           std::string& response)
+    {
+        response.clear();
+
+        if (!_socket) return false;
+
+        std::stringstream ss;
+        ss << "AUTH ";
+        ss << password;
+        ss << "\r\n";
+
+        bool sent = _socket->writeBytes(ss.str(), nullptr);
+        if (!sent)
+        {
+            return false;
+        }
+
+        auto pollResult = _socket->isReadyToRead(-1);
+        if (pollResult == PollResultType::Error)
+        {
+            return false;
+        }
+
+        auto lineResult = _socket->readLine(nullptr);
+        auto lineValid = lineResult.first;
+        auto line = lineResult.second;
+
+        response = line;
+        return lineValid;
+    }
+
+
+    bool RedisClient::publish(const std::string& channel,
+                              const std::string& message)
+    {
+        if (!_socket) return false;
+
+        std::stringstream ss;
+        ss << "PUBLISH ";
+        ss << channel;
+        ss << " ";
+        ss << message;
+        ss << "\r\n";
+
+        bool sent = _socket->writeBytes(ss.str(), nullptr);
+        if (!sent)
+        {
+            return false;
+        }
+
+        auto pollResult = _socket->isReadyToRead(-1);
+        if (pollResult == PollResultType::Error)
+        {
+            return false;
+        }
+
+        auto lineResult = _socket->readLine(nullptr);
+        auto lineValid = lineResult.first;
+        auto line = lineResult.second;
+
+        return lineValid;
+    }
+
+    //
+    // FIXME: we assume that redis never return errors...
+    //
+    bool RedisClient::subscribe(const std::string& channel,
+                                const OnRedisSubscribeResponseCallback& responseCallback,
+                                const OnRedisSubscribeCallback& callback)
+    {
+        if (!_socket) return false;
+
+        std::stringstream ss;
+        ss << "SUBSCRIBE ";
+        ss << channel;
+        ss << "\r\n";
+
+        bool sent = _socket->writeBytes(ss.str(), nullptr);
+        if (!sent)
+        {
+            return false;
+        }
+
+        // Wait 1s for the response
+        auto pollResult = _socket->isReadyToRead(-1);
+        if (pollResult == PollResultType::Error)
+        {
+            return false;
+        }
+
+        // build the response as a single string
+        std::stringstream oss;
+
+        // Read the first line of the response
+        auto lineResult = _socket->readLine(nullptr);
+        auto lineValid = lineResult.first;
+        auto line = lineResult.second;
+        oss << line;
+
+        if (!lineValid) return false;
+
+        // There are 5 items for the subscribe repply
+        for (int i = 0; i < 5; ++i)
+        {
+            auto lineResult = _socket->readLine(nullptr);
+            auto lineValid = lineResult.first;
+            auto line = lineResult.second;
+            oss << line;
+
+            if (!lineValid) return false;
+        }
+
+        responseCallback(oss.str());
+
+        // Wait indefinitely for new messages
+        while (true)
+        {
+            // Wait until something is ready to read
+            auto pollResult = _socket->isReadyToRead(-1);
+            if (pollResult == PollResultType::Error)
+            {
+                return false;
+            }
+
+            // The first line of the response describe the return type,
+            // => *3 (an array of 3 elements)
+            auto lineResult = _socket->readLine(nullptr);
+            auto lineValid = lineResult.first;
+            auto line = lineResult.second;
+
+            if (!lineValid) return false;
+
+            int arraySize;
+            {
+                std::stringstream ss;
+                ss << line.substr(1, line.size()-1);
+                ss >> arraySize;
+            }
+
+            // There are 6 items for each received message
+            for (int i = 0; i < arraySize; ++i)
+            {
+                auto lineResult = _socket->readLine(nullptr);
+                auto lineValid = lineResult.first;
+                auto line = lineResult.second;
+
+                if (!lineValid) return false;
+
+                // Messages are string, which start with a string size
+                // => $7 (7 bytes)
+                int stringSize;
+                std::stringstream ss;
+                ss << line.substr(1, line.size()-1);
+                ss >> stringSize;
+
+                auto readResult = _socket->readBytes(stringSize, nullptr, nullptr);
+                if (!readResult.first) return false;
+
+                if (i == 2)
+                {
+                    // The message is the 3rd element.
+                    callback(readResult.second);
+                }
+
+                // read last 2 bytes (\r\n)
+                char c;
+                _socket->readByte(&c, nullptr);
+                _socket->readByte(&c, nullptr);
+            }
+        }
+
+        return true;
+    }
+}

ws/IXRedisClient.h

@@ -0,0 +1,41 @@
+/*
+ *  IXRedisClient.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <memory>
+#include <functional>
+
+namespace ix
+{
+    class Socket;
+
+    class RedisClient {
+    public:
+        using OnRedisSubscribeResponseCallback = std::function<void(const std::string&)>;
+        using OnRedisSubscribeCallback = std::function<void(const std::string&)>;
+
+        RedisClient() = default;
+        ~RedisClient() = default;
+
+        bool connect(const std::string& hostname,
+                     int port);
+
+        bool auth(const std::string& password,
+                  std::string& response);
+
+        bool publish(const std::string& channel,
+                     const std::string& message);
+
+        bool subscribe(const std::string& channel,
+                       const OnRedisSubscribeResponseCallback& responseCallback,
+                       const OnRedisSubscribeCallback& callback);
+
+    private:
+        std::shared_ptr<Socket> _socket;
+    };
+}
+

ws/IXSentryClient.cpp

@@ -0,0 +1,178 @@
+/*
+ *  IXSentryClient.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone. All rights reserved.
+ */
+
+#include "IXSentryClient.h"
+
+#include <chrono>
+#include <iostream>
+
+#include <ixwebsocket/IXWebSocketHttpHeaders.h>
+
+
+namespace ix
+{
+    SentryClient::SentryClient(const std::string& dsn) :
+        _dsn(dsn),
+        _validDsn(false),
+        _luaFrameRegex("\t([^/]+):([0-9]+): in function '([^/]+)'")
+    {
+        const std::regex dsnRegex("(http[s]?)://([^:]+):([^@]+)@([^/]+)/([0-9]+)");
+        std::smatch group;
+
+        if (std::regex_match(dsn, group, dsnRegex) and group.size() == 6)
+        {
+            _validDsn = true;
+
+            const auto scheme = group.str(1);
+            const auto host = group.str(4);
+            const auto project_id = group.str(5);
+            _url = scheme + "://" + host + "/api/" + project_id + "/store/";
+
+            _publicKey = group.str(2);
+            _secretKey = group.str(3);
+        }
+    }
+
+    int64_t SentryClient::getTimestamp()
+    {
+        const auto tp = std::chrono::system_clock::now();
+        const auto dur = tp.time_since_epoch();
+        return std::chrono::duration_cast<std::chrono::seconds>(dur).count();
+    }
+
+    std::string SentryClient::getIso8601()
+    {
+        std::time_t now;
+        std::time(&now);
+        char buf[sizeof "2011-10-08T07:07:09Z"];
+        std::strftime(buf, sizeof buf, "%Y-%m-%dT%H:%M:%SZ", std::gmtime(&now));
+        return buf;
+    }
+
+    std::string SentryClient::computeAuthHeader()
+    {
+        std::string securityHeader("Sentry sentry_version=5");
+        securityHeader += ",sentry_client=ws/1.0.0";
+        securityHeader += ",sentry_timestamp=" + std::to_string(SentryClient::getTimestamp());
+        securityHeader += ",sentry_key=" + _publicKey;
+        securityHeader += ",sentry_secret=" + _secretKey;
+
+        return securityHeader;
+    }
+
+    Json::Value SentryClient::parseLuaStackTrace(const std::string& stack)
+    {
+        Json::Value frames;
+
+        // Split by lines
+        std::string line;
+        std::stringstream tokenStream(stack);
+
+        std::stringstream ss;
+        std::smatch group;
+
+        while (std::getline(tokenStream, line))
+        {
+            //	MapScene.lua:2169: in function 'singleCB'
+            if (std::regex_match(line, group, _luaFrameRegex))
+            {
+                const auto fileName = group.str(1);
+                const auto linenoStr = group.str(2);
+                const auto function = group.str(3);
+
+                ss << linenoStr;
+                uint64_t lineno;
+                ss >> lineno;
+
+                Json::Value frame;
+                frame["lineno"] = lineno;
+                frame["filename"] = fileName;
+                frame["function"] = function;
+
+                frames.append(frame);
+            }
+        }
+
+        return frames;
+    }
+
+    std::string SentryClient::computePayload(const Json::Value& msg)
+    {
+        Json::Value payload;
+        payload["platform"] = "python";
+        payload["sdk"]["name"] = "ws";
+        payload["sdk"]["version"] = "1.0.0";
+        payload["timestamp"] = SentryClient::getIso8601();
+
+        Json::Value exception;
+        exception["value"] = msg["data"]["message"];
+
+        std::string stackTraceFieldName = 
+            (msg["id"].asString() == "game_noisytypes_id") ? "traceback" : "stack";
+                             
+        exception["stacktrace"]["frames"] = 
+            parseLuaStackTrace(msg["data"][stackTraceFieldName].asString());
+
+        payload["exception"].append(exception);
+
+        Json::Value extra;
+        extra["cobra_event"] = msg;
+
+        exception["extra"] = extra;
+
+        return _jsonWriter.write(payload);
+    }
+
+    bool SentryClient::send(const Json::Value& msg,
+                            bool verbose)
+    {
+        HttpRequestArgs args;
+        args.extraHeaders["X-Sentry-Auth"] = SentryClient::computeAuthHeader();
+        args.connectTimeout = 60;
+        args.transferTimeout = 5 * 60;
+        args.followRedirects = true;
+        args.verbose = verbose;
+        args.logger = [](const std::string& msg)
+        {
+            std::cout << msg;
+        };
+
+        std::string body = computePayload(msg);
+        HttpResponse out = _httpClient.post(_url, body, args);
+
+        auto statusCode = std::get<0>(out);
+        auto errorCode = std::get<1>(out);
+        auto responseHeaders = std::get<2>(out);
+        auto payload = std::get<3>(out);
+        auto errorMsg = std::get<4>(out);
+        auto uploadSize = std::get<5>(out);
+        auto downloadSize = std::get<6>(out);
+
+        if (verbose)
+        {
+            for (auto it : responseHeaders)
+            {
+                std::cerr << it.first << ": " << it.second << std::endl;
+            }
+
+            std::cerr << "Upload size: " << uploadSize << std::endl;
+            std::cerr << "Download size: " << downloadSize << std::endl;
+
+            std::cerr << "Status: " << statusCode << std::endl;
+            if (errorCode != HttpErrorCode_Ok)
+            {
+                std::cerr << "error message: " << errorMsg << std::endl;
+            }
+
+            if (responseHeaders["Content-Type"] != "application/octet-stream")
+            {
+                std::cerr << "payload: " << payload << std::endl;
+            }
+        }
+
+        return statusCode == 200;
+    }
+} // namespace ix

ws/IXSentryClient.h

@@ -0,0 +1,47 @@
+/*
+ *  IXSentryClient.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone. All rights reserved.
+ */
+
+#pragma once
+
+#include <jsoncpp/json/json.h>
+#include <regex>
+
+#include <ixwebsocket/IXHttpClient.h>
+
+namespace ix
+{
+    class SentryClient
+    {
+    public:
+        SentryClient(const std::string& dsn);
+        ~SentryClient() = default;
+
+        bool send(const Json::Value& msg, bool verbose);
+
+    private:
+        int64_t getTimestamp();
+        std::string computeAuthHeader();
+        std::string getIso8601();
+        std::string computePayload(const Json::Value& msg);
+
+        Json::Value parseLuaStackTrace(const std::string& stack);
+
+        std::string _dsn;
+        bool _validDsn;
+        std::string _url;
+
+        // Used for authentication with a header
+        std::string _publicKey;
+        std::string _secretKey;
+
+        Json::FastWriter _jsonWriter;
+
+        std::regex _luaFrameRegex;
+
+        HttpClient _httpClient;
+    };
+    
+} // namespace ix

ws/README.md

@@ -1,10 +1,64 @@
+# General
+
+ws is a command line tool that should exercise most of the IXWebSocket code, and provide example code.
+
+```
+$ ws --help
+ws is a websocket tool
+Usage: ws [OPTIONS] SUBCOMMAND
+
+Options:
+  -h,--help                   Print this help message and exit
+
+Subcommands:
+  send                        Send a file
+  receive                     Receive a file
+  transfer                    Broadcasting server
+  connect                     Connect to a remote server
+  chat                        Group chat
+  echo_server                 Echo server
+  broadcast_server            Broadcasting server
+  ping                        Ping pong
+  curl                        HTTP Client
+  redis_publish               Redis publisher
+  redis_subscribe             Redis subscriber
+```
+
+## file transfer
+
 ```
 # Start transfer server, which is just a broadcast server at this point
-./ws transfer # running on port 8080.
+ws transfer # running on port 8080.
 
 # Start receiver first
-./ws receive ws://localhost:8080 
+ws receive ws://localhost:8080
 
 # Then send a file. File will be received and written to disk by the receiver process
-./ws send ws://localhost:8080 /file/to/path
+ws send ws://localhost:8080 /file/to/path
+```
+
+## curl
+
+```
+$ ws curl --help
+HTTP Client
+Usage: ws curl [OPTIONS] url
+
+Positionals:
+  url TEXT REQUIRED           Connection url
+
+Options:
+  -h,--help                   Print this help message and exit
+  -d TEXT                     Form data
+  -F TEXT                     Form data
+  -H TEXT                     Header
+  --output TEXT               Output file
+  -I                          Send a HEAD request
+  -L                          Follow redirects
+  --max-redirects INT         Max Redirects
+  -v                          Verbose
+  -O                          Save output to disk
+  --compress                  Enable gzip compression
+  --connect-timeout INT       Connection timeout
+  --transfer-timeout INT      Transfer timeout
 ```

ws/cobra_publisher/.gitignore

@@ -1,3 +0,0 @@
-venv
-build
-node_modules

ws/cobra_publisher/CMakeLists.txt

@@ -1,38 +0,0 @@
-#
-# Author: Benjamin Sergeant
-# Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
-#
-
-cmake_minimum_required (VERSION 3.4.1)
-project (cobra_publisher)
-
-# There's -Weverything too for clang
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
-
-set (OPENSSL_PREFIX /usr/local/opt/openssl) # Homebrew openssl
-
-set (CMAKE_CXX_STANDARD 14)
-
-option(USE_TLS "Add TLS support" ON)
-
-include_directories(cobra_publisher ${OPENSSL_PREFIX}/include)
-include_directories(cobra_publisher .)
-
-add_executable(cobra_publisher 
-  jsoncpp/jsoncpp.cpp
-  ixcrypto/IXHMac.cpp
-  ixcrypto/IXBase64.cpp
-  IXCobraConnection.cpp
-  cobra_publisher.cpp)
-
-if (APPLE AND USE_TLS)
-    target_link_libraries(cobra_publisher "-framework foundation" "-framework security")
-endif()
-
-get_filename_component(crypto_lib_path ${OPENSSL_PREFIX}/lib/libcrypto.a ABSOLUTE)
-add_library(lib_crypto STATIC IMPORTED)
-set_target_properties(lib_crypto PROPERTIES IMPORTED_LOCATION ${crypto_lib_path})
-
-link_directories(/usr/local/opt/openssl/lib)
-target_link_libraries(cobra_publisher ixwebsocket lib_crypto)
-install(TARGETS cobra_publisher DESTINATION bin)

ws/cobra_publisher/README.md

@@ -1,6 +0,0 @@
-```
-mkdir build
-cd build
-cmake ..
-make && (cd .. ; sh cobra_publisher.sh)
-```

ws/cobra_publisher/cobra_publisher.cpp

@@ -1,123 +0,0 @@
-/*
- *  cobra_publisher.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include <fstream>
-#include <atomic>
-#include <ixwebsocket/IXWebSocket.h>
-#include "IXCobraConnection.h"
-#include "jsoncpp/json/json.h"
-
-void msleep(int ms)
-{
-    std::chrono::duration<double, std::milli> duration(ms);
-    std::this_thread::sleep_for(duration);
-}
-
-int main(int argc, char* argv[])
-{
-    if (argc != 7)
-    {
-        std::cerr << "Usage error: need 6 arguments." << std::endl;
-    }
-
-    std::string endpoint = argv[1];
-    std::string appkey = argv[2];
-    std::string channel = argv[3];
-    std::string rolename = argv[4];
-    std::string rolesecret = argv[5];
-    std::string path = argv[6];
-
-    std::atomic<size_t> incomingBytes(0);
-    std::atomic<size_t> outgoingBytes(0);
-    ix::CobraConnection::setTrafficTrackerCallback(
-        [&incomingBytes, &outgoingBytes](size_t size, bool incoming)
-        {
-            if (incoming)
-            {
-                incomingBytes += size;
-            }
-            else
-            {
-                outgoingBytes += size;
-            }
-        }
-    );
-
-    bool done = false;
-    ix::CobraConnection cobraConnection;
-    ix::WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions(
-        true, false, false, 15, 15);
-    cobraConnection.configure(appkey, endpoint, rolename, rolesecret,
-                               webSocketPerMessageDeflateOptions);
-    cobraConnection.connect();
-    cobraConnection.setEventCallback(
-        [&cobraConnection, channel, path, &done]
-        (ix::CobraConnectionEventType eventType,
-         const std::string& errMsg,
-         const ix::WebSocketHttpHeaders& headers)
-        {
-            if (eventType == ix::CobraConnection_EventType_Open)
-            {
-                std::cout << "Handshake Headers:" << std::endl;
-                for (auto it : headers)
-                {
-                    std::cout << it.first << ": " << it.second << std::endl;
-                }
-            }
-            else if (eventType == ix::CobraConnection_EventType_Authenticated)
-            {
-                std::cout << "Authenticated" << std::endl;
-
-                std::string line;
-                std::ifstream f(path);
-                if (!f.is_open())
-                {
-                    std::cerr << "Error while opening file: " << path << std::endl;
-                }
-
-                int n = 0;
-                while (getline(f, line))
-                {
-                    Json::Value value;
-                    Json::Reader reader;
-                    reader.parse(line, value);
-
-                    cobraConnection.publish(channel, value);
-                    n++;
-                }
-                std::cerr << "#published messages: " << n << std::endl;
-
-                if (f.bad())
-                {
-                    std::cerr << "Error while opening file: " << path << std::endl;
-                }
-
-                done = true;
-            }
-            else if (eventType == ix::CobraConnection_EventType_Error)
-            {
-                std::cerr << "Cobra Error received: " << errMsg << std::endl;
-                done = true;
-            }
-            else if (eventType == ix::CobraConnection_EventType_Closed)
-            {
-                std::cerr << "Cobra connection closed" << std::endl;
-            }
-        }
-    );
-
-    while (!done)
-    {
-        msleep(1);
-    }
-
-    std::cout << "Incoming bytes: " << incomingBytes << std::endl;
-    std::cout << "Outgoing bytes: " << outgoingBytes << std::endl;
-
-    return 0;
-}

ws/cobra_publisher/cobra_publisher.sh

@@ -1,11 +0,0 @@
-#!/bin/sh
-
-endpoint="ws://127.0.0.1:8765"
-endpoint="ws://127.0.0.1:5678"
-appkey="appkey"
-channel="foo"
-rolename="a_role"
-rolesecret="a_secret"
-filename=${FILENAME:=events.jsonl}
-
-build/cobra_publisher $endpoint $appkey $channel $rolename $rolesecret $filename

ws/cobra_publisher/devnull_server.js

@@ -1,45 +0,0 @@
-/*
- *  devnull_server.js
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
- */
-const WebSocket = require('ws');
-
-let wss = new WebSocket.Server({ port: 5678, perMessageDeflate: true })
-
-wss.on('connection', (ws) => {
-
-  let handshake = false
-  let authenticated = false
-
-  ws.on('message', (data) => {
-
-    console.log(data.toString('utf-8'))
-
-    if (!handshake) {
-      let response = {
-          "action": "auth/handshake/ok",
-          "body": {
-              "data": {
-                  "nonce": "MTI0Njg4NTAyMjYxMzgxMzgzMg==",
-                  "version": "0.0.24"
-              }
-          },
-          "id": 1
-      }
-      ws.send(JSON.stringify(response))
-      handshake = true
-    } else if (!authenticated) {
-      let response = {
-        "action": "auth/authenticate/ok",
-        "body": {},
-        "id": 2
-      }
-      
-      ws.send(JSON.stringify(response))
-      authenticated = true
-    } else {
-      console.log(data)
-    }
-  });
-})

ws/cobra_publisher/devnull_server.py

@@ -1,43 +0,0 @@
-#!/usr/bin/env python
-
-import os
-import json
-import asyncio
-import websockets
-
-
-async def echo(websocket, path):
-    handshake = False
-    authenticated = False
-
-    async for message in websocket:
-        print(message)
-
-        if not handshake:
-            response = {
-                "action": "auth/handshake/ok",
-                "body": {
-                    "data": {
-                        "nonce": "MTI0Njg4NTAyMjYxMzgxMzgzMg==",
-                        "version": "0.0.24"
-                    }
-                },
-                "id": 1
-            }
-            await websocket.send(json.dumps(response))
-            handshake = True
-
-        elif not authenticated:
-            response = {
-                "action": "auth/authenticate/ok",
-                "body": {},
-                "id": 2
-            }
-          
-            await websocket.send(json.dumps(response))
-            authenticated = True
-
-
-asyncio.get_event_loop().run_until_complete(
-    websockets.serve(echo, 'localhost', 5678))
-asyncio.get_event_loop().run_forever()

ws/cobra_publisher/events.jsonl

@@ -1,3 +0,0 @@
-{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Foo"}
-{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Bar"}
-{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Baz"}

ws/cobra_publisher/jsoncpp/json/json-forwards.h

@@ -1,333 +0,0 @@
-/// Json-cpp amalgated forward header (http://jsoncpp.sourceforge.net/).
-/// It is intended to be used with #include "json/json-forwards.h"
-/// This header provides forward declaration for all JsonCpp types.
-
-// //////////////////////////////////////////////////////////////////////
-// Beginning of content of file: LICENSE
-// //////////////////////////////////////////////////////////////////////
-
-/*
-The JsonCpp library's source code, including accompanying documentation,
-tests and demonstration applications, are licensed under the following
-conditions...
-
-Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
-jurisdictions which recognize such a disclaimer. In such jurisdictions,
-this software is released into the Public Domain.
-
-In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
-2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
-The JsonCpp Authors, and is released under the terms of the MIT License (see below).
-
-In jurisdictions which recognize Public Domain property, the user of this
-software may choose to accept it either as 1) Public Domain, 2) under the
-conditions of the MIT License (see below), or 3) under the terms of dual
-Public Domain/MIT License conditions described here, as they choose.
-
-The MIT License is about as close to Public Domain as a license can get, and is
-described in clear, concise terms at:
-
-   http://en.wikipedia.org/wiki/MIT_License
-
-The full text of the MIT License follows:
-
-========================================================================
-Copyright (c) 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
-
-Permission is hereby granted, free of charge, to any person
-obtaining a copy of this software and associated documentation
-files (the "Software"), to deal in the Software without
-restriction, including without limitation the rights to use, copy,
-modify, merge, publish, distribute, sublicense, and/or sell copies
-of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be
-included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
-NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
-BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
-ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
-CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
-========================================================================
-(END LICENSE TEXT)
-
-The MIT license is compatible with both the GPL and commercial
-software, affording one all of the rights of Public Domain with the
-minor nuisance of being required to keep the above copyright notice
-and license text in the source code. Note also that by accepting the
-Public Domain "license" you can re-license your copy using whatever
-license you like.
-
-*/
-
-// //////////////////////////////////////////////////////////////////////
-// End of content of file: LICENSE
-// //////////////////////////////////////////////////////////////////////
-
-
-
-
-
-#ifndef JSON_FORWARD_AMALGATED_H_INCLUDED
-# define JSON_FORWARD_AMALGATED_H_INCLUDED
-/// If defined, indicates that the source file is amalgated
-/// to prevent private header inclusion.
-#define JSON_IS_AMALGAMATION
-
-// //////////////////////////////////////////////////////////////////////
-// Beginning of content of file: include/json/config.h
-// //////////////////////////////////////////////////////////////////////
-
-// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
-// Distributed under MIT license, or public domain if desired and
-// recognized in your jurisdiction.
-// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
-
-#ifndef JSON_CONFIG_H_INCLUDED
-#define JSON_CONFIG_H_INCLUDED
-#include <stddef.h>
-#include <string> //typedef String
-#include <stdint.h> //typedef int64_t, uint64_t
-
-/// If defined, indicates that json library is embedded in CppTL library.
-//# define JSON_IN_CPPTL 1
-
-/// If defined, indicates that json may leverage CppTL library
-//#  define JSON_USE_CPPTL 1
-/// If defined, indicates that cpptl vector based map should be used instead of
-/// std::map
-/// as Value container.
-//#  define JSON_USE_CPPTL_SMALLMAP 1
-
-// If non-zero, the library uses exceptions to report bad input instead of C
-// assertion macros. The default is to use exceptions.
-#ifndef JSON_USE_EXCEPTION
-#define JSON_USE_EXCEPTION 1
-#endif
-
-/// If defined, indicates that the source file is amalgated
-/// to prevent private header inclusion.
-/// Remarks: it is automatically defined in the generated amalgated header.
-// #define JSON_IS_AMALGAMATION
-
-#ifdef JSON_IN_CPPTL
-#include <cpptl/config.h>
-#ifndef JSON_USE_CPPTL
-#define JSON_USE_CPPTL 1
-#endif
-#endif
-
-#ifdef JSON_IN_CPPTL
-#define JSON_API CPPTL_API
-#elif defined(JSON_DLL_BUILD)
-#if defined(_MSC_VER) || defined(__MINGW32__)
-#define JSON_API __declspec(dllexport)
-#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
-#endif // if defined(_MSC_VER)
-#elif defined(JSON_DLL)
-#if defined(_MSC_VER) || defined(__MINGW32__)
-#define JSON_API __declspec(dllimport)
-#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
-#endif // if defined(_MSC_VER)
-#endif // ifdef JSON_IN_CPPTL
-#if !defined(JSON_API)
-#define JSON_API
-#endif
-
-// If JSON_NO_INT64 is defined, then Json only support C++ "int" type for
-// integer
-// Storages, and 64 bits integer support is disabled.
-// #define JSON_NO_INT64 1
-
-#if defined(_MSC_VER) // MSVC
-#  if _MSC_VER <= 1200 // MSVC 6
-    // Microsoft Visual Studio 6 only support conversion from __int64 to double
-    // (no conversion from unsigned __int64).
-#    define JSON_USE_INT64_DOUBLE_CONVERSION 1
-    // Disable warning 4786 for VS6 caused by STL (identifier was truncated to '255'
-    // characters in the debug information)
-    // All projects I've ever seen with VS6 were using this globally (not bothering
-    // with pragma push/pop).
-#    pragma warning(disable : 4786)
-#  endif // MSVC 6
-
-#  if _MSC_VER >= 1500 // MSVC 2008
-    /// Indicates that the following function is deprecated.
-#    define JSONCPP_DEPRECATED(message) __declspec(deprecated(message))
-#  endif
-
-#endif // defined(_MSC_VER)
-
-// In c++11 the override keyword allows you to explicity define that a function
-// is intended to override the base-class version.  This makes the code more
-// managable and fixes a set of common hard-to-find bugs.
-#if __cplusplus >= 201103L
-# define JSONCPP_OVERRIDE override
-# define JSONCPP_NOEXCEPT noexcept
-#elif defined(_MSC_VER) && _MSC_VER > 1600 && _MSC_VER < 1900
-# define JSONCPP_OVERRIDE override
-# define JSONCPP_NOEXCEPT throw()
-#elif defined(_MSC_VER) && _MSC_VER >= 1900
-# define JSONCPP_OVERRIDE override
-# define JSONCPP_NOEXCEPT noexcept
-#else
-# define JSONCPP_OVERRIDE
-# define JSONCPP_NOEXCEPT throw()
-#endif
-
-#ifndef JSON_HAS_RVALUE_REFERENCES
-
-#if defined(_MSC_VER) && _MSC_VER >= 1600 // MSVC >= 2010
-#define JSON_HAS_RVALUE_REFERENCES 1
-#endif // MSVC >= 2010
-
-#ifdef __clang__
-#if __has_feature(cxx_rvalue_references)
-#define JSON_HAS_RVALUE_REFERENCES 1
-#endif  // has_feature
-
-#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
-#if defined(__GXX_EXPERIMENTAL_CXX0X__) || (__cplusplus >= 201103L)
-#define JSON_HAS_RVALUE_REFERENCES 1
-#endif  // GXX_EXPERIMENTAL
-
-#endif // __clang__ || __GNUC__
-
-#endif // not defined JSON_HAS_RVALUE_REFERENCES
-
-#ifndef JSON_HAS_RVALUE_REFERENCES
-#define JSON_HAS_RVALUE_REFERENCES 0
-#endif
-
-#ifdef __clang__
-#  if __has_extension(attribute_deprecated_with_message)
-#    define JSONCPP_DEPRECATED(message)  __attribute__ ((deprecated(message)))
-#  endif
-#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
-#  if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 5))
-#    define JSONCPP_DEPRECATED(message)  __attribute__ ((deprecated(message)))
-#  elif (__GNUC__ > 3 || (__GNUC__ == 3 && __GNUC_MINOR__ >= 1))
-#    define JSONCPP_DEPRECATED(message)  __attribute__((__deprecated__))
-#  endif  // GNUC version
-#endif // __clang__ || __GNUC__
-
-#if !defined(JSONCPP_DEPRECATED)
-#define JSONCPP_DEPRECATED(message)
-#endif // if !defined(JSONCPP_DEPRECATED)
-
-#if __GNUC__ >= 6
-#  define JSON_USE_INT64_DOUBLE_CONVERSION 1
-#endif
-
-#if !defined(JSON_IS_AMALGAMATION)
-
-# include "version.h"
-
-# if JSONCPP_USING_SECURE_MEMORY
-#  include "allocator.h" //typedef Allocator
-# endif
-
-#endif // if !defined(JSON_IS_AMALGAMATION)
-
-namespace Json {
-typedef int Int;
-typedef unsigned int UInt;
-#if defined(JSON_NO_INT64)
-typedef int LargestInt;
-typedef unsigned int LargestUInt;
-#undef JSON_HAS_INT64
-#else                 // if defined(JSON_NO_INT64)
-// For Microsoft Visual use specific types as long long is not supported
-#if defined(_MSC_VER) // Microsoft Visual Studio
-typedef __int64 Int64;
-typedef unsigned __int64 UInt64;
-#else                 // if defined(_MSC_VER) // Other platforms, use long long
-typedef int64_t Int64;
-typedef uint64_t UInt64;
-#endif // if defined(_MSC_VER)
-typedef Int64 LargestInt;
-typedef UInt64 LargestUInt;
-#define JSON_HAS_INT64
-#endif // if defined(JSON_NO_INT64)
-#if JSONCPP_USING_SECURE_MEMORY
-#define JSONCPP_STRING        std::basic_string<char, std::char_traits<char>, Json::SecureAllocator<char> >
-#define JSONCPP_OSTRINGSTREAM std::basic_ostringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
-#define JSONCPP_OSTREAM       std::basic_ostream<char, std::char_traits<char>>
-#define JSONCPP_ISTRINGSTREAM std::basic_istringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
-#define JSONCPP_ISTREAM       std::istream
-#else
-#define JSONCPP_STRING        std::string
-#define JSONCPP_OSTRINGSTREAM std::ostringstream
-#define JSONCPP_OSTREAM       std::ostream
-#define JSONCPP_ISTRINGSTREAM std::istringstream
-#define JSONCPP_ISTREAM       std::istream
-#endif // if JSONCPP_USING_SECURE_MEMORY
-} // end namespace Json
-
-#endif // JSON_CONFIG_H_INCLUDED
-
-// //////////////////////////////////////////////////////////////////////
-// End of content of file: include/json/config.h
-// //////////////////////////////////////////////////////////////////////
-
-
-
-
-
-
-// //////////////////////////////////////////////////////////////////////
-// Beginning of content of file: include/json/forwards.h
-// //////////////////////////////////////////////////////////////////////
-
-// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
-// Distributed under MIT license, or public domain if desired and
-// recognized in your jurisdiction.
-// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
-
-#ifndef JSON_FORWARDS_H_INCLUDED
-#define JSON_FORWARDS_H_INCLUDED
-
-#if !defined(JSON_IS_AMALGAMATION)
-#include "config.h"
-#endif // if !defined(JSON_IS_AMALGAMATION)
-
-namespace Json {
-
-// writer.h
-class FastWriter;
-class StyledWriter;
-
-// reader.h
-class Reader;
-
-// features.h
-class Features;
-
-// value.h
-typedef unsigned int ArrayIndex;
-class StaticString;
-class Path;
-class PathArgument;
-class Value;
-class ValueIteratorBase;
-class ValueIterator;
-class ValueConstIterator;
-
-} // namespace Json
-
-#endif // JSON_FORWARDS_H_INCLUDED
-
-// //////////////////////////////////////////////////////////////////////
-// End of content of file: include/json/forwards.h
-// //////////////////////////////////////////////////////////////////////
-
-
-
-
-
-#endif //ifndef JSON_FORWARD_AMALGATED_H_INCLUDED

ws/docker_build.sh

@@ -13,6 +13,7 @@ g++ --std=c++14 \
     ../ixwebsocket/IXSocket.cpp \
     ../ixwebsocket/IXSocketServer.cpp \
     ../ixwebsocket/IXSocketConnect.cpp \
+    ../ixwebsocket/IXSocketFactory.cpp \
     ../ixwebsocket/IXDNSLookup.cpp \
     ../ixwebsocket/IXCancellationRequest.cpp \
     ../ixwebsocket/IXWebSocket.cpp \
@@ -22,12 +23,16 @@ g++ --std=c++14 \
     ../ixwebsocket/IXWebSocketPerMessageDeflate.cpp \
     ../ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp \
     ../ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp \
+    ../ixwebsocket/IXWebSocketHttpHeaders.cpp \
+    ../ixwebsocket/IXHttpClient.cpp \
+    ../ixwebsocket/IXUrlParser.cpp \
     ../ixwebsocket/IXSocketOpenSSL.cpp \
     ../ixwebsocket/linux/IXSetThreadName_linux.cpp \
-    ../third_party/jsoncpp/jsoncpp.cpp \
+    ../third_party/msgpack11/msgpack11.cpp \
     ixcrypto/IXBase64.cpp \
     ixcrypto/IXHash.cpp \
     ixcrypto/IXUuid.cpp \
+    ws_http_client.cpp \
     ws_ping_pong.cpp \
     ws_broadcast_server.cpp \
     ws_echo_server.cpp \

ws/ixcrypto/IXHMac.cpp

@@ -6,7 +6,11 @@
 #include "IXHMac.h"
 #include "IXBase64.h"
 
+#ifdef __APPLE__
+# include <CommonCrypto/CommonHMAC.h>
+#else
 # include <openssl/hmac.h>
+#endif
 
 namespace ix
 {
@@ -15,10 +19,17 @@ namespace ix
         constexpr size_t hashSize = 16;
         unsigned char hash[hashSize];
 
+#ifdef __APPLE__
+        CCHmac(kCCHmacAlgMD5,
+               key.c_str(), key.size(),
+               data.c_str(), data.size(),
+               &hash);
+#else
         HMAC(EVP_md5(),
              key.c_str(), (int) key.size(),
              (unsigned char *) data.c_str(), (int) data.size(),
              (unsigned char *) hash, nullptr);
+#endif
 
         std::string hashString(reinterpret_cast<char*>(hash), hashSize);
 

ws/ixcrypto/IXHash.h

@@ -6,6 +6,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <vector>
 
 namespace ix

ws/package-lock.json

@@ -8,9 +8,9 @@
       "integrity": "sha512-jp/uFnooOiO+L211eZOoSyzpOITMXx1rBITauYykG3BRYPu8h0UcxsPNB04RR5vo4Tyz3+ay17tR6JVf9qzYWg=="
     },
     "ws": {
-      "version": "6.1.0",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-6.1.0.tgz",
-      "integrity": "sha512-H3dGVdGvW2H8bnYpIDc3u3LH8Wue3Qh+Zto6aXXFzvESkTVT6rAfKR6tR/+coaUvxs8yHtmNV0uioBF62ZGSTg==",
+      "version": "6.2.0",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-6.2.0.tgz",
+      "integrity": "sha512-deZYUNlt2O4buFCa3t5bKLf8A7FPP/TVjwOeVNpw818Ma5nk4MLXls2eoEGS39o8119QIYxTrTDoPQ5B/gTD6w==",
       "requires": {
         "async-limiter": "1.0.0"
       }

ws/test_ws.sh

@@ -0,0 +1,64 @@
+#!/bin/sh
+
+# Handle Ctrl-C by killing all sub-processing AND exiting
+trap cleanup INT
+
+function cleanup {
+    kill `cat /tmp/ws_test/pidfile.transfer`
+    kill `cat /tmp/ws_test/pidfile.receive`
+    kill `cat /tmp/ws_test/pidfile.send`
+    exit 1
+}
+
+rm -rf /tmp/ws_test
+mkdir -p /tmp/ws_test
+
+# Start a transport server
+cd /tmp/ws_test
+ws transfer --port 8090 --pidfile /tmp/ws_test/pidfile.transfer &
+
+# Wait until the transfer server is up 
+while true
+do
+    nc -zv 127.0.0.1 8090 && {
+        echo "Transfer server up and running"
+        break
+    }
+    echo "sleep ... wait for transfer server"
+    sleep 0.1
+done
+
+# Start a receiver
+mkdir -p /tmp/ws_test/receive
+cd /tmp/ws_test/receive
+ws receive --delay 10 ws://127.0.0.1:8090 --pidfile /tmp/ws_test/pidfile.receive &
+
+mkdir /tmp/ws_test/send
+cd /tmp/ws_test/send
+dd if=/dev/urandom of=20M_file count=20000 bs=1024
+
+# Start the sender job
+ws send --pidfile /tmp/ws_test/pidfile.send ws://127.0.0.1:8090 20M_file
+
+# Wait until the file has been written to disk
+while true
+do
+    if test -f /tmp/ws_test/receive/20M_file ; then
+        echo "Received file does exists, exiting loop"
+        break
+    fi
+    echo "sleep ... wait for output file"
+    sleep 0.1
+done
+
+cksum /tmp/ws_test/send/20M_file
+cksum /tmp/ws_test/receive/20M_file
+
+# Give some time to ws receive to terminate
+sleep 2
+
+# Cleanup
+kill `cat /tmp/ws_test/pidfile.transfer`
+kill `cat /tmp/ws_test/pidfile.receive`
+kill `cat /tmp/ws_test/pidfile.send`
+

ws/test_ws_redis.sh

@@ -0,0 +1,25 @@
+#!/bin/sh
+
+# Handle Ctrl-C by killing all sub-processing AND exiting
+trap cleanup INT
+
+function cleanup {
+    kill `cat /tmp/pidfile.subscribe`
+    exit 1
+}
+
+REDIS_HOST=${REDIS_HOST:=localhost}
+
+ws redis_subscribe --pidfile /tmp/pidfile.subscribe --host $REDIS_HOST foo &
+
+# Wait for the subscriber to be ready
+sleep 0.5
+
+# Now publish messages
+ws redis_publish -c 100000 --host ${REDIS_HOST} foo bar
+
+# Wait a little for all messages to be received
+sleep 1.5
+
+# Cleanup
+cleanup

ws/ws.cpp

@@ -1,9 +1,14 @@
 /*
  *  ws.cpp
  *  Author: Benjamin Sergeant
- *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
  */
 
+//
+// Main driver for websocket utilities
+//
+#include "ws.h"
+
 //
 // Main drive for websocket utilities
 //
@@ -11,32 +16,12 @@
 #include <string>
 #include <sstream>
 #include <iostream>
+#include <fstream>
+#include <unistd.h>
 
 #include <cli11/CLI11.hpp>
 #include <ixwebsocket/IXSocket.h>
 
-namespace ix
-{
-    int ws_ping_pong_main(const std::string& url);
-
-    int ws_echo_server_main(int port);
-
-    int ws_broadcast_server_main(int port);
-
-    int ws_chat_main(const std::string& url,
-                     const std::string& user);
-
-    int ws_connect_main(const std::string& url);
-
-    int ws_receive_main(const std::string& url,
-                        bool enablePerMessageDeflate);
-
-    int ws_transfer_main(int port);
-
-    int ws_send_main(const std::string& url,
-                     const std::string& path);
-}
-
 int main(int argc, char** argv)
 {
     CLI::App app{"ws is a websocket tool"};
@@ -45,17 +30,53 @@ int main(int argc, char** argv)
     std::string url("ws://127.0.0.1:8080");
     std::string path;
     std::string user;
+    std::string data;
+    std::string headers;
+    std::string output;
+    std::string hostname("127.0.0.1");
+    std::string pidfile;
+    std::string channel;
+    std::string message;
+    std::string password;
+    std::string appkey;
+    std::string endpoint;
+    std::string rolename;
+    std::string rolesecret;
+    std::string prefix("ws.test.v0");
+    std::string fields;
+    std::string dsn;
+    bool headersOnly = false;
+    bool followRedirects = false;
+    bool verbose = false;
+    bool save = false;
+    bool compress = false;
+    bool strict = false;
     int port = 8080;
+    int redisPort = 6379;
+    int statsdPort = 8125;
+    int connectTimeOut = 60;
+    int transferTimeout = 1800;
+    int maxRedirects = 5;
+    int delayMs = -1;
+    int count = 1;
+    int jobs = 4;
 
     CLI::App* sendApp = app.add_subcommand("send", "Send a file");
     sendApp->add_option("url", url, "Connection url")->required();
-    sendApp->add_option("path", path, "Path to the file to send")->required();
+    sendApp->add_option("path", path, "Path to the file to send")
+        ->required()->check(CLI::ExistingPath);
+    sendApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* receiveApp = app.add_subcommand("receive", "Receive a file");
     receiveApp->add_option("url", url, "Connection url")->required();
+    receiveApp->add_option("--delay", delayMs, "Delay (ms) to wait after receiving a fragment"
+                                               " to artificially slow down the receiver");
+    receiveApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
     transferApp->add_option("--port", port, "Connection url");
+    transferApp->add_option("--host", hostname, "Hostname");
+    transferApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
     connectApp->add_option("url", url, "Connection url")->required();
@@ -65,21 +86,97 @@ int main(int argc, char** argv)
     chatApp->add_option("user", user, "User name")->required();
 
     CLI::App* echoServerApp = app.add_subcommand("echo_server", "Echo server");
-    echoServerApp->add_option("--port", port, "Connection url");
+    echoServerApp->add_option("--port", port, "Port");
+    echoServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
-    broadcastServerApp->add_option("--port", port, "Connection url");
+    broadcastServerApp->add_option("--port", port, "Port");
+    broadcastServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* pingPongApp = app.add_subcommand("ping", "Ping pong");
     pingPongApp->add_option("url", url, "Connection url")->required();
 
+    CLI::App* httpClientApp = app.add_subcommand("curl", "HTTP Client");
+    httpClientApp->add_option("url", url, "Connection url")->required();
+    httpClientApp->add_option("-d", data, "Form data")->join();
+    httpClientApp->add_option("-F", data, "Form data")->join();
+    httpClientApp->add_option("-H", headers, "Header")->join();
+    httpClientApp->add_option("--output", output, "Output file");
+    httpClientApp->add_flag("-I", headersOnly, "Send a HEAD request");
+    httpClientApp->add_flag("-L", followRedirects, "Follow redirects");
+    httpClientApp->add_option("--max-redirects", maxRedirects, "Max Redirects");
+    httpClientApp->add_flag("-v", verbose, "Verbose");
+    httpClientApp->add_flag("-O", save, "Save output to disk");
+    httpClientApp->add_flag("--compress", compress, "Enable gzip compression");
+    httpClientApp->add_option("--connect-timeout", connectTimeOut, "Connection timeout");
+    httpClientApp->add_option("--transfer-timeout", transferTimeout, "Transfer timeout");
+
+    CLI::App* redisPublishApp = app.add_subcommand("redis_publish", "Redis publisher");
+    redisPublishApp->add_option("--port", redisPort, "Port");
+    redisPublishApp->add_option("--host", hostname, "Hostname");
+    redisPublishApp->add_option("--password", password, "Password");
+    redisPublishApp->add_option("channel", channel, "Channel")->required();
+    redisPublishApp->add_option("message", message, "Message")->required();
+    redisPublishApp->add_option("-c", count, "Count");
+
+    CLI::App* redisSubscribeApp = app.add_subcommand("redis_subscribe", "Redis subscriber");
+    redisSubscribeApp->add_option("--port", redisPort, "Port");
+    redisSubscribeApp->add_option("--host", hostname, "Hostname");
+    redisSubscribeApp->add_option("--password", password, "Password");
+    redisSubscribeApp->add_option("channel", channel, "Channel")->required();
+    redisSubscribeApp->add_flag("-v", verbose, "Verbose");
+    redisSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
+
+    CLI::App* cobraSubscribeApp = app.add_subcommand("cobra_subscribe", "Cobra subscriber");
+    cobraSubscribeApp->add_option("--appkey", appkey, "Appkey");
+    cobraSubscribeApp->add_option("--endpoint", endpoint, "Endpoint");
+    cobraSubscribeApp->add_option("--rolename", rolename, "Role name");
+    cobraSubscribeApp->add_option("--rolesecret", rolesecret, "Role secret");
+    cobraSubscribeApp->add_option("channel", channel, "Channel")->required();
+    cobraSubscribeApp->add_flag("-v", verbose, "Verbose");
+    cobraSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
+
+    CLI::App* cobra2statsd = app.add_subcommand("cobra_to_statsd", "Cobra to statsd");
+    cobra2statsd->add_option("--appkey", appkey, "Appkey");
+    cobra2statsd->add_option("--endpoint", endpoint, "Endpoint");
+    cobra2statsd->add_option("--rolename", rolename, "Role name");
+    cobra2statsd->add_option("--rolesecret", rolesecret, "Role secret");
+    cobra2statsd->add_option("--host", hostname, "Statsd host");
+    cobra2statsd->add_option("--port", statsdPort, "Statsd port");
+    cobra2statsd->add_option("--prefix", prefix, "Statsd prefix");
+    cobra2statsd->add_option("--fields", fields, "Extract fields for naming the event")->join();
+    cobra2statsd->add_option("channel", channel, "Channel")->required();
+    cobra2statsd->add_flag("-v", verbose, "Verbose");
+    cobra2statsd->add_option("--pidfile", pidfile, "Pid file");
+
+    CLI::App* cobra2sentry = app.add_subcommand("cobra_to_sentry", "Cobra to sentry");
+    cobra2sentry->add_option("--appkey", appkey, "Appkey");
+    cobra2sentry->add_option("--endpoint", endpoint, "Endpoint");
+    cobra2sentry->add_option("--rolename", rolename, "Role name");
+    cobra2sentry->add_option("--rolesecret", rolesecret, "Role secret");
+    cobra2sentry->add_option("--dsn", dsn, "Sentry DSN");
+    cobra2sentry->add_option("--jobs", jobs, "Number of thread sending events to Sentry");
+    cobra2sentry->add_option("channel", channel, "Channel")->required();
+    cobra2sentry->add_flag("-v", verbose, "Verbose");
+    cobra2sentry->add_flag("-s", strict, "Strict mode. Error out when sending to sentry fails");
+    cobra2sentry->add_option("--pidfile", pidfile, "Pid file");
+
     CLI11_PARSE(app, argc, argv);
 
-    ix::Socket::init();
+    // pid file handling
+    if (!pidfile.empty())
+    {
+        unlink(pidfile.c_str());
+
+        std::ofstream f;
+        f.open(pidfile);
+        f << getpid();
+        f.close();
+    }
 
     if (app.got_subcommand("transfer"))
     {
-        return ix::ws_transfer_main(port);
+        return ix::ws_transfer_main(port, hostname);
     }
     else if (app.got_subcommand("send"))
     {
@@ -88,7 +185,7 @@ int main(int argc, char** argv)
     else if (app.got_subcommand("receive"))
     {
         bool enablePerMessageDeflate = false;
-        return ix::ws_receive_main(url, enablePerMessageDeflate);
+        return ix::ws_receive_main(url, enablePerMessageDeflate, delayMs);
     }
     else if (app.got_subcommand("connect"))
     {
@@ -100,19 +197,51 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("echo_server"))
     {
-        return ix::ws_echo_server_main(port);
+        return ix::ws_echo_server_main(port, hostname);
     }
     else if (app.got_subcommand("broadcast_server"))
     {
-        return ix::ws_broadcast_server_main(port);
+        return ix::ws_broadcast_server_main(port, hostname);
     }
     else if (app.got_subcommand("ping"))
     {
         return ix::ws_ping_pong_main(url);
     }
-    else
+    else if (app.got_subcommand("curl"))
     {
-        assert(false);
+        return ix::ws_http_client_main(url, headers, data, headersOnly,
+                                       connectTimeOut, transferTimeout,
+                                       followRedirects, maxRedirects, verbose,
+                                       save, output, compress);
+    }
+    else if (app.got_subcommand("redis_publish"))
+    {
+        return ix::ws_redis_publish_main(hostname, redisPort, password,
+                                         channel, message, count);
+    }
+    else if (app.got_subcommand("redis_subscribe"))
+    {
+        return ix::ws_redis_subscribe_main(hostname, redisPort, password, channel, verbose);
+    }
+    else if (app.got_subcommand("cobra_subscribe"))
+    {
+        return ix::ws_cobra_subscribe_main(appkey, endpoint,
+                                           rolename, rolesecret,
+                                           channel, verbose);
+    }
+    else if (app.got_subcommand("cobra_to_statsd"))
+    {
+        return ix::ws_cobra_to_statsd_main(appkey, endpoint,
+                                           rolename, rolesecret,
+                                           channel, hostname, statsdPort,
+                                           prefix, fields, verbose);
+    }
+    else if (app.got_subcommand("cobra_to_sentry"))
+    {
+        return ix::ws_cobra_to_sentry_main(appkey, endpoint,
+                                           rolename, rolesecret,
+                                           channel, dsn,
+                                           verbose, strict, jobs);
     }
 
     return 1;

ws/ws.h

@@ -0,0 +1,83 @@
+/*
+ *  ws.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+#pragma once
+
+#include <string>
+
+namespace ix
+{
+    int ws_http_client_main(const std::string& url,
+                            const std::string& headers,
+                            const std::string& data,
+                            bool headersOnly,
+                            int connectTimeout,
+                            int transferTimeout,
+                            bool followRedirects,
+                            int maxRedirects,
+                            bool verbose,
+                            bool save,
+                            const std::string& output,
+                            bool compress);
+
+    int ws_ping_pong_main(const std::string& url);
+
+    int ws_echo_server_main(int port, const std::string& hostname);
+    int ws_broadcast_server_main(int port, const std::string& hostname);
+    int ws_transfer_main(int port, const std::string& hostname);
+
+    int ws_chat_main(const std::string& url,
+                     const std::string& user);
+
+    int ws_connect_main(const std::string& url);
+
+    int ws_receive_main(const std::string& url,
+                        bool enablePerMessageDeflate,
+                        int delayMs);
+
+    int ws_send_main(const std::string& url,
+                     const std::string& path);
+
+    int ws_redis_publish_main(const std::string& hostname,
+                              int port,
+                              const std::string& password,
+                              const std::string& channel,
+                              const std::string& message,
+                              int count);
+
+    int ws_redis_subscribe_main(const std::string& hostname,
+                                int port,
+                                const std::string& password,
+                                const std::string& channel,
+                                bool verbose);
+
+    int ws_cobra_subscribe_main(const std::string& appkey,
+                                const std::string& endpoint,
+                                const std::string& rolename,
+                                const std::string& rolesecret,
+                                const std::string& channel,
+                                bool verbose);
+
+    int ws_cobra_to_statsd_main(const std::string& appkey,
+                                const std::string& endpoint,
+                                const std::string& rolename,
+                                const std::string& rolesecret,
+                                const std::string& channel,
+                                const std::string& host,
+                                int port,
+                                const std::string& prefix,
+                                const std::string& fields,
+                                bool verbose);
+    
+    int ws_cobra_to_sentry_main(const std::string& appkey,
+                                const std::string& endpoint,
+                                const std::string& rolename,
+                                const std::string& rolesecret,
+                                const std::string& channel,
+                                const std::string& dsn,
+                                bool verbose,
+                                bool strict,
+                                int jobs);
+}

ws/ws_broadcast_server.cpp

@@ -10,17 +10,18 @@
 
 namespace ix
 {
-    int ws_broadcast_server_main(int port)
+    int ws_broadcast_server_main(int port, const std::string& hostname)
     {
-        std::cout << "Listening on port " << port << std::endl;
+        std::cout << "Listening on " << hostname << ":" << port << std::endl;
 
-        ix::WebSocketServer server(port);
+        ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket)
+            [&server](std::shared_ptr<WebSocket> webSocket,
+                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -30,6 +31,7 @@ namespace ix
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             std::cerr << "New connection" << std::endl;
+                            std::cerr << "id: " << connectionState->getId() << std::endl;
                             std::cerr << "Uri: " << openInfo.uri << std::endl;
                             std::cerr << "Headers:" << std::endl;
                             for (auto it : openInfo.headers)
@@ -39,16 +41,47 @@ namespace ix
                         }
                         else if (messageType == ix::WebSocket_MessageType_Close)
                         {
-                            std::cerr << "Closed connection" << std::endl;
+                            std::cerr << "Closed connection"
+                                      << " code " << closeInfo.code
+                                      << " reason " << closeInfo.reason << std::endl;
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Error)
+                        {
+                            std::stringstream ss;
+                            ss << "Connection error: " << error.reason      << std::endl;
+                            ss << "#retries: "         << error.retries     << std::endl;
+                            ss << "Wait time(ms): "    << error.wait_time   << std::endl;
+                            ss << "HTTP Status: "      << error.http_status << std::endl;
+                            std::cerr << ss.str();
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Fragment)
+                        {
+                            std::cerr << "Received message fragment" << std::endl;
                         }
                         else if (messageType == ix::WebSocket_MessageType_Message)
                         {
                             std::cerr << "Received " << wireSize << " bytes" << std::endl;
+
                             for (auto&& client : server.getClients())
                             {
                                 if (client != webSocket)
                                 {
-                                    client->send(str);
+                                    client->send(str,
+                                                 [](int current, int total) -> bool
+                                    {
+                                        std::cerr << "Step " << current
+                                                  << " out of " << total << std::endl;
+                                        return true;
+                                    });
+
+                                    do
+                                    {
+                                        size_t bufferedAmount = client->bufferedAmount();
+                                        std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
+
+                                        std::chrono::duration<double, std::milli> duration(10);
+                                        std::this_thread::sleep_for(duration);
+                                    } while (client->bufferedAmount() != 0);
                                 }
                             }
                         }

ws/ws_chat.cpp

@@ -5,9 +5,10 @@
  */
 
 //
-// Simple chat program that talks to the node.js server at
-// websocket_chat_server/broacast-server.js
+// Simple chat program that talks to a broadcast server
+// Broadcast server can be ran with `ws broadcast_server`
 //
+
 #include <iostream>
 #include <sstream>
 #include <queue>
@@ -93,16 +94,26 @@ namespace ix
                 std::stringstream ss;
                 if (messageType == ix::WebSocket_MessageType_Open)
                 {
-                    ss << "cmd_websocket_chat: user "
+                    log("ws chat: connected");
+                    std::cout << "Uri: " << openInfo.uri << std::endl;
+                    std::cout << "Handshake Headers:" << std::endl;
+                    for (auto it : openInfo.headers)
+                    {
+                        std::cout << it.first << ": " << it.second << std::endl;
+                    }
+
+                    ss << "ws chat: user "
                        << _user
                        << " Connected !";
                        log(ss.str());
                 }
                 else if (messageType == ix::WebSocket_MessageType_Close)
                 {
-                    ss << "cmd_websocket_chat: user "
+                    ss << "ws chat: user "
                        << _user
-                       << " disconnected !";
+                       << " disconnected !"
+                       << " code " << closeInfo.code
+                       << " reason " << closeInfo.reason;
                        log(ss.str());
                 }
                 else if (messageType == ix::WebSocket_MessageType_Message)
@@ -116,7 +127,7 @@ namespace ix
                     _receivedQueue.push(result.second);
 
                     ss << std::endl
-                       << result.first << " > " << result.second
+                       << result.first << "(" << wireSize << " bytes)" << " > " << result.second
                        << std::endl
                        << _user << " > ";
                     log(ss.str());
@@ -187,5 +198,7 @@ namespace ix
 
         std::cout << std::endl;
         webSocketChat.stop();
+
+        return 0;
     }
 }