do not busy loop while sending

CMakeLists.txt

@@ -15,11 +15,8 @@ if (NOT WIN32)
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
 endif()
 
-if ("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang")
-  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wshorten-64-to-32")
-endif()
-
 set( IXWEBSOCKET_SOURCES
+    ixwebsocket/IXEventFd.cpp
     ixwebsocket/IXSocket.cpp
     ixwebsocket/IXSocketServer.cpp
     ixwebsocket/IXSocketConnect.cpp
@@ -36,13 +33,10 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXWebSocketHttpHeaders.cpp
     ixwebsocket/IXHttpClient.cpp
     ixwebsocket/IXUrlParser.cpp
-    ixwebsocket/IXSelectInterrupt.cpp
-    ixwebsocket/IXSelectInterruptPipe.cpp
-    ixwebsocket/IXSelectInterruptFactory.cpp
-    ixwebsocket/IXConnectionState.cpp
 )
 
 set( IXWEBSOCKET_HEADERS
+    ixwebsocket/IXEventFd.h
     ixwebsocket/IXSocket.h
     ixwebsocket/IXSocketServer.h
     ixwebsocket/IXSocketConnect.h
@@ -64,10 +58,6 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/libwshandshake.hpp
     ixwebsocket/IXHttpClient.h
     ixwebsocket/IXUrlParser.h
-    ixwebsocket/IXSelectInterrupt.h
-    ixwebsocket/IXSelectInterruptPipe.h
-    ixwebsocket/IXSelectInterruptFactory.h
-    ixwebsocket/IXConnectionState.h
 )
 
 # Platform specific code
@@ -77,8 +67,6 @@ elseif (WIN32)
     list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/windows/IXSetThreadName_windows.cpp)
 else()
     list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/linux/IXSetThreadName_linux.cpp)
-    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptEventFd.cpp)
-    list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSelectInterruptEventFd.h)
 endif()
 
 if (USE_TLS)

README.md

@@ -11,6 +11,7 @@ communication channels over a single TCP connection. *IXWebSocket* is a C++ libr
 * iOS
 * Linux
 * Android 
+* Windows (no TLS support yet)
 
 ## Examples
 
@@ -63,11 +64,10 @@ Here is what the server API looks like. Note that server support is very recent
 ix::WebSocketServer server(port);
 
 server.setOnConnectionCallback(
-    [&server](std::shared_ptr<WebSocket> webSocket,
+    [&server](std::shared_ptr<ix::WebSocket> webSocket)
-              std::shared_ptr<ConnectionState> connectionState)
     {
         webSocket->setOnMessageCallback(
-            [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+            [webSocket, &server](ix::WebSocketMessageType messageType,
                const std::string& str,
                size_t wireSize,
                const ix::WebSocketErrorInfo& error,
@@ -78,12 +78,6 @@ server.setOnConnectionCallback(
                 {
                     std::cerr << "New connection" << std::endl;
 
-                    // A connection state object is available, and has a default id
-                    // You can subclass ConnectionState and pass an alternate factory
-                    // to override it. It is useful if you want to store custom
-                    // attributes per connection (authenticated bool flag, attributes, etc...)
-                    std::cerr << "id: " << connectionState->getId() << std::endl;
-
                     // The uri the client did connect to.
                     std::cerr << "Uri: " << openInfo.uri << std::endl;
 
@@ -131,7 +125,7 @@ HttpRequestArgs args;
 // Custom headers can be set
 WebSocketHttpHeaders headers;
 headers["Foo"] = "bar";
-args.extraHeaders = headers;
+args.extraHeaders = parseHeaders(headersData);
 
 // Timeout options
 args.connectTimeout = connectTimeout;
@@ -190,7 +184,7 @@ There is a Dockerfile for running some code on Linux, and a unittest which can b
 You can build and install the ws command line tool with Homebrew.
 
 ```
-brew tap bsergean/IXWebSocket
+brew create --cmake https://github.com/machinezone/IXWebSocket/archive/v1.1.0.tar.gz
 brew install IXWebSocket
 ```
 
@@ -386,7 +380,7 @@ websocket.ping("ping data, optional (empty string is ok): limited to 125 bytes l
 ### Heartbeat.
 
 You can configure an optional heart beat / keep-alive, sent every 45 seconds
-when there is no any traffic to make sure that load balancers do not kill an
+when there is not any traffic to make sure that load balancers do not kill an
 idle connection.
 
 ```

examples/cobra_publisher/ixcrypto/IXHash.cpp

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

examples/cobra_publisher/ixcrypto/IXHash.h

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

examples/cobra_publisher/ixcrypto/IXUuid.cpp

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

examples/cobra_publisher/ixcrypto/IXUuid.h

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

examples/ws_receive/.gitignore

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

examples/ws_receive/CMakeLists.txt

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

examples/ws_receive/README.md

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

examples/ws_receive/ixcrypto

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

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

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

examples/ws_receive/package-lock.json

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

examples/ws_receive/ws_receive.cpp

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

examples/ws_receive/ws_receive.js

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

examples/ws_send/.gitignore

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

examples/ws_send/CMakeLists.txt

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

examples/ws_send/README.md

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

examples/ws_send/ixcrypto

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

examples/ws_send/jsoncpp

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

examples/ws_send/ws_send.cpp

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

ixwebsocket/IXConnectionState.cpp

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

ixwebsocket/IXConnectionState.h

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

ixwebsocket/IXDNSLookup.cpp

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

ixwebsocket/IXEventFd.cpp

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

ixwebsocket/IXEventFd.h

@@ -0,0 +1,33 @@
+/*
+ *  IXEventFd.h
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#pragma once
+
+#include <stdint.h>
+
+namespace ix
+{
+    class EventFd {
+    public:
+        EventFd();
+        virtual ~EventFd();
+
+        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
+    };
+}

ixwebsocket/IXSelectInterrupt.cpp

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

ixwebsocket/IXSelectInterrupt.h

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

ixwebsocket/IXSelectInterruptEventFd.cpp

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

ixwebsocket/IXSelectInterruptEventFd.h

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

ixwebsocket/IXSelectInterruptFactory.cpp

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

ixwebsocket/IXSelectInterruptFactory.h

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

ixwebsocket/IXSelectInterruptPipe.cpp

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

ixwebsocket/IXSelectInterruptPipe.h

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

ixwebsocket/IXSocket.cpp

@@ -7,8 +7,6 @@
 #include "IXSocket.h"
 #include "IXSocketConnect.h"
 #include "IXNetSystem.h"
-#include "IXSelectInterrupt.h"
-#include "IXSelectInterruptFactory.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -25,13 +23,12 @@ namespace ix
 {
     const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
     const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
-    const uint64_t Socket::kSendRequest = 1;
+    const uint8_t Socket::kSendRequest = 1;
-    const uint64_t Socket::kCloseRequest = 2;
+    const uint8_t Socket::kCloseRequest = 2;
     constexpr size_t Socket::kChunkSize;
 
     Socket::Socket(int fd) :
-        _sockfd(fd),
+        _sockfd(fd)
-        _selectInterrupt(createSelectInterrupt())
     {
         ;
     }
@@ -45,93 +42,69 @@ namespace ix
     {
         if (_sockfd == -1)
         {
-            if (onPollCallback) onPollCallback(PollResultType::Error);
+            if (onPollCallback) onPollCallback(PollResultType_Error);
             return;
         }
 
-        PollResultType pollResult = isReadyToRead(1000 * timeoutSecs);
+        PollResultType pollResult = select(timeoutSecs, 0);
 
         if (onPollCallback) onPollCallback(pollResult);
     }
 
-    PollResultType Socket::select(bool readyToRead, int timeoutMs)
+    PollResultType Socket::select(int timeoutSecs, int timeoutMs)
     {
         fd_set rfds;
-        fd_set wfds;
         FD_ZERO(&rfds);
-        FD_ZERO(&wfds);
+        FD_SET(_sockfd, &rfds);
 
-        fd_set* fds = (readyToRead) ? &rfds : & wfds;
+        // File descriptor at index 0 in _fildes is the read end of the pipe
-        FD_SET(_sockfd, fds);
+        int eventfd = _eventfd.getFd();
-
+        if (eventfd != -1)
-        // File descriptor used to interrupt select when needed
-        int interruptFd = _selectInterrupt->getFd();
-        if (interruptFd != -1)
         {
-            FD_SET(interruptFd, fds);
+            FD_SET(eventfd, &rfds);
         }
 
         struct timeval timeout;
-        timeout.tv_sec = timeoutMs / 1000;
+        timeout.tv_sec = timeoutSecs;
-        timeout.tv_usec = (timeoutMs < 1000) ? 0 : 1000 * (timeoutMs % 1000);
+        timeout.tv_usec = 1000 * timeoutMs;
 
         // Compute the highest fd.
         int sockfd = _sockfd;
-        int nfds = (std::max)(sockfd, interruptFd);
+        int nfds = (std::max)(sockfd, eventfd);
 
-        int ret = ::select(nfds + 1, &rfds, &wfds, nullptr,
+        int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
-                           (timeoutMs < 0) ? nullptr : &timeout);
+                           (timeoutSecs < 0) ? nullptr : &timeout);
 
-        PollResultType pollResult = PollResultType::ReadyForRead;
+        PollResultType pollResult = PollResultType_ReadyForRead;
         if (ret < 0)
         {
-            pollResult = PollResultType::Error;
+            pollResult = PollResultType_Error;
         }
         else if (ret == 0)
         {
-            pollResult = PollResultType::Timeout;
+            pollResult = PollResultType_Timeout;
         }
-        else if (interruptFd != -1 && FD_ISSET(interruptFd, &rfds))
+        else if (eventfd != -1 && FD_ISSET(eventfd, &rfds))
         {
-            uint64_t value = _selectInterrupt->read();
+            uint8_t value = _eventfd.read();
 
             if (value == kSendRequest)
             {
-                pollResult = PollResultType::SendRequest;
+                pollResult = PollResultType_SendRequest;
             }
             else if (value == kCloseRequest)
             {
-                pollResult = PollResultType::CloseRequest;
+                pollResult = PollResultType_CloseRequest;
             }
         }
-        else if (sockfd != -1 && readyToRead && FD_ISSET(sockfd, &rfds))
-        {
-            pollResult = PollResultType::ReadyForRead;
-        }
-        else if (sockfd != -1 && !readyToRead && FD_ISSET(sockfd, &wfds))
-        {
-            pollResult = PollResultType::ReadyForWrite;
-        }
 
         return pollResult;
     }
 
-    PollResultType Socket::isReadyToRead(int timeoutMs)
-    {
-        bool readyToRead = true;
-        return select(readyToRead, timeoutMs);
-    }
-
-    PollResultType Socket::isReadyToWrite(int timeoutMs)
-    {
-        bool readyToRead = false;
-        return select(readyToRead, timeoutMs);
-    }
-
     // Wake up from poll/select by writing to the pipe which is watched by select
     bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
     {
-        return _selectInterrupt->notify(wakeUpCode);
+        return _eventfd.notify(wakeUpCode);
     }
 
     bool Socket::connect(const std::string& host,
@@ -141,7 +114,7 @@ namespace ix
     {
         std::lock_guard<std::mutex> lock(_socketMutex);
 
-        if (!_selectInterrupt->clear()) return false;
+        if (!_eventfd.clear()) return false;
 
         _sockfd = SocketConnect::connect(host, port, errMsg, isCancellationRequested);
         return _sockfd != -1;
@@ -200,9 +173,24 @@ namespace ix
 #endif
     }
 
-    bool Socket::init(std::string& errorMsg)
+    bool Socket::init()
     {
-        return _selectInterrupt->init(errorMsg);
+#ifdef _WIN32
+        INT rc;
+        WSADATA wsaData;
+
+        rc = WSAStartup(MAKEWORD(2, 2), &wsaData);
+        return rc != 0;
+#else
+        return true;
+#endif
+    }
+
+    void Socket::cleanup()
+    {
+#ifdef _WIN32
+        WSACleanup();
+#endif
     }
 
     bool Socket::writeBytes(const std::string& str,
@@ -210,7 +198,7 @@ namespace ix
     {
         while (true)
         {
-            if (isCancellationRequested && isCancellationRequested()) return false;
+            if (isCancellationRequested()) return false;
 
             char* buffer = const_cast<char*>(str.c_str());
             int len = (int) str.size();
@@ -222,7 +210,7 @@ namespace ix
             {
                 return ret == len;
             }
-            // There is possibly something to be writen, try again
+            // There is possibly something to be write, try again
             else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
                                  getErrno() == EAGAIN))
             {
@@ -241,7 +229,7 @@ namespace ix
     {
         while (true)
         {
-            if (isCancellationRequested && isCancellationRequested()) return false;
+            if (isCancellationRequested()) return false;
 
             ssize_t ret;
             ret = recv(buffer, 1);
@@ -255,9 +243,10 @@ namespace ix
             else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
                                  getErrno() == EAGAIN))
             {
-                // Wait with a 1ms timeout until the socket is ready to read.
+                // Wait with a timeout until something is ready to read.
                 // This way we are not busy looping
-                if (isReadyToRead(1) == PollResultType::Error)
+                int res = select(0, 1);
+                if (res < 0 && (errno == EBADF || errno == EINVAL))
                 {
                     return false;
                 }
@@ -304,12 +293,9 @@ namespace ix
         std::vector<uint8_t> output;
         while (output.size() != length)
         {
-            if (isCancellationRequested && isCancellationRequested())
+            if (isCancellationRequested()) return std::make_pair(false, std::string());
-            {
-                return std::make_pair(false, std::string());
-            }
 
-            size_t size = std::min(kChunkSize, length - output.size());
+            int size = std::min(kChunkSize, length - output.size());
             ssize_t ret = recv((char*)&_readBuffer[0], size);
 
             if (ret <= 0 && (getErrno() != EWOULDBLOCK &&
@@ -327,12 +313,9 @@ namespace ix
 
             if (onProgressCallback) onProgressCallback((int) output.size(), (int) length);
 
-            // Wait with a 1ms timeout until the socket is ready to read.
+            // Wait with a timeout until something is ready to read.
             // This way we are not busy looping
-            if (isReadyToRead(1) == PollResultType::Error)
+            select(0, 1);
-            {
-                return std::make_pair(false, std::string());
-            }
         }
 
         return std::make_pair(true, std::string(output.begin(),

ixwebsocket/IXSocket.h

@@ -11,7 +11,6 @@
 #include <mutex>
 #include <atomic>
 #include <vector>
-#include <memory>
 
 #ifdef _WIN32
 #include <BaseTsd.h>
@@ -20,19 +19,17 @@ typedef SSIZE_T ssize_t;
 
 #include "IXCancellationRequest.h"
 #include "IXProgressCallback.h"
+#include "IXEventFd.h"
 
 namespace ix
 {
-    class SelectInterrupt;
+    enum PollResultType
-
-    enum class PollResultType
     {
-        ReadyForRead = 0,
+        PollResultType_ReadyForRead = 0,
-        ReadyForWrite = 1,
+        PollResultType_Timeout = 1,
-        Timeout = 2,
+        PollResultType_Error = 2,
-        Error = 3,
+        PollResultType_SendRequest = 3,
-        SendRequest = 4,
+        PollResultType_CloseRequest = 4
-        CloseRequest = 5
     };
 
     class Socket {
@@ -41,17 +38,13 @@ namespace ix
 
         Socket(int fd = -1);
         virtual ~Socket();
-        bool init(std::string& errorMsg);
 
         void configure();
 
-        // Functions to check whether there is activity on the socket
+        PollResultType select(int timeoutSecs, int timeoutMs);
-        void poll(const OnPollCallback& onPollCallback,
+        virtual void poll(const OnPollCallback& onPollCallback,
                           int timeoutSecs = kDefaultPollTimeout);
-        bool wakeUpFromPoll(uint8_t wakeUpCode);
+        virtual bool wakeUpFromPoll(uint8_t wakeUpCode);
-
-        PollResultType isReadyToWrite(int timeoutMs);
-        PollResultType isReadyToRead(int timeoutMs);
 
         // Virtual methods
         virtual bool connect(const std::string& url,
@@ -79,10 +72,12 @@ namespace ix
             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 uint8_t kSendRequest;
-        static const uint64_t kCloseRequest;
+        static const uint8_t kCloseRequest;
 
     protected:
         void closeSocket(int fd);
@@ -91,8 +86,6 @@ namespace ix
         std::mutex _socketMutex;
 
     private:
-        PollResultType select(bool readyToRead, int timeoutMs);
-
         static const int kDefaultPollTimeout;
         static const int kDefaultPollNoTimeout;
 
@@ -100,6 +93,6 @@ namespace ix
         std::vector<uint8_t> _readBuffer;
         static constexpr size_t kChunkSize = 1 << 15;
 
-        std::shared_ptr<SelectInterrupt> _selectInterrupt;
+        EventFd _eventfd;
     };
 }

ixwebsocket/IXSocketConnect.cpp

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

ixwebsocket/IXSocketFactory.cpp

@@ -20,45 +20,23 @@ namespace ix
                                          std::string& errorMsg)
     {
         errorMsg.clear();
-        std::shared_ptr<Socket> socket;
 
         if (!tls)
         {
-            socket = std::make_shared<Socket>();
+            return std::make_shared<Socket>();
         }
         else
         {
 #ifdef IXWEBSOCKET_USE_TLS
 # ifdef __APPLE__
-            socket = std::make_shared<SocketAppleSSL>();
+            return std::make_shared<SocketAppleSSL>();
 # else
-            socket = std::make_shared<SocketOpenSSL>();
+            return std::make_shared<SocketOpenSSL>();
 # endif
 #else
             errorMsg = "TLS support is not enabled on this platform.";
             return nullptr;
 #endif
         }
-
-        if (!socket->init(errorMsg))
-        {
-            socket.reset();
-        }
-
-        return socket;
-    }
-
-    std::shared_ptr<Socket> createSocket(int fd,
-                                         std::string& errorMsg)
-    {
-        errorMsg.clear();
-
-        std::shared_ptr<Socket> socket = std::make_shared<Socket>(fd);
-        if (!socket->init(errorMsg))
-        {
-            socket.reset();
-        }
-
-        return socket;
     }
 }

ixwebsocket/IXSocketFactory.h

@@ -14,7 +14,4 @@ namespace ix
     class Socket;
     std::shared_ptr<Socket> createSocket(bool tls,
                                          std::string& errorMsg);
-
-    std::shared_ptr<Socket> createSocket(int fd,
-                                         std::string& errorMsg);
 }

ixwebsocket/IXSocketOpenSSL.cpp

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

ixwebsocket/IXSocketOpenSSL.h

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

ixwebsocket/IXSocketServer.cpp

@@ -29,8 +29,7 @@ namespace ix
         _host(host),
         _backlog(backlog),
         _maxConnections(maxConnections),
-        _stop(false),
+        _stop(false)
-        _connectionStateFactory(&ConnectionState::createConnectionState)
     {
 
     }
@@ -146,12 +145,6 @@ namespace ix
         ::close(_serverFd);
     }
 
-    void SocketServer::setConnectionStateFactory(
-        const ConnectionStateFactory& connectionStateFactory)
-    {
-        _connectionStateFactory = connectionStateFactory;
-    }
-
     void SocketServer::run()
     {
         // Set the socket to non blocking mode, so that accept calls are not blocking
@@ -221,12 +214,6 @@ namespace ix
                 continue;
             }
 
-            std::shared_ptr<ConnectionState> connectionState;
-            if (_connectionStateFactory)
-            {
-                connectionState = _connectionStateFactory();
-            }
-
             // Launch the handleConnection work asynchronously in its own thread.
             //
             // the destructor of a future returned by std::async blocks,
@@ -234,8 +221,7 @@ namespace ix
             f = std::async(std::launch::async,
                            &SocketServer::handleConnection,
                            this,
-                           clientFd,
+                           clientFd);
-                           connectionState);
         }
     }
 }

ixwebsocket/IXSocketServer.h

@@ -6,8 +6,6 @@
 
 #pragma once
 
-#include "IXConnectionState.h"
-
 #include <utility> // pair
 #include <string>
 #include <set>
@@ -22,8 +20,6 @@ namespace ix
 {
     class SocketServer {
     public:
-        using ConnectionStateFactory = std::function<std::shared_ptr<ConnectionState>()>;
-
         SocketServer(int port = SocketServer::kDefaultPort,
                      const std::string& host = SocketServer::kDefaultHost,
                      int backlog = SocketServer::kDefaultTcpBacklog,
@@ -31,8 +27,6 @@ namespace ix
         virtual ~SocketServer();
         virtual void stop();
 
-        void setConnectionStateFactory(const ConnectionStateFactory& connectionStateFactory);
-
         const static int kDefaultPort;
         const static std::string kDefaultHost;
         const static int kDefaultTcpBacklog;
@@ -66,13 +60,9 @@ namespace ix
         std::condition_variable _conditionVariable;
         std::mutex _conditionVariableMutex;
 
-        //
-        ConnectionStateFactory _connectionStateFactory;
-
         // Methods
         void run();
-        virtual void handleConnection(int fd,
+        virtual void handleConnection(int fd) = 0;
-                                      std::shared_ptr<ConnectionState> connectionState) = 0;
         virtual size_t getConnectedClientsCount() = 0;
     };
 }

ixwebsocket/IXWebSocket.cpp

@@ -79,10 +79,10 @@ namespace ix
         return _perMessageDeflateOptions;
     }
 
-    void WebSocket::setHeartBeatPeriod(int heartBeatPeriod)
+    void WebSocket::setHeartBeatPeriod(int hearBeatPeriod)
     {
         std::lock_guard<std::mutex> lock(_configMutex);
-        _heartBeatPeriod = heartBeatPeriod;
+        _heartBeatPeriod = hearBeatPeriod;
     }
 
     int WebSocket::getHeartBeatPeriod() const

ixwebsocket/IXWebSocket.h

@@ -89,7 +89,7 @@ namespace ix
         void setUrl(const std::string& url);
         void setPerMessageDeflateOptions(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
         void setHandshakeTimeout(int handshakeTimeoutSecs);
-        void setHeartBeatPeriod(int heartBeatPeriod);
+        void setHeartBeatPeriod(int hearBeatPeriod);
 
         // Run asynchronously, by calling start and stop.
         void start();

ixwebsocket/IXWebSocketServer.cpp

@@ -49,12 +49,10 @@ namespace ix
         _onConnectionCallback = callback;
     }
 
-    void WebSocketServer::handleConnection(
+    void WebSocketServer::handleConnection(int fd)
-        int fd,
-        std::shared_ptr<ConnectionState> connectionState)
     {
         auto webSocket = std::make_shared<WebSocket>();
-        _onConnectionCallback(webSocket, connectionState);
+        _onConnectionCallback(webSocket);
 
         webSocket->disableAutomaticReconnection();
 

ixwebsocket/IXWebSocketServer.h

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

ixwebsocket/IXWebSocketTransport.cpp

@@ -53,7 +53,7 @@
 
 namespace ix
 {
-    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::heartbeat");
+    const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::hearbeat");
     const int WebSocketTransport::kDefaultHeartBeatPeriod(-1);
     constexpr size_t WebSocketTransport::kChunkSize;
 
@@ -75,11 +75,11 @@ namespace ix
     }
 
     void WebSocketTransport::configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                                       int heartBeatPeriod)
+                                       int hearBeatPeriod)
     {
         _perMessageDeflateOptions = perMessageDeflateOptions;
         _enablePerMessageDeflate = _perMessageDeflateOptions.enabled();
-        _heartBeatPeriod = heartBeatPeriod;
+        _heartBeatPeriod = hearBeatPeriod;
     }
 
     // Client
@@ -123,13 +123,8 @@ namespace ix
     // Server
     WebSocketInitResult WebSocketTransport::connectToSocket(int fd, int timeoutSecs)
     {
-        std::string errorMsg;
+        _socket.reset();
-        _socket = createSocket(fd, errorMsg);
+        _socket = std::make_shared<Socket>(fd);
-
-        if (!_socket)
-        {
-            return WebSocketInitResult(false, 0, errorMsg);
-        }
 
         WebSocketHandshake webSocketHandshake(_requestInitCancellation,
                                               _socket,
@@ -189,7 +184,7 @@ namespace ix
                 // If (1) heartbeat is enabled, and (2) no data was received or
                 // send for a duration exceeding our heart-beat period, send a
                 // ping to the server.
-                if (pollResult == PollResultType::Timeout &&
+                if (pollResult == PollResultType_Timeout &&
                     heartBeatPeriodExceeded())
                 {
                     std::stringstream ss;
@@ -198,27 +193,20 @@ namespace ix
                 }
                 // Make sure we send all the buffered data
                 // there can be a lot of it for large messages.
-                else if (pollResult == PollResultType::SendRequest)
+                else if (pollResult == PollResultType_SendRequest)
                 {
                     while (!isSendBufferEmpty() && !_requestInitCancellation)
-                    {
-                        // Wait with a 10ms timeout until the socket is ready to write.
-                        // This way we are not busy looping
-                        PollResultType result = _socket->isReadyToWrite(10);
-
-                        if (result == PollResultType::Error)
-                        {
-                            _socket->close();
-                            setReadyState(CLOSED);
-                            break;
-                        }
-                        else if (result == PollResultType::ReadyForWrite)
                     {
                         sendOnSocket();
+
+                        // 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)
-                else if (pollResult == PollResultType::ReadyForRead)
                 {
                     while (true)
                     {
@@ -244,21 +232,15 @@ namespace ix
                         }
                     }
                 }
-                else if (pollResult == PollResultType::Error)
+                else if (pollResult == PollResultType_Error)
                 {
                     _socket->close();
                 }
-                else if (pollResult == PollResultType::CloseRequest)
+                else if (pollResult == PollResultType_CloseRequest)
                 {
-                    _socket->close();
+                    ;
                 }
 
-                // Avoid a race condition where we get stuck in select
-                // while closing.
-                if (_readyState == CLOSING)
-                {
-                    _socket->close();
-                }
             },
             _heartBeatPeriod);
     }
@@ -784,7 +766,6 @@ namespace ix
         _socket->close();
 
         _closeCode = 1000;
-        _closeReason = "Normal Closure";
         setReadyState(CLOSED);
     }
 

ixwebsocket/IXWebSocketTransport.h

@@ -61,7 +61,7 @@ namespace ix
         ~WebSocketTransport();
 
         void configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
-                       int heartBeatPeriod);
+                       int hearBeatPeriod);
 
         WebSocketInitResult connectToUrl(const std::string& url, // Client
                                          int timeoutSecs);
@@ -148,7 +148,7 @@ namespace ix
         mutable std::mutex _lastSendTimePointMutex;
         std::chrono::time_point<std::chrono::steady_clock> _lastSendTimePoint;
 
-        // No data was send through the socket for longer than the heartbeat period
+        // No data was send through the socket for longer that the hearbeat period
         bool heartBeatPeriodExceeded();
 
         void sendOnSocket();

makefile

@@ -3,8 +3,6 @@
 #
 all: brew
 
-install: brew
-
 brew:
 	mkdir -p build && (cd build ; cmake .. ; make -j install)
 
@@ -38,8 +36,8 @@ test_server:
 test:
 	python test/run.py
 
-ws_test: all
+ws_test:
-	(cd ws ; bash test_ws.sh)
+	(cd ws ; sh test_ws.sh)
 
 # For the fork that is configured with appveyor
 rebase_upstream:

test/IXSocketTest.cpp

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

test/IXWebSocketHeartBeatTest.cpp

@@ -65,7 +65,7 @@ namespace
         _webSocket.setUrl(url);
 
         // The important bit for this test.
-        // Set a 1 second heartbeat ; if no traffic is present on the connection for 1 second
+        // Set a 1 second hearbeat ; if no traffic is present on the connection for 1 second
         // a ping message will be sent by the client.
         _webSocket.setHeartBeatPeriod(1);
 
@@ -128,11 +128,10 @@ namespace
     {
         // A dev/null server
         server.setOnConnectionCallback(
-            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket,
+            [&server, &receivedPingMessages](std::shared_ptr<ix::WebSocket> webSocket)
-                                             std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
+                    [webSocket, &server, &receivedPingMessages](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -142,7 +141,6 @@ namespace
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             Logger() << "New server connection";
-                            Logger() << "id: " << connectionState->getId();
                             Logger() << "Uri: " << openInfo.uri;
                             Logger() << "Headers:";
                             for (auto it : openInfo.headers)

test/IXWebSocketServerTest.cpp

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

test/cmd_websocket_chat.cpp

@@ -217,11 +217,10 @@ namespace
     bool startServer(ix::WebSocketServer& server)
     {
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket,
+            [&server](std::shared_ptr<ix::WebSocket> webSocket)
-                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -231,7 +230,6 @@ namespace
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             Logger() << "New connection";
-                            Logger() << "id: " << connectionState->getId();
                             Logger() << "Uri: " << openInfo.uri;
                             Logger() << "Headers:";
                             for (auto it : openInfo.headers)

test/run.py

@@ -2,39 +2,6 @@ import os
 import platform
 import shutil
 
-import subprocess
-import threading
-
-
-class Command(object):
-    """Run system commands with timeout
-    
-    From http://www.bo-yang.net/2016/12/01/python-run-command-with-timeout
-    Python3 might have a builtin way to do that.
-    """
-    def __init__(self, cmd):
-        self.cmd = cmd
-        self.process = None
-
-    def run_command(self, capture = False):
-        self.process = subprocess.Popen(self.cmd, shell=True)
-        self.process.communicate()
-
-    def run(self, timeout = 5 * 60):
-        '''5 minutes default timeout'''
-        thread = threading.Thread(target=self.run_command, args=())
-        thread.start()
-        thread.join(timeout)
-
-        if thread.is_alive():
-            print('Command timeout, kill it: ' + self.cmd)
-            self.process.terminate()
-            thread.join()
-            return False, 255
-        else:
-            return True, self.process.returncode
-
-
 osName = platform.system()
 print('os name = {}'.format(osName))
 
@@ -111,9 +78,8 @@ shutil.copy(os.path.join(
     'bin',
     'zlib.dll'), '.')
 
-# lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
+lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
 lldb = ""  # Disabled for now
 testCommand = '{} {} {}'.format(lldb, testBinary, os.getenv('TEST', ''))
-command = Command(testCommand)
+ret = os.system(testCommand)
-timedout, ret = command.run()
 assert ret == 0, 'Test command failed'

test/test_runner.cpp

@@ -11,6 +11,10 @@
 
 int main(int argc, char* argv[])
 {
+    ix::Socket::init(); // for Windows
+
     int result = Catch::Session().run(argc, argv);
+
+    ix::Socket::cleanup(); // for Windows
     return result;
 }

third_party/remote_trailing_whitespaces.sh

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

ws/.gitignore

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

ws/CMakeLists.txt

@@ -23,8 +23,6 @@ add_executable(ws
   ixcrypto/IXHash.cpp
   ixcrypto/IXUuid.cpp
 
-  IXRedisClient.cpp
-
   ws_http_client.cpp
   ws_ping_pong.cpp
   ws_broadcast_server.cpp
@@ -34,8 +32,6 @@ add_executable(ws
   ws_transfer.cpp
   ws_send.cpp
   ws_receive.cpp
-  ws_redis_publish.cpp
-  ws_redis_subscribe.cpp
   ws.cpp)
 
 if (APPLE AND USE_TLS)

ws/IXRedisClient.cpp

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

ws/IXRedisClient.h

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

ws/package-lock.json

@@ -1,19 +0,0 @@
-{
-  "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=="
-    },
-    "ws": {
-      "version": "6.2.0",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-6.2.0.tgz",
-      "integrity": "sha512-deZYUNlt2O4buFCa3t5bKLf8A7FPP/TVjwOeVNpw818Ma5nk4MLXls2eoEGS39o8119QIYxTrTDoPQ5B/gTD6w==",
-      "requires": {
-        "async-limiter": "1.0.0"
-      }
-    }
-  }
-}

ws/test_ws.sh

@@ -1,21 +1,11 @@
 #!/bin/sh
 
-# Handle Ctrl-C by killing all sub-processing AND exiting
-trap cleanup INT
-
-function cleanup {
-    kill `cat /tmp/ws_test/pidfile.transfer`
-    kill `cat /tmp/ws_test/pidfile.receive`
-    kill `cat /tmp/ws_test/pidfile.send`
-    exit 1
-}
-
 rm -rf /tmp/ws_test
 mkdir -p /tmp/ws_test
 
 # Start a transport server
 cd /tmp/ws_test
-ws transfer --port 8090 --pidfile /tmp/ws_test/pidfile.transfer &
+ws transfer --port 8090 --pidfile /tmp/ws_test/pidfile &
 
 # Wait until the transfer server is up 
 while true
@@ -24,41 +14,39 @@ do
         echo "Transfer server up and running"
         break
     }
-    echo "sleep ... wait for transfer server"
+    echo "sleep ..."
     sleep 0.1
 done
 
 # Start a receiver
 mkdir -p /tmp/ws_test/receive
 cd /tmp/ws_test/receive
-ws receive --delay 10 ws://127.0.0.1:8090 --pidfile /tmp/ws_test/pidfile.receive &
+ws receive ws://127.0.0.1:8090 &
 
 mkdir /tmp/ws_test/send
 cd /tmp/ws_test/send
-dd if=/dev/urandom of=20M_file count=20000 bs=1024
+# mkfile 10m 10M_file
+dd if=/dev/urandom of=10M_file count=10000 bs=1024
 
 # Start the sender job
-ws send --pidfile /tmp/ws_test/pidfile.send ws://127.0.0.1:8090 20M_file
+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/20M_file ; then
+    if test -f /tmp/ws_test/receive/10M_file ; then
         echo "Received file does exists, exiting loop"
         break
     fi
-    echo "sleep ... wait for output file"
+    echo "sleep ..."
     sleep 0.1
 done
 
-cksum /tmp/ws_test/send/20M_file
+cksum /tmp/ws_test/send/10M_file
-cksum /tmp/ws_test/receive/20M_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.transfer`
+kill `cat /tmp/ws_test/pidfile`
-kill `cat /tmp/ws_test/pidfile.receive`
-kill `cat /tmp/ws_test/pidfile.send`
-

ws/ws.cpp

@@ -35,31 +35,22 @@ int main(int argc, char** argv)
     std::string output;
     std::string hostname("127.0.0.1");
     std::string pidfile;
-    std::string channel;
-    std::string message;
     bool headersOnly = false;
     bool followRedirects = false;
     bool verbose = false;
     bool save = false;
     bool compress = false;
     int port = 8080;
-    int redisPort = 6379;
     int connectTimeOut = 60;
     int transferTimeout = 1800;
     int maxRedirects = 5;
-    int delayMs = -1;
 
     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()->check(CLI::ExistingPath);
-    sendApp->add_option("--pidfile", pidfile, "Pid file");
-
     CLI::App* receiveApp = app.add_subcommand("receive", "Receive a file");
     receiveApp->add_option("url", url, "Connection url")->required();
-    receiveApp->add_option("--delay", delayMs, "Delay (ms) to wait after receiving a fragment"
-                                               " to artificially slow down the receiver");
-    receiveApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
     transferApp->add_option("--port", port, "Connection url");
@@ -99,21 +90,14 @@ int main(int argc, char** argv)
     httpClientApp->add_option("--connect-timeout", connectTimeOut, "Connection timeout");
     httpClientApp->add_option("--transfer-timeout", transferTimeout, "Transfer timeout");
 
-    CLI::App* redisPublishApp = app.add_subcommand("redis_publish", "Redis publisher");
-    redisPublishApp->add_option("--port", redisPort, "Port");
-    redisPublishApp->add_option("--host", hostname, "Hostname");
-    redisPublishApp->add_option("channel", channel, "Channel")->required();
-    redisPublishApp->add_option("message", message, "Message")->required();
-
-    CLI::App* redisSubscribeApp = app.add_subcommand("redis_subscribe", "Redis subscriber");
-    redisSubscribeApp->add_option("--port", redisPort, "Port");
-    redisSubscribeApp->add_option("--host", hostname, "Hostname");
-    redisSubscribeApp->add_option("channel", channel, "Channel")->required();
-    redisSubscribeApp->add_flag("-v", verbose, "Verbose");
-
     CLI11_PARSE(app, argc, argv);
 
+    ix::Socket::init();
+
     // pid file handling
+
+    if (app.got_subcommand("transfer"))
+    {
         if (!pidfile.empty())
         {
             unlink(pidfile.c_str());
@@ -124,8 +108,6 @@ int main(int argc, char** argv)
             f.close();
         }
 
-    if (app.got_subcommand("transfer"))
-    {
         return ix::ws_transfer_main(port, hostname);
     }
     else if (app.got_subcommand("send"))
@@ -135,7 +117,7 @@ int main(int argc, char** argv)
     else if (app.got_subcommand("receive"))
     {
         bool enablePerMessageDeflate = false;
-        return ix::ws_receive_main(url, enablePerMessageDeflate, delayMs);
+        return ix::ws_receive_main(url, enablePerMessageDeflate);
     }
     else if (app.got_subcommand("connect"))
     {
@@ -164,14 +146,6 @@ int main(int argc, char** argv)
                                        followRedirects, maxRedirects, verbose,
                                        save, output, compress);
     }
-    else if (app.got_subcommand("redis_publish"))
-    {
-        return ix::ws_redis_publish_main(hostname, redisPort, channel, message);
-    }
-    else if (app.got_subcommand("redis_subscribe"))
-    {
-        return ix::ws_redis_subscribe_main(hostname, redisPort, channel, verbose);
-    }
 
     return 1;
 }

ws/ws.h

@@ -34,19 +34,8 @@ namespace ix
     int ws_connect_main(const std::string& url);
 
     int ws_receive_main(const std::string& url,
-                        bool enablePerMessageDeflate,
+                        bool enablePerMessageDeflate);
-                        int delayMs);
 
     int ws_send_main(const std::string& url,
                      const std::string& path);
-
-    int ws_redis_publish_main(const std::string& hostname,
-                              int port,
-                              const std::string& channel,
-                              const std::string& message);
-
-    int ws_redis_subscribe_main(const std::string& hostname,
-                                int port,
-                                const std::string& channel,
-                                bool verbose);
 }

ws/ws_broadcast_server.cpp

@@ -17,11 +17,10 @@ namespace ix
         ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<WebSocket> webSocket,
+            [&server](std::shared_ptr<ix::WebSocket> webSocket)
-                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -31,7 +30,6 @@ namespace ix
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             std::cerr << "New connection" << std::endl;
-                            std::cerr << "id: " << connectionState->getId() << std::endl;
                             std::cerr << "Uri: " << openInfo.uri << std::endl;
                             std::cerr << "Headers:" << std::endl;
                             for (auto it : openInfo.headers)

ws/ws_connect.cpp

@@ -153,6 +153,7 @@ namespace ix
 
     int ws_connect_main(const std::string& url)
     {
+        Socket::init();
         interactiveMain(url);
         return 0;
     }

ws/ws_echo_server.cpp

@@ -17,11 +17,10 @@ namespace ix
         ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
-            [](std::shared_ptr<ix::WebSocket> webSocket,
+            [](std::shared_ptr<ix::WebSocket> webSocket)
-               std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState](ix::WebSocketMessageType messageType,
+                    [webSocket](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -31,7 +30,6 @@ namespace ix
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             std::cerr << "New connection" << std::endl;
-                            std::cerr << "id: " << connectionState->getId() << std::endl;
                             std::cerr << "Uri: " << openInfo.uri << std::endl;
                             std::cerr << "Headers:" << std::endl;
                             for (auto it : openInfo.headers)

ws/ws_receive.cpp

@@ -26,8 +26,7 @@ namespace ix
     {
         public:
             WebSocketReceiver(const std::string& _url,
-                              bool enablePerMessageDeflate,
+                              bool enablePerMessageDeflate);
-                              int delayMs);
 
             void subscribe(const std::string& channel);
             void start();
@@ -42,8 +41,6 @@ namespace ix
             std::string _id;
             ix::WebSocket _webSocket;
             bool _enablePerMessageDeflate;
-            int _delayMs;
-            int _receivedFragmentCounter;
 
             std::mutex _conditionVariableMutex;
             std::condition_variable _condition;
@@ -54,12 +51,9 @@ namespace ix
     };
 
     WebSocketReceiver::WebSocketReceiver(const std::string& url,
-                                         bool enablePerMessageDeflate,
+                                         bool enablePerMessageDeflate) :
-                                         int delayMs) :
         _url(url),
-        _enablePerMessageDeflate(enablePerMessageDeflate),
+        _enablePerMessageDeflate(enablePerMessageDeflate)
-        _delayMs(delayMs),
-        _receivedFragmentCounter(0)
     {
         ;
     }
@@ -219,19 +213,11 @@ namespace ix
                 }
                 else if (messageType == ix::WebSocket_MessageType_Fragment)
                 {
-                    ss << "ws_receive: received fragment " << _receivedFragmentCounter++;
+                    ss << "ws_receive: received fragment";
                     log(ss.str());
-
-                    if (_delayMs > 0)
-                    {
-                        // Introduce an arbitrary delay, to simulate a slow connection
-                        std::chrono::duration<double, std::milli> duration(_delayMs);
-                        std::this_thread::sleep_for(duration);
-                    }
                 }
                 else if (messageType == ix::WebSocket_MessageType_Error)
                 {
-                    ss << "ws_receive ";
                     ss << "Connection error: " << error.reason      << std::endl;
                     ss << "#retries: "         << error.retries     << std::endl;
                     ss << "Wait time(ms): "    << error.wait_time   << std::endl;
@@ -249,10 +235,9 @@ namespace ix
     }
 
     void wsReceive(const std::string& url,
-                   bool enablePerMessageDeflate,
+                   bool enablePerMessageDeflate)
-                   int delayMs)
     {
-        WebSocketReceiver webSocketReceiver(url, enablePerMessageDeflate, delayMs);
+        WebSocketReceiver webSocketReceiver(url, enablePerMessageDeflate);
         webSocketReceiver.start();
 
         webSocketReceiver.waitForConnection();
@@ -267,10 +252,10 @@ namespace ix
     }
 
     int ws_receive_main(const std::string& url,
-                        bool enablePerMessageDeflate,
+                        bool enablePerMessageDeflate)
-                        int delayMs)
     {
-        wsReceive(url, enablePerMessageDeflate, delayMs);
+        Socket::init();
+        wsReceive(url, enablePerMessageDeflate);
         return 0;
     }
 }

ws/ws_redis_publish.cpp

@@ -1,35 +0,0 @@
-/*
- *  ws_redis_publish.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include "IXRedisClient.h"
-
-namespace ix
-{
-    int ws_redis_publish_main(const std::string& hostname,
-                              int port,
-                              const std::string& channel,
-                              const std::string& message)
-    {
-        RedisClient redisClient;
-        if (!redisClient.connect(hostname, port))
-        {
-            std::cerr << "Cannot connect to redis host" << std::endl;
-            return 1;
-        }
-
-        std::cerr << "Publishing message " << message
-                  << " to " << channel << "..." << std::endl;
-        if (!redisClient.publish(channel, message))
-        {
-            std::cerr << "Error publishing to channel " << channel << std::endl;
-            return 1;
-        }
-
-        return 0;
-    }
-}

ws/ws_redis_subscribe.cpp

@@ -1,66 +0,0 @@
-/*
- *  ws_redis_subscribe.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include <chrono>
-#include "IXRedisClient.h"
-
-namespace ix
-{
-    int ws_redis_subscribe_main(const std::string& hostname,
-                                int port,
-                                const std::string& channel,
-                                bool verbose)
-    {
-        RedisClient redisClient;
-        if (!redisClient.connect(hostname, port))
-        {
-            std::cerr << "Cannot connect to redis host" << std::endl;
-            return 1;
-        }
-
-        std::chrono::time_point<std::chrono::steady_clock> lastTimePoint;
-        int msgPerSeconds = 0;
-        int msgCount = 0;
-
-        auto callback = [&lastTimePoint, &msgPerSeconds, &msgCount, verbose]
-                         (const std::string& message)
-        {
-            if (verbose)
-            {
-                std::cout << message << std::endl;
-            }
-
-            msgPerSeconds++;
-
-            auto now = std::chrono::steady_clock::now();
-            if (now - lastTimePoint > std::chrono::seconds(1))
-            {
-                lastTimePoint = std::chrono::steady_clock::now();
-
-                msgCount += msgPerSeconds;
-
-                // #messages 901 msg/s 150
-                std::cout << "#messages " << msgCount << " "
-                          << "msg/s " << msgPerSeconds
-                          << std::endl;
-
-                msgPerSeconds = 0;
-            }
-        };
-
-        std::cerr << "Subscribing to " << channel << "..." << std::endl;
-        if (!redisClient.subscribe(channel, callback))
-        {
-            std::cerr << "Error subscribing to channel " << channel << std::endl;
-            return 1;
-        }
-
-        return 0;
-    }
-}
-

ws/ws_send.cpp

@@ -162,7 +162,6 @@ namespace ix
                 }
                 else if (messageType == ix::WebSocket_MessageType_Error)
                 {
-                    ss << "ws_send ";
                     ss << "Connection error: " << error.reason      << std::endl;
                     ss << "#retries: "         << error.retries     << std::endl;
                     ss << "Wait time(ms): "    << error.wait_time   << std::endl;
@@ -247,7 +246,7 @@ namespace ix
         _webSocket.send(msg.dump(),
                         [throttle](int current, int total) -> bool
         {
-            std::cout << "ws_send: Step " << current << " out of " << total << std::endl;
+            std::cout << "Step " << current << " out of " << total << std::endl;
 
             if (throttle)
             {
@@ -261,8 +260,7 @@ namespace ix
         do
         {
             size_t bufferedAmount = _webSocket.bufferedAmount();
-            std::cout << "ws_send: " << bufferedAmount
+            std::cout << bufferedAmount << " bytes left to be sent" << std::endl;
-                      << " bytes left to be sent" << std::endl;
 
             std::chrono::duration<double, std::milli> duration(10);
             std::this_thread::sleep_for(duration);
@@ -300,6 +298,7 @@ namespace ix
         bool throttle = false;
         bool enablePerMessageDeflate = false;
 
+        Socket::init();
         wsSend(url, path, enablePerMessageDeflate, throttle);
         return 0;
     }

ws/ws_transfer.cpp

@@ -17,11 +17,10 @@ namespace ix
         ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
-            [&server](std::shared_ptr<ix::WebSocket> webSocket,
+            [&server](std::shared_ptr<ix::WebSocket> webSocket)
-                      std::shared_ptr<ConnectionState> connectionState)
             {
                 webSocket->setOnMessageCallback(
-                    [webSocket, connectionState, &server](ix::WebSocketMessageType messageType,
+                    [webSocket, &server](ix::WebSocketMessageType messageType,
                        const std::string& str,
                        size_t wireSize,
                        const ix::WebSocketErrorInfo& error,
@@ -31,7 +30,6 @@ namespace ix
                         if (messageType == ix::WebSocket_MessageType_Open)
                         {
                             std::cerr << "New connection" << std::endl;
-                            std::cerr << "id: " << connectionState->getId() << std::endl;
                             std::cerr << "Uri: " << openInfo.uri << std::endl;
                             std::cerr << "Headers:" << std::endl;
                             for (auto it : openInfo.headers)
@@ -56,8 +54,7 @@ namespace ix
                         }
                         else if (messageType == ix::WebSocket_MessageType_Fragment)
                         {
-                            std::cerr << "Received message fragment "
+                            std::cerr << "Received message fragment" << std::endl;
-                                      << std::endl;
                         }
                         else if (messageType == ix::WebSocket_MessageType_Message)
                         {
@@ -69,7 +66,7 @@ namespace ix
                                     client->send(str,
                                                  [](int current, int total) -> bool
                                     {
-                                        std::cerr << "ws_transfer: Step " << current
+                                        std::cerr << "Step " << current
                                                   << " out of " << total << std::endl;
                                         return true;
                                     });
@@ -77,8 +74,7 @@ namespace ix
                                     do
                                     {
                                         size_t bufferedAmount = client->bufferedAmount();
-                                        std::cerr << "ws_transfer: " << bufferedAmount
+                                        std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
-                                                  << " bytes left to be sent" << std::endl;
 
                                         std::chrono::duration<double, std::milli> duration(10);
                                         std::this_thread::sleep_for(duration);