Merge branch 'master' into feature/http

* introduce send fragment

* can pass a fin frame

* can send messages which are a perfect multiple of the chunk size

* set fin only for last fragment

* cleanup

* last fragment should be of type CONTINUATION

* Add simple send and receive programs

* speedups receiving + better way to wait for thing

* receive speedup by using linked list of chunks instead of large array

* document bug

* use chunks to receive data

* trailing spaces

CMakeLists.txt

@@ -15,11 +15,8 @@ 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
@@ -36,12 +33,10 @@ set( IXWEBSOCKET_SOURCES
     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
@@ -63,17 +58,8 @@ set( IXWEBSOCKET_HEADERS
     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)
@@ -81,11 +67,8 @@ 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)
 
@@ -96,7 +79,6 @@ 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()
@@ -107,16 +89,14 @@ add_library( ixwebsocket STATIC
     ${IXWEBSOCKET_HEADERS}
 )
 
-if (APPLE AND USE_TLS)
-  target_link_libraries(ixwebsocket "-framework foundation" "-framework security")
-endif()
+# gcc/Linux needs -pthread
+find_package(Threads)
 
-if(USE_OPEN_SSL)
+if(UNIX AND NOT APPLE)
   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)
@@ -131,21 +111,15 @@ if (WIN32)
 
   target_link_libraries(ixwebsocket libz wsock32 ws2_32)
   add_definitions(-D_CRT_SECURE_NO_WARNINGS)
-
+  
 else()
-  # gcc/Linux needs -pthread
-  find_package(Threads)
-
   target_link_libraries(ixwebsocket 
-    z ${CMAKE_THREAD_LIBS_INIT})
+    z ${OPENSSL_LIBRARIES} ${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()
+add_subdirectory(ws)

DOCKER_VERSION

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

Dockerfile

@@ -1,47 +0,0 @@
-# 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"]

Dockerfile

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

README.md

@@ -4,14 +4,14 @@
 
 ## Introduction
 
-[*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.
+[*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, 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
-
-The code was made to compile once on Windows but support is currently broken on this platform.
+* Android 
+* Windows (no TLS support yet)
 
 ## Examples
 
@@ -47,12 +47,9 @@ 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 (default to BINARY mode)
+// Send a message to the server
 webSocket.send("hello world");
 
-// The message can be sent in TEXT mode
-webSocket.sendText("hello again");
-
 // ... finally ...
 
 // Stop the connection
@@ -67,11 +64,10 @@ 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<WebSocket> webSocket,
-              std::shared_ptr<ConnectionState> connectionState)
+    [&server](std::shared_ptr<ix::WebSocket> webSocket)
     {
         webSocket->setOnMessageCallback(
-            [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+            [webSocket, &server](ix::WebSocketMessageType messageType,
                const std::string& str,
                size_t wireSize,
                const ix::WebSocketErrorInfo& error,
@@ -81,16 +77,7 @@ 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)
                     {
@@ -135,7 +122,7 @@ HttpRequestArgs args;
 // Custom headers can be set
 WebSocketHttpHeaders headers;
 headers["Foo"] = "bar";
-args.extraHeaders = headers;
+args.extraHeaders = parseHeaders(headersData);
 
 // Timeout options
 args.connectTimeout = connectTimeout;
@@ -191,13 +178,6 @@ CMakefiles for the library and the examples are available. This library has few
 
 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.
@@ -218,11 +198,11 @@ 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 1G 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 500M were sent and received succesfully.
 
 ## Limitations
 
-* No utf-8 validation is made when sending TEXT message with sendText()
+* There is no text support for sending data, only the binary protocol is supported. Sending json or text over the binary protocol works well.
 * 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 is not as scalable as strategies using epoll or kqueue.
 
@@ -234,13 +214,13 @@ Here is a simplistic diagram which explains how the code is structured in term o
 +-----------------------+ --- Public
 |                       | Start the receiving Background thread. Auto reconnection. Simple websocket Ping.
 |  IXWebSocket          | Interface used by C++ test clients. No IX dependencies.
-|                       |
+|                       | 
 +-----------------------+
 |                       |
 |  IXWebSocketServer    | Run a server and give each connections its own WebSocket object.
 |                       | Each connection is handled in a new OS thread.
 |                       |
-+-----------------------+ --- Private
++-----------------------+ --- Private 
 |                       |
 |  IXWebSocketTransport | Low level websocket code, framing, managing raw socket. Adapted from easywsclient.
 |                       |
@@ -390,7 +370,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 is no any traffic to make sure that load balancers do not kill an
+when there is not any traffic to make sure that load balancers do not kill an
 idle connection.
 
 ```

docker-compose.yml

@@ -1,21 +0,0 @@
-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

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

@@ -0,0 +1,11 @@
+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

@@ -0,0 +1,11 @@
+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

@@ -0,0 +1,22 @@
+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

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

examples/cobra_publisher/ixcrypto/IXHash.cpp

@@ -0,0 +1,22 @@
+/*
+ *  IXHash.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone. All rights reserved.
+ */
+
+#include <string>
+
+namespace ix
+{
+    uint64_t djb2Hash(const std::string& data)
+    {
+        uint64_t hashAddress = 5381;
+
+        for (auto& c : data)
+        {
+            hashAddress = ((hashAddress << 5) + hashAddress) + c;
+        }
+
+        return hashAddress;
+    }
+}

examples/cobra_publisher/ixcrypto/IXHash.h

@@ -0,0 +1,15 @@
+/*
+ *  IXHash.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone. All rights reserved.
+ */
+
+#pragma once
+
+#include <string>
+
+namespace ix
+{
+    uint64_t djb2Hash(const std::string& data);
+}
+

examples/cobra_publisher/ixcrypto/IXUuid.cpp

@@ -0,0 +1,75 @@
+/*
+ *  IXUuid.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone. All rights reserved.
+ */
+
+/**
+ * Generate a random uuid similar to the uuid python module
+ *
+ * >>> import uuid
+ * >>> uuid.uuid4().hex
+ * 'bec08155b37d4050a1f3c3fa0276bf12'
+ *
+ * Code adapted from https://github.com/r-lyeh-archived/sole
+ */
+
+#include "IXUuid.h"
+
+#include <sstream>
+#include <string>
+#include <iomanip>
+#include <random>
+
+
+namespace ix
+{
+    class Uuid
+    {
+        public:
+            Uuid();
+            std::string toString() const;
+
+        private:
+            uint64_t _ab;
+            uint64_t _cd;
+    };
+
+    Uuid::Uuid()
+    {
+        static std::random_device rd;
+        static std::uniform_int_distribution<uint64_t> dist(0, (uint64_t)(~0));
+
+        _ab = dist(rd);
+        _cd = dist(rd);
+
+        _ab = (_ab & 0xFFFFFFFFFFFF0FFFULL) | 0x0000000000004000ULL;
+        _cd = (_cd & 0x3FFFFFFFFFFFFFFFULL) | 0x8000000000000000ULL;
+    }
+
+    std::string Uuid::toString() const
+    {
+        std::stringstream ss;
+        ss << std::hex << std::nouppercase << std::setfill('0');
+
+        uint32_t a = (_ab >> 32);
+        uint32_t b = (_ab & 0xFFFFFFFF);
+        uint32_t c = (_cd >> 32);
+        uint32_t d = (_cd & 0xFFFFFFFF);
+
+        ss << std::setw(8) << (a);
+        ss << std::setw(4) << (b >> 16);
+        ss << std::setw(4) << (b & 0xFFFF);
+        ss << std::setw(4) << (c >> 16 );
+        ss << std::setw(4) << (c & 0xFFFF);
+        ss << std::setw(8) << d;
+
+        return ss.str();
+    }
+
+    std::string uuid4()
+    {
+        Uuid id;
+        return id.toString();
+    }
+}

examples/cobra_publisher/ixcrypto/IXUuid.h

@@ -0,0 +1,17 @@
+/*
+ *  IXUuid.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2017 Machine Zone. All rights reserved.
+ */
+#pragma once
+
+#include <string>
+
+namespace ix
+{
+   /**
+    * Generate a random uuid
+    */
+   std::string uuid4();
+
+}

examples/ws_receive/.gitignore

@@ -0,0 +1 @@
+build

examples/ws_receive/CMakeLists.txt

@@ -0,0 +1,30 @@
+#
+# Author: Benjamin Sergeant
+# Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+#
+
+cmake_minimum_required (VERSION 3.4.1)
+project (ws_receive)
+
+# There's -Weverything too for clang
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
+
+set (CMAKE_CXX_STANDARD 14)
+
+option(USE_TLS "Add TLS support" ON)
+
+add_subdirectory(${PROJECT_SOURCE_DIR}/../.. ixwebsocket)
+
+include_directories(ws_receive .)
+
+add_executable(ws_receive 
+  jsoncpp/jsoncpp.cpp
+  ixcrypto/IXBase64.cpp
+  ixcrypto/IXHash.cpp
+  ws_receive.cpp)
+
+if (APPLE AND USE_TLS)
+    target_link_libraries(ws_receive "-framework foundation" "-framework security")
+endif()
+
+target_link_libraries(ws_receive ixwebsocket)

examples/ws_receive/README.md

@@ -0,0 +1 @@
+ws_receive is a simple server upload program. It needs to be used in conjonction with ws_send.

examples/ws_receive/ixcrypto

@@ -0,0 +1 @@
+../cobra_publisher/ixcrypto

examples/ws_receive/jsoncpp/json/json-forwards.h

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

examples/ws_receive/package-lock.json

@@ -0,0 +1,29 @@
+{
+  "requires": true,
+  "lockfileVersion": 1,
+  "dependencies": {
+    "async-limiter": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/async-limiter/-/async-limiter-1.0.0.tgz",
+      "integrity": "sha512-jp/uFnooOiO+L211eZOoSyzpOITMXx1rBITauYykG3BRYPu8h0UcxsPNB04RR5vo4Tyz3+ay17tR6JVf9qzYWg=="
+    },
+    "base-64": {
+      "version": "0.1.0",
+      "resolved": "https://registry.npmjs.org/base-64/-/base-64-0.1.0.tgz",
+      "integrity": "sha1-eAqZyE59YAJgNhURxId2E78k9rs="
+    },
+    "djb2": {
+      "version": "0.0.2",
+      "resolved": "https://registry.npmjs.org/djb2/-/djb2-0.0.2.tgz",
+      "integrity": "sha1-RAs4kao6uBQrVNRpsXe66p6W5O8="
+    },
+    "ws": {
+      "version": "6.1.4",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-6.1.4.tgz",
+      "integrity": "sha512-eqZfL+NE/YQc1/ZynhojeV8q+H050oR8AZ2uIev7RU10svA9ZnJUddHcOUZTJLinZ9yEfdA2kSATS2qZK5fhJA==",
+      "requires": {
+        "async-limiter": "1.0.0"
+      }
+    }
+  }
+}

examples/ws_receive/ws_receive.cpp

@@ -0,0 +1,153 @@
+/*
+ *  ws_receive.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <iostream>
+#include <sstream>
+#include <fstream>
+#include <ixwebsocket/IXWebSocketServer.h>
+#include <jsoncpp/json/json.h>
+#include <ixcrypto/IXBase64.h>
+#include <ixcrypto/IXHash.h>
+
+
+namespace
+{
+    // We should cleanup the file name and full path further to remove .. as well
+    std::string extractFilename(const std::string& path)
+    {
+        std::string filename("filename.conf");
+        std::string::size_type idx;
+
+        idx = path.rfind('/');
+        if (idx != std::string::npos)
+        {
+            std::string filename = path.substr(idx+1);
+            return filename;
+        }
+        else
+        {
+            return std::string();
+        }
+    }
+}
+
+namespace ix
+{
+    void errorHandler(const std::string& errMsg,
+                      const std::string& id,
+                      std::shared_ptr<ix::WebSocket> webSocket)
+    {
+        Json::Value pdu;
+        pdu["kind"] = "error";
+        pdu["id"] = id;
+        pdu["message"] = errMsg;
+        webSocket->send(pdu.toStyledString());
+    }
+
+    void messageHandler(const std::string& str,
+                        std::shared_ptr<ix::WebSocket> webSocket)
+    {
+        std::cerr << "Received message: " << str.size() << std::endl;
+
+        Json::Value data;
+        Json::Reader reader;
+        if (!reader.parse(str, data))
+        {
+            errorHandler("Invalid JSON", std::string(), webSocket);
+            return;
+        }
+
+        std::cout << "id: " << data["id"].asString() << std::endl;
+
+        std::string content = ix::base64_decode(data["content"].asString());
+        std::cout << "Content size: " << content.size() << std::endl;
+
+        // Validate checksum
+        uint64_t cksum = ix::djb2Hash(data["content"].asString());
+        uint64_t cksumRef = data["djb2_hash"].asUInt64();
+
+        std::cout << "Computed hash: " << cksum << std::endl;
+        std::cout << "Reference hash: " << cksumRef << std::endl;
+
+        if (cksum != cksumRef)
+        {
+            errorHandler("Hash mismatch.", std::string(), webSocket);
+            return;
+        }
+
+        std::string filename = data["filename"].asString();
+        filename = extractFilename(filename);
+
+        std::ofstream out(filename);
+        out << content;
+        out.close();
+
+        Json::Value pdu;
+        pdu["ack"] = true;
+        pdu["id"] = data["id"];
+        pdu["filename"] = data["filename"];
+        webSocket->send(pdu.toStyledString());
+    }
+}
+
+int main(int argc, char** argv)
+{
+    int port = 8080;
+    if (argc == 2)
+    {
+        std::stringstream ss;
+        ss << argv[1];
+        ss >> port;
+    }
+
+    ix::WebSocketServer server(port);
+
+    server.setOnConnectionCallback(
+        [&server](std::shared_ptr<ix::WebSocket> webSocket)
+        {
+            webSocket->setOnMessageCallback(
+                [webSocket, &server](ix::WebSocketMessageType messageType,
+                                     const std::string& str,
+                                     size_t wireSize,
+                                     const ix::WebSocketErrorInfo& error,
+                                     const ix::WebSocketOpenInfo& openInfo,
+                                     const ix::WebSocketCloseInfo& closeInfo)
+                {
+                    if (messageType == ix::WebSocket_MessageType_Open)
+                    {
+                        std::cerr << "New connection" << std::endl;
+                        std::cerr << "Uri: " << openInfo.uri << std::endl;
+                        std::cerr << "Headers:" << std::endl;
+                        for (auto it : openInfo.headers)
+                        {
+                            std::cerr << it.first << ": " << it.second << std::endl;
+                        }
+                    }
+                    else if (messageType == ix::WebSocket_MessageType_Close)
+                    {
+                        std::cerr << "Closed connection" << std::endl;
+                    }
+                    else if (messageType == ix::WebSocket_MessageType_Message)
+                    {
+                        messageHandler(str, webSocket);
+                    }
+                }
+            );
+        }
+    );
+
+    auto res = server.listen();
+    if (!res.first)
+    {
+        std::cerr << res.second << std::endl;
+        return 1;
+    }
+
+    server.start();
+    server.wait();
+
+    return 0;
+}

examples/ws_receive/ws_receive.js

@@ -0,0 +1,43 @@
+/*
+ *  ws_receive.js
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+const WebSocket = require('ws')
+const djb2 = require('djb2')
+const fs = require('fs')
+
+const wss = new WebSocket.Server({ port: 8080,
+                                   perMessageDeflate: false,
+                                   maxPayload: 1024 * 1024 * 1024 * 1024});
+
+wss.on('connection', function connection(ws) {
+  ws.on('message', function incoming(data) {
+    console.log('Received message')
+
+    let str = data.toString()
+    let obj = JSON.parse(str)
+
+    console.log(obj.id)
+    console.log(obj.djb2_hash)
+    console.log(djb2(obj.content))
+
+    var content = Buffer.from(obj.content, 'base64')
+    // let bytes = base64.decode(obj.content)
+
+    let path = obj.filename
+    fs.writeFile(path, content, function(err) {
+      if (err) {
+        throw err
+      } else {
+        console.log('wrote data to disk')
+      }
+    });
+
+    let response = {
+      id: obj.id
+    }
+
+    ws.send(JSON.stringify(response))
+  });
+});

examples/ws_send/.gitignore

@@ -0,0 +1 @@
+build

examples/ws_send/CMakeLists.txt

@@ -0,0 +1,31 @@
+#
+# Author: Benjamin Sergeant
+# Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+#
+
+cmake_minimum_required (VERSION 3.4.1)
+project (ws_send)
+
+# There's -Weverything too for clang
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
+
+set (CMAKE_CXX_STANDARD 14)
+
+option(USE_TLS "Add TLS support" ON)
+
+add_subdirectory(${PROJECT_SOURCE_DIR}/../.. ixwebsocket)
+
+include_directories(ws_send .)
+
+add_executable(ws_send 
+  jsoncpp/jsoncpp.cpp
+  ixcrypto/IXBase64.cpp
+  ixcrypto/IXUuid.cpp
+  ixcrypto/IXHash.cpp
+  ws_send.cpp)
+
+if (APPLE AND USE_TLS)
+    target_link_libraries(ws_send "-framework foundation" "-framework security")
+endif()
+
+target_link_libraries(ws_send ixwebsocket)

examples/ws_send/README.md

@@ -0,0 +1 @@
+ws_send is a simple upload program. It needs to be used in conjonction with ws_receive.

examples/ws_send/ixcrypto

@@ -0,0 +1 @@
+../cobra_publisher/ixcrypto

examples/ws_send/jsoncpp

@@ -0,0 +1 @@
+../cobra_publisher/jsoncpp

examples/ws_send/ws_send.cpp

@@ -0,0 +1,306 @@
+/*
+ *  ws_send.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <iostream>
+#include <fstream>
+#include <sstream>
+#include <vector>
+#include <condition_variable>
+#include <mutex>
+#include <chrono>
+#include <ixwebsocket/IXWebSocket.h>
+#include <ixwebsocket/IXSocket.h>
+#include <ixcrypto/IXUuid.h>
+#include <ixcrypto/IXBase64.h>
+#include <ixcrypto/IXHash.h>
+#include <jsoncpp/json/json.h>
+
+using namespace ix;
+
+namespace
+{
+    void log(const std::string& msg)
+    {
+        std::cout << msg << std::endl;
+    }
+
+    class WebSocketSender
+    {
+        public:
+            WebSocketSender(const std::string& _url,
+                            bool enablePerMessageDeflate);
+
+            void subscribe(const std::string& channel);
+            void start();
+            void stop();
+
+            void waitForConnection();
+            void waitForAck();
+
+            void sendMessage(const std::string& filename, bool throttle);
+
+        private:
+            std::string _url;
+            std::string _id;
+            ix::WebSocket _webSocket;
+            bool _enablePerMessageDeflate;
+
+            std::mutex _conditionVariableMutex;
+            std::condition_variable _condition;
+    };
+
+    WebSocketSender::WebSocketSender(const std::string& url,
+                                     bool enablePerMessageDeflate) :
+        _url(url),
+        _enablePerMessageDeflate(enablePerMessageDeflate)
+    {
+        ;
+    }
+
+    void WebSocketSender::stop()
+    {
+        _webSocket.stop();
+    }
+
+    void WebSocketSender::waitForConnection()
+    {
+        std::cout << "Connecting..." << std::endl;
+
+        std::unique_lock<std::mutex> lock(_conditionVariableMutex);
+        _condition.wait(lock);
+    }
+
+    void WebSocketSender::waitForAck()
+    {
+        std::cout << "Waiting for ack..." << std::endl;
+
+        std::unique_lock<std::mutex> lock(_conditionVariableMutex);
+        _condition.wait(lock);
+    }
+
+    std::string load(const std::string& path)
+    {
+        // std::vector<uint8_t> memblock;
+        std::string str;
+
+        std::ifstream file(path);
+        if (!file.is_open()) return std::string();
+
+        file.seekg(0, file.end);
+        std::streamoff size = file.tellg();
+        file.seekg(0, file.beg);
+
+        str.resize(size);
+        file.read((char*)&str.front(), static_cast<std::streamsize>(size));
+
+        return str;
+    }
+
+    void WebSocketSender::start()
+    {
+        _webSocket.setUrl(_url);
+
+        ix::WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions(
+            _enablePerMessageDeflate, false, false, 15, 15);
+        _webSocket.setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
+
+        std::stringstream ss;
+        log(std::string("Connecting to url: ") + _url);
+
+        _webSocket.setOnMessageCallback(
+            [this](ix::WebSocketMessageType messageType,
+               const std::string& str,
+               size_t wireSize,
+               const ix::WebSocketErrorInfo& error,
+               const ix::WebSocketOpenInfo& openInfo,
+               const ix::WebSocketCloseInfo& closeInfo)
+            {
+                std::stringstream ss;
+                if (messageType == ix::WebSocket_MessageType_Open)
+                {
+                    _condition.notify_one();
+
+                    log("ws_send: 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;
+                    }
+                }
+                else if (messageType == ix::WebSocket_MessageType_Close)
+                {
+                    ss << "ws_send: connection closed:";
+                    ss << " code " << closeInfo.code;
+                    ss << " reason " << closeInfo.reason << std::endl;
+                    log(ss.str());
+                }
+                else if (messageType == ix::WebSocket_MessageType_Message)
+                {
+                    _condition.notify_one();
+
+                    ss << "ws_send: received message: "
+                       << str;
+                    log(ss.str());
+
+                    Json::Value data;
+                    Json::Reader reader;
+                    if (!reader.parse(str, data))
+                    {
+                        std::cerr << "Invalid JSON response" << std::endl;
+                        return;
+                    }
+
+                    std::string id = data["id"].asString();
+                    if (_id != id)
+                    {
+                        std::cerr << "Invalid id" << std::endl;
+                    }
+                }
+                else if (messageType == ix::WebSocket_MessageType_Error)
+                {
+                    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;
+                    log(ss.str());
+                }
+                else
+                {
+                    ss << "Invalid ix::WebSocketMessageType";
+                    log(ss.str());
+                }
+            });
+
+        _webSocket.start();
+    }
+
+    class Bench
+    {
+        public:
+            Bench(const std::string& description) :
+                _description(description),
+                _start(std::chrono::system_clock::now()),
+                _reported(false)
+            {
+                ;
+            }
+
+            ~Bench()
+            {
+                if (!_reported)
+                {
+                    report();
+                }
+            }
+
+            void report()
+            {
+                auto now = std::chrono::system_clock::now();
+                auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - _start);
+
+                _ms = milliseconds.count();
+                std::cout << _description << " completed in "
+                          << _ms << "ms" << std::endl;
+
+                _reported = true;
+            }
+
+            uint64_t getDuration() const
+            {
+                return _ms;
+            }
+
+        private:
+            std::string _description;
+            std::chrono::time_point<std::chrono::system_clock> _start;
+            uint64_t _ms;
+            bool _reported;
+    };
+
+    void WebSocketSender::sendMessage(const std::string& filename,
+                                      bool throttle)
+    {
+        std::string content;
+        {
+            Bench bench("load file from disk");
+            content = load(filename);
+        }
+
+        _id = uuid4();
+
+        std::string b64Content;
+        {
+            Bench bench("base 64 encode file");
+            b64Content = base64_encode(content, content.size());
+        }
+
+        Json::Value pdu;
+        pdu["kind"] = "send";
+        pdu["id"] = _id;
+        pdu["content"] = b64Content;
+        pdu["djb2_hash"] = djb2Hash(b64Content);
+        pdu["filename"] = filename;
+
+        Bench bench("Sending file through websocket");
+        _webSocket.send(pdu.toStyledString(),
+                        [throttle](int current, int total) -> bool
+        {
+            std::cout << "Step " << current << " out of " << total << std::endl;
+
+            if (throttle)
+            {
+                std::chrono::duration<double, std::milli> duration(10);
+                std::this_thread::sleep_for(duration);
+            }
+
+            return true;
+        });
+
+        bench.report();
+        auto duration = bench.getDuration();
+        auto transferRate = 1000 * b64Content.size() / duration;
+        transferRate /= (1024 * 1024);
+        std::cout << "Send transfer rate: " << transferRate << "MB/s" << std::endl;
+    }
+
+    void wsSend(const std::string& url,
+                const std::string& path,
+                bool enablePerMessageDeflate,
+                bool throttle)
+    {
+        WebSocketSender webSocketSender(url, enablePerMessageDeflate);
+        webSocketSender.start();
+
+        webSocketSender.waitForConnection();
+
+        std::cout << "Sending..." << std::endl;
+        webSocketSender.sendMessage(path, throttle);
+
+        webSocketSender.waitForAck();
+
+        std::cout << "Done !" << std::endl;
+        webSocketSender.stop();
+    }
+}
+
+int main(int argc, char** argv)
+{
+    if (argc != 3)
+    {
+        std::cerr << "Usage: ws_send <url> <path>" << std::endl;
+        return 1;
+    }
+    std::string url = argv[1];
+    std::string path = argv[2];
+
+    bool throttle = false;
+    bool enablePerMessageDeflate = false;
+
+    Socket::init();
+    wsSend(url, path, enablePerMessageDeflate, throttle);
+    return 0;
+}

ixwebsocket/IXConnectionState.cpp

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

@@ -1,37 +0,0 @@
-/*
- *  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 && isCancellationRequested())
+        if (isCancellationRequested())
         {
             errMsg = "cancellation requested";
             return nullptr;
@@ -121,7 +121,7 @@ namespace ix
             }
 
             // Were we cancelled ?
-            if (isCancellationRequested && isCancellationRequested())
+            if (isCancellationRequested())
             {
                 errMsg = "cancellation requested";
                 return nullptr;
@@ -129,7 +129,7 @@ namespace ix
         }
 
         // Maybe a cancellation request got in before the bg terminated ?
-        if (isCancellationRequested && isCancellationRequested())
+        if (isCancellationRequested())
         {
             errMsg = "cancellation requested";
             return nullptr;

ixwebsocket/IXEventFd.cpp

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

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

@@ -231,17 +231,19 @@ namespace ix
 
             payload.reserve(contentLength);
 
-            auto chunkResult = _socket->readBytes(contentLength,
-                                                  args.onProgressCallback,
-                                                  isCancellationRequested);
-            if (!chunkResult.first)
+            // FIXME: very inefficient way to read bytes, but it works...
+            for (int i = 0; i < contentLength; ++i)
             {
-                errorMsg = "Cannot read chunk";
-                return std::make_tuple(code, HttpErrorCode_ChunkReadError,
-                                       headers, payload, errorMsg,
-                                       uploadSize, downloadSize);
+                char c;
+                if (!_socket->readByte(&c, isCancellationRequested))
+                {
+                    return std::make_tuple(code, HttpErrorCode_ReadError,
+                                           headers, payload, "Cannot read byte",
+                                           uploadSize, downloadSize);
+                }
+
+                payload += c;
             }
-            payload += chunkResult.second;
         }
         else if (headers.find("Transfer-Encoding") != headers.end() &&
                  headers["Transfer-Encoding"] == "chunked")
@@ -275,20 +277,22 @@ namespace ix
 
                 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 another line
+
+                for (uint64_t i = 0; i < chunkSize; ++i)
+                {
+                    char c;
+                    if (!_socket->readByte(&c, isCancellationRequested))
+                    {
+                        errorMsg = "Cannot read byte";
+                        return std::make_tuple(code, HttpErrorCode_ChunkReadError,
+                                               headers, payload, errorMsg,
+                                               uploadSize, downloadSize);
+                    }
+
+                    payload += c;
+                }
 
-                // Read the line that terminates the chunk (\r\n)
                 lineResult = _socket->readLine(isCancellationRequested);
 
                 if (!lineResult.first)

ixwebsocket/IXHttpClient.h

@@ -61,7 +61,6 @@ namespace ix
         bool verbose;
         bool compress;
         Logger logger;
-        OnProgressCallback onProgressCallback;
     };
 
     class HttpClient {

ixwebsocket/IXSelectInterrupt.cpp

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

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

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

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

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

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

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

@@ -1,39 +0,0 @@
-/*
- *  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,8 +7,6 @@
 #include "IXSocket.h"
 #include "IXSocketConnect.h"
 #include "IXNetSystem.h"
-#include "IXSelectInterrupt.h"
-#include "IXSelectInterruptFactory.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -21,23 +19,15 @@
 #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),
-        _selectInterrupt(createSelectInterrupt())
+        _sockfd(fd)
     {
-        ;
+
     }
 
     Socket::~Socket()
@@ -49,93 +39,44 @@ namespace ix
     {
         if (_sockfd == -1)
         {
-            if (onPollCallback) onPollCallback(PollResultType::Error);
+            onPollCallback(PollResultType_Error);
             return;
         }
 
-        PollResultType pollResult = isReadyToRead(1000 * timeoutSecs);
-
-        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(_sockfd, &rfds);
 
-        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);
-        }
+#ifdef __linux__
+        FD_SET(_eventfd.getFd(), &rfds);
+#endif
 
         struct timeval timeout;
-        timeout.tv_sec = timeoutMs / 1000;
-        timeout.tv_usec = (timeoutMs < 1000) ? 0 : 1000 * (timeoutMs % 1000);
+        timeout.tv_sec = timeoutSecs;
+        timeout.tv_usec = 0;
 
-        // Compute the highest fd.
         int sockfd = _sockfd;
-        int nfds = (std::max)(sockfd, interruptFd);
+        int nfds = (std::max)(sockfd, _eventfd.getFd());
+        int ret = select(nfds + 1, &rfds, nullptr, nullptr,
+                         (timeoutSecs < 0) ? nullptr : &timeout);
 
-        int ret = ::select(nfds + 1, &rfds, &wfds, nullptr,
-                           (timeoutMs < 0) ? nullptr : &timeout);
-
-        PollResultType pollResult = PollResultType::ReadyForRead;
+        PollResultType pollResult = PollResultType_ReadyForRead;
         if (ret < 0)
         {
-            pollResult = PollResultType::Error;
+            pollResult = PollResultType_Error;
         }
         else if (ret == 0)
         {
-            pollResult = PollResultType::Timeout;
-        }
-        else if (interruptFd != -1 && FD_ISSET(interruptFd, &rfds))
-        {
-            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;
+            pollResult = PollResultType_Timeout;
         }
 
-        return pollResult;
+        onPollCallback(pollResult);
     }
 
-    PollResultType Socket::isReadyToRead(int timeoutMs)
+    void Socket::wakeUpFromPoll()
     {
-        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);
+        // this will wake up the thread blocked on select, only needed on Linux
+        _eventfd.notify();
     }
 
     bool Socket::connect(const std::string& host,
@@ -145,7 +86,7 @@ namespace ix
     {
         std::lock_guard<std::mutex> lock(_socketMutex);
 
-        if (!_selectInterrupt->clear()) return false;
+        if (!_eventfd.clear()) return false;
 
         _sockfd = SocketConnect::connect(host, port, errMsg, isCancellationRequested);
         return _sockfd != -1;
@@ -204,40 +145,24 @@ namespace ix
 #endif
     }
 
-    bool Socket::init(std::string& errorMsg)
+    bool Socket::init()
     {
-        return _selectInterrupt->init(errorMsg);
+#ifdef _WIN32
+        INT rc;
+        WSADATA wsaData;
+
+        rc = WSAStartup(MAKEWORD(2, 2), &wsaData);
+        return rc != 0;
+#else
+        return true;
+#endif
     }
 
-    bool Socket::writeBytes(const std::string& str,
-                            const CancellationRequest& isCancellationRequested)
+    void Socket::cleanup()
     {
-        while (true)
-        {
-            if (isCancellationRequested && isCancellationRequested()) return false;
-
-            char* buffer = const_cast<char*>(str.c_str());
-            int len = (int) str.size();
-
-            ssize_t ret = send(buffer, len);
-
-            // We wrote some bytes, as needed, all good.
-            if (ret > 0)
-            {
-                return ret == len;
-            }
-            // There is possibly something to be writen, try again
-            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
-                                 getErrno() == EAGAIN))
-            {
-                continue;
-            }
-            // There was an error during the write, abort
-            else
-            {
-                return false;
-            }
-        }
+#ifdef _WIN32
+        WSACleanup();
+#endif
     }
 
     bool Socket::readByte(void* buffer,
@@ -245,7 +170,7 @@ namespace ix
     {
         while (true)
         {
-            if (isCancellationRequested && isCancellationRequested()) return false;
+            if (isCancellationRequested()) return false;
 
             ssize_t ret;
             ret = recv(buffer, 1);
@@ -259,12 +184,23 @@ namespace ix
             else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
                                  getErrno() == EAGAIN))
             {
-                // Wait with a 1ms timeout until the socket is ready to read.
+                // Wait with a timeout until something is written.
                 // This way we are not busy looping
-                if (isReadyToRead(1) == PollResultType::Error)
+                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
@@ -274,8 +210,38 @@ namespace ix
         }
     }
 
-    std::pair<bool, std::string> Socket::readLine(
-        const CancellationRequest& isCancellationRequested)
+    bool Socket::writeBytes(const std::string& str,
+                            const CancellationRequest& isCancellationRequested)
+    {
+        while (true)
+        {
+            if (isCancellationRequested()) return false;
+
+            char* buffer = const_cast<char*>(str.c_str());
+            int len = (int) str.size();
+
+            ssize_t ret = send(buffer, len);
+
+            // We wrote some bytes, as needed, all good.
+            if (ret > 0)
+            {
+                return ret == len;
+            }
+            // There is possibly something to be write, try again
+            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
+                                 getErrno() == EAGAIN))
+            {
+                continue;
+            }
+            // There was an error during the write, abort
+            else
+            {
+                return false;
+            }
+        }
+    }
+
+    std::pair<bool, std::string> Socket::readLine(const CancellationRequest& isCancellationRequested)
     {
         char c;
         std::string line;
@@ -285,8 +251,7 @@ namespace ix
         {
             if (!readByte(&c, isCancellationRequested))
             {
-                // Return what we were able to read
-                return std::make_pair(false, line);
+                return std::make_pair(false, std::string());
             }
 
             line += c;
@@ -294,52 +259,4 @@ 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,29 +10,22 @@
 #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
 {
-    class SelectInterrupt;
-
-    enum class PollResultType
+    enum PollResultType
     {
-        ReadyForRead = 0,
-        ReadyForWrite = 1,
-        Timeout = 2,
-        Error = 3,
-        SendRequest = 4,
-        CloseRequest = 5
+        PollResultType_ReadyForRead = 0,
+        PollResultType_Timeout = 1,
+        PollResultType_Error = 2
     };
 
     class Socket {
@@ -41,17 +34,12 @@ namespace ix
 
         Socket(int fd = -1);
         virtual ~Socket();
-        bool init(std::string& errorMsg);
 
         void configure();
 
-        // Functions to check whether there is activity on the socket
-        void poll(const OnPollCallback& onPollCallback,
-                  int timeoutSecs = kDefaultPollTimeout);
-        bool wakeUpFromPoll(uint8_t wakeUpCode);
-
-        PollResultType isReadyToWrite(int timeoutMs);
-        PollResultType isReadyToRead(int timeoutMs);
+        virtual void poll(const OnPollCallback& onPollCallback,
+                          int timeoutSecs = kDefaultPollTimeout);
+        virtual void wakeUpFromPoll();
 
         // Virtual methods
         virtual bool connect(const std::string& url,
@@ -70,36 +58,21 @@ 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> readBytes(
-            size_t length,
-            const OnProgressCallback& onProgressCallback,
-            const CancellationRequest& isCancellationRequested);
+        std::pair<bool, std::string> readLine(const CancellationRequest& isCancellationRequested);
 
         static int getErrno();
-
-        // Used as special codes for pipe communication
-        static const uint64_t kSendRequest;
-        static const uint64_t kCloseRequest;
+        static bool init(); // Required on Windows to initialize WinSocket
+        static void cleanup(); // Required on Windows to cleanup WinSocket
 
     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 && isCancellationRequested()) // Must handle timeout as well
+            if (isCancellationRequested()) // Must handle timeout as well
             {
                 closeSocket(fd);
                 errMsg = "Cancelled";

ixwebsocket/IXSocketFactory.cpp

@@ -6,20 +6,12 @@
 
 #include "IXSocketFactory.h"
 
-#ifdef IXWEBSOCKET_USE_TLS
-
+#if defined(__APPLE__) or defined(__linux__)
 # 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
@@ -28,47 +20,23 @@ namespace ix
                                          std::string& errorMsg)
     {
         errorMsg.clear();
-        std::shared_ptr<Socket> socket;
 
         if (!tls)
         {
-            socket = std::make_shared<Socket>();
+            return std::make_shared<Socket>();
         }
         else
         {
 #ifdef IXWEBSOCKET_USE_TLS
 # ifdef __APPLE__
-            socket = std::make_shared<SocketAppleSSL>();
-# elif defined(_WIN32)
-            socket = std::make_shared<SocketSChannel>();
+            return std::make_shared<SocketAppleSSL>();
 # else
-            socket = std::make_shared<SocketOpenSSL>();
+            return 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

@@ -8,14 +8,10 @@
 #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,7 +21,6 @@
 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
 
-        static std::once_flag _openSSLInitFlag;
+        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, nullptr);
+        return Socket::connect(host, port, errMsg);
     }
 
 
@@ -89,17 +89,17 @@ namespace ix
         Socket::close();
     }
 
-    ssize_t SocketSChannel::send(char* buf, size_t nbyte)
+    int SocketSChannel::send(char* buf, size_t nbyte)
     {
         return Socket::send(buf, nbyte);
     }
 
-    ssize_t SocketSChannel::send(const std::string& buffer)
+    int SocketSChannel::send(const std::string& buffer)
     {
         return Socket::send(buffer);
     }
 
-    ssize_t SocketSChannel::recv(void* buf, size_t nbyte)
+    int 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 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;
+        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;
 
     private:
     };

ixwebsocket/IXSocketServer.cpp

@@ -29,8 +29,7 @@ namespace ix
         _host(host),
         _backlog(backlog),
         _maxConnections(maxConnections),
-        _stop(false),
-        _connectionStateFactory(&ConnectionState::createConnectionState)
+        _stop(false)
     {
 
     }
@@ -136,9 +135,6 @@ namespace ix
 
     void SocketServer::stop()
     {
-        closeTerminatedThreads();
-        assert(_connectionsThreads.empty());
-
         if (!_thread.joinable()) return; // nothing to do
 
         _stop = true;
@@ -149,50 +145,18 @@ 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;
@@ -250,19 +214,14 @@ namespace ix
                 continue;
             }
 
-            std::shared_ptr<ConnectionState> connectionState;
-            if (_connectionStateFactory)
-            {
-                connectionState = _connectionStateFactory();
-            }
-
             // Launch the handleConnection work asynchronously in its own thread.
-            _connectionsThreads.push_back(std::make_pair(
-                    connectionState,
-                    std::thread(&SocketServer::handleConnection,
-                                this,
-                                clientFd,
-                                connectionState)));
+            //
+            // 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,
+                           this,
+                           clientFd);
         }
     }
 }

ixwebsocket/IXSocketServer.h

@@ -6,13 +6,10 @@
 
 #pragma once
 
-#include "IXConnectionState.h"
-
 #include <utility> // pair
 #include <string>
 #include <set>
 #include <thread>
-#include <list>
 #include <mutex>
 #include <functional>
 #include <memory>
@@ -23,12 +20,6 @@ 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,
@@ -36,8 +27,6 @@ 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;
@@ -68,20 +57,12 @@ 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,
-                                      std::shared_ptr<ConnectionState> connectionState) = 0;
+        virtual void handleConnection(int fd) = 0;
         virtual size_t getConnectedClientsCount() = 0;
-
-        void closeTerminatedThreads();
     };
 }

ixwebsocket/IXWebSocket.cpp

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

ixwebsocket/IXWebSocket.h

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

ixwebsocket/IXWebSocketHandshake.cpp

@@ -114,7 +114,7 @@ namespace ix
         std::stringstream ss;
         ss << "HTTP/1.1 ";
         ss << code;
-        ss << " ";
+        ss << "\r\n";
         ss << reason;
         ss << "\r\n";
 
@@ -353,7 +353,7 @@ namespace ix
         WebSocketHandshakeKeyGen::generate(headers["sec-websocket-key"].c_str(), output);
 
         std::stringstream ss;
-        ss << "HTTP/1.1 101 Switching Protocols\r\n";
+        ss << "HTTP/1.1 101\r\n";
         ss << "Sec-WebSocket-Accept: " << std::string(output) << "\r\n";
         ss << "Upgrade: websocket\r\n";
         ss << "Connection: Upgrade\r\n";

ixwebsocket/IXWebSocketHttpHeaders.cpp

@@ -6,28 +6,12 @@
 
 #include "IXWebSocketHttpHeaders.h"
 #include "IXSocket.h"
-#include <algorithm>
-#include <locale>
+
+#include <string>
+#include <unordered_map>
 
 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)

ixwebsocket/IXWebSocketHttpHeaders.h

@@ -11,6 +11,7 @@
 #include <string>
 #include <map>
 #include <memory>
+#include <algorithm>
 
 namespace ix
 {
@@ -21,10 +22,19 @@ namespace ix
         // Case Insensitive compare_less binary function
         struct NocaseCompare
         {
-            bool operator() (const unsigned char& c1, const unsigned char& c2) const;
+            bool operator() (const unsigned char& c1, const unsigned char& c2) const
+            {
+                return std::tolower(c1) < std::tolower(c2);
+            }
         };
 
-        bool operator() (const std::string & s1, const std::string & s2) const;
+        bool 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
+        }
     };
 
     using WebSocketHttpHeaders = std::map<std::string, std::string, CaseInsensitiveLess>;

ixwebsocket/IXWebSocketServer.cpp

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

ixwebsocket/IXWebSocketServer.h

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

ixwebsocket/IXWebSocketTransport.cpp

@@ -1,31 +1,7 @@
-/*
- * 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-2019 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
  */
 
 //
@@ -38,6 +14,14 @@
 #include "IXUrlParser.h"
 #include "IXSocketFactory.h"
 
+#ifdef IXWEBSOCKET_USE_TLS
+# ifdef __APPLE__
+#  include "IXSocketAppleSSL.h"
+# else
+#  include "IXSocketOpenSSL.h"
+# endif
+#endif
+
 #include <string.h>
 #include <stdlib.h>
 
@@ -53,12 +37,11 @@
 
 namespace ix
 {
-    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::heartbeat");
+    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::hearbeat");
     const int WebSocketTransport::kDefaultHeartBeatPeriod(-1);
     constexpr size_t WebSocketTransport::kChunkSize;
 
     WebSocketTransport::WebSocketTransport() :
-        _useMask(true),
         _readyState(CLOSED),
         _closeCode(0),
         _closeWireSize(0),
@@ -76,11 +59,11 @@ namespace ix
     }
 
     void WebSocketTransport::configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                                       int heartBeatPeriod)
+                                       int hearBeatPeriod)
     {
         _perMessageDeflateOptions = perMessageDeflateOptions;
         _enablePerMessageDeflate = _perMessageDeflateOptions.enabled();
-        _heartBeatPeriod = heartBeatPeriod;
+        _heartBeatPeriod = hearBeatPeriod;
     }
 
     // Client
@@ -97,6 +80,16 @@ namespace ix
                                        std::string("Could not parse URL ") + url);
         }
 
+        if (protocol != "ws" && protocol != "wss")
+        {
+            std::stringstream ss;
+            ss << "Invalid protocol: " << protocol
+               << " for url " << url
+               << " . Supported protocols are ws and wss";
+
+            return WebSocketInitResult(false, 0, ss.str());
+        }
+
         bool tls = protocol == "wss";
         std::string errorMsg;
         _socket = createSocket(tls, errorMsg);
@@ -124,16 +117,8 @@ namespace ix
     // Server
     WebSocketInitResult WebSocketTransport::connectToSocket(int fd, int timeoutSecs)
     {
-        // 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);
-        }
+        _socket.reset();
+        _socket = std::make_shared<Socket>(fd);
 
         WebSocketHandshake webSocketHandshake(_requestInitCancellation,
                                               _socket,
@@ -193,75 +178,43 @@ 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();
-                        }
+                while (true)
+                {
+                    ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
+
+                    if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
+                                    _socket->getErrno() == EAGAIN))
+                    {
+                        break;
+                    }
+                    else if (ret <= 0)
+                    {
+                        _rxbuf.clear();
+                        _socket->close();
+                        setReadyState(CLOSED);
+                        break;
+                    }
+                    else
+                    {
+                        _rxbuf.insert(_rxbuf.end(),
+                                      _readbuf.begin(),
+                                      _readbuf.begin() + ret);
                     }
                 }
-                else if (pollResult == PollResultType::ReadyForRead)
-                {
-                    while (true)
-                    {
-                        ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
 
-                        if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
-                                        _socket->getErrno() == EAGAIN))
-                        {
-                            break;
-                        }
-                        else if (ret <= 0)
-                        {
-                            _rxbuf.clear();
-                            _socket->close();
-                            setReadyState(CLOSED);
-                            break;
-                        }
-                        else
-                        {
-                            _rxbuf.insert(_rxbuf.end(),
-                                          _readbuf.begin(),
-                                          _readbuf.begin() + ret);
-                        }
-                    }
-                }
-                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)
+                if (isSendBufferEmpty() && _readyState == CLOSING)
                 {
                     _socket->close();
+                    setReadyState(CLOSED);
                 }
             },
             _heartBeatPeriod);
@@ -284,15 +237,19 @@ namespace ix
         _txbuf.insert(_txbuf.end(), header.begin(), header.end());
         _txbuf.insert(_txbuf.end(), begin, end);
 
-        if (_useMask)
+        // Masking
+        for (size_t i = 0; i != (size_t) message_size; ++i)
         {
-            for (size_t i = 0; i != (size_t) message_size; ++i)
-            {
-                *(_txbuf.end() - (size_t) message_size + i) ^= masking_key[i&0x3];
-            }
+            *(_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)
     {
         if (ws.mask)
@@ -435,10 +392,6 @@ namespace ix
                         emitMessage(MSG, getMergedChunks(), ws, onMessageCallback);
                         _chunks.clear();
                     }
-                    else
-                    {
-                        emitMessage(FRAGMENT, std::string(), ws, onMessageCallback);
-                    }
                 }
             }
             else if (ws.opcode == wsheader_type::PING)
@@ -474,8 +427,14 @@ 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(code, reason, _rxbuf.size());
+                close();
             }
             else
             {
@@ -516,7 +475,7 @@ namespace ix
         size_t wireSize = message.size();
 
         // When the RSV1 bit is 1 it means the message is compressed
-        if (_enablePerMessageDeflate && ws.rsv1 && messageKind != FRAGMENT)
+        if (_enablePerMessageDeflate && ws.rsv1)
         {
             std::string decompressedMessage;
             bool success = _perMessageDeflate.decompress(message, decompressedMessage);
@@ -614,7 +573,7 @@ namespace ix
                 // Send message
                 sendFragment(opcodeType, fin, begin, end, compress);
 
-                if (onProgressCallback && !onProgressCallback((int)i, (int) steps))
+                if (onProgressCallback && !onProgressCallback(i, steps))
                 {
                     break;
                 }
@@ -623,12 +582,6 @@ 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);
     }
 
@@ -638,7 +591,7 @@ namespace ix
                                           std::string::const_iterator message_end,
                                           bool compress)
     {
-        uint64_t message_size = static_cast<uint64_t>(message_end - message_begin);
+        auto message_size = message_end - message_begin;
 
         unsigned x = getRandomUnsigned();
         uint8_t masking_key[4] = {};
@@ -650,8 +603,7 @@ namespace ix
         std::vector<uint8_t> header;
         header.assign(2 +
                       (message_size >= 126 ? 2 : 0) +
-                      (message_size >= 65536 ? 6 : 0) +
-                      (_useMask ? 4 : 0), 0);
+                      (message_size >= 65536 ? 6 : 0) + 4, 0);
         header[0] = type;
 
         // The fin bit indicate that this is the last fragment. Fin is French for end.
@@ -668,33 +620,27 @@ namespace ix
 
         if (message_size < 126)
         {
-            header[1] = (message_size & 0xff) | (_useMask ? 0x80 : 0);
+            header[1] = (message_size & 0xff) | 0x80;
 
-            if (_useMask)
-            {
-                header[2] = masking_key[0];
-                header[3] = masking_key[1];
-                header[4] = masking_key[2];
-                header[5] = masking_key[3];
-            }
+            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 | (_useMask ? 0x80 : 0);
+            header[1] = 126 | 0x80;
             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];
-            }
+            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 | (_useMask ? 0x80 : 0);
+            header[1] = 127 | 0x80;
             header[2] = (message_size >> 56) & 0xff;
             header[3] = (message_size >> 48) & 0xff;
             header[4] = (message_size >> 40) & 0xff;
@@ -704,13 +650,10 @@ 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];
-            }
+            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:
@@ -736,15 +679,6 @@ 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);
@@ -775,7 +709,7 @@ namespace ix
         _lastSendTimePoint = std::chrono::steady_clock::now();
     }
 
-    void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)
+    void WebSocketTransport::close()
     {
         _requestInitCancellation = true;
 
@@ -783,29 +717,18 @@ namespace ix
 
         // See list of close events here:
         // https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
-        const std::string closure{(char)(code >> 8), (char)(code & 0xff)};
-
+        // We use 1000: normal closure.
+        //
+        // >>> struct.pack('!H', 1000)
+        // b'\x03\xe8'
+        //
+        const std::string normalClosure = std::string("\x03\xe8");
         bool compress = false;
-        sendData(wsheader_type::CLOSE, closure, compress);
+        sendData(wsheader_type::CLOSE, normalClosure, compress);
         setReadyState(CLOSING);
 
-        _socket->wakeUpFromPoll(Socket::kCloseRequest);
+        _socket->wakeUpFromPoll();
         _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,13 +30,6 @@ namespace ix
 {
     class Socket;
 
-    enum class SendMessageKind
-    {
-        Text,
-        Binary,
-        Ping
-    };
-
     class WebSocketTransport
     {
     public:
@@ -52,8 +45,7 @@ namespace ix
         {
             MSG,
             PING,
-            PONG,
-            FRAGMENT
+            PONG
         };
 
         using OnMessageCallback = std::function<void(const std::string&,
@@ -68,7 +60,7 @@ namespace ix
         ~WebSocketTransport();
 
         void configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                       int heartBeatPeriod);
+                       int hearBeatPeriod);
 
         WebSocketInitResult connectToUrl(const std::string& url, // Client
                                          int timeoutSecs);
@@ -78,19 +70,12 @@ 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(uint16_t code = 1000,
-                   const std::string& reason = "Normal closure",
-                   size_t closeWireSize = 0);
-
+        void close();
         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;
@@ -113,10 +98,6 @@ 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;
 
@@ -165,7 +146,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 than the heartbeat period
+        // No data was send through the socket for longer that the hearbeat period
         bool heartBeatPeriodExceeded();
 
         void sendOnSocket();
@@ -191,6 +172,7 @@ 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,29 +3,15 @@
 #
 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 ${IMG} .
-	docker tag ${IMG} ${BUILD}
-
-docker_push:
-	docker tag ${IMG} ${LATEST}
-	docker push ${LATEST}
+	docker build -t broadcast_server:latest .
 
 run:
-	docker run --cap-add sys_ptrace -it ws:latest
+	docker run --cap-add sys_ptrace -it broadcast_server:latest bash
 
 # this is helpful to remove trailing whitespaces
 trail:
@@ -48,10 +34,7 @@ test_server:
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
 test:
-	(cd test ; python2.7 run.py)
-
-ws_test: all
-	(cd ws ; bash test_ws.sh)
+	python test/run.py
 
 # For the fork that is configured with appveyor
 rebase_upstream:
@@ -60,9 +43,5 @@ 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,7 +29,6 @@ set (SOURCES
 
   IXDNSLookupTest.cpp
   IXSocketTest.cpp
-  IXSocketConnectTest.cpp
 )
 
 # Some unittest don't work on windows yet

test/IXSocketConnectTest.cpp

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

test/IXTest.cpp

@@ -16,7 +16,6 @@
 #include <iostream>
 #include <stdlib.h>
 #include <stack>
-#include <iomanip>
 
 namespace ix
 {
@@ -70,15 +69,10 @@ 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)
         {
@@ -109,7 +103,7 @@ namespace ix
         }
 
         struct sockaddr_in sa; // server address information
-        socklen_t len;
+        unsigned int len;
         if (getsockname(sockfd, (struct sockaddr *) &sa, &len) < 0)
         {
             log("Cannot compute a free port. getsockname error.");
@@ -128,15 +122,8 @@ 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...
@@ -149,21 +136,4 @@ 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 heartbeat ; if no traffic is present on the connection for 1 second
+        // Set a 1 second hearbeat ; if no traffic is present on the connection for 1 second
         // a ping message will be sent by the client.
         _webSocket.setHeartBeatPeriod(1);
 
@@ -128,11 +128,10 @@ namespace
     {
         // A dev/null server
         server.setOnConnectionCallback(
-            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket,
-                                             std::shared_ptr<ConnectionState> connectionState)
+            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
+                    [webSocket, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -142,7 +141,6 @@ 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)
@@ -212,10 +210,6 @@ 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,7 +8,6 @@
 #include <ixwebsocket/IXSocket.h>
 #include <ixwebsocket/IXWebSocket.h>
 #include <ixwebsocket/IXWebSocketServer.h>
-#include <ixwebsocket/IXSocketFactory.h>
 
 #include "IXTest.h"
 
@@ -18,32 +17,13 @@ using namespace ix;
 
 namespace ix
 {
-    // Test that we can override the connectionState impl to provide our own
-    class ConnectionStateCustom : public ConnectionState
+    bool startServer(ix::WebSocketServer& server)
     {
-        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, &connectionId](std::shared_ptr<ix::WebSocket> webSocket,
-                      std::shared_ptr<ConnectionState> connectionState)
+            [&server](std::shared_ptr<ix::WebSocket> webSocket)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState,
-                     &connectionId, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -53,16 +33,12 @@ 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)
                         {
@@ -101,21 +77,19 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        std::string connectionId;
-        REQUIRE(startServer(server, connectionId));
+        REQUIRE(startServer(server));
 
-        std::string errMsg;
-        bool tls = false;
-        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        Socket socket;
         std::string host("localhost");
+        std::string errMsg;
         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;
 
@@ -135,24 +109,22 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        std::string connectionId;
-        REQUIRE(startServer(server, connectionId));
+        REQUIRE(startServer(server));
 
-        std::string errMsg;
-        bool tls = false;
-        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        Socket socket;
         std::string host("localhost");
+        std::string errMsg;
         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;
 
@@ -172,28 +144,26 @@ TEST_CASE("Websocket_server", "[websocket_server]")
     {
         int port = getFreePort();
         ix::WebSocketServer server(port);
-        std::string connectionId;
-        REQUIRE(startServer(server, connectionId));
+        REQUIRE(startServer(server));
 
-        std::string errMsg;
-        bool tls = false;
-        std::shared_ptr<Socket> socket = createSocket(tls, errMsg);
+        Socket socket;
         std::string host("localhost");
+        std::string errMsg;
         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"
-                           "Upgrade: websocket\r\n"
-                           "Sec-WebSocket-Version: 13\r\n"
-                           "Sec-WebSocket-Key: foobar\r\n"
-                           "\r\n",
-                           isCancellationRequested);
+        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;
 
@@ -204,8 +174,6 @@ 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,21 +164,10 @@ 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
                 {
-                    ss << "Unexpected ix::WebSocketMessageType";
+                    // FIXME: missing ping/pong messages
+                    ss << "Invalid ix::WebSocketMessageType";
                     log(ss.str());
                 }
             });
@@ -217,11 +206,10 @@ namespace
     bool startServer(ix::WebSocketServer& server)
     {
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket,
-                      std::shared_ptr<ConnectionState> connectionState)
+            [&server](std::shared_ptr<ix::WebSocket> webSocket)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -231,7 +219,6 @@ 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,484 +1,82 @@
-#!/usr/bin/env python2.7
-'''
-'''
-
-from __future__ import print_function
-
 import os
-import sys
 import platform
-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
-
-hasClick = True
-try:
-    import click
-except ImportError:
-    hasClick = False
-
-
-DEFAULT_EXE = 'ixwebsocket_unittest'
-
-
-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:
-            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',
-        'ubsan': '-DSANITIZE_UNDEFINED=On',
-        'tsan': '-DSANITIZE_THREAD=On',
-        'none': ''
-    }
-    sanitizerFlag = 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'
-
-    fmt = '''
-{cmakeExecutable} -H. \
-    {sanitizerFlag} \
-    -B{buildDir} \
-    -DCMAKE_BUILD_TYPE=Debug \
-    -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
-'''
-    cmakeCmd = fmt.format(**locals())
-    runCommand(cmakeCmd)
-
-
-def runTest(args, buildDir, xmlOutput, testRunName):
-    '''Execute the unittest.
-    '''
-    if args is None:
-        args = ''
-
-    fmt = '{buildDir}/{DEFAULT_EXE} -o {xmlOutput} -n "{testRunName}" -r junit "{args}"'
-    testCommand = fmt.format(**locals())
-    runCommand(testCommand,
-               assertOnFailure=False)
-
-
-def validateTestSuite(xmlOutput):
-    '''
-    Parse the output XML file to validate that all tests passed.
-
-    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
-
-        testName = testcase.attrib['name']
-        systemOutput = None
-
-        for child in testcase:
-            if child.tag == 'system-out':
-                systemOutput = child.text
-
-            if child.tag == 'failure':
-                success = False
-
-                print("Testcase '{}' failed".format(testName))
-                print(' ', systemOutput)
-
-    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()
+import shutil
+
+osName = platform.system()
+print('os name = {}'.format(osName))
+
+root = os.path.dirname(os.path.realpath(__file__))
+buildDir = os.path.join(root, 'build')
+
+if not os.path.exists(buildDir):
+    os.mkdir(buildDir)
+
+os.chdir(buildDir)
+
+if osName == 'Windows':
+    generator = '-G"NMake Makefiles"'
+    make = 'nmake'
+    testBinary ='ixwebsocket_unittest.exe'
+else:
+    generator = ''
+    make = 'make -j6'
+    testBinary ='./ixwebsocket_unittest'
+
+sanitizersFlags = {
+    'asan': '-DSANITIZE_ADDRESS=On',
+    'ubsan': '-DSANITIZE_UNDEFINED=On',
+    'tsan': '-DSANITIZE_THREAD=On',
+    'none': ''
+}
+sanitizer = 'tsan'
+if osName == 'Linux':
+    sanitizer = 'none'
+
+sanitizerFlags = sanitizersFlags[sanitizer]
+
+# if osName == 'Windows':
+#     os.environ['CC'] = 'clang-cl'
+#     os.environ['CXX'] = 'clang-cl'
+
+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 findFiles(prefix):
+    '''Find all files under a given directory'''
+
+    paths = []
+
+    for root, _, files in os.walk(prefix):
+        for path in files:
+            fullPath = os.path.join(root, path)
+
+            if os.path.islink(fullPath):
+                continue
+
+            paths.append(fullPath)
+
+    return paths
+
+#for path in findFiles('.'):
+#    print(path)
+
+# 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'), '.')
+
+testCommand = '{} {}'.format(testBinary, os.getenv('TEST', ''))
+ret = os.system(testCommand)
+assert ret == 0, 'Test command failed'

test/test_runner.cpp

@@ -7,8 +7,14 @@
 #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

@@ -1 +0,0 @@
-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/remove_trailing_whitespaces.sh

@@ -1,3 +1,2 @@
 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

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

third_party/statsd-client-cpp/CMakeLists.txt

@@ -1,18 +0,0 @@
-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

@@ -1,27 +0,0 @@
-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

@@ -1,34 +0,0 @@
-# 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

@@ -1,164 +0,0 @@
-
-#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

@@ -1,28 +0,0 @@
-
-#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

@@ -1,246 +0,0 @@
-#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

@@ -1,66 +0,0 @@
-
-#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,2 +1 @@
 build
-node_modules

ws/CMakeLists.txt

@@ -7,12 +7,7 @@ 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_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
 
 set (CMAKE_CXX_STANDARD 14)
 
@@ -21,20 +16,12 @@ 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
@@ -45,21 +32,11 @@ 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)
 
-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})
+if (APPLE AND USE_TLS)
+    target_link_libraries(ws "-framework foundation" "-framework security")
 endif()
 
+target_link_libraries(ws ixwebsocket)
 install(TARGETS ws RUNTIME DESTINATION bin)

ws/IXRedisClient.cpp

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

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

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

@@ -1,47 +0,0 @@
-/*
- *  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,64 +1,10 @@
-# 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
-```
-
-## 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 send ws://localhost:8080 /file/to/path
 ```

ws/cobra_publisher/.gitignore

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

ws/cobra_publisher/CMakeLists.txt

@@ -0,0 +1,38 @@
+#
+# 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/IXCobraConnection.cpp

@@ -6,7 +6,6 @@
 
 #include "IXCobraConnection.h"
 #include <ixcrypto/IXHMac.h>
-#include <ixwebsocket/IXWebSocket.h>
 
 #include <algorithm>
 #include <stdexcept>
@@ -21,10 +20,9 @@ namespace ix
     constexpr size_t CobraConnection::kQueueMaxSize;
 
     CobraConnection::CobraConnection() :
-        _webSocket(new WebSocket()),
-        _publishMode(CobraConnection_PublishMode_Immediate),
         _authenticated(false),
-        _eventCallback(nullptr)
+        _eventCallback(nullptr),
+        _publishMode(CobraConnection_PublishMode_Immediate)
     {
         _pdu["action"] = "rtm/publish";
 
@@ -34,7 +32,6 @@ namespace ix
     CobraConnection::~CobraConnection()
     {
         disconnect();
-        setEventCallback(nullptr);
     }
 
     void CobraConnection::setTrafficTrackerCallback(const TrafficTrackerCallback& callback)
@@ -62,40 +59,36 @@ namespace ix
     }
 
     void CobraConnection::invokeEventCallback(ix::CobraConnectionEventType eventType,
-                                               const std::string& errorMsg,
-                                               const WebSocketHttpHeaders& headers,
-                                               const std::string& subscriptionId)
+                                              const std::string& errorMsg,
+                                              const WebSocketHttpHeaders& headers)
     {
         std::lock_guard<std::mutex> lock(_eventCallbackMutex);
         if (_eventCallback)
         {
-            _eventCallback(eventType, errorMsg, headers, subscriptionId);
+            _eventCallback(eventType, errorMsg, headers);
         }
     }
 
-    void CobraConnection::invokeErrorCallback(const std::string& errorMsg,
-                                               const std::string& serializedPdu)
+    void CobraConnection::invokeErrorCallback(const std::string& errorMsg)
     {
-        std::stringstream ss;
-        ss << errorMsg << " : received pdu => " << serializedPdu;
-        invokeEventCallback(ix::CobraConnection_EventType_Error, ss.str());
+        invokeEventCallback(ix::CobraConnection_EventType_Error, errorMsg);
     }
 
     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::WebSocketOpenInfo& openInfo,
-                   const ix::WebSocketCloseInfo& closeInfo)
+                   const ix::WebSocketCloseInfo& closeInfo,
+                   const ix::WebSocketHttpHeaders& headers)
             {
                 CobraConnection::invokeTrafficTrackerCallback(wireSize, true);
 
@@ -104,7 +97,7 @@ namespace ix
                 {
                     invokeEventCallback(ix::CobraConnection_EventType_Open,
                                         std::string(),
-                                        openInfo.headers);
+                                        headers);
                     sendHandshakeMessage();
                 }
                 else if (messageType == ix::WebSocket_MessageType_Close)
@@ -123,13 +116,13 @@ namespace ix
                     Json::Reader reader;
                     if (!reader.parse(str, data))
                     {
-                        invokeErrorCallback("Invalid json", str);
+                        invokeErrorCallback(std::string("Invalid json: ") + str);
                         return;
                     }
 
                     if (!data.isMember("action"))
                     {
-                        invokeErrorCallback("Missing action", str);
+                        invokeErrorCallback("Missing action");
                         return;
                     }
 
@@ -139,12 +132,12 @@ namespace ix
                     {
                         if (!handleHandshakeResponse(data))
                         {
-                            invokeErrorCallback("Error extracting nonce from handshake response", str);
+                            invokeErrorCallback("Error extracting nonce from handshake response");
                         }
                     }
                     else if (action == "auth/handshake/error")
                     {
-                        invokeErrorCallback("Handshake error", str);
+                        invokeErrorCallback("Handshake error."); // print full message ?
                     }
                     else if (action == "auth/authenticate/ok")
                     {
@@ -154,37 +147,15 @@ namespace ix
                     }
                     else if (action == "auth/authenticate/error")
                     {
-                        invokeErrorCallback("Authentication error", str);
+                        invokeErrorCallback("Authentication error."); // print full message ?
                     }
                     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("Un-handled message type", str);
+                        invokeErrorCallback(std::string("Un-handled message type: ") + action);
                     }
                 }
                 else if (messageType == ix::WebSocket_MessageType_Error)
@@ -194,7 +165,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(), std::string());
+                    invokeErrorCallback(ss.str());
                 }
         });
     }
@@ -205,10 +176,10 @@ namespace ix
     }
 
     void CobraConnection::configure(const std::string& appkey,
-                                     const std::string& endpoint,
-                                     const std::string& rolename,
-                                     const std::string& rolesecret,
-                                     WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
+                                    const std::string& endpoint,
+                                    const std::string& rolename,
+                                    const std::string& rolesecret,
+                                    WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
     {
         _appkey = appkey;
         _endpoint = endpoint;
@@ -221,8 +192,8 @@ namespace ix
         ss << _appkey;
 
         std::string url = ss.str();
-        _webSocket->setUrl(url);
-        _webSocket->setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
+        _webSocket.setUrl(url);
+        _webSocket.setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
     }
 
     //
@@ -255,10 +226,10 @@ namespace ix
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
 
-        return _webSocket->send(serializedJson).success;
+        return _webSocket.send(serializedJson).success;
     }
 
-    // 
+    //
     // Extract the nonce from the handshake response
     // use it to compute a hash during authentication
     //
@@ -317,47 +288,16 @@ 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)
     {
         if (!pdu.isMember("body")) return false;
         Json::Value body = pdu["body"];
 
-        // Identify subscription_id, so that we can find 
+        // Identify subscription_id, so that we can find
         // which callback to execute
         if (!body.isMember("subscription_id")) return false;
         Json::Value subscriptionId = body["subscription_id"];
@@ -380,13 +320,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)
@@ -399,7 +339,7 @@ namespace ix
     // publish is not thread safe as we are trying to reuse some Json objects.
     //
     bool CobraConnection::publish(const Json::Value& channels,
-                                   const Json::Value& msg)
+                                  const Json::Value& msg)
     {
         _body["channels"] = channels;
         _body["message"] = msg;
@@ -431,7 +371,7 @@ namespace ix
     }
 
     void CobraConnection::subscribe(const std::string& channel,
-                                     SubscriptionCallback cb)
+                                    SubscriptionCallback cb)
     {
         // Create and send a subscribe pdu
         Json::Value body;
@@ -441,7 +381,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);
@@ -460,13 +400,13 @@ namespace ix
 
         // Create and send an unsubscribe pdu
         Json::Value body;
-        body["subscription_id"] = channel;
+        body["channel"] = channel;
 
         Json::Value pdu;
         pdu["action"] = "rtm/unsubscribe";
         pdu["body"] = body;
 
-        _webSocket->send(pdu.toStyledString());
+        _webSocket.send(pdu.toStyledString());
     }
 
     //
@@ -516,7 +456,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;
@@ -531,5 +471,5 @@ namespace ix
     {
         connect();
     }
-    
+
 } // namespace ix

ws/cobra_publisher/IXCobraConnection.h

@@ -11,24 +11,19 @@
 #include <string>
 #include <thread>
 #include <unordered_map>
-#include <memory>
 
 #include <jsoncpp/json/json.h>
-#include <ixwebsocket/IXWebSocketHttpHeaders.h>
+#include <ixwebsocket/IXWebSocket.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_Subscribed = 4,
-        CobraConnection_EventType_UnSubscribed = 5
+        CobraConnection_EventType_Closed = 3
     };
 
     enum CobraConnectionPublishMode
@@ -40,8 +35,7 @@ namespace ix
     using SubscriptionCallback = std::function<void(const Json::Value&)>;
     using EventCallback = std::function<void(CobraConnectionEventType,
                                              const std::string&,
-                                             const WebSocketHttpHeaders&,
-                                             const std::string&)>;
+                                             const WebSocketHttpHeaders&)>;
     using TrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
 
     class CobraConnection
@@ -90,7 +84,7 @@ namespace ix
 
         /// Returns true only if we're connected
         bool isConnected() const;
-        
+
         /// Flush the publish queue
         bool flushQueue();
 
@@ -106,8 +100,6 @@ 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();
 
@@ -121,15 +113,13 @@ namespace ix
         /// Invoke event callbacks
         void invokeEventCallback(CobraConnectionEventType eventType,
                                  const std::string& errorMsg = std::string(),
-                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders(),
-                                 const std::string& subscriptionId = std::string());
-        void invokeErrorCallback(const std::string& errorMsg,
-                                 const std::string& serializedPdu);
+                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders());
+        void invokeErrorCallback(const std::string& errorMsg);
 
         ///
         /// Member variables
-        /// 
-        std::unique_ptr<WebSocket> _webSocket;
+        ///
+        WebSocket _webSocket;
 
         /// Configuration data
         std::string _appkey;
@@ -158,10 +148,10 @@ namespace ix
         std::unordered_map<std::string, SubscriptionCallback> _cbs;
         mutable std::mutex _cbsMutex;
 
-        // Message Queue can be touched on control+background thread, 
+        // Message Queue can be touched on control+background thread,
         // protecting with a mutex.
         //
-        // Message queue is used when there are problems sending messages so 
+        // Message queue is used when there are problems sending messages so
         // that sending can be retried later.
         std::deque<std::string> _messageQueue;
         mutable std::mutex _queueMutex;
@@ -169,5 +159,5 @@ namespace ix
         // Cap the queue size (100 elems so far -> ~100k)
         static constexpr size_t kQueueMaxSize = 256;
     };
-    
+
 } // namespace ix

ws/cobra_publisher/README.md

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

ws/cobra_publisher/cobra_publisher.cpp

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