do not busy loop while sending

Dockerfile

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

Dockerfile

@@ -0,0 +1,31 @@
+FROM debian:stretch
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update 
+RUN apt-get -y install g++ 
+RUN apt-get -y install libssl-dev
+RUN apt-get -y install gdb
+RUN apt-get -y install screen
+RUN apt-get -y install procps
+RUN apt-get -y install lsof
+RUN apt-get -y install libz-dev
+RUN apt-get -y install vim
+RUN apt-get -y install make
+RUN apt-get -y install cmake
+RUN apt-get -y install curl
+RUN apt-get -y install python
+RUN apt-get -y install netcat
+
+# debian strech cmake is too old for building with Docker
+COPY makefile .
+RUN ["make", "install_cmake_for_linux"]
+
+COPY . .
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-rc4-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+# RUN ["make"]
+
+EXPOSE 8765
+CMD ["/ws/ws", "transfer", "--port", "8765", "--host", "0.0.0.0"]

docker/Dockerfile

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

docker/Dockerfile.alpine

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

docker/Dockerfile.centos

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

docker/Dockerfile.debian

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

docker/Dockerfile.gcc

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

ixwebsocket/IXEventFd.cpp

@@ -17,6 +17,8 @@
 // cf Android/Kernel table here
 // https://android.stackexchange.com/questions/51651/which-android-runs-which-linux-kernel
 //
+// On macOS we use UNIX pipes to wake up select.
+//
 
 #include "IXEventFd.h"
 
@@ -24,17 +26,24 @@
 # include <sys/eventfd.h>
 #endif
 
-#ifndef _WIN32
 #include <unistd.h> // for write
-#endif
+#include <fcntl.h>
 
 namespace ix
 {
-    EventFd::EventFd() :
-        _eventfd(-1)
+    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
     }
 
@@ -42,22 +51,44 @@ namespace ix
     {
 #ifdef __linux__
         ::close(_eventfd);
+#else
+        ::close(_fildes[0]);
+        ::close(_fildes[1]);
+        _fildes[0] = -1;
+        _fildes[1] = -1;
 #endif
     }
 
-    bool EventFd::notify()
+    bool EventFd::notify(uint64_t value)
     {
-#if defined(__linux__)
-        if (_eventfd == -1) return false;
+        int fd;
 
-        // select will wake up when a non-zero value is written to our eventfd
-        uint64_t value = 1;
+#if defined(__linux__)
+        fd = _eventfd;
+#else
+        // 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(_eventfd, &value, sizeof(value)) == 8;
+        return write(fd, &value, sizeof(value)) == 8;
+    }
+
+    // TODO: return max uint64_t for errors ?
+    uint64_t EventFd::read()
+    {
+        int fd;
+
+#if defined(__linux__)
+        fd = _eventfd;
 #else
-        return true;
+        fd = _fildes[0];
 #endif
+        uint64_t value = 0;
+        ::read(fd, &value, sizeof(value));
+        return value;
     }
 
     bool EventFd::clear()
@@ -77,6 +108,10 @@ namespace ix
 
     int EventFd::getFd()
     {
+#if defined(__linux__)
         return _eventfd;
+#else
+        return _fildes[0];
+#endif
     }
 }

ixwebsocket/IXEventFd.h

@@ -6,6 +6,8 @@
 
 #pragma once
 
+#include <stdint.h>
+
 namespace ix
 {
     class EventFd {
@@ -13,11 +15,19 @@ namespace ix
         EventFd();
         virtual ~EventFd();
 
-        bool notify();
+        bool notify(uint64_t value);
         bool clear();
+        uint64_t read();
         int getFd();
 
     private:
+#if defined(__linux__)
         int _eventfd;
+#else
+        // Store file descriptors used by the communication pipe. Communication
+        // happens between a control thread and a background thread, which is
+        // blocked on select.
+        int _fildes[2];
+#endif
     };
 }

ixwebsocket/IXSocket.cpp

@@ -23,6 +23,8 @@ namespace ix
 {
     const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
     const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
+    const uint8_t Socket::kSendRequest = 1;
+    const uint8_t Socket::kCloseRequest = 2;
     constexpr size_t Socket::kChunkSize;
 
     Socket::Socket(int fd) :
@@ -44,7 +46,34 @@ namespace ix
             return;
         }
 
-        int ret = select(timeoutSecs, 0);
+        PollResultType pollResult = select(timeoutSecs, 0);
+
+        if (onPollCallback) onPollCallback(pollResult);
+    }
+
+    PollResultType Socket::select(int timeoutSecs, int timeoutMs)
+    {
+        fd_set rfds;
+        FD_ZERO(&rfds);
+        FD_SET(_sockfd, &rfds);
+
+        // File descriptor at index 0 in _fildes is the read end of the pipe
+        int eventfd = _eventfd.getFd();
+        if (eventfd != -1)
+        {
+            FD_SET(eventfd, &rfds);
+        }
+
+        struct timeval timeout;
+        timeout.tv_sec = timeoutSecs;
+        timeout.tv_usec = 1000 * timeoutMs;
+
+        // Compute the highest fd.
+        int sockfd = _sockfd;
+        int nfds = (std::max)(sockfd, eventfd);
+
+        int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
+                           (timeoutSecs < 0) ? nullptr : &timeout);
 
         PollResultType pollResult = PollResultType_ReadyForRead;
         if (ret < 0)
@@ -55,35 +84,27 @@ namespace ix
         {
             pollResult = PollResultType_Timeout;
         }
+        else if (eventfd != -1 && FD_ISSET(eventfd, &rfds))
+        {
+            uint8_t value = _eventfd.read();
 
-        if (onPollCallback) onPollCallback(pollResult);
+            if (value == kSendRequest)
+            {
+                pollResult = PollResultType_SendRequest;
+            }
+            else if (value == kCloseRequest)
+            {
+                pollResult = PollResultType_CloseRequest;
+            }
+        }
+
+        return pollResult;
     }
 
-    int Socket::select(int timeoutSecs, int timeoutMs)
+    // Wake up from poll/select by writing to the pipe which is watched by select
+    bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
     {
-        fd_set rfds;
-        FD_ZERO(&rfds);
-        FD_SET(_sockfd, &rfds);
-
-#ifdef __linux__
-        FD_SET(_eventfd.getFd(), &rfds);
-#endif
-
-        struct timeval timeout;
-        timeout.tv_sec = timeoutSecs;
-        timeout.tv_usec = 1000 * timeoutMs;
-
-        int sockfd = _sockfd;
-        int nfds = (std::max)(sockfd, _eventfd.getFd());
-        int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
-                           (timeoutSecs < 0) ? nullptr : &timeout);
-        return ret;
-    }
-
-    void Socket::wakeUpFromPoll()
-    {
-        // this will wake up the thread blocked on select, only needed on Linux
-        _eventfd.notify();
+        return _eventfd.notify(wakeUpCode);
     }
 
     bool Socket::connect(const std::string& host,

ixwebsocket/IXSocket.h

@@ -17,9 +17,9 @@
 typedef SSIZE_T ssize_t;
 #endif
 
-#include "IXEventFd.h"
 #include "IXCancellationRequest.h"
 #include "IXProgressCallback.h"
+#include "IXEventFd.h"
 
 namespace ix
 {
@@ -27,7 +27,9 @@ namespace ix
     {
         PollResultType_ReadyForRead = 0,
         PollResultType_Timeout = 1,
-        PollResultType_Error = 2
+        PollResultType_Error = 2,
+        PollResultType_SendRequest = 3,
+        PollResultType_CloseRequest = 4
     };
 
     class Socket {
@@ -39,10 +41,10 @@ namespace ix
 
         void configure();
 
-        int select(int timeoutSecs, int timeoutMs);
+        PollResultType select(int timeoutSecs, int timeoutMs);
         virtual void poll(const OnPollCallback& onPollCallback,
                           int timeoutSecs = kDefaultPollTimeout);
-        virtual void wakeUpFromPoll();
+        virtual bool wakeUpFromPoll(uint8_t wakeUpCode);
 
         // Virtual methods
         virtual bool connect(const std::string& url,
@@ -73,12 +75,15 @@ namespace ix
         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 uint8_t kSendRequest;
+        static const uint8_t kCloseRequest;
+
     protected:
         void closeSocket(int fd);
 
         std::atomic<int> _sockfd;
         std::mutex _socketMutex;
-        EventFd _eventfd;
 
     private:
         static const int kDefaultPollTimeout;
@@ -87,5 +92,7 @@ namespace ix
         // Buffer for reading from our socket. That buffer is never resized.
         std::vector<uint8_t> _readBuffer;
         static constexpr size_t kChunkSize = 1 << 15;
+
+        EventFd _eventfd;
     };
 }

ixwebsocket/IXWebSocket.cpp

@@ -379,4 +379,9 @@ namespace ix
     {
         _automaticReconnection = false;
     }
+
+    size_t WebSocket::bufferedAmount() const
+    {
+        return _ws.bufferedAmount();
+    }
 }

ixwebsocket/IXWebSocket.h

@@ -112,6 +112,7 @@ namespace ix
         const std::string& getUrl() const;
         const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
         int getHeartBeatPeriod() const;
+        size_t bufferedAmount() const;
 
         void enableAutomaticReconnection();
         void disableAutomaticReconnection();

ixwebsocket/IXWebSocketTransport.cpp

@@ -1,7 +1,31 @@
+/*
+ * The MIT License (MIT)
+ * 
+ * Copyright (c) 2012, 2013 <dhbaird@gmail.com>
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
 /*
  *  IXWebSocketTransport.cpp
  *  Author: Benjamin Sergeant
- *  Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
+ *  Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
  */
 
 //
@@ -14,14 +38,6 @@
 #include "IXUrlParser.h"
 #include "IXSocketFactory.h"
 
-#ifdef IXWEBSOCKET_USE_TLS
-# ifdef __APPLE__
-#  include "IXSocketAppleSSL.h"
-# else
-#  include "IXSocketOpenSSL.h"
-# endif
-#endif
-
 #include <string.h>
 #include <stdlib.h>
 
@@ -80,16 +96,6 @@ namespace ix
                                        std::string("Could not parse URL ") + url);
         }
 
-        if (protocol != "ws" && protocol != "wss")
-        {
-            std::stringstream ss;
-            ss << "Invalid protocol: " << protocol
-               << " for url " << url
-               << " . Supported protocols are ws and wss";
-
-            return WebSocketInitResult(false, 0, ss.str());
-        }
-
         bool tls = protocol == "wss";
         std::string errorMsg;
         _socket = createSocket(tls, errorMsg);
@@ -184,38 +190,57 @@ namespace ix
                     std::stringstream ss;
                     ss << kHeartBeatPingMessage << "::" << _heartBeatPeriod << "s";
                     sendPing(ss.str());
-                    return;
                 }
-
-                while (true)
+                // Make sure we send all the buffered data
+                // there can be a lot of it for large messages.
+                else if (pollResult == PollResultType_SendRequest)
                 {
-                    ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
+                    while (!isSendBufferEmpty() && !_requestInitCancellation)
+                    {
+                        sendOnSocket();
 
-                    if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
-                                    _socket->getErrno() == EAGAIN))
-                    {
-                        break;
-                    }
-                    else if (ret <= 0)
-                    {
-                        _rxbuf.clear();
-                        _socket->close();
-                        setReadyState(CLOSED);
-                        break;
-                    }
-                    else
-                    {
-                        _rxbuf.insert(_rxbuf.end(),
-                                      _readbuf.begin(),
-                                      _readbuf.begin() + ret);
+                        // Sleep 10ms between each send so that we dont busy loop
+                        // A better strategy would be to select on the socket to 
+                        // check whether we can write to it without blocking
+                        std::chrono::duration<double, std::micro> duration(10);
+                        std::this_thread::sleep_for(duration);
                     }
                 }
+                else if (pollResult == PollResultType_ReadyForRead)
+                {
+                    while (true)
+                    {
+                        ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
 
-                if (isSendBufferEmpty() && _readyState == CLOSING)
+                        if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
+                                        _socket->getErrno() == EAGAIN))
+                        {
+                            break;
+                        }
+                        else if (ret <= 0)
+                        {
+                            _rxbuf.clear();
+                            _socket->close();
+                            setReadyState(CLOSED);
+                            break;
+                        }
+                        else
+                        {
+                            _rxbuf.insert(_rxbuf.end(),
+                                          _readbuf.begin(),
+                                          _readbuf.begin() + ret);
+                        }
+                    }
+                }
+                else if (pollResult == PollResultType_Error)
                 {
                     _socket->close();
-                    setReadyState(CLOSED);
                 }
+                else if (pollResult == PollResultType_CloseRequest)
+                {
+                    ;
+                }
+
             },
             _heartBeatPeriod);
     }
@@ -586,11 +611,11 @@ namespace ix
             }
         }
 
-        // Make sure we send all the buffered data ; there can be a lot of it
-        // for large messages.
-        // TODO / this will block the sending thread ; we need to eval whether
-        //        this is the right fix
-        while (!isSendBufferEmpty()) sendOnSocket();
+        // Request to flush the send buffer on the background thread if it isn't empty
+        if (!isSendBufferEmpty())
+        {
+            _socket->wakeUpFromPoll(Socket::kSendRequest);
+        }
 
         return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
     }
@@ -737,8 +762,17 @@ namespace ix
         sendData(wsheader_type::CLOSE, normalClosure, compress);
         setReadyState(CLOSING);
 
-        _socket->wakeUpFromPoll();
+        _socket->wakeUpFromPoll(Socket::kCloseRequest);
         _socket->close();
+
+        _closeCode = 1000;
+        setReadyState(CLOSED);
+    }
+
+    size_t WebSocketTransport::bufferedAmount() const
+    {
+        std::lock_guard<std::mutex> lock(_txbufMutex);
+        return _txbuf.size();
     }
 
 } // namespace ix

ixwebsocket/IXWebSocketTransport.h

@@ -77,6 +77,7 @@ namespace ix
         void setReadyState(ReadyStateValues readyStateValue);
         void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
         void dispatch(const OnMessageCallback& onMessageCallback);
+        size_t bufferedAmount() const;
 
     private:
         std::string _url;

makefile

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

test/IXSocketTest.cpp

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

test/cmd_websocket_chat.cpp

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

test/run.py

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

ws/test_ws.sh

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

ws/ws.cpp

@@ -16,6 +16,8 @@
 #include <string>
 #include <sstream>
 #include <iostream>
+#include <fstream>
+#include <unistd.h>
 
 #include <cli11/CLI11.hpp>
 #include <ixwebsocket/IXSocket.h>
@@ -32,6 +34,7 @@ int main(int argc, char** argv)
     std::string headers;
     std::string output;
     std::string hostname("127.0.0.1");
+    std::string pidfile;
     bool headersOnly = false;
     bool followRedirects = false;
     bool verbose = false;
@@ -51,6 +54,8 @@ int main(int argc, char** argv)
 
     CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
     transferApp->add_option("--port", port, "Connection url");
+    transferApp->add_option("--host", hostname, "Hostname");
+    transferApp->add_option("--pidfile", pidfile, "Pid file");
 
     CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
     connectApp->add_option("url", url, "Connection url")->required();
@@ -60,11 +65,12 @@ int main(int argc, char** argv)
     chatApp->add_option("user", user, "User name")->required();
 
     CLI::App* echoServerApp = app.add_subcommand("echo_server", "Echo server");
-    echoServerApp->add_option("--port", port, "Connection url");
+    echoServerApp->add_option("--port", port, "Port");
+    echoServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
-    broadcastServerApp->add_option("--port", port, "Connection url");
-    broadcastServerApp->add_option("--hostname", hostname, "Hostname");
+    broadcastServerApp->add_option("--port", port, "Port");
+    broadcastServerApp->add_option("--host", hostname, "Hostname");
 
     CLI::App* pingPongApp = app.add_subcommand("ping", "Ping pong");
     pingPongApp->add_option("url", url, "Connection url")->required();
@@ -88,9 +94,21 @@ int main(int argc, char** argv)
 
     ix::Socket::init();
 
+    // pid file handling
+
     if (app.got_subcommand("transfer"))
     {
-        return ix::ws_transfer_main(port);
+        if (!pidfile.empty())
+        {
+            unlink(pidfile.c_str());
+
+            std::ofstream f;
+            f.open(pidfile);
+            f << getpid();
+            f.close();
+        }
+
+        return ix::ws_transfer_main(port, hostname);
     }
     else if (app.got_subcommand("send"))
     {
@@ -111,7 +129,7 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("echo_server"))
     {
-        return ix::ws_echo_server_main(port);
+        return ix::ws_echo_server_main(port, hostname);
     }
     else if (app.got_subcommand("broadcast_server"))
     {

ws/ws.h

@@ -24,9 +24,9 @@ namespace ix
 
     int ws_ping_pong_main(const std::string& url);
 
-    int ws_echo_server_main(int port);
-
+    int ws_echo_server_main(int port, const std::string& hostname);
     int ws_broadcast_server_main(int port, const std::string& hostname);
+    int ws_transfer_main(int port, const std::string& hostname);
 
     int ws_chat_main(const std::string& url,
                      const std::string& user);
@@ -36,8 +36,6 @@ namespace ix
     int ws_receive_main(const std::string& url,
                         bool enablePerMessageDeflate);
 
-    int ws_transfer_main(int port);
-
     int ws_send_main(const std::string& url,
                      const std::string& path);
 }

ws/ws_broadcast_server.cpp

@@ -71,6 +71,15 @@ namespace ix
                                                   << " out of " << total << std::endl;
                                         return true;
                                     });
+
+                                    do
+                                    {
+                                        size_t bufferedAmount = client->bufferedAmount();
+                                        std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
+
+                                        std::chrono::duration<double, std::milli> duration(10);
+                                        std::this_thread::sleep_for(duration);
+                                    } while (client->bufferedAmount() != 0);
                                 }
                             }
                         }

ws/ws_echo_server.cpp

@@ -10,11 +10,11 @@
 
 namespace ix
 {
-    int ws_echo_server_main(int port)
+    int ws_echo_server_main(int port, const std::string& hostname)
     {
-        std::cout << "Listening on port " << port << std::endl;
+        std::cout << "Listening on " << hostname << ":" << port << std::endl;
 
-        ix::WebSocketServer server(port);
+        ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
             [](std::shared_ptr<ix::WebSocket> webSocket)

ws/ws_receive.cpp

@@ -146,11 +146,16 @@ namespace ix
         std::string filename = data["filename"].string_value();
         filename = extractFilename(filename);
 
-        std::cout << "Writing to disk: " << filename << std::endl;
-        std::ofstream out(filename);
+        std::string filenameTmp = filename + ".tmp";
+
+        std::cout << "Writing to disk: " << filenameTmp << std::endl;
+        std::ofstream out(filenameTmp);
         out.write((char*)&content.front(), content.size());
         out.close();
 
+        std::cout << "Renaming " << filenameTmp << " to " << filename << std::endl;
+        rename(filenameTmp.c_str(), filename.c_str());
+
         std::map<MsgPack, MsgPack> pdu;
         pdu["ack"] = true;
         pdu["id"] = data["id"];

ws/ws_send.cpp

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

ws/ws_transfer.cpp

@@ -10,11 +10,11 @@
 
 namespace ix
 {
-    int ws_transfer_main(int port)
+    int ws_transfer_main(int port, const std::string& hostname)
     {
-        std::cout << "Listening on port " << port << std::endl;
+        std::cout << "Listening on " << hostname << ":" << port << std::endl;
 
-        ix::WebSocketServer server(port);
+        ix::WebSocketServer server(port, hostname);
 
         server.setOnConnectionCallback(
             [&server](std::shared_ptr<ix::WebSocket> webSocket)
@@ -70,6 +70,15 @@ namespace ix
                                                   << " out of " << total << std::endl;
                                         return true;
                                     });
+
+                                    do
+                                    {
+                                        size_t bufferedAmount = client->bufferedAmount();
+                                        std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
+
+                                        std::chrono::duration<double, std::milli> duration(10);
+                                        std::this_thread::sleep_for(duration);
+                                    } while (client->bufferedAmount() != 0);
                                 }
                             }
                         }