(websocket server) Handle programmer error when the server callback is not registered properly (fix #227)

.github/workflows/unittest_uwp.yml

@@ -10,12 +10,10 @@ jobs:
     steps:
     - uses: actions/checkout@v1
     - uses: seanmiddleditch/gha-setup-vsdevenv@master
-    - run: |
-        vcpkg install zlib:x64-uwp
     - run: |
         mkdir build
         cd build
-        cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_SYSTEM_NAME=WindowsStore -DCMAKE_SYSTEM_VERSION="10.0" -DCMAKE_CXX_COMPILER=cl.exe -DUSE_TEST=1 ..
+        cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_SYSTEM_NAME=WindowsStore -DCMAKE_SYSTEM_VERSION="10.0" -DCMAKE_CXX_COMPILER=cl.exe -DUSE_TEST=1 -DUSE_ZLIB=0 ..
     - run: cmake --build build
 
 #

.github/workflows/unittest_windows.yml

@@ -10,10 +10,11 @@ jobs:
     steps:
     - uses: actions/checkout@v1
     - uses: seanmiddleditch/gha-setup-vsdevenv@master
-    - run: |
-        vcpkg install zlib:x64-windows
     - run: |
         mkdir build
         cd build
-        cmake -DCMAKE_TOOLCHAIN_FILE=c:/vcpkg/scripts/buildsystems/vcpkg.cmake -DCMAKE_CXX_COMPILER=cl.exe -DUSE_WS=1 -DUSE_TEST=1 ..
+        cmake -DCMAKE_CXX_COMPILER=cl.exe -DUSE_WS=1 -DUSE_TEST=1 -DUSE_ZLIB=0 ..
     - run: cmake --build build
+
+#- run: ../build/test/ixwebsocket_unittest.exe
+# working-directory: test

CMakeLists.txt

@@ -190,10 +190,16 @@ if (USE_TLS)
   endif()
 endif()
 
-# Use ZLIB_ROOT CMake variable if you need to use your own zlib
+option(USE_ZLIB "Enable zlib support" TRUE)
-find_package(ZLIB REQUIRED)
+
-include_directories(${ZLIB_INCLUDE_DIRS})
+if (USE_ZLIB)
-target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})
+  # Use ZLIB_ROOT CMake variable if you need to use your own zlib
+  find_package(ZLIB REQUIRED)
+  include_directories(${ZLIB_INCLUDE_DIRS})
+  target_link_libraries(ixwebsocket ${ZLIB_LIBRARIES})
+
+  target_compile_definitions(ixwebsocket PUBLIC IXWEBSOCKET_USE_ZLIB)
+endif()
 
 if (WIN32)
   target_link_libraries(ixwebsocket wsock32 ws2_32 shlwapi)

docs/CHANGELOG.md

@@ -1,6 +1,27 @@
 # Changelog
+
 All changes to this project will be documented in this file.
 
+## [10.1.6] - 2020-08-06
+
+(websocket server) Handle programmer error when the server callback is not registered properly (fix #227)
+
+## [10.1.5] - 2020-08-02
+
+(ws) Add a new ws sub-command, push_server. This command runs a server which sends many messages in a loop to a websocket client. We can receive above 200,000 messages per second (cf #235).
+
+## [10.1.4] - 2020-08-02
+
+(ws) Add a new ws sub-command, echo_client. This command sends a message to an echo server, and send back to a server whatever message it does receive. When connecting to a local ws echo_server, on my MacBook Pro 2015 I can send/receive around 30,000 messages per second. (cf #235)
+
+## [10.1.3] - 2020-08-02
+
+(ws) ws echo_server. Add a -q option to only enable warning and error log levels. This is useful for bench-marking so that we do not print a lot of things on the console. (cf #235)
+
+## [10.1.2] - 2020-07-31
+
+(build) make using zlib optional, with the caveat that some http and websocket features are not available when zlib is absent
+
 ## [10.1.1] - 2020-07-29
 
 (websocket client) onProgressCallback not called for short messages on a websocket (fix #233)

docs/build.md

@@ -17,6 +17,7 @@ There is a unittest which can be executed by typing `make test`.
 
 Options for building:
 
+* `-DUSE_ZLIB=1` will enable zlib support, required for http client + server + websocket per message deflate extension
 * `-DUSE_TLS=1` will enable TLS support
 * `-DUSE_OPEN_SSL=1` will use [openssl](https://www.openssl.org/) for the TLS support (default on Linux and Windows)
 * `-DUSE_MBED_TLS=1` will use [mbedlts](https://tls.mbed.org/) for the TLS support

docs/performance.md

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

ixwebsocket/IXHttpClient.cpp

@@ -16,7 +16,10 @@
 #include <random>
 #include <sstream>
 #include <vector>
+
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
+#endif
 
 namespace ix
 {
@@ -174,11 +177,13 @@ namespace ix
         ss << verb << " " << path << " HTTP/1.1\r\n";
         ss << "Host: " << host << "\r\n";
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         if (args->compress)
         {
             ss << "Accept-Encoding: gzip"
                << "\r\n";
         }
+#endif
 
         // Append extra headers
         for (auto&& it : args->extraHeaders)
@@ -495,6 +500,7 @@ namespace ix
 
         downloadSize = payload.size();
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         // If the content was compressed with gzip, decode it
         if (headers["Content-Encoding"] == "gzip")
         {
@@ -513,6 +519,7 @@ namespace ix
             }
             payload = decompressedPayload;
         }
+#endif
 
         return std::make_shared<HttpResponse>(code,
                                               description,
@@ -672,6 +679,7 @@ namespace ix
         return ss.str();
     }
 
+#ifdef IXWEBSOCKET_USE_ZLIB
     bool HttpClient::gzipInflate(const std::string& in, std::string& out)
     {
         z_stream inflateState;
@@ -716,6 +724,7 @@ namespace ix
         inflateEnd(&inflateState);
         return true;
     }
+#endif
 
     void HttpClient::log(const std::string& msg, HttpRequestArgsPtr args)
     {

ixwebsocket/IXHttpClient.h

@@ -90,7 +90,9 @@ namespace ix
     private:
         void log(const std::string& msg, HttpRequestArgsPtr args);
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         bool gzipInflate(const std::string& in, std::string& out);
+#endif
 
         // Async API background thread runner
         void run();

ixwebsocket/IXHttpServer.cpp

@@ -13,7 +13,10 @@
 #include <fstream>
 #include <sstream>
 #include <vector>
+
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include <zlib.h>
+#endif
 
 namespace
 {
@@ -41,6 +44,7 @@ namespace
         return std::make_pair(res.first, std::string(vec.begin(), vec.end()));
     }
 
+#ifdef IXWEBSOCKET_USE_ZLIB
     std::string gzipCompress(const std::string& str)
     {
         z_stream zs; // z_stream is zlib's control structure
@@ -83,6 +87,7 @@ namespace
 
         return outstring;
     }
+#endif
 } // namespace
 
 namespace ix
@@ -168,12 +173,14 @@ namespace ix
 
                 std::string content = res.second;
 
+#ifdef IXWEBSOCKET_USE_ZLIB
                 std::string acceptEncoding = request->headers["Accept-encoding"];
                 if (acceptEncoding == "*" || acceptEncoding.find("gzip") != std::string::npos)
                 {
                     content = gzipCompress(content);
                     headers["Content-Encoding"] = "gzip";
                 }
+#endif
 
                 // Log request
                 std::stringstream ss;

ixwebsocket/IXUserAgent.cpp

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

ixwebsocket/IXWebSocket.cpp

@@ -405,6 +405,11 @@ namespace ix
         _onMessageCallback = callback;
     }
 
+    bool WebSocket::isOnMessageCallbackRegistered() const
+    {
+        return _onMessageCallback != nullptr;
+    }
+
     void WebSocket::setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback)
     {
         _onTrafficTrackerCallback = callback;

ixwebsocket/IXWebSocket.h

@@ -84,6 +84,7 @@ namespace ix
                    const std::string& reason = WebSocketCloseConstants::kNormalClosureMessage);
 
         void setOnMessageCallback(const OnMessageCallback& callback);
+        bool isOnMessageCallbackRegistered() const;
         static void setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback);
         static void resetTrafficTrackerCallback();
 

ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp

@@ -28,21 +28,26 @@ namespace ix
     WebSocketPerMessageDeflateCompressor::WebSocketPerMessageDeflateCompressor()
         : _compressBufferSize(kBufferSize)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_deflateState, 0, sizeof(_deflateState));
 
         _deflateState.zalloc = Z_NULL;
         _deflateState.zfree = Z_NULL;
         _deflateState.opaque = Z_NULL;
+#endif
     }
 
     WebSocketPerMessageDeflateCompressor::~WebSocketPerMessageDeflateCompressor()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         deflateEnd(&_deflateState);
+#endif
     }
 
     bool WebSocketPerMessageDeflateCompressor::init(uint8_t deflateBits,
                                                     bool clientNoContextTakeOver)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         int ret = deflateInit2(&_deflateState,
                                Z_DEFAULT_COMPRESSION,
                                Z_DEFLATED,
@@ -57,6 +62,9 @@ namespace ix
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
+#else
+        return false;
+#endif
     }
 
     template<typename T>
@@ -96,6 +104,7 @@ namespace ix
     template<typename T, typename S>
     bool WebSocketPerMessageDeflateCompressor::compressData(const T& in, S& out)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         //
         // 7.2.1.  Compression
         //
@@ -152,6 +161,9 @@ namespace ix
         }
 
         return true;
+#else
+        return false;
+#endif
     }
 
     //
@@ -160,6 +172,7 @@ namespace ix
     WebSocketPerMessageDeflateDecompressor::WebSocketPerMessageDeflateDecompressor()
         : _compressBufferSize(kBufferSize)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         memset(&_inflateState, 0, sizeof(_inflateState));
 
         _inflateState.zalloc = Z_NULL;
@@ -167,16 +180,20 @@ namespace ix
         _inflateState.opaque = Z_NULL;
         _inflateState.avail_in = 0;
         _inflateState.next_in = Z_NULL;
+#endif
     }
 
     WebSocketPerMessageDeflateDecompressor::~WebSocketPerMessageDeflateDecompressor()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         inflateEnd(&_inflateState);
+#endif
     }
 
     bool WebSocketPerMessageDeflateDecompressor::init(uint8_t inflateBits,
                                                       bool clientNoContextTakeOver)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         int ret = inflateInit2(&_inflateState, -1 * inflateBits);
 
         if (ret != Z_OK) return false;
@@ -186,10 +203,14 @@ namespace ix
         _flush = (clientNoContextTakeOver) ? Z_FULL_FLUSH : Z_SYNC_FLUSH;
 
         return true;
+#else
+        return false;
+#endif
     }
 
     bool WebSocketPerMessageDeflateDecompressor::decompress(const std::string& in, std::string& out)
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         //
         // 7.2.2.  Decompression
         //
@@ -226,5 +247,8 @@ namespace ix
         } while (_inflateState.avail_out == 0);
 
         return true;
+#else
+        return false;
+#endif
     }
 } // namespace ix

ixwebsocket/IXWebSocketPerMessageDeflateCodec.h

@@ -6,7 +6,9 @@
 
 #pragma once
 
+#ifdef IXWEBSOCKET_USE_ZLIB
 #include "zlib.h"
+#endif
 #include <memory>
 #include <string>
 #include <vector>
@@ -34,7 +36,10 @@ namespace ix
         int _flush;
         size_t _compressBufferSize;
         std::unique_ptr<unsigned char[]> _compressBuffer;
+
+#ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _deflateState;
+#endif
     };
 
     class WebSocketPerMessageDeflateDecompressor
@@ -50,7 +55,10 @@ namespace ix
         int _flush;
         size_t _compressBufferSize;
         std::unique_ptr<unsigned char[]> _compressBuffer;
+
+#ifdef IXWEBSOCKET_USE_ZLIB
         z_stream _inflateState;
+#endif
     };
 
 } // namespace ix

ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp

@@ -61,6 +61,7 @@ namespace ix
         _clientMaxWindowBits = kDefaultClientMaxWindowBits;
         _serverMaxWindowBits = kDefaultServerMaxWindowBits;
 
+#ifdef IXWEBSOCKET_USE_ZLIB
         // Split by ;
         std::string token;
         std::stringstream tokenStream(extension);
@@ -112,6 +113,7 @@ namespace ix
                 sanitizeClientMaxWindowBits();
             }
         }
+#endif
     }
 
     void WebSocketPerMessageDeflateOptions::sanitizeClientMaxWindowBits()
@@ -126,6 +128,7 @@ namespace ix
 
     std::string WebSocketPerMessageDeflateOptions::generateHeader()
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         std::stringstream ss;
         ss << "Sec-WebSocket-Extensions: permessage-deflate";
 
@@ -138,11 +141,18 @@ namespace ix
         ss << "\r\n";
 
         return ss.str();
+#else
+        return std::string();
+#endif
     }
 
     bool WebSocketPerMessageDeflateOptions::enabled() const
     {
+#ifdef IXWEBSOCKET_USE_ZLIB
         return _enabled;
+#else
+        return false;
+#endif
     }
 
     bool WebSocketPerMessageDeflateOptions::getClientNoContextTakeover() const

ixwebsocket/IXWebSocketServer.cpp

@@ -86,6 +86,15 @@ namespace ix
         if (_onConnectionCallback)
         {
             _onConnectionCallback(webSocket, connectionState, std::move(connectionInfo));
+
+            if (!webSocket->isOnMessageCallbackRegistered())
+            {
+                logError("WebSocketServer Application developer error: Server callback improperly "
+                         "registerered.");
+                logError("Missing call to setOnMessageCallback inside setOnConnectionCallback.");
+                connectionState->setTerminated();
+                return;
+            }
         }
         else if (_onClientMessageCallback)
         {

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "10.1.1"
+#define IX_WEBSOCKET_VERSION "10.1.5"

makefile

@@ -34,7 +34,10 @@ ws:
 	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && ninja install)
 
 ws_install:
-	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. && make -j 4 install)
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. -DUSE_TEST=0 && ninja install)
+
+ws_install_release:
+	mkdir -p build && (cd build ; cmake -GNinja -DCMAKE_BUILD_TYPE=MinSizeRel -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 .. -DUSE_TEST=0 && ninja install)
 
 ws_openssl:
 	mkdir -p build && (cd build ; cmake -DCMAKE_BUILD_TYPE=Debug -DUSE_PYTHON=1 -DUSE_TLS=1 -DUSE_WS=1 -DUSE_OPEN_SSL=1 .. ; make -j 4)

test/compatibility/cpp/libwebsockets/devnull_client.cpp

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

test/compatibility/csharp/.gitignore

@@ -0,0 +1,2 @@
+bin
+obj

test/compatibility/csharp/Main.cs

@@ -0,0 +1,99 @@
+//
+// Main.cs
+// Author: Benjamin Sergeant
+// Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+//
+// In a different terminal, start a push server:
+// $ ws push_server -q
+//
+// $ dotnet run
+// messages received per second: 145157
+// messages received per second: 141405
+// messages received per second: 152202
+// messages received per second: 157149
+// messages received per second: 157673
+// messages received per second: 153594
+// messages received per second: 157830
+// messages received per second: 158422
+//
+
+using System;
+using System.Net.WebSockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+public class DevNullClientCli
+{
+    private static int receivedMessage = 0;
+
+    public static async Task<byte[]> ReceiveAsync(ClientWebSocket ws, CancellationToken token)
+    {
+        int bufferSize = 8192; // 8K
+        var buffer = new byte[bufferSize];
+        var offset = 0;
+        var free = buffer.Length;
+
+        while (true)
+        {
+            var result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer, offset, free), token).ConfigureAwait(false);
+
+            offset += result.Count;
+            free -= result.Count;
+            if (result.EndOfMessage) break;
+
+            if (free == 0)
+            {
+                // No free space
+                // Resize the outgoing buffer
+                var newSize = buffer.Length + bufferSize;
+
+                var newBuffer = new byte[newSize];
+                Array.Copy(buffer, 0, newBuffer, 0, offset);
+                buffer = newBuffer;
+                free = buffer.Length - offset;
+            }
+        }
+
+        return buffer;
+    }
+
+    private static void OnTimedEvent(object source, EventArgs e)
+    {
+        Console.WriteLine($"messages received per second: {receivedMessage}");
+        receivedMessage = 0;
+    }
+
+    public static async Task ReceiveMessagesAsync(string url)
+    {
+        var ws = new ClientWebSocket();
+
+        System.Uri uri = new System.Uri(url);
+        var cancellationToken = CancellationToken.None;
+
+        try 
+        {
+            await ws.ConnectAsync(uri, cancellationToken).ConfigureAwait(false);
+            while (true)
+            {
+                var data = await DevNullClientCli.ReceiveAsync(ws, cancellationToken);
+                receivedMessage += 1;
+            }
+        }
+        catch (System.Net.WebSockets.WebSocketException e)
+        {
+            Console.WriteLine($"WebSocket error: {e}");
+            return;
+        }
+    }
+
+    public static async Task Main()
+    {
+        var timer = new System.Timers.Timer(1000);
+        timer.Elapsed += OnTimedEvent;
+        timer.Enabled = true;
+        timer.Start();
+
+        var url = "ws://localhost:8008";
+        await ReceiveMessagesAsync(url);
+    }
+}

test/compatibility/csharp/devnull_client.csproj

@@ -0,0 +1,6 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+</Project>

test/compatibility/node/devnull_client.js

@@ -0,0 +1,42 @@
+//
+// With ws@7.3.1
+// and 
+// node --version
+// v13.11.0
+//
+// In a different terminal, start a push server:
+// $ ws push_server -q
+//
+// $ node devnull_client.js
+// messages received per second: 16643
+// messages received per second: 28065
+// messages received per second: 28432
+// messages received per second: 22207
+// messages received per second: 28805
+// messages received per second: 28694
+// messages received per second: 28180
+// messages received per second: 28601
+// messages received per second: 28698
+// messages received per second: 28931
+// messages received per second: 27975
+//
+const WebSocket = require('ws');
+
+const ws = new WebSocket('ws://localhost:8008');
+
+ws.on('open', function open() {
+  ws.send('hello from node');
+});
+
+var receivedMessages = 0;
+
+setInterval(function timeout() {
+  console.log(`messages received per second: ${receivedMessages}`)
+  receivedMessages = 0;
+}, 1000);
+
+ws.on('message', function incoming(data) {
+  receivedMessages += 1;
+});
+
+

test/compatibility/python/websockets/devnull_client.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+
+# websocket send client
+
+import argparse
+import asyncio
+import websockets
+
+try:
+    import uvloop
+    uvloop.install()
+except ImportError:
+    print('uvloop not available')
+    pass
+
+msgCount = 0
+
+async def timer():
+    global msgCount
+
+    while True:
+        print(f'Received messages: {msgCount}')
+        msgCount = 0
+
+        await asyncio.sleep(1)
+
+
+async def client(url):
+    global msgCount
+
+    asyncio.ensure_future(timer())
+
+    async with websockets.connect(url) as ws:
+        async for message in ws:
+            msgCount += 1
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='websocket proxy.')
+    parser.add_argument('--url', help='Remote websocket url',
+                        default='wss://echo.websocket.org')
+    args = parser.parse_args()
+
+    asyncio.get_event_loop().run_until_complete(client(args.url))

test/compatibility/python/websockets/echo_server.py

@@ -10,7 +10,7 @@ import websockets
 async def echo(websocket, path):
     while True:
         msg = await websocket.recv()
-        print(f'Received {len(msg)} bytes')
+        # print(f'Received {len(msg)} bytes')
         await websocket.send(msg)
 
 host = os.getenv('BIND_HOST', 'localhost')

ws/CMakeLists.txt

@@ -50,7 +50,9 @@ add_executable(ws
   ws_http_client.cpp
   ws_ping_pong.cpp
   ws_broadcast_server.cpp
+  ws_push_server.cpp
   ws_echo_server.cpp
+  ws_echo_client.cpp
   ws_chat.cpp
   ws_connect.cpp
   ws_transfer.cpp

ws/ws.cpp

@@ -125,6 +125,7 @@ int main(int argc, char** argv)
     std::string logfile;
     std::string scriptPath;
     std::string republishChannel;
+    std::string sendMsg("hello world");
     ix::SocketTLSOptions tlsOptions;
     ix::CobraConfig cobraConfig;
     ix::CobraBotConfig cobraBotConfig;
@@ -147,6 +148,7 @@ int main(int argc, char** argv)
     bool version = false;
     bool verifyNone = false;
     bool disablePong = false;
+    bool noSend = false;
     int port = 8008;
     int redisPort = 6379;
     int statsdPort = 8125;
@@ -243,6 +245,19 @@ int main(int argc, char** argv)
     connectApp->add_option("--subprotocol", subprotocol, "Subprotocol");
     addTLSOptions(connectApp);
 
+    CLI::App* echoClientApp =
+        app.add_subcommand("echo_client", "Echo messages sent by a remote server");
+    echoClientApp->fallthrough();
+    echoClientApp->add_option("url", url, "Connection url")->required();
+    echoClientApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
+    echoClientApp->add_flag("-b", binaryMode, "Send in binary mode");
+    echoClientApp->add_option(
+        "--ping_interval", pingIntervalSecs, "Interval between sending pings");
+    echoClientApp->add_option("--subprotocol", subprotocol, "Subprotocol");
+    echoClientApp->add_option("--send_msg", sendMsg, "Send message");
+    echoClientApp->add_flag("-m", noSend, "Do not send messages, only receive messages");
+    addTLSOptions(echoClientApp);
+
     CLI::App* chatApp = app.add_subcommand("chat", "Group chat");
     chatApp->fallthrough();
     chatApp->add_option("url", url, "Connection url")->required();
@@ -252,12 +267,25 @@ int main(int argc, char** argv)
     echoServerApp->fallthrough();
     echoServerApp->add_option("--port", port, "Port");
     echoServerApp->add_option("--host", hostname, "Hostname");
-    echoServerApp->add_flag("-g", greetings, "Verbose");
+    echoServerApp->add_flag("-q", quiet, "Quiet / only display warnings and errors");
+    echoServerApp->add_flag("-g", greetings, "Greet");
     echoServerApp->add_flag("-6", ipv6, "IpV6");
     echoServerApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
     echoServerApp->add_flag("-p", disablePong, "Disable sending PONG in response to PING");
     addTLSOptions(echoServerApp);
 
+    CLI::App* pushServerApp = app.add_subcommand("push_server", "Push server");
+    pushServerApp->fallthrough();
+    pushServerApp->add_option("--port", port, "Port");
+    pushServerApp->add_option("--host", hostname, "Hostname");
+    pushServerApp->add_flag("-q", quiet, "Quiet / only display warnings and errors");
+    pushServerApp->add_flag("-g", greetings, "Greet");
+    pushServerApp->add_flag("-6", ipv6, "IpV6");
+    pushServerApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
+    pushServerApp->add_flag("-p", disablePong, "Disable sending PONG in response to PING");
+    pushServerApp->add_option("--send_msg", sendMsg, "Send message");
+    addTLSOptions(pushServerApp);
+
     CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
     broadcastServerApp->fallthrough();
     broadcastServerApp->add_option("--port", port, "Port");
@@ -477,6 +505,11 @@ int main(int argc, char** argv)
         }
     }
 
+    if (quiet)
+    {
+        spdlog::set_level(spdlog::level::warn);
+    }
+
     // Cobra config
     cobraConfig.webSocketPerMessageDeflateOptions = ix::WebSocketPerMessageDeflateOptions(true);
     cobraConfig.socketTLSOptions = tlsOptions;
@@ -498,11 +531,33 @@ int main(int argc, char** argv)
                                   subprotocol,
                                   pingIntervalSecs);
     }
+    else if (app.got_subcommand("echo_client"))
+    {
+        ret = ix::ws_echo_client(url,
+                                 disablePerMessageDeflate,
+                                 binaryMode,
+                                 tlsOptions,
+                                 subprotocol,
+                                 pingIntervalSecs,
+                                 sendMsg,
+                                 noSend);
+    }
     else if (app.got_subcommand("echo_server"))
     {
         ret = ix::ws_echo_server_main(
             port, greetings, hostname, tlsOptions, ipv6, disablePerMessageDeflate, disablePong);
     }
+    else if (app.got_subcommand("push_server"))
+    {
+        ret = ix::ws_push_server(port,
+                                 greetings,
+                                 hostname,
+                                 tlsOptions,
+                                 ipv6,
+                                 disablePerMessageDeflate,
+                                 disablePong,
+                                 sendMsg);
+    }
     else if (app.got_subcommand("transfer"))
     {
         ret = ix::ws_transfer_main(port, hostname, tlsOptions);

ws/ws.h

@@ -35,6 +35,15 @@ namespace ix
                             bool disablePerMessageDeflate,
                             bool disablePong);
 
+    int ws_push_server(int port,
+                       bool greetings,
+                       const std::string& hostname,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       bool ipv6,
+                       bool disablePerMessageDeflate,
+                       bool disablePong,
+                       const std::string& sendMsg);
+
     int ws_broadcast_server_main(int port,
                                  const std::string& hostname,
                                  const ix::SocketTLSOptions& tlsOptions);
@@ -54,6 +63,15 @@ namespace ix
                         const std::string& subprotocol,
                         int pingIntervalSecs);
 
+    int ws_echo_client(const std::string& url,
+                       bool disablePerMessageDeflate,
+                       bool binaryMode,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       const std::string& subprotocol,
+                       int pingIntervalSecs,
+                       const std::string& sendMsg,
+                       bool noSend);
+
     int ws_receive_main(const std::string& url,
                         bool enablePerMessageDeflate,
                         int delayMs,

ws/ws_connect.cpp

@@ -160,7 +160,7 @@ namespace ix
             std::stringstream ss;
             if (msg->type == ix::WebSocketMessageType::Open)
             {
-                log("ws_connect: connected");
+                spdlog::info("ws_connect: connected");
                 spdlog::info("Uri: {}", msg->openInfo.uri);
                 spdlog::info("Headers:");
                 for (auto it : msg->openInfo.headers)

ws/ws_echo_client.cpp

@@ -0,0 +1,121 @@
+/*
+ *  ws_echo_client.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <iostream>
+#include <ixcore/utils/IXCoreLogger.h>
+#include <ixwebsocket/IXNetSystem.h>
+#include <ixwebsocket/IXSetThreadName.h>
+#include <ixwebsocket/IXWebSocket.h>
+#include <spdlog/spdlog.h>
+#include <sstream>
+#include <thread>
+
+namespace ix
+{
+    int ws_echo_client(const std::string& url,
+                       bool disablePerMessageDeflate,
+                       bool binaryMode,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       const std::string& subprotocol,
+                       int pingIntervalSecs,
+                       const std::string& sendMsg,
+                       bool noSend)
+    {
+        // Our websocket object
+        ix::WebSocket webSocket;
+
+        webSocket.setUrl(url);
+        webSocket.setTLSOptions(tlsOptions);
+        webSocket.setPingInterval(pingIntervalSecs);
+
+        if (disablePerMessageDeflate)
+        {
+            webSocket.disablePerMessageDeflate();
+        }
+
+        if (!subprotocol.empty())
+        {
+            webSocket.addSubProtocol(subprotocol);
+        }
+
+        std::atomic<uint64_t> receivedCount(0);
+        uint64_t receivedCountTotal(0);
+        uint64_t receivedCountPerSecs(0);
+
+        // Setup a callback to be fired (in a background thread, watch out for race conditions !)
+        // when a message or an event (open, close, error) is received
+        webSocket.setOnMessageCallback([&webSocket, &receivedCount, &sendMsg, noSend, binaryMode](
+                                           const ix::WebSocketMessagePtr& msg) {
+            if (msg->type == ix::WebSocketMessageType::Message)
+            {
+                if (!noSend)
+                {
+                    webSocket.send(msg->str, msg->binary);
+                }
+                receivedCount++;
+            }
+            else if (msg->type == ix::WebSocketMessageType::Open)
+            {
+                spdlog::info("ws_echo_client: connected");
+                spdlog::info("Uri: {}", msg->openInfo.uri);
+                spdlog::info("Headers:");
+                for (auto it : msg->openInfo.headers)
+                {
+                    spdlog::info("{}: {}", it.first, it.second);
+                }
+
+                webSocket.send(sendMsg, binaryMode);
+            }
+            else if (msg->type == ix::WebSocketMessageType::Pong)
+            {
+                spdlog::info("Received pong {}", msg->str);
+            }
+        });
+
+        auto timer = [&receivedCount, &receivedCountTotal, &receivedCountPerSecs] {
+            setThreadName("Timer");
+            while (true)
+            {
+                //
+                // We cannot write to sentCount and receivedCount
+                // as those are used externally, so we need to introduce
+                // our own counters
+                //
+                std::stringstream ss;
+                ss << "messages received: " << receivedCountPerSecs << " per second "
+                   << receivedCountTotal << " total";
+
+                CoreLogger::info(ss.str());
+
+                receivedCountPerSecs = receivedCount - receivedCountTotal;
+                receivedCountTotal += receivedCountPerSecs;
+
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+            }
+        };
+
+        std::thread t1(timer);
+
+        // Now that our callback is setup, we can start our background thread and receive messages
+        std::cout << "Connecting to " << url << "..." << std::endl;
+        webSocket.start();
+
+        // Send a message to the server (default to TEXT mode)
+        webSocket.send("hello world");
+
+        while (true)
+        {
+            std::string text;
+            std::cout << "> " << std::flush;
+            std::getline(std::cin, text);
+
+            webSocket.send(text);
+        }
+
+        return 0;
+    }
+} // namespace ix

ws/ws_push_server.cpp

@@ -0,0 +1,108 @@
+/*
+ *  ws_push_server.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <ixwebsocket/IXNetSystem.h>
+#include <ixwebsocket/IXWebSocketServer.h>
+#include <spdlog/spdlog.h>
+#include <sstream>
+
+namespace ix
+{
+    int ws_push_server(int port,
+                       bool greetings,
+                       const std::string& hostname,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       bool ipv6,
+                       bool disablePerMessageDeflate,
+                       bool disablePong,
+                       const std::string& sendMsg)
+    {
+        spdlog::info("Listening on {}:{}", hostname, port);
+
+        ix::WebSocketServer server(port,
+                                   hostname,
+                                   SocketServer::kDefaultTcpBacklog,
+                                   SocketServer::kDefaultMaxConnections,
+                                   WebSocketServer::kDefaultHandShakeTimeoutSecs,
+                                   (ipv6) ? AF_INET6 : AF_INET);
+
+        server.setTLSOptions(tlsOptions);
+
+        if (disablePerMessageDeflate)
+        {
+            spdlog::info("Disable per message deflate");
+            server.disablePerMessageDeflate();
+        }
+
+        if (disablePong)
+        {
+            spdlog::info("Disable responding to PING messages with PONG");
+            server.disablePong();
+        }
+
+        server.setOnClientMessageCallback(
+            [greetings, &sendMsg](std::shared_ptr<ConnectionState> connectionState,
+                                  ConnectionInfo& connectionInfo,
+                                  WebSocket& webSocket,
+                                  const WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
+                if (msg->type == ix::WebSocketMessageType::Open)
+                {
+                    spdlog::info("New connection");
+                    spdlog::info("remote ip: {}", remoteIp);
+                    spdlog::info("id: {}", connectionState->getId());
+                    spdlog::info("Uri: {}", msg->openInfo.uri);
+                    spdlog::info("Headers:");
+                    for (auto it : msg->openInfo.headers)
+                    {
+                        spdlog::info("{}: {}", it.first, it.second);
+                    }
+
+                    if (greetings)
+                    {
+                        webSocket.sendText("Welcome !");
+                    }
+
+                    bool binary = false;
+                    while (true)
+                    {
+                        webSocket.send(sendMsg, binary);
+                    }
+                }
+                else if (msg->type == ix::WebSocketMessageType::Close)
+                {
+                    spdlog::info("Closed connection: client id {} code {} reason {}",
+                                 connectionState->getId(),
+                                 msg->closeInfo.code,
+                                 msg->closeInfo.reason);
+                }
+                else if (msg->type == ix::WebSocketMessageType::Error)
+                {
+                    spdlog::error("Connection error: {}", msg->errorInfo.reason);
+                    spdlog::error("#retries: {}", msg->errorInfo.retries);
+                    spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
+                    spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
+                }
+                else if (msg->type == ix::WebSocketMessageType::Message)
+                {
+                    spdlog::info("Received {} bytes", msg->wireSize);
+                    webSocket.send(msg->str, msg->binary);
+                }
+            });
+
+        auto res = server.listen();
+        if (!res.first)
+        {
+            spdlog::error(res.second);
+            return 1;
+        }
+
+        server.start();
+        server.wait();
+
+        return 0;
+    }
+} // namespace ix