linux build fix

Stop lying about Windows support ...

CMakeLists.txt

@@ -20,6 +20,7 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXSocket.cpp
     ixwebsocket/IXSocketServer.cpp
     ixwebsocket/IXSocketConnect.cpp
+    ixwebsocket/IXSocketFactory.cpp
     ixwebsocket/IXDNSLookup.cpp
     ixwebsocket/IXCancellationRequest.cpp
     ixwebsocket/IXWebSocket.cpp
@@ -29,6 +30,9 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXWebSocketPerMessageDeflate.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
+    ixwebsocket/IXWebSocketHttpHeaders.cpp
+    ixwebsocket/IXHttpClient.cpp
+    ixwebsocket/IXUrlParser.cpp
 )
 
 set( IXWEBSOCKET_HEADERS
@@ -36,6 +40,7 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketServer.h
     ixwebsocket/IXSocketConnect.h
+    ixwebsocket/IXSocketFactory.h
     ixwebsocket/IXSetThreadName.h
     ixwebsocket/IXDNSLookup.h
     ixwebsocket/IXCancellationRequest.h
@@ -51,6 +56,8 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXWebSocketPerMessageDeflateOptions.h
     ixwebsocket/IXWebSocketHttpHeaders.h
     ixwebsocket/libwshandshake.hpp
+    ixwebsocket/IXHttpClient.h
+    ixwebsocket/IXUrlParser.h
 )
 
 # Platform specific code

Dockerfile

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

Dockerfile

@@ -0,0 +1,31 @@
+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
+RUN apt-get -y install curl
+RUN apt-get -y install python
+RUN apt-get -y install netcat
+
+# debian strech cmake is too old for building with Docker
+COPY makefile .
+RUN ["make", "install_cmake_for_linux"]
+
+COPY . .
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-rc4-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+# RUN ["make"]
+
+EXPOSE 8765
+CMD ["/ws/ws", "transfer", "--port", "8765", "--host", "0.0.0.0"]

README.md

@@ -5,17 +5,16 @@
 ## Introduction
 
 [*WebSocket*](https://en.wikipedia.org/wiki/WebSocket) is a computer communications protocol, providing full-duplex
-communication channels over a single TCP connection. *IXWebSocket* is a C++ library for client and server Websocket communication. The code is derived from [easywsclient](https://github.com/dhbaird/easywsclient) and from the [Satori C SDK](https://github.com/satori-com/satori-rtm-sdk-c). It has been tested on the following platforms.
+communication channels over a single TCP connection. *IXWebSocket* is a C++ library for client and server Websocket communication, and for client HTTP communication. The code is derived from [easywsclient](https://github.com/dhbaird/easywsclient) and from the [Satori C SDK](https://github.com/satori-com/satori-rtm-sdk-c). It has been tested on the following platforms.
 
 * macOS
 * iOS
 * Linux
 * Android 
-* Windows (no TLS support yet)
 
 ## Examples
 
-The ws folder countains many interactive programs for chat and file transfers demonstrating client and server usage.
+The [*ws*](https://github.com/machinezone/IXWebSocket/tree/master/ws) folder countains many interactive programs for chat, [file transfers](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_send.cpp), [curl like](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_http_client.cpp) http clients, demonstrating client and server usage.
 
 Here is what the client API looks like.
 
@@ -25,7 +24,7 @@ ix::WebSocket webSocket;
 std::string url("ws://localhost:8080/");
 webSocket.setUrl(url);
 
-// Optional heart beat, sent every 45 seconds when there isn't any traffic
+// Optional heart beat, sent every 45 seconds when there is not any traffic
 // to make sure that load balancers do not kill an idle connection.
 webSocket.setHeartBeatPeriod(45);
 
@@ -77,7 +76,10 @@ server.setOnConnectionCallback(
                 if (messageType == ix::WebSocket_MessageType_Open)
                 {
                     std::cerr << "New connection" << std::endl;
+
+                    // The uri the client did connect to.
                     std::cerr << "Uri: " << openInfo.uri << std::endl;
+
                     std::cerr << "Headers:" << std::endl;
                     for (auto it : openInfo.headers)
                     {
@@ -110,12 +112,81 @@ server.wait();
 
 ```
 
+Here is what the HTTP client API looks like. Note that HTTP client support is very recent and subject to changes.
+
+```
+//
+// Preparation
+//
+HttpClient httpClient;
+HttpRequestArgs args;
+
+// Custom headers can be set
+WebSocketHttpHeaders headers;
+headers["Foo"] = "bar";
+args.extraHeaders = parseHeaders(headersData);
+
+// Timeout options
+args.connectTimeout = connectTimeout;
+args.transferTimeout = transferTimeout;
+
+// Redirect options
+args.followRedirects = followRedirects;
+args.maxRedirects = maxRedirects;
+
+// Misc
+args.compress = compress; // Enable gzip compression
+args.verbose = verbose;
+args.logger = [](const std::string& msg)
+{
+    std::cout << msg;
+};
+
+//
+// Request
+//
+HttpResponse out;
+std::string url = "https://www.google.com";
+
+// HEAD request
+out = httpClient.head(url, args);
+
+// GET request
+out = httpClient.get(url, args);
+
+// POST request with parameters
+HttpParameters httpParameters;
+httpParameters["foo"] = "bar";
+out = httpClient.post(url, httpParameters, args);
+
+// POST request with a body
+out = httpClient.post(url, std::string("foo=bar"), args);
+
+//
+// Result
+//
+auto statusCode = std::get<0>(out);
+auto errorCode = std::get<1>(out);
+auto responseHeaders = std::get<2>(out);
+auto payload = std::get<3>(out);
+auto errorMsg = std::get<4>(out);
+auto uploadSize = std::get<5>(out);
+auto downloadSize = std::get<6>(out);
+```
+
 ## Build
 
 CMakefiles for the library and the examples are available. This library has few dependencies, so it is possible to just add the source files into your project.
 
 There is a Dockerfile for running some code on Linux, and a unittest which can be executed by typing `make test`.
 
+You can build and install the ws command line tool with Homebrew.
+
+```
+brew create --cmake https://github.com/machinezone/IXWebSocket/archive/v1.1.0.tar.gz
+brew install IXWebSocket
+```
+
 ## Implementation details
 
 ### Per Message Deflate compression.
@@ -142,11 +213,11 @@ Large frames are broken up into smaller chunks or messages to avoid filling up t
 
 * 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 isn't as scalable as strategies using epoll or kqueue.
+* The server code is using select to detect incoming data, and creates one OS thread per connection. This is not as scalable as strategies using epoll or kqueue.
 
 ## C++ code organization
 
-Here's a simplistic diagram which explains how the code is structured in term of class/modules.
+Here is a simplistic diagram which explains how the code is structured in term of class/modules.
 
 ```
 +-----------------------+ --- Public
@@ -198,7 +269,7 @@ If the connection was closed and sending failed, the return value will be set to
 1. WebSocket_ReadyState_Connecting - The connection is not yet open.
 2. WebSocket_ReadyState_Open       - The connection is open and ready to communicate.
 3. WebSocket_ReadyState_Closing    - The connection is in the process of closing.
-4. WebSocket_MessageType_Close     - The connection is closed or couldn't be opened.
+4. WebSocket_MessageType_Close     - The connection is closed or could not be opened.
 
 ### Open and Close notifications
 
@@ -308,7 +379,7 @@ websocket.ping("ping data, optional (empty string is ok): limited to 125 bytes l
 ### Heartbeat.
 
 You can configure an optional heart beat / keep-alive, sent every 45 seconds
-when there isn't any traffic to make sure that load balancers do not kill an
+when there is not any traffic to make sure that load balancers do not kill an
 idle connection.
 
 ```

docker/Dockerfile

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

docker/Dockerfile.alpine

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

docker/Dockerfile.centos

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

docker/Dockerfile.debian

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

docker/Dockerfile.gcc

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

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/IXEventFd.cpp

@@ -17,6 +17,8 @@
 // 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 "IXEventFd.h"
 
@@ -24,17 +26,29 @@
 # include <sys/eventfd.h>
 #endif
 
-#ifndef _WIN32
 #include <unistd.h> // for write
-#endif
+#include <fcntl.h>
 
 namespace ix
 {
-    EventFd::EventFd() :
-        _eventfd(-1)
+    // 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 EventFd::kPipeReadIndex = 0;
+    const int EventFd::kPipeWriteIndex = 1;
+
+    EventFd::EventFd()
     {
 #ifdef __linux__
+        _eventfd = -1;
         _eventfd = eventfd(0, 0);
+        fcntl(_eventfd, F_SETFL, O_NONBLOCK);
+#else
+        _fildes[kPipeReadIndex] = -1;
+        _fildes[kPipeWriteIndex] = -1;
+
+        pipe(_fildes);
+        fcntl(_fildes[kPipeReadIndex], F_SETFL, O_NONBLOCK);
+        fcntl(_fildes[kPipeWriteIndex], F_SETFL, O_NONBLOCK);
 #endif
     }
 
@@ -42,22 +56,43 @@ namespace ix
     {
 #ifdef __linux__
         ::close(_eventfd);
+#else
+        ::close(_fildes[kPipeReadIndex]);
+        ::close(_fildes[kPipeWriteIndex]);
+        _fildes[kPipeReadIndex] = -1;
+        _fildes[kPipeWriteIndex] = -1;
 #endif
     }
 
-    bool EventFd::notify()
+    bool EventFd::notify(uint64_t value)
     {
-#if defined(__linux__)
-        if (_eventfd == -1) return false;
+        int fd;
 
-        // select will wake up when a non-zero value is written to our eventfd
-        uint64_t value = 1;
+#if defined(__linux__)
+        fd = _eventfd;
+#else
+        fd = _fildes[kPipeWriteIndex];
+#endif
+
+        if (fd == -1) return false;
 
         // we should write 8 bytes for an uint64_t
-        return write(_eventfd, &value, sizeof(value)) == 8;
+        return write(fd, &value, sizeof(value)) == 8;
+    }
+
+    // TODO: return max uint64_t for errors ?
+    uint64_t EventFd::read()
+    {
+        int fd;
+
+#if defined(__linux__)
+        fd = _eventfd;
 #else
-        return true;
+        fd = _fildes[kPipeReadIndex];
 #endif
+        uint64_t value = 0;
+        ::read(fd, &value, sizeof(value));
+        return value;
     }
 
     bool EventFd::clear()
@@ -77,6 +112,10 @@ namespace ix
 
     int EventFd::getFd()
     {
+#if defined(__linux__)
         return _eventfd;
+#else
+        return _fildes[kPipeReadIndex];
+#endif
     }
 }

ixwebsocket/IXEventFd.h

@@ -6,6 +6,8 @@
 
 #pragma once
 
+#include <stdint.h>
+
 namespace ix
 {
     class EventFd {
@@ -13,11 +15,23 @@ namespace ix
         EventFd();
         virtual ~EventFd();
 
-        bool notify();
+        bool notify(uint64_t value);
         bool clear();
+        uint64_t read();
         int getFd();
 
     private:
+#if defined(__linux__)
         int _eventfd;
+#else
+        // 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];
+#endif
+
+        // Used to identify the read/write idx
+        static const int kPipeReadIndex;
+        static const int kPipeWriteIndex;
     };
 }

ixwebsocket/IXHttpClient.cpp

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

ixwebsocket/IXHttpClient.h

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

ixwebsocket/IXSocket.cpp

@@ -23,11 +23,14 @@ 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)
     {
-
+        ;
     }
 
     Socket::~Socket()
@@ -39,26 +42,38 @@ namespace ix
     {
         if (_sockfd == -1)
         {
-            onPollCallback(PollResultType_Error);
+            if (onPollCallback) onPollCallback(PollResultType_Error);
             return;
         }
 
+        PollResultType pollResult = select(timeoutSecs, 0);
+
+        if (onPollCallback) onPollCallback(pollResult);
+    }
+
+    PollResultType Socket::select(int timeoutSecs, int timeoutMs)
+    {
         fd_set rfds;
         FD_ZERO(&rfds);
         FD_SET(_sockfd, &rfds);
 
-#ifdef __linux__
-        FD_SET(_eventfd.getFd(), &rfds);
-#endif
+        // File descriptor at index 0 in _fildes is the read end of the pipe
+        int eventfd = _eventfd.getFd();
+        if (eventfd != -1)
+        {
+            FD_SET(eventfd, &rfds);
+        }
 
         struct timeval timeout;
         timeout.tv_sec = timeoutSecs;
-        timeout.tv_usec = 0;
+        timeout.tv_usec = 1000 * timeoutMs;
 
+        // Compute the highest fd.
         int sockfd = _sockfd;
-        int nfds = (std::max)(sockfd, _eventfd.getFd());
-        int ret = select(nfds + 1, &rfds, nullptr, nullptr,
-                         (timeoutSecs < 0) ? nullptr : &timeout);
+        int nfds = (std::max)(sockfd, eventfd);
+
+        int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
+                           (timeoutSecs < 0) ? nullptr : &timeout);
 
         PollResultType pollResult = PollResultType_ReadyForRead;
         if (ret < 0)
@@ -69,14 +84,27 @@ namespace ix
         {
             pollResult = PollResultType_Timeout;
         }
+        else if (eventfd != -1 && FD_ISSET(eventfd, &rfds))
+        {
+            uint64_t value = _eventfd.read();
 
-        onPollCallback(pollResult);
+            if (value == kSendRequest)
+            {
+                pollResult = PollResultType_SendRequest;
+            }
+            else if (value == kCloseRequest)
+            {
+                pollResult = PollResultType_CloseRequest;
+            }
+        }
+
+        return pollResult;
     }
 
-    void Socket::wakeUpFromPoll()
+    // Wake up from poll/select by writing to the pipe which is watched by select
+    bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
     {
-        // this will wake up the thread blocked on select, only needed on Linux
-        _eventfd.notify();
+        return _eventfd.notify(wakeUpCode);
     }
 
     bool Socket::connect(const std::string& host,
@@ -165,51 +193,6 @@ namespace ix
 #endif
     }
 
-    bool Socket::readByte(void* buffer,
-                          const CancellationRequest& isCancellationRequested)
-    {
-        while (true)
-        {
-            if (isCancellationRequested()) return false;
-
-            ssize_t ret;
-            ret = recv(buffer, 1);
-
-            // We read one byte, as needed, all good.
-            if (ret == 1)
-            {
-                return true;
-            }
-            // There is possibly something to be read, try again
-            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
-                                 getErrno() == EAGAIN))
-            {
-                // Wait with a timeout until something is written.
-                // This way we are not busy looping
-                fd_set rfds;
-                struct timeval timeout;
-                timeout.tv_sec = 0;
-                timeout.tv_usec = 1 * 1000; // 1ms timeout
-
-                FD_ZERO(&rfds);
-                FD_SET(_sockfd, &rfds);
-
-                if (select(_sockfd + 1, &rfds, nullptr, nullptr, &timeout) < 0 &&
-                    (errno == EBADF || errno == EINVAL))
-                {
-                    return false;
-                }
-
-                continue;
-            }
-            // There was an error during the read, abort
-            else
-            {
-                return false;
-            }
-        }
-    }
-
     bool Socket::writeBytes(const std::string& str,
                             const CancellationRequest& isCancellationRequested)
     {
@@ -241,7 +224,43 @@ namespace ix
         }
     }
 
-    std::pair<bool, std::string> Socket::readLine(const CancellationRequest& isCancellationRequested)
+    bool Socket::readByte(void* buffer,
+                          const CancellationRequest& isCancellationRequested)
+    {
+        while (true)
+        {
+            if (isCancellationRequested()) return false;
+
+            ssize_t ret;
+            ret = recv(buffer, 1);
+
+            // We read one byte, as needed, all good.
+            if (ret == 1)
+            {
+                return true;
+            }
+            // There is possibly something to be read, try again
+            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
+                                 getErrno() == EAGAIN))
+            {
+                // Wait with a timeout until something is ready to read.
+                // This way we are not busy looping
+                int res = select(0, 1);
+                if (res < 0 && (errno == EBADF || errno == EINVAL))
+                {
+                    return false;
+                }
+            }
+            // There was an error during the read, abort
+            else
+            {
+                return false;
+            }
+        }
+    }
+
+    std::pair<bool, std::string> Socket::readLine(
+        const CancellationRequest& isCancellationRequested)
     {
         char c;
         std::string line;
@@ -251,7 +270,8 @@ namespace ix
         {
             if (!readByte(&c, isCancellationRequested))
             {
-                return std::make_pair(false, std::string());
+                // Return what we were able to read
+                return std::make_pair(false, line);
             }
 
             line += c;
@@ -259,4 +279,46 @@ 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()) return std::make_pair(false, std::string());
+
+            int 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 timeout until something is ready to read.
+            // This way we are not busy looping
+            select(0, 1);
+        }
+
+        return std::make_pair(true, std::string(output.begin(),
+                                                output.end()));
+    }
 }

ixwebsocket/IXSocket.h

@@ -10,14 +10,16 @@
 #include <functional>
 #include <mutex>
 #include <atomic>
+#include <vector>
 
 #ifdef _WIN32
 #include <BaseTsd.h>
 typedef SSIZE_T ssize_t;
 #endif
 
-#include "IXEventFd.h"
 #include "IXCancellationRequest.h"
+#include "IXProgressCallback.h"
+#include "IXEventFd.h"
 
 namespace ix
 {
@@ -25,7 +27,9 @@ namespace ix
     {
         PollResultType_ReadyForRead = 0,
         PollResultType_Timeout = 1,
-        PollResultType_Error = 2
+        PollResultType_Error = 2,
+        PollResultType_SendRequest = 3,
+        PollResultType_CloseRequest = 4
     };
 
     class Socket {
@@ -37,9 +41,10 @@ namespace ix
 
         void configure();
 
+        PollResultType select(int timeoutSecs, int timeoutMs);
         virtual void poll(const OnPollCallback& onPollCallback,
                           int timeoutSecs = kDefaultPollTimeout);
-        virtual void wakeUpFromPoll();
+        virtual bool wakeUpFromPoll(uint8_t wakeUpCode);
 
         // Virtual methods
         virtual bool connect(const std::string& url,
@@ -58,21 +63,36 @@ namespace ix
                       const CancellationRequest& isCancellationRequested);
         bool writeBytes(const std::string& str,
                         const CancellationRequest& isCancellationRequested);
-        std::pair<bool, std::string> readLine(const CancellationRequest& isCancellationRequested);
+
+        std::pair<bool, std::string> readLine(
+            const CancellationRequest& isCancellationRequested);
+        std::pair<bool, std::string> readBytes(
+            size_t length,
+            const OnProgressCallback& onProgressCallback,
+            const CancellationRequest& isCancellationRequested);
 
         static int getErrno();
         static bool init(); // Required on Windows to initialize WinSocket
         static void cleanup(); // Required on Windows to cleanup WinSocket
 
+        // Used as special codes for pipe communication
+        static const uint64_t kSendRequest;
+        static const uint64_t kCloseRequest;
+
     protected:
         void closeSocket(int fd);
 
         std::atomic<int> _sockfd;
         std::mutex _socketMutex;
-        EventFd _eventfd;
 
     private:
         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;
+
+        EventFd _eventfd;
     };
 }

ixwebsocket/IXSocketFactory.cpp

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

ixwebsocket/IXSocketFactory.h

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

ixwebsocket/IXUrlParser.cpp

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

ixwebsocket/IXUrlParser.h

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

ixwebsocket/IXWebSocket.cpp

@@ -252,6 +252,11 @@ namespace ix
                         {
                             webSocketMessageType = WebSocket_MessageType_Pong;
                         } break;
+
+                        case WebSocketTransport::FRAGMENT:
+                        {
+                            webSocketMessageType = WebSocket_MessageType_Fragment;
+                        } break;
                     }
 
                     WebSocketErrorInfo webSocketErrorInfo;
@@ -374,4 +379,9 @@ namespace ix
     {
         _automaticReconnection = false;
     }
+
+    size_t WebSocket::bufferedAmount() const
+    {
+        return _ws.bufferedAmount();
+    }
 }

ixwebsocket/IXWebSocket.h

@@ -39,7 +39,8 @@ namespace ix
         WebSocket_MessageType_Close = 2,
         WebSocket_MessageType_Error = 3,
         WebSocket_MessageType_Ping = 4,
-        WebSocket_MessageType_Pong = 5
+        WebSocket_MessageType_Pong = 5,
+        WebSocket_MessageType_Fragment = 6
     };
 
     struct WebSocketOpenInfo
@@ -111,6 +112,7 @@ namespace ix
         const std::string& getUrl() const;
         const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
         int getHeartBeatPeriod() const;
+        size_t bufferedAmount() const;
 
         void enableAutomaticReconnection();
         void disableAutomaticReconnection();

ixwebsocket/IXWebSocketHandshake.cpp

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

ixwebsocket/IXWebSocketHandshake.h

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

ixwebsocket/IXWebSocketHttpHeaders.cpp

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

ixwebsocket/IXWebSocketHttpHeaders.h

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

ixwebsocket/IXWebSocketTransport.cpp

@@ -1,7 +1,31 @@
+/*
+ * The MIT License (MIT)
+ * 
+ * Copyright (c) 2012, 2013 <dhbaird@gmail.com>
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
 /*
  *  IXWebSocketTransport.cpp
  *  Author: Benjamin Sergeant
- *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
  */
 
 //
@@ -11,14 +35,8 @@
 #include "IXWebSocketTransport.h"
 #include "IXWebSocketHandshake.h"
 #include "IXWebSocketHttpHeaders.h"
-
-#ifdef IXWEBSOCKET_USE_TLS
-# ifdef __APPLE__
-#  include "IXSocketAppleSSL.h"
-# else
-#  include "IXSocketOpenSSL.h"
-# endif
-#endif
+#include "IXUrlParser.h"
+#include "IXSocketFactory.h"
 
 #include <string.h>
 #include <stdlib.h>
@@ -70,31 +88,21 @@ namespace ix
     {
         std::string protocol, host, path, query;
         int port;
+        bool websocket = true;
 
-        if (!WebSocketHandshake::parseUrl(url, protocol, host,
-                                          path, query, port))
+        if (!UrlParser::parse(url, protocol, host, path, query, port, websocket))
         {
             return WebSocketInitResult(false, 0,
                                        std::string("Could not parse URL ") + url);
         }
 
-        if (protocol == "wss")
+        bool tls = protocol == "wss";
+        std::string errorMsg;
+        _socket = createSocket(tls, errorMsg);
+
+        if (!_socket)
         {
-            _socket.reset();
-#ifdef IXWEBSOCKET_USE_TLS
-# ifdef __APPLE__
-             _socket = std::make_shared<SocketAppleSSL>();
-# else
-             _socket = std::make_shared<SocketOpenSSL>();
-# endif
-#else
-            return WebSocketInitResult(false, 0, "TLS is not supported.");
-#endif
-        }
-        else
-        {
-            _socket.reset();
-            _socket = std::make_shared<Socket>();
+            return WebSocketInitResult(false, 0, errorMsg);
         }
 
         WebSocketHandshake webSocketHandshake(_requestInitCancellation,
@@ -182,38 +190,57 @@ namespace ix
                     std::stringstream ss;
                     ss << kHeartBeatPingMessage << "::" << _heartBeatPeriod << "s";
                     sendPing(ss.str());
-                    return;
                 }
-
-                while (true)
+                // Make sure we send all the buffered data
+                // there can be a lot of it for large messages.
+                else if (pollResult == PollResultType_SendRequest)
                 {
-                    ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
+                    while (!isSendBufferEmpty() && !_requestInitCancellation)
+                    {
+                        sendOnSocket();
 
-                    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);
+                        // Sleep 10ms between each send so that we dont busy loop
+                        // A better strategy would be to select on the socket to 
+                        // check whether we can write to it without blocking
+                        std::chrono::duration<double, std::micro> duration(10);
+                        std::this_thread::sleep_for(duration);
                     }
                 }
+                else if (pollResult == PollResultType_ReadyForRead)
+                {
+                    while (true)
+                    {
+                        ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
 
-                if (isSendBufferEmpty() && _readyState == CLOSING)
+                        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();
-                    setReadyState(CLOSED);
                 }
+                else if (pollResult == PollResultType_CloseRequest)
+                {
+                    ;
+                }
+
             },
             _heartBeatPeriod);
     }
@@ -390,6 +417,10 @@ namespace ix
                         emitMessage(MSG, getMergedChunks(), ws, onMessageCallback);
                         _chunks.clear();
                     }
+                    else
+                    {
+                        emitMessage(FRAGMENT, std::string(), ws, onMessageCallback);
+                    }
                 }
             }
             else if (ws.opcode == wsheader_type::PING)
@@ -473,7 +504,7 @@ namespace ix
         size_t wireSize = message.size();
 
         // When the RSV1 bit is 1 it means the message is compressed
-        if (_enablePerMessageDeflate && ws.rsv1)
+        if (_enablePerMessageDeflate && ws.rsv1 && messageKind != FRAGMENT)
         {
             std::string decompressedMessage;
             bool success = _perMessageDeflate.decompress(message, decompressedMessage);
@@ -571,7 +602,7 @@ namespace ix
                 // Send message
                 sendFragment(opcodeType, fin, begin, end, compress);
 
-                if (onProgressCallback && !onProgressCallback(i, steps))
+                if (onProgressCallback && !onProgressCallback((int)i, (int) steps))
                 {
                     break;
                 }
@@ -580,6 +611,12 @@ namespace ix
             }
         }
 
+        // Request to flush the send buffer on the background thread if it isn't empty
+        if (!isSendBufferEmpty())
+        {
+            _socket->wakeUpFromPoll(Socket::kSendRequest);
+        }
+
         return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
     }
 
@@ -725,8 +762,18 @@ namespace ix
         sendData(wsheader_type::CLOSE, normalClosure, compress);
         setReadyState(CLOSING);
 
-        _socket->wakeUpFromPoll();
+        _socket->wakeUpFromPoll(Socket::kCloseRequest);
         _socket->close();
+
+        _closeCode = 1000;
+        _closeReason = "Normal Closure";
+        setReadyState(CLOSED);
+    }
+
+    size_t WebSocketTransport::bufferedAmount() const
+    {
+        std::lock_guard<std::mutex> lock(_txbufMutex);
+        return _txbuf.size();
     }
 
 } // namespace ix

ixwebsocket/IXWebSocketTransport.h

@@ -45,7 +45,8 @@ namespace ix
         {
             MSG,
             PING,
-            PONG
+            PONG,
+            FRAGMENT
         };
 
         using OnMessageCallback = std::function<void(const std::string&,
@@ -76,6 +77,7 @@ namespace ix
         void setReadyState(ReadyStateValues readyStateValue);
         void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
         void dispatch(const OnMessageCallback& onMessageCallback);
+        size_t bufferedAmount() const;
 
     private:
         std::string _url;

makefile

@@ -8,14 +8,14 @@ brew:
 
 .PHONY: docker
 docker:
-	docker build -t broadcast_server:latest .
+	docker build -t ws:latest .
 
 run:
-	docker run --cap-add sys_ptrace -it broadcast_server:latest bash
+	docker run --cap-add sys_ptrace -it ws:latest
 
 # this is helpful to remove trailing whitespaces
 trail:
-	sh third_party/remove_trailing_whitespaces.sh
+	sh third_party/remote_trailing_whitespaces.sh
 
 build:
 	(cd examples/satori_publisher ; mkdir -p build ; cd build ; cmake .. ; make)
@@ -36,6 +36,9 @@ test_server:
 test:
 	python test/run.py
 
+ws_test:
+	(cd ws ; sh test_ws.sh)
+
 # For the fork that is configured with appveyor
 rebase_upstream:
 	git fetch upstream
@@ -43,5 +46,9 @@ rebase_upstream:
 	git reset --hard upstream/master
 	git push origin master --force
 
+install_cmake_for_linux:
+	mkdir -p /tmp/cmake
+	(cd /tmp/cmake ; curl -L -O https://github.com/Kitware/CMake/releases/download/v3.14.0-rc4/cmake-3.14.0-rc4-Linux-x86_64.tar.gz ; tar zxf cmake-3.14.0-rc4-Linux-x86_64.tar.gz)
+
 .PHONY: test
 .PHONY: build

test/IXSocketTest.cpp

@@ -18,6 +18,7 @@
 
 #include "IXTest.h"
 #include "catch.hpp"
+#include <string.h>
 
 using namespace ix;
 
@@ -65,7 +66,13 @@ TEST_CASE("socket", "[socket]")
         std::shared_ptr<Socket> socket(new Socket);
         std::string host("www.google.com");
         int port = 80;
-        std::string request("GET / HTTP/1.1\r\n\r\n");
+
+        std::stringstream ss;
+        ss << "GET / HTTP/1.1\r\n";
+        ss << "Host: " << host << "\r\n";
+        ss << "\r\n";
+        std::string request(ss.str());
+
         int expectedStatus = 200;
         int timeoutSecs = 3;
 

test/IXTest.cpp

@@ -69,10 +69,15 @@ namespace ix
         Logger() << msg;
     }
 
+    int getAnyFreePortSimple()
+    {
+        static int defaultPort = 8090;
+        return defaultPort++;
+    }
+
     int getAnyFreePort()
     {
         int defaultPort = 8090;
-
         int sockfd;
         if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
         {
@@ -122,8 +127,15 @@ namespace ix
     {
         while (true)
         {
+#if defined(__has_feature)
+# if __has_feature(address_sanitizer)
+            int port = getAnyFreePortSimple();
+# else
             int port = getAnyFreePort();
-
+# endif
+#else
+            int port = getAnyFreePort();
+#endif
             //
             // Only port above 1024 can be used by non root users, but for some
             // reason I got port 7 returned with macOS when binding on port 0...

test/cmd_websocket_chat.cpp

@@ -164,10 +164,21 @@ namespace
                     ss << "cmd_websocket_chat: Error ! " << error.reason;
                     log(ss.str());
                 }
+                else if (messageType == ix::WebSocket_MessageType_Ping)
+                {
+                    log("cmd_websocket_chat: received ping message");
+                }
+                else if (messageType == ix::WebSocket_MessageType_Pong)
+                {
+                    log("cmd_websocket_chat: received pong message");
+                }
+                else if (messageType == ix::WebSocket_MessageType_Fragment)
+                {
+                    log("cmd_websocket_chat: received message fragment");
+                }
                 else
                 {
-                    // FIXME: missing ping/pong messages
-                    ss << "Invalid ix::WebSocketMessageType";
+                    ss << "Unexpected ix::WebSocketMessageType";
                     log(ss.str());
                 }
             });

test/run.py

@@ -6,10 +6,10 @@ osName = platform.system()
 print('os name = {}'.format(osName))
 
 root = os.path.dirname(os.path.realpath(__file__))
-buildDir = os.path.join(root, 'build')
+buildDir = os.path.join(root, 'build', osName)
 
 if not os.path.exists(buildDir):
-    os.mkdir(buildDir)
+    os.makedirs(buildDir)
 
 os.chdir(buildDir)
 
@@ -38,7 +38,7 @@ sanitizerFlags = sanitizersFlags[sanitizer]
 #     os.environ['CC'] = 'clang-cl'
 #     os.environ['CXX'] = 'clang-cl'
 
-cmakeCmd = 'cmake -DCMAKE_BUILD_TYPE=Debug {} {} ..'.format(generator, sanitizerFlags)
+cmakeCmd = 'cmake -DCMAKE_BUILD_TYPE=Debug {} {} ../..'.format(generator, sanitizerFlags)
 print(cmakeCmd)
 ret = os.system(cmakeCmd)
 assert ret == 0, 'CMake failed, exiting'
@@ -67,6 +67,7 @@ def findFiles(prefix):
 
 # We need to copy the zlib DLL in the current work directory
 shutil.copy(os.path.join(
+    '..',
     '..',
     '..',
     'third_party',
@@ -77,6 +78,8 @@ shutil.copy(os.path.join(
     'bin',
     'zlib.dll'), '.')
 
-testCommand = '{} {}'.format(testBinary, os.getenv('TEST', ''))
+lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
+lldb = ""  # Disabled for now
+testCommand = '{} {} {}'.format(lldb, testBinary, os.getenv('TEST', ''))
 ret = os.system(testCommand)
 assert ret == 0, 'Test command failed'

third_party/homebrew_formula.rb

@@ -0,0 +1,20 @@
+class Ixwebsocket < Formula
+  desc "WebSocket client and server, and HTTP client command-line tool"
+  homepage "https://github.com/machinezone/IXWebSocket"
+  url "https://github.com/machinezone/IXWebSocket/archive/v1.1.0.tar.gz"
+  sha256 "52592ce3d0a67ad0f90ac9e8a458f61724175d95a01a38d1bad3fcdc5c7b6666"
+  depends_on "cmake" => :build
+
+  def install
+    system "cmake", ".", *std_cmake_args
+    system "make", "install"
+  end
+
+  test do
+    system "#{bin}/ws", "--help"
+    system "#{bin}/ws", "send", "--help"
+    system "#{bin}/ws", "receive", "--help"
+    system "#{bin}/ws", "transfer", "--help"
+    system "#{bin}/ws", "curl", "--help"
+  end
+end

ws/CMakeLists.txt

@@ -23,6 +23,7 @@ add_executable(ws
   ixcrypto/IXHash.cpp
   ixcrypto/IXUuid.cpp
 
+  ws_http_client.cpp
   ws_ping_pong.cpp
   ws_broadcast_server.cpp
   ws_echo_server.cpp

ws/README.md

@@ -1,10 +1,62 @@
+# 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
+```
+
+## file transfer
+
 ```
 # Start transfer server, which is just a broadcast server at this point
-./ws transfer # running on port 8080.
+ws transfer # running on port 8080.
 
 # Start receiver first
-./ws receive ws://localhost:8080 
+ws receive ws://localhost:8080 
 
 # Then send a file. File will be received and written to disk by the receiver process
-./ws send ws://localhost:8080 /file/to/path
+ws send ws://localhost:8080 /file/to/path
+```
+
+## curl
+
+```
+$ ws curl --help
+HTTP Client
+Usage: ws curl [OPTIONS] url
+
+Positionals:
+  url TEXT REQUIRED           Connection url
+
+Options:
+  -h,--help                   Print this help message and exit
+  -d TEXT                     Form data
+  -F TEXT                     Form data
+  -H TEXT                     Header
+  --output TEXT               Output file
+  -I                          Send a HEAD request
+  -L                          Follow redirects
+  --max-redirects INT         Max Redirects
+  -v                          Verbose
+  -O                          Save output to disk
+  --compress                  Enable gzip compression
+  --connect-timeout INT       Connection timeout
+  --transfer-timeout INT      Transfer timeout
 ```

ws/docker_build.sh

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

ws/ixcrypto/IXHash.h

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

ws/test_ws.sh

@@ -0,0 +1,52 @@
+#!/bin/sh
+
+rm -rf /tmp/ws_test
+mkdir -p /tmp/ws_test
+
+# Start a transport server
+cd /tmp/ws_test
+ws transfer --port 8090 --pidfile /tmp/ws_test/pidfile &
+
+# Wait until the transfer server is up 
+while true
+do
+    nc -zv 127.0.0.1 8090 && {
+        echo "Transfer server up and running"
+        break
+    }
+    echo "sleep ..."
+    sleep 0.1
+done
+
+# Start a receiver
+mkdir -p /tmp/ws_test/receive
+cd /tmp/ws_test/receive
+ws receive ws://127.0.0.1:8090 &
+
+mkdir /tmp/ws_test/send
+cd /tmp/ws_test/send
+# mkfile 10m 10M_file
+dd if=/dev/urandom of=10M_file count=10000 bs=1024
+
+# Start the sender job
+ws send ws://127.0.0.1:8090 10M_file
+
+# Wait until the file has been written to disk
+while true
+do
+    if test -f /tmp/ws_test/receive/10M_file ; then
+        echo "Received file does exists, exiting loop"
+        break
+    fi
+    echo "sleep ..."
+    sleep 0.1
+done
+
+cksum /tmp/ws_test/send/10M_file
+cksum /tmp/ws_test/receive/10M_file
+
+# Give some time to ws receive to terminate
+sleep 2
+
+# Cleanup
+kill `cat /tmp/ws_test/pidfile`

ws/ws.cpp

@@ -1,9 +1,14 @@
 /*
  *  ws.cpp
  *  Author: Benjamin Sergeant
- *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
  */
 
+//
+// Main driver for websocket utilities
+//
+#include "ws.h"
+
 //
 // Main drive for websocket utilities
 //
@@ -11,32 +16,12 @@
 #include <string>
 #include <sstream>
 #include <iostream>
+#include <fstream>
+#include <unistd.h>
 
 #include <cli11/CLI11.hpp>
 #include <ixwebsocket/IXSocket.h>
 
-namespace ix
-{
-    int ws_ping_pong_main(const std::string& url);
-
-    int ws_echo_server_main(int port);
-
-    int ws_broadcast_server_main(int port);
-
-    int ws_chat_main(const std::string& url,
-                     const std::string& user);
-
-    int ws_connect_main(const std::string& url);
-
-    int ws_receive_main(const std::string& url,
-                        bool enablePerMessageDeflate);
-
-    int ws_transfer_main(int port);
-
-    int ws_send_main(const std::string& url,
-                     const std::string& path);
-}
-
 int main(int argc, char** argv)
 {
     CLI::App app{"ws is a websocket tool"};
@@ -45,17 +30,32 @@ int main(int argc, char** argv)
     std::string url("ws://127.0.0.1:8080");
     std::string path;
     std::string user;
+    std::string data;
+    std::string headers;
+    std::string output;
+    std::string hostname("127.0.0.1");
+    std::string pidfile;
+    bool headersOnly = false;
+    bool followRedirects = false;
+    bool verbose = false;
+    bool save = false;
+    bool compress = false;
     int port = 8080;
+    int connectTimeOut = 60;
+    int transferTimeout = 1800;
+    int maxRedirects = 5;
 
     CLI::App* sendApp = app.add_subcommand("send", "Send a file");
     sendApp->add_option("url", url, "Connection url")->required();
-    sendApp->add_option("path", path, "Path to the file to send")->required();
-
+    sendApp->add_option("path", path, "Path to the file to send")
+        ->required()->check(CLI::ExistingPath);
     CLI::App* receiveApp = app.add_subcommand("receive", "Receive a file");
     receiveApp->add_option("url", url, "Connection url")->required();
 
     CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
     transferApp->add_option("--port", port, "Connection url");
+    transferApp->add_option("--host", hostname, "Hostname");
+    transferApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
     connectApp->add_option("url", url, "Connection url")->required();
@@ -65,21 +65,50 @@ int main(int argc, char** argv)
     chatApp->add_option("user", user, "User name")->required();
 
     CLI::App* echoServerApp = app.add_subcommand("echo_server", "Echo server");
-    echoServerApp->add_option("--port", port, "Connection url");
+    echoServerApp->add_option("--port", port, "Port");
+    echoServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
-    broadcastServerApp->add_option("--port", port, "Connection url");
+    broadcastServerApp->add_option("--port", port, "Port");
+    broadcastServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* pingPongApp = app.add_subcommand("ping", "Ping pong");
     pingPongApp->add_option("url", url, "Connection url")->required();
 
+    CLI::App* httpClientApp = app.add_subcommand("curl", "HTTP Client");
+    httpClientApp->add_option("url", url, "Connection url")->required();
+    httpClientApp->add_option("-d", data, "Form data")->join();
+    httpClientApp->add_option("-F", data, "Form data")->join();
+    httpClientApp->add_option("-H", headers, "Header")->join();
+    httpClientApp->add_option("--output", output, "Output file");
+    httpClientApp->add_flag("-I", headersOnly, "Send a HEAD request");
+    httpClientApp->add_flag("-L", followRedirects, "Follow redirects");
+    httpClientApp->add_option("--max-redirects", maxRedirects, "Max Redirects");
+    httpClientApp->add_flag("-v", verbose, "Verbose");
+    httpClientApp->add_flag("-O", save, "Save output to disk");
+    httpClientApp->add_flag("--compress", compress, "Enable gzip compression");
+    httpClientApp->add_option("--connect-timeout", connectTimeOut, "Connection timeout");
+    httpClientApp->add_option("--transfer-timeout", transferTimeout, "Transfer timeout");
+
     CLI11_PARSE(app, argc, argv);
 
     ix::Socket::init();
 
+    // pid file handling
+
     if (app.got_subcommand("transfer"))
     {
-        return ix::ws_transfer_main(port);
+        if (!pidfile.empty())
+        {
+            unlink(pidfile.c_str());
+
+            std::ofstream f;
+            f.open(pidfile);
+            f << getpid();
+            f.close();
+        }
+
+        return ix::ws_transfer_main(port, hostname);
     }
     else if (app.got_subcommand("send"))
     {
@@ -100,19 +129,22 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("echo_server"))
     {
-        return ix::ws_echo_server_main(port);
+        return ix::ws_echo_server_main(port, hostname);
     }
     else if (app.got_subcommand("broadcast_server"))
     {
-        return ix::ws_broadcast_server_main(port);
+        return ix::ws_broadcast_server_main(port, hostname);
     }
     else if (app.got_subcommand("ping"))
     {
         return ix::ws_ping_pong_main(url);
     }
-    else
+    else if (app.got_subcommand("curl"))
     {
-        assert(false);
+        return ix::ws_http_client_main(url, headers, data, headersOnly,
+                                       connectTimeOut, transferTimeout,
+                                       followRedirects, maxRedirects, verbose,
+                                       save, output, compress);
     }
 
     return 1;

ws/ws.h

@@ -0,0 +1,41 @@
+/*
+ *  ws.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+#pragma once
+
+#include <string>
+
+namespace ix
+{
+    int ws_http_client_main(const std::string& url,
+                            const std::string& headers,
+                            const std::string& data,
+                            bool headersOnly,
+                            int connectTimeout,
+                            int transferTimeout,
+                            bool followRedirects,
+                            int maxRedirects,
+                            bool verbose,
+                            bool save,
+                            const std::string& output,
+                            bool compress);
+
+    int ws_ping_pong_main(const std::string& url);
+
+    int ws_echo_server_main(int port, const std::string& hostname);
+    int ws_broadcast_server_main(int port, const std::string& hostname);
+    int ws_transfer_main(int port, const std::string& hostname);
+
+    int ws_chat_main(const std::string& url,
+                     const std::string& user);
+
+    int ws_connect_main(const std::string& url);
+
+    int ws_receive_main(const std::string& url,
+                        bool enablePerMessageDeflate);
+
+    int ws_send_main(const std::string& url,
+                     const std::string& path);
+}

ws/ws_broadcast_server.cpp

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

ws/ws_chat.cpp

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

ws/ws_connect.cpp

@@ -84,6 +84,8 @@ namespace ix
                 }
                 else if (messageType == ix::WebSocket_MessageType_Message)
                 {
+                    std::cerr << "Received " << wireSize << " bytes" << std::endl;
+
                     ss << "ws_connect: received message: "
                        << str;
                     log(ss.str());

ws/ws_echo_server.cpp

@@ -10,17 +10,17 @@
 
 namespace ix
 {
-    int ws_echo_server_main(int port)
+    int ws_echo_server_main(int port, const std::string& hostname)
     {
-        std::cout << "Listening on port " << port << std::endl;
+        std::cout << "Listening on " << hostname << ":" << port << std::endl;
 
-        ix::WebSocketServer server(port);
+        ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket)
+            [](std::shared_ptr<ix::WebSocket> webSocket)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, &server](ix::WebSocketMessageType messageType,
+                    [webSocket](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -39,7 +39,18 @@ namespace ix
                         }
                         else if (messageType == ix::WebSocket_MessageType_Close)
                         {
-                            std::cerr << "Closed connection" << std::endl;
+                            std::cerr << "Closed connection"
+                                      << " code " << closeInfo.code
+                                      << " reason " << closeInfo.reason << std::endl;
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Error)
+                        {
+                            std::stringstream ss;
+                            ss << "Connection error: " << error.reason      << std::endl;
+                            ss << "#retries: "         << error.retries     << std::endl;
+                            ss << "Wait time(ms): "    << error.wait_time   << std::endl;
+                            ss << "HTTP Status: "      << error.http_status << std::endl;
+                            std::cerr << ss.str();
                         }
                         else if (messageType == ix::WebSocket_MessageType_Message)
                         {

ws/ws_http_client.cpp

@@ -0,0 +1,191 @@
+/*
+ *  http_client.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <iostream>
+#include <sstream>
+#include <fstream>
+#include <ixwebsocket/IXHttpClient.h>
+#include <ixwebsocket/IXWebSocketHttpHeaders.h>
+
+namespace ix
+{
+    std::string extractFilename(const std::string& path)
+    {
+        std::string::size_type idx;
+
+        idx = path.rfind('/');
+        if (idx != std::string::npos)
+        {
+            std::string filename = path.substr(idx+1);
+            return filename;
+        }
+        else
+        {
+            return path;
+        }
+    }
+
+    WebSocketHttpHeaders parseHeaders(const std::string& data)
+    {
+        WebSocketHttpHeaders headers;
+
+        // Split by \n
+        std::string token;
+        std::stringstream tokenStream(data);
+
+        while (std::getline(tokenStream, token))
+        {
+            std::size_t pos = token.rfind(':');
+
+            // Bail out if last '.' is found
+            if (pos == std::string::npos) continue;
+
+            auto key = token.substr(0, pos);
+            auto val = token.substr(pos+2);
+
+            std::cerr << key << ": " << val << std::endl;
+            headers[key] = val;
+        }
+
+        return headers;
+    }
+
+    //
+    // Useful endpoint to test HTTP post
+    // https://postman-echo.com/post
+    //
+    HttpParameters parsePostParameters(const std::string& data)
+    {
+        HttpParameters httpParameters;
+
+        // Split by \n
+        std::string token;
+        std::stringstream tokenStream(data);
+
+        while (std::getline(tokenStream, token))
+        {
+            std::size_t pos = token.rfind('=');
+
+            // Bail out if last '.' is found
+            if (pos == std::string::npos) continue;
+
+            auto key = token.substr(0, pos);
+            auto val = token.substr(pos+1);
+
+            std::cerr << key << ": " << val << std::endl;
+            httpParameters[key] = val;
+        }
+
+        return httpParameters;
+    }
+
+    int ws_http_client_main(const std::string& url,
+                            const std::string& headersData,
+                            const std::string& data,
+                            bool headersOnly,
+                            int connectTimeout,
+                            int transferTimeout,
+                            bool followRedirects,
+                            int maxRedirects,
+                            bool verbose,
+                            bool save,
+                            const std::string& output,
+                            bool compress)
+    {
+        HttpRequestArgs args;
+        args.extraHeaders = parseHeaders(headersData);
+        args.connectTimeout = connectTimeout;
+        args.transferTimeout = transferTimeout;
+        args.followRedirects = followRedirects;
+        args.maxRedirects = maxRedirects;
+        args.verbose = verbose;
+        args.compress = compress;
+        args.logger = [](const std::string& msg)
+        {
+            std::cout << msg;
+        };
+        args.onProgressCallback = [](int current, int total) -> bool
+        {
+            std::cerr << "\r" << "Downloaded "
+                      << current << " bytes out of " << total;
+            return true;
+        };
+
+        HttpParameters httpParameters = parsePostParameters(data);
+
+        HttpClient httpClient;
+        HttpResponse out;
+        if (headersOnly)
+        {
+            out = httpClient.head(url, args);
+        }
+        else if (data.empty())
+        {
+            out = httpClient.get(url, args);
+        }
+        else
+        {
+            out = httpClient.post(url, httpParameters, args);
+        }
+
+        std::cerr << std::endl;
+
+        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);
+
+        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 (!headersOnly && errorCode == HttpErrorCode_Ok)
+        {
+            if (save || !output.empty())
+            {
+                // FIMXE we should decode the url first
+                std::string filename = extractFilename(url);
+                if (!output.empty())
+                {
+                    filename = output;
+                }
+
+                std::cout << "Writing to disk: " << filename << std::endl;
+                std::ofstream out(filename);
+                out.write((char*)&payload.front(), payload.size());
+                out.close();
+            }
+            else
+            {
+                if (responseHeaders["Content-Type"] != "application/octet-stream")
+                {
+                    std::cout << "payload: " << payload << std::endl;
+                }
+                else
+                {
+                    std::cerr << "Binary output can mess up your terminal." << std::endl;
+                    std::cerr << "Use the -O flag to save the file to disk." << std::endl;
+                    std::cerr << "You can also use the --output option to specify a filename." << std::endl;
+                }
+            }
+        }
+
+        return 0;
+    }
+}

ws/ws_ping_pong.cpp

@@ -61,10 +61,19 @@ namespace ix
                const ix::WebSocketOpenInfo& openInfo,
                const ix::WebSocketCloseInfo& closeInfo)
             {
+                std::cerr << "Received " << wireSize << " bytes" << std::endl;
+
                 std::stringstream ss;
                 if (messageType == ix::WebSocket_MessageType_Open)
                 {
                     log("ping_pong: 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)
                 {
@@ -153,5 +162,7 @@ namespace ix
 
         std::cout << std::endl;
         webSocketPingPong.stop();
+
+        return 0;
     }
 }

ws/ws_receive.cpp

@@ -146,11 +146,16 @@ namespace ix
         std::string filename = data["filename"].string_value();
         filename = extractFilename(filename);
 
-        std::cout << "Writing to disk: " << filename << std::endl;
-        std::ofstream out(filename);
+        std::string filenameTmp = filename + ".tmp";
+
+        std::cout << "Writing to disk: " << filenameTmp << std::endl;
+        std::ofstream out(filenameTmp);
         out.write((char*)&content.front(), content.size());
         out.close();
 
+        std::cout << "Renaming " << filenameTmp << " to " << filename << std::endl;
+        rename(filenameTmp.c_str(), filename.c_str());
+
         std::map<MsgPack, MsgPack> pdu;
         pdu["ack"] = true;
         pdu["id"] = data["id"];
@@ -206,6 +211,11 @@ namespace ix
                     handleMessage(str);
                     _condition.notify_one();
                 }
+                else if (messageType == ix::WebSocket_MessageType_Fragment)
+                {
+                    ss << "ws_receive: received fragment";
+                    log(ss.str());
+                }
                 else if (messageType == ix::WebSocket_MessageType_Error)
                 {
                     ss << "Connection error: " << error.reason      << std::endl;

ws/ws_send.cpp

@@ -257,6 +257,15 @@ namespace ix
             return true;
         });
 
+        do
+        {
+            size_t bufferedAmount = _webSocket.bufferedAmount();
+            std::cout << bufferedAmount << " bytes left to be sent" << std::endl;
+
+            std::chrono::duration<double, std::milli> duration(10);
+            std::this_thread::sleep_for(duration);
+        } while (_webSocket.bufferedAmount() != 0);
+
         bench.report();
         auto duration = bench.getDuration();
         auto transferRate = 1000 * content.size() / duration;

ws/ws_transfer.cpp

@@ -10,11 +10,11 @@
 
 namespace ix
 {
-    int ws_transfer_main(int port)
+    int ws_transfer_main(int port, const std::string& hostname)
     {
-        std::cout << "Listening on port " << port << std::endl;
+        std::cout << "Listening on " << hostname << ":" << port << std::endl;
 
-        ix::WebSocketServer server(port);
+        ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
             [&server](std::shared_ptr<ix::WebSocket> webSocket)
@@ -39,7 +39,22 @@ namespace ix
                         }
                         else if (messageType == ix::WebSocket_MessageType_Close)
                         {
-                            std::cerr << "Closed connection" << std::endl;
+                            std::cerr << "Closed connection"
+                                      << " code " << closeInfo.code
+                                      << " reason " << closeInfo.reason << std::endl;
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Error)
+                        {
+                            std::stringstream ss;
+                            ss << "Connection error: " << error.reason      << std::endl;
+                            ss << "#retries: "         << error.retries     << std::endl;
+                            ss << "Wait time(ms): "    << error.wait_time   << std::endl;
+                            ss << "HTTP Status: "      << error.http_status << std::endl;
+                            std::cerr << ss.str();
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Fragment)
+                        {
+                            std::cerr << "Received message fragment" << std::endl;
                         }
                         else if (messageType == ix::WebSocket_MessageType_Message)
                         {
@@ -48,7 +63,22 @@ namespace ix
                             {
                                 if (client != webSocket)
                                 {
-                                    client->send(str);
+                                    client->send(str,
+                                                 [](int current, int total) -> bool
+                                    {
+                                        std::cerr << "Step " << current
+                                                  << " out of " << total << std::endl;
+                                        return true;
+                                    });
+
+                                    do
+                                    {
+                                        size_t bufferedAmount = client->bufferedAmount();
+                                        std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
+
+                                        std::chrono::duration<double, std::milli> duration(10);
+                                        std::this_thread::sleep_for(duration);
+                                    } while (client->bufferedAmount() != 0);
                                 }
                             }
                         }