bump timeout

* init Net system on Windows

* propagate DNS error

* Add zlib 1.2.11 sources

* link zlib statically for windows

* remove not implemented function declaration

* fix connect on Windows

Dockerfile

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

Dockerfile

@@ -0,0 +1,42 @@
+FROM fedora:30 as build
+
+RUN yum install -y gcc-g++
+RUN yum install -y cmake
+RUN yum install -y make
+RUN yum install -y openssl-devel
+
+RUN yum install -y wget
+RUN mkdir -p /tmp/cmake
+WORKDIR /tmp/cmake
+RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
+RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+RUN yum install -y python
+RUN yum install -y libtsan
+
+COPY . .
+# RUN ["make", "test"]
+RUN ["make"]
+
+# Runtime
+FROM fedora:30 as runtime
+
+RUN yum install -y libtsan
+
+RUN groupadd app && useradd -g app app 
+COPY --chown=app:app --from=build /usr/local/bin/ws /usr/local/bin/ws
+RUN chmod +x /usr/local/bin/ws
+RUN ldd /usr/local/bin/ws
+
+# Now run in usermode
+USER app
+WORKDIR /home/app
+
+COPY --chown=app:app ws/snake/appsConfig.json .
+COPY --chown=app:app ws/cobraMetricsSample.json .
+
+ENTRYPOINT ["ws"]
+CMD ["--help"]

README.md

@@ -199,8 +199,6 @@ make install # will install to /usr/local on Unix, on macOS it is a good idea to
 
 Headers and a static library will be installed to the target dir.
 
-A [conan](https://conan.io/) file is available at [conan-IXWebSocket](https://github.com/Zinnion/conan-IXWebSocket).
-
 There is a unittest which can be executed by typing `make test`.
 
 There is a Dockerfile for running some code on Linux. To use docker-compose you must make a docker container first.

docker/Dockerfile.debian

@@ -1,52 +0,0 @@
-# Build time
-FROM debian:buster as build
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-RUN apt-get -y install wget 
-RUN mkdir -p /tmp/cmake
-WORKDIR /tmp/cmake
-RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
-RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
-
-RUN apt-get -y install g++ 
-RUN apt-get -y install libssl-dev
-RUN apt-get -y install libz-dev
-RUN apt-get -y install make
-
-COPY . .
-
-ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
-ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
-
-RUN ["make"]
-
-# Runtime
-FROM debian:buster as runtime
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-# Runtime 
-RUN apt-get install -y libssl1.1 
-RUN apt-get install -y ca-certificates
-RUN ["update-ca-certificates"]
-
-# Debugging
-RUN apt-get install -y strace
-RUN apt-get install -y procps
-RUN apt-get install -y htop
-
-RUN adduser --disabled-password --gecos '' app
-COPY --chown=app:app --from=build /usr/local/bin/ws /usr/local/bin/ws
-RUN chmod +x /usr/local/bin/ws
-RUN ldd /usr/local/bin/ws
-
-# Now run in usermode
-USER app
-WORKDIR /home/app
-
-COPY --chown=app:app ws/snake/appsConfig.json .
-COPY --chown=app:app ws/cobraMetricsSample.json .
-
-ENTRYPOINT ["ws"]
-CMD ["--help"]

docker/Dockerfile.fedora

@@ -1,42 +0,0 @@
-FROM fedora:30 as build
-
-RUN yum install -y gcc-g++
-RUN yum install -y cmake
-RUN yum install -y make
-RUN yum install -y openssl-devel
-
-RUN yum install -y wget
-RUN mkdir -p /tmp/cmake
-WORKDIR /tmp/cmake
-RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
-RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
-
-ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
-ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
-
-RUN yum install -y python
-RUN yum install -y libtsan
-
-COPY . .
-# RUN ["make", "test"]
-RUN ["make"]
-
-# Runtime
-FROM fedora:30 as runtime
-
-RUN yum install -y libtsan
-
-RUN groupadd app && useradd -g app app 
-COPY --chown=app:app --from=build /usr/local/bin/ws /usr/local/bin/ws
-RUN chmod +x /usr/local/bin/ws
-RUN ldd /usr/local/bin/ws
-
-# Now run in usermode
-USER app
-WORKDIR /home/app
-
-COPY --chown=app:app ws/snake/appsConfig.json .
-COPY --chown=app:app ws/cobraMetricsSample.json .
-
-ENTRYPOINT ["ws"]
-CMD ["--help"]

docker/Dockerfile.ubuntu_xenial

@@ -1,24 +0,0 @@
-# Build time
-FROM ubuntu:xenial as build
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-RUN apt-get -y install wget 
-RUN mkdir -p /tmp/cmake
-WORKDIR /tmp/cmake
-RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
-RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
-
-RUN apt-get -y install g++ 
-RUN apt-get -y install libssl-dev
-RUN apt-get -y install libz-dev
-RUN apt-get -y install make
-RUN apt-get -y install python
-
-COPY . .
-
-ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
-ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
-
-# RUN ["make"]
-RUN ["make", "test"]

ixwebsocket/IXDNSLookup.cpp

@@ -17,25 +17,28 @@ namespace ix
     std::atomic<uint64_t> DNSLookup::_nextId(0);
     std::set<uint64_t> DNSLookup::_activeJobs;
     std::mutex DNSLookup::_activeJobsMutex;
+    std::mutex DNSLookup::_resMutex;
 
     DNSLookup::DNSLookup(const std::string& hostname, int port, int64_t wait) :
+        _hostname(hostname),
         _port(port),
         _wait(wait),
         _res(nullptr),
         _done(false),
         _id(_nextId++)
     {
-        setHostname(hostname);
+        ;
     }
 
     DNSLookup::~DNSLookup()
     {
         // Remove this job from the active jobs list
-        std::lock_guard<std::mutex> lock(_activeJobsMutex);
+        std::unique_lock<std::mutex> lock(_activeJobsMutex);
         _activeJobs.erase(_id);
     }
 
-    struct addrinfo* DNSLookup::getAddrInfo(const std::string& hostname,
+    // we want hostname to be copied, not passed as a const reference
+    struct addrinfo* DNSLookup::getAddrInfo(std::string hostname,
                                             int port,
                                             std::string& errMsg)
     {
@@ -78,7 +81,7 @@ namespace ix
             return nullptr;
         }
 
-        return getAddrInfo(getHostname(), _port, errMsg);
+        return getAddrInfo(_hostname, _port, errMsg);
     }
 
     struct addrinfo* DNSLookup::resolveAsync(std::string& errMsg,
@@ -96,7 +99,7 @@ namespace ix
 
         // Record job in the active Job set
         {
-            std::lock_guard<std::mutex> lock(_activeJobsMutex);
+            std::unique_lock<std::mutex> lock(_activeJobsMutex);
             _activeJobs.insert(_id);
         }
 
@@ -104,7 +107,7 @@ namespace ix
         // Good resource on thread forced termination
         // https://www.bo-yang.net/2017/11/19/cpp-kill-detached-thread
         //
-        _thread = std::thread(&DNSLookup::run, this, _id, getHostname(), _port);
+        _thread = std::thread(&DNSLookup::run, this, _id, _hostname, _port);
         _thread.detach();
 
         std::unique_lock<std::mutex> lock(_conditionVariableMutex);
@@ -134,8 +137,13 @@ namespace ix
             return nullptr;
         }
 
-        errMsg = getErrMsg();
+        if (!_errMsg.empty())
-        return getRes();
+        {
+            errMsg = _errMsg;
+        }
+
+        std::unique_lock<std::mutex> rlock(_resMutex);
+        return _res;
     }
 
     void DNSLookup::run(uint64_t id, const std::string& hostname, int port) // thread runner
@@ -147,55 +155,21 @@ namespace ix
         struct addrinfo* res = getAddrInfo(hostname, port, errMsg);
 
         // if this isn't an active job, and the control thread is gone
-        // there is nothing to do, and we don't want to touch the defunct
+        // there is not thing to do, and we don't want to touch the defunct
         // object data structure such as _errMsg or _condition
-        std::lock_guard<std::mutex> lock(_activeJobsMutex);
+        std::unique_lock<std::mutex> lock(_activeJobsMutex);
         if (_activeJobs.count(id) == 0)
         {
             return;
         }
 
         // Copy result into the member variables
-        setRes(res);
+        {
-        setErrMsg(errMsg);
+            std::unique_lock<std::mutex> rlock(_resMutex);
-
+            _res = res;
+        }
+        _errMsg = errMsg;
         _condition.notify_one();
         _done = true;
     }
-
-    void DNSLookup::setHostname(const std::string& hostname)
-    {
-        std::lock_guard<std::mutex> lock(_hostnameMutex);
-        _hostname = hostname;
-    }
-
-    const std::string& DNSLookup::getHostname()
-    {
-        std::lock_guard<std::mutex> lock(_hostnameMutex);
-        return _hostname;
-    }
-
-    void DNSLookup::setErrMsg(const std::string& errMsg)
-    {
-        std::lock_guard<std::mutex> lock(_errMsgMutex);
-        _errMsg = errMsg;
-    }
-
-    const std::string& DNSLookup::getErrMsg()
-    {
-        std::lock_guard<std::mutex> lock(_errMsgMutex);
-        return _errMsg;
-    }
-
-    void DNSLookup::setRes(struct addrinfo* addr)
-    {
-        std::lock_guard<std::mutex> lock(_resMutex);
-        _res = addr;
-    }
-
-    struct addrinfo* DNSLookup::getRes()
-    {
-        std::lock_guard<std::mutex> lock(_resMutex);
-        return _res;
-    }
 }

ixwebsocket/IXDNSLookup.h

@@ -39,32 +39,18 @@ namespace ix
         struct addrinfo* resolveBlocking(std::string& errMsg,
                                          const CancellationRequest& isCancellationRequested);
 
-        static struct addrinfo* getAddrInfo(const std::string& hostname,
+        static struct addrinfo* getAddrInfo(std::string hostname,
                                             int port,
                                             std::string& errMsg);
 
         void run(uint64_t id, const std::string& hostname, int port); // thread runner
 
-        void setHostname(const std::string& hostname);
-        const std::string& getHostname();
-
-        void setErrMsg(const std::string& errMsg);
-        const std::string& getErrMsg();
-
-        void setRes(struct addrinfo* addr);
-        struct addrinfo* getRes();
-
         std::string _hostname;
-        std::mutex _hostnameMutex;
         int _port;
-
         int64_t _wait;
-
-        struct addrinfo* _res;
-        std::mutex _resMutex;
-
         std::string _errMsg;
-        std::mutex _errMsgMutex;
+        struct addrinfo* _res;
+        static std::mutex _resMutex;
 
         std::atomic<bool> _done;
         std::thread _thread;

ixwebsocket/IXSocket.cpp

@@ -196,25 +196,6 @@ namespace ix
 #endif
     }
 
-    bool Socket::isWaitNeeded()
-    {
-        int err = getErrno();
-
-        if (err == EWOULDBLOCK || err == EAGAIN)
-        {
-            return true;
-        }
-
-#ifdef _WIN32
-        if (err == WSAEWOULDBLOCK || err == WSATRY_AGAIN)
-        {
-            return true;
-        }
-#endif
-
-        return false;
-    }
-
     void Socket::closeSocket(int fd)
     {
 #ifdef _WIN32
@@ -247,7 +228,8 @@ namespace ix
                 return ret == len;
             }
             // There is possibly something to be writen, try again
-            else if (ret < 0 && Socket::isWaitNeeded())
+            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
+                                 getErrno() == EAGAIN))
             {
                 continue;
             }
@@ -275,7 +257,8 @@ namespace ix
                 return true;
             }
             // There is possibly something to be read, try again
-            else if (ret < 0 && Socket::isWaitNeeded())
+            else if (ret < 0 && (getErrno() == EWOULDBLOCK ||
+                                 getErrno() == EAGAIN))
             {
                 // Wait with a 1ms timeout until the socket is ready to read.
                 // This way we are not busy looping
@@ -334,12 +317,13 @@ namespace ix
             size_t size = std::min(kChunkSize, length - output.size());
             ssize_t ret = recv((char*)&_readBuffer[0], size);
 
-            if (ret <= 0 && !Socket::isWaitNeeded())
+            if (ret <= 0 && (getErrno() != EWOULDBLOCK &&
+                             getErrno() != EAGAIN))
             {
                 // Error
                 return std::make_pair(false, std::string());
             }
-            else
+            else if (ret > 0)
             {
                 output.insert(output.end(),
                               _readBuffer.begin(),

ixwebsocket/IXSocket.h

@@ -41,6 +41,8 @@ namespace ix
         virtual ~Socket();
         bool init(std::string& errorMsg);
 
+        void configure();
+
         // Functions to check whether there is activity on the socket
         PollResultType poll(int timeoutSecs = kDefaultPollTimeout);
         bool wakeUpFromPoll(uint8_t wakeUpCode);
@@ -74,7 +76,6 @@ namespace ix
             const CancellationRequest& isCancellationRequested);
 
         static int getErrno();
-        static bool isWaitNeeded();
 
         // Used as special codes for pipe communication
         static const uint64_t kSendRequest;

ixwebsocket/IXSocketServer.cpp

@@ -247,7 +247,7 @@ namespace ix
 
             if ((clientFd = accept(_serverFd, (struct sockaddr *)&client, &addressLen)) < 0)
             {
-                if (!Socket::isWaitNeeded())
+                if (Socket::getErrno() != EWOULDBLOCK)
                 {
                     // FIXME: that error should be propagated
                     std::stringstream ss;

ixwebsocket/IXWebSocket.cpp

@@ -216,9 +216,14 @@ namespace ix
         return getReadyState() == WebSocket_ReadyState_Closing;
     }
 
-    void WebSocket::close()
+    bool WebSocket::isConnectedOrClosing() const
     {
-        _ws.close();
+        return isConnected() || isClosing();
+    }
+
+    void WebSocket::close(uint16_t code, const std::string& reason)
+    {
+        _ws.close(code, reason);
     }
 
     void WebSocket::reconnectPerpetuallyIfDisconnected()
@@ -229,12 +234,16 @@ namespace ix
         using millis = std::chrono::duration<double, std::milli>;
         millis duration;
 
-        // Try to connect only once when we don't have automaticReconnection setup
+        while (true)
-        if (!isConnected() && !isClosing() && !_stop && !_automaticReconnection)
         {
+            if (isConnected() || isClosing() || _stop || !_automaticReconnection)
+            {
+                break;
+            }
+
             status = connect(_handshakeTimeoutSecs);
 
-            if (!status.success)
+            if (!status.success && !_stop)
             {
                 duration = millis(calculateRetryWaitMilliseconds(retries++));
 
@@ -245,38 +254,8 @@ namespace ix
                 _onMessageCallback(WebSocket_MessageType_Error, "", 0,
                                    connectErr, WebSocketOpenInfo(),
                                    WebSocketCloseInfo());
-            }
-        }
-        else
-        {
-            // Otherwise try to reconnect perpertually
-            while (true)
-            {
-                if (isConnected() || isClosing() || _stop || !_automaticReconnection)
-                {
-                    break;
-                }
 
-                status = connect(_handshakeTimeoutSecs);
+                std::this_thread::sleep_for(duration);
-
-                if (!status.success)
-                {
-                    duration = millis(calculateRetryWaitMilliseconds(retries++));
-
-                    connectErr.retries = retries;
-                    connectErr.wait_time = duration.count();
-                    connectErr.reason = status.errorStr;
-                    connectErr.http_status = status.http_status;
-                    _onMessageCallback(WebSocket_MessageType_Error, "", 0,
-                                       connectErr, WebSocketOpenInfo(),
-                                       WebSocketCloseInfo());
-
-                    // Only sleep if we aren't in the middle of stopping
-                    if (!_stop)
-                    {
-                        std::this_thread::sleep_for(duration);
-                    }
-                }
             }
         }
     }
@@ -287,20 +266,17 @@ namespace ix
 
         while (true)
         {
-            if (_stop) return;
+            if (_stop && !isClosing()) return;
 
             // 1. Make sure we are always connected
             reconnectPerpetuallyIfDisconnected();
 
-            if (_stop) return;
-
             // 2. Poll to see if there's any new data available
-            _ws.poll();
+            WebSocketTransport::PollPostTreatment pollPostTreatment = _ws.poll();
-
-            if (_stop) return;
 
             // 3. Dispatch the incoming messages
             _ws.dispatch(
+                pollPostTreatment,
                 [this](const std::string& msg,
                        size_t wireSize,
                        bool decompressionError,
@@ -340,8 +316,10 @@ namespace ix
                     WebSocket::invokeTrafficTrackerCallback(msg.size(), true);
                 });
 
-            // If we aren't trying to reconnect automatically, exit if we aren't connected
+            // 4. In blocking mode, getting out of this function is triggered by
-            if (!isConnected() && !_automaticReconnection) return;
+            //    an explicit disconnection from the callback, or by the remote end
+            //    closing the connection, ie isConnectedOrClosing() == false.
+            if (!_thread.joinable() && !isConnectedOrClosing() && !_automaticReconnection) return;
         }
     }
 

ixwebsocket/IXWebSocket.h

@@ -111,7 +111,8 @@ namespace ix
         WebSocketSendInfo sendText(const std::string& text,
                                    const OnProgressCallback& onProgressCallback = nullptr);
         WebSocketSendInfo ping(const std::string& text);
-        void close();
+        void close(uint16_t code = 1000,
+                   const std::string& reason = "Normal closure");
 
         void setOnMessageCallback(const OnMessageCallback& callback);
         static void setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback);
@@ -136,6 +137,7 @@ namespace ix
 
         bool isConnected() const;
         bool isClosing() const;
+        bool isConnectedOrClosing() const;
         void reconnectPerpetuallyIfDisconnected();
         std::string readyStateToString(ReadyState readyState);
         static void invokeTrafficTrackerCallback(size_t size, bool incoming);

ixwebsocket/IXWebSocketTransport.cpp

@@ -72,7 +72,9 @@ namespace ix
     const int WebSocketTransport::kDefaultPingIntervalSecs(-1);
     const int WebSocketTransport::kDefaultPingTimeoutSecs(-1);
     const bool WebSocketTransport::kDefaultEnablePong(true);
+    const int WebSocketTransport::kClosingMaximumWaitingDelayInMs(200);
     constexpr size_t WebSocketTransport::kChunkSize;
+
     const uint16_t WebSocketTransport::kInternalErrorCode(1011);
     const uint16_t WebSocketTransport::kAbnormalCloseCode(1006);
     const uint16_t WebSocketTransport::kProtocolErrorCode(1002);
@@ -90,6 +92,7 @@ namespace ix
         _closeRemote(false),
         _enablePerMessageDeflate(false),
         _requestInitCancellation(false),
+        _closingTimePoint(std::chrono::steady_clock::now()),
         _enablePong(kDefaultEnablePong),
         _pingIntervalSecs(kDefaultPingIntervalSecs),
         _pingTimeoutSecs(kDefaultPingTimeoutSecs),
@@ -246,9 +249,19 @@ namespace ix
         return now - _lastReceivePongTimePoint > std::chrono::seconds(_pingTimeoutSecs);
     }
 
-    void WebSocketTransport::poll()
+    bool WebSocketTransport::closingDelayExceeded()
     {
-        PollResultType pollResult = _socket->poll(_pingIntervalOrTimeoutGCDSecs);
+        std::lock_guard<std::mutex> lock(_closingTimePointMutex);
+        auto now = std::chrono::steady_clock::now();
+        return now - _closingTimePoint > std::chrono::milliseconds(kClosingMaximumWaitingDelayInMs);
+    }
+
+    WebSocketTransport::PollPostTreatment WebSocketTransport::poll()
+    {
+        // we need to have no timeout if state is CLOSING
+        int timeoutDelaySecs = (_readyState == CLOSING) ? 0 : _pingIntervalOrTimeoutGCDSecs;
+
+        PollResultType pollResult = _socket->poll(timeoutDelaySecs);
 
         if (_readyState == OPEN)
         {
@@ -296,23 +309,19 @@ namespace ix
             {
                 ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
 
-                if (ret < 0 && Socket::isWaitNeeded())
+                if (ret < 0 && _readyState != CLOSING && (_socket->getErrno() == EWOULDBLOCK ||
+                                                          _socket->getErrno() == EAGAIN))
                 {
                     break;
                 }
                 else if (ret <= 0)
                 {
-                    _rxbuf.clear();
+                    // if there are received data pending to be processed, then delay the abnormal closure
+                    // to after dispatch (other close code/reason could be read from the buffer)
+                    
                     _socket->close();
-                    {
+
-                        std::lock_guard<std::mutex> lock(_closeDataMutex);
+                    return CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH;
-                        _closeCode = kAbnormalCloseCode;
-                        _closeReason = kAbnormalCloseMessage;
-                        _closeWireSize = 0;
-                        _closeRemote = true;
-                    }
-                    setReadyState(CLOSED);
-                    break;
                 }
                 else
                 {
@@ -331,12 +340,15 @@ namespace ix
             _socket->close();
         }
 
-        // Avoid a race condition where we get stuck in select
+        if (_readyState == CLOSING && closingDelayExceeded())
-        // while closing.
-        if (_readyState == CLOSING)
         {
+            _rxbuf.clear();
+            // close code and reason were set when calling close()
             _socket->close();
+            setReadyState(CLOSED);
         }
+
+        return NONE;
     }
 
     bool WebSocketTransport::isSendBufferEmpty() const
@@ -398,12 +410,13 @@ namespace ix
     // |                     Payload Data continued ...                |
     // +---------------------------------------------------------------+
     //
-    void WebSocketTransport::dispatch(const OnMessageCallback& onMessageCallback)
+    void WebSocketTransport::dispatch(WebSocketTransport::PollPostTreatment pollPostTreatment,
+                                      const OnMessageCallback& onMessageCallback)
     {
         while (true)
         {
             wsheader_type ws;
-            if (_rxbuf.size() < 2) return; /* Need at least 2 */
+            if (_rxbuf.size() < 2) break; /* Need at least 2 */
             const uint8_t * data = (uint8_t *) &_rxbuf[0]; // peek, but don't consume
             ws.fin = (data[0] & 0x80) == 0x80;
             ws.rsv1 = (data[0] & 0x40) == 0x40;
@@ -411,7 +424,7 @@ namespace ix
             ws.mask = (data[1] & 0x80) == 0x80;
             ws.N0 = (data[1] & 0x7f);
             ws.header_size = 2 + (ws.N0 == 126? 2 : 0) + (ws.N0 == 127? 8 : 0) + (ws.mask? 4 : 0);
-            if (_rxbuf.size() < ws.header_size) return; /* Need: ws.header_size - _rxbuf.size() */
+            if (_rxbuf.size() < ws.header_size) break; /* Need: ws.header_size - _rxbuf.size() */
 
             //
             // Calculate payload length:
@@ -553,9 +566,25 @@ namespace ix
                 std::string reason(_rxbuf.begin()+ws.header_size + 2,
                                    _rxbuf.begin()+ws.header_size + (size_t) ws.N);
                 
-                bool remote = true;
                 
-                close(code, reason, _rxbuf.size(), remote);
+                // We receive a CLOSE frame from remote and are NOT the ones who triggered the close
+                if (_readyState != CLOSING)
+                {
+                    // send back the CLOSE frame
+                    sendCloseFrame(code, reason);
+
+                    _socket->wakeUpFromPoll(Socket::kCloseRequest);
+                    
+                    bool remote = true;
+                    closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
+                }
+                // we got the CLOSE frame answer from our close, so we can close the connection if
+                // the code/reason are the same
+                else if (_closeCode == code && _closeReason == reason)
+                {
+                    bool remote = false;
+                    closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
+                }
             }
             else
             {
@@ -568,6 +597,25 @@ namespace ix
             _rxbuf.erase(_rxbuf.begin(),
                          _rxbuf.begin() + ws.header_size + (size_t) ws.N);
         }
+
+        // if an abnormal closure was raised in poll, and nothing else triggered a CLOSED state in
+        // the received and processed data then close the connection
+        if (pollPostTreatment == CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH)
+        {
+            _rxbuf.clear();
+
+            // if we previously closed the connection (CLOSING state), then set state to CLOSED (code/reason were set before)
+            if (_readyState == CLOSING)
+            {
+                _socket->close();
+                setReadyState(CLOSED);
+            }
+            // if we weren't closing, then close using abnormal close code and message 
+            else if (_readyState != CLOSED)
+            {
+                closeSocketAndSwitchToClosedState(kAbnormalCloseCode, kAbnormalCloseMessage, 0, false);
+            }
+        }
     }
 
     std::string WebSocketTransport::getMergedChunks() const
@@ -843,7 +891,8 @@ namespace ix
         {
             ssize_t ret = _socket->send((char*)&_txbuf[0], _txbuf.size());
 
-            if (ret < 0 && Socket::isWaitNeeded())
+            if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
+                            _socket->getErrno() == EAGAIN))
             {
                 break;
             }
@@ -861,12 +910,9 @@ namespace ix
         }
     }
 
-    void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
+
+    void WebSocketTransport::sendCloseFrame(uint16_t code, const std::string& reason)
     {
-        _requestInitCancellation = true;
-
-        if (_readyState == CLOSING || _readyState == CLOSED) return;
-
         // See list of close events here:
         // https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
 
@@ -879,11 +925,11 @@ namespace ix
 
         bool compress = false;
         sendData(wsheader_type::CLOSE, closure, compress);
-        setReadyState(CLOSING);
+    }
 
-        _socket->wakeUpFromPoll(Socket::kCloseRequest);
+    void WebSocketTransport::closeSocketAndSwitchToClosedState(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
+    {
         _socket->close();
-
         {
             std::lock_guard<std::mutex> lock(_closeDataMutex);
             _closeCode = code;
@@ -891,10 +937,33 @@ namespace ix
             _closeWireSize = closeWireSize;
             _closeRemote = remote;
         }
-
         setReadyState(CLOSED);
     }
 
+    void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
+    {
+        _requestInitCancellation = true;
+
+        if (_readyState == CLOSING || _readyState == CLOSED) return;
+
+        sendCloseFrame(code, reason);
+        {
+            std::lock_guard<std::mutex> lock(_closeDataMutex);
+            _closeCode = code;
+            _closeReason = reason;
+            _closeWireSize = closeWireSize;
+            _closeRemote = remote;
+        }
+        {
+            std::lock_guard<std::mutex> lock(_closingTimePointMutex);
+            _closingTimePoint = std::chrono::steady_clock::now();
+        }
+        setReadyState(CLOSING);
+
+        // wake up the poll, but do not close yet
+        _socket->wakeUpFromPoll(Socket::kSendRequest);
+    }
+
     size_t WebSocketTransport::bufferedAmount() const
     {
         std::lock_guard<std::mutex> lock(_txbufMutex);

ixwebsocket/IXWebSocketTransport.h

@@ -56,6 +56,12 @@ namespace ix
             FRAGMENT
         };
 
+        enum PollPostTreatment
+        {
+            NONE,
+            CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH
+        };
+
         using OnMessageCallback = std::function<void(const std::string&,
                                                      size_t,
                                                      bool,
@@ -78,7 +84,7 @@ namespace ix
         WebSocketInitResult connectToSocket(int fd,              // Server
                                             int timeoutSecs);
 
-        void poll();
+        PollPostTreatment poll();
         WebSocketSendInfo sendBinary(const std::string& message,
                                      const OnProgressCallback& onProgressCallback);
         WebSocketSendInfo sendText(const std::string& message,
@@ -93,7 +99,8 @@ namespace ix
         ReadyStateValues getReadyState() const;
         void setReadyState(ReadyStateValues readyStateValue);
         void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
-        void dispatch(const OnMessageCallback& onMessageCallback);
+        void dispatch(PollPostTreatment pollPostTreatment,
+                      const OnMessageCallback& onMessageCallback);
         size_t bufferedAmount() const;
 
     private:
@@ -119,7 +126,7 @@ namespace ix
 
         // Tells whether we should mask the data we send.
         // client should mask but server should not
-        std::atomic<bool> _useMask;
+        bool _useMask;
 
         // Buffer for reading from our socket. That buffer is never resized.
         std::vector<uint8_t> _readbuf;
@@ -163,6 +170,10 @@ namespace ix
         // Used to cancel dns lookup + socket connect + http upgrade
         std::atomic<bool> _requestInitCancellation;
               
+        mutable std::mutex _closingTimePointMutex;
+        std::chrono::time_point<std::chrono::steady_clock>_closingTimePoint;
+        static const int kClosingMaximumWaitingDelayInMs;
+
         // Constants for dealing with closing conneections
         static const uint16_t kInternalErrorCode;
         static const uint16_t kAbnormalCloseCode;
@@ -201,6 +212,16 @@ namespace ix
         // No PONG data was received through the socket for longer than ping timeout delay
         bool pingTimeoutExceeded();
 
+        // after calling close(), if no CLOSE frame answer is received back from the remote, we should close the connexion
+        bool closingDelayExceeded();
+
+        void sendCloseFrame(uint16_t code, const std::string& reason);
+
+        void closeSocketAndSwitchToClosedState(uint16_t code,
+                                               const std::string& reason,
+                                               size_t closeWireSize,
+                                               bool remote);
+
         void sendOnSocket();
         WebSocketSendInfo sendData(wsheader_type::opcode_type type,
                                    const std::string& message,

test/CMakeLists.txt

@@ -8,11 +8,12 @@ project (ixwebsocket_unittest)
 set(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/../third_party/sanitizers-cmake/cmake" ${CMAKE_MODULE_PATH})
 find_package(Sanitizers)
 
+# set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
+# set(CMAKE_LD_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
+
 set (CMAKE_CXX_STANDARD 14)
 
 if (NOT WIN32)
-  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
-  set(CMAKE_LD_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
   option(USE_TLS "Add TLS support" ON)
 endif()
 
@@ -37,6 +38,7 @@ set (SOURCES
 # Some unittest don't work on windows yet
 if (NOT WIN32)
   list(APPEND SOURCES 
+    IXWebSocketCloseTest.cpp
     IXWebSocketServerTest.cpp
     IXWebSocketPingTest.cpp
     IXWebSocketPingTimeoutTest.cpp

test/IXWebSocketCloseTest.cpp

@@ -0,0 +1,407 @@
+/*
+ *  IXWebSocketCloseTest.cpp
+ *  Author: Alexandre Konieczny
+ *  Copyright (c) 2019 Machine Zone. All rights reserved.
+ */
+
+#include <iostream>
+#include <sstream>
+#include <queue>
+#include <ixwebsocket/IXWebSocket.h>
+#include <ixwebsocket/IXWebSocketServer.h>
+
+#include "IXTest.h"
+
+#include "catch.hpp"
+
+using namespace ix;
+
+namespace
+{
+    class WebSocketClient
+    {
+        public:
+            WebSocketClient(int port);
+
+            void subscribe(const std::string& channel);
+            void start();
+            void stop();
+            void stop(uint16_t code, const std::string& reason);
+            bool isReady() const;
+            void sendMessage(const std::string& text);
+
+            uint16_t getCloseCode();
+            const std::string& getCloseReason();
+            bool getCloseRemote();
+
+        private:
+            ix::WebSocket _webSocket;
+            int _port;
+
+            mutable std::mutex _mutexCloseData;
+            uint16_t _closeCode;
+            std::string _closeReason;
+            bool _closeRemote;
+    };
+
+    WebSocketClient::WebSocketClient(int port)
+        : _port(port)
+        , _closeCode(0)
+        , _closeReason(std::string(""))
+        , _closeRemote(false)
+    {
+        ;
+    }
+
+    bool WebSocketClient::isReady() const
+    {
+        return _webSocket.getReadyState() == ix::WebSocket_ReadyState_Open;
+    }
+
+    uint16_t WebSocketClient::getCloseCode()
+    {
+        std::lock_guard<std::mutex> lck(_mutexCloseData);
+        
+        return _closeCode;
+    }
+
+    const std::string& WebSocketClient::getCloseReason()
+    {
+        std::lock_guard<std::mutex> lck(_mutexCloseData);
+
+        return _closeReason;
+    }
+
+    bool WebSocketClient::getCloseRemote()
+    {
+        std::lock_guard<std::mutex> lck(_mutexCloseData);
+        
+        return _closeRemote;
+    }
+
+    void WebSocketClient::stop()
+    {
+        _webSocket.stop();
+    }
+
+    void WebSocketClient::stop(uint16_t code, const std::string& reason)
+    {
+        _webSocket.close(code, reason);
+        _webSocket.stop();
+    }
+
+    void WebSocketClient::start()
+    {
+        std::string url;
+        {
+            std::stringstream ss;
+            ss << "ws://localhost:"
+               << _port
+               << "/";
+
+            url = ss.str();
+        }
+
+        _webSocket.setUrl(url);
+
+        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)
+                {
+                    log("client connected");
+
+                    _webSocket.disableAutomaticReconnection();
+                }
+                else if (messageType == ix::WebSocket_MessageType_Close)
+                {
+                    log("client disconnected");
+
+                    std::lock_guard<std::mutex> lck(_mutexCloseData);
+                    
+                    _closeCode = closeInfo.code;
+                    _closeReason = std::string(closeInfo.reason);
+                    _closeRemote = closeInfo.remote;
+                    
+                    _webSocket.disableAutomaticReconnection();
+                }
+                else if (messageType == ix::WebSocket_MessageType_Error)
+                {
+                    ss << "Error ! " << error.reason;
+                    log(ss.str());
+
+                    _webSocket.disableAutomaticReconnection();
+                }
+                else if (messageType == ix::WebSocket_MessageType_Pong)
+                {
+                    ss << "Received pong message " << str;
+                    log(ss.str());
+                }
+                else if (messageType == ix::WebSocket_MessageType_Ping)
+                {
+                    ss << "Received ping message " << str;
+                    log(ss.str());
+                }
+                else if (messageType == ix::WebSocket_MessageType_Message)
+                {
+                    ss << "Received message " << str;
+                    log(ss.str());
+                }
+                else
+                {
+                    ss << "Invalid ix::WebSocketMessageType";
+                    log(ss.str());
+                }
+            });
+
+        _webSocket.start();
+    }
+
+    void WebSocketClient::sendMessage(const std::string& text)
+    {
+        _webSocket.send(text);
+    }
+
+    bool startServer(ix::WebSocketServer& server,
+                     uint16_t& receivedCloseCode,
+                     std::string& receivedCloseReason,
+                     bool& receivedCloseRemote,
+                     std::mutex& mutexWrite)
+    {
+        // A dev/null server
+        server.setOnConnectionCallback(
+            [&server, &receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](std::shared_ptr<ix::WebSocket> webSocket,
+                                             std::shared_ptr<ConnectionState> connectionState)
+            {
+                webSocket->setOnMessageCallback(
+                    [webSocket, connectionState, &server, &receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](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)
+                        {
+                            Logger() << "New server connection";
+                            Logger() << "id: " << connectionState->getId();
+                            Logger() << "Uri: " << openInfo.uri;
+                            Logger() << "Headers:";
+                            for (auto it : openInfo.headers)
+                            {
+                                Logger() << it.first << ": " << it.second;
+                            }
+                        }
+                        else if (messageType == ix::WebSocket_MessageType_Close)
+                        {
+                            log("Server closed connection");
+
+                            //Logger() << closeInfo.code;
+                            //Logger() << closeInfo.reason;
+                            //Logger() << closeInfo.remote;
+
+                            std::lock_guard<std::mutex> lck(mutexWrite);
+                            
+                            receivedCloseCode = closeInfo.code;
+                            receivedCloseReason = std::string(closeInfo.reason);
+                            receivedCloseRemote = closeInfo.remote;
+                        }
+                    }
+                );
+            }
+        );
+
+        auto res = server.listen();
+        if (!res.first)
+        {
+            log(res.second);
+            return false;
+        }
+
+        server.start();
+        return true;
+    }
+}
+
+TEST_CASE("Websocket_client_close_default", "[close]")
+{
+    SECTION("Make sure that close code and reason was used and sent to server.")
+    {
+        ix::setupWebSocketTrafficTrackerCallback();
+
+        int port = getFreePort();
+        ix::WebSocketServer server(port);
+        
+        uint16_t serverReceivedCloseCode(0);
+        bool serverReceivedCloseRemote(false);
+        std::string serverReceivedCloseReason("");
+        std::mutex mutexWrite;
+
+        REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
+
+        std::string session = ix::generateSessionId();
+        WebSocketClient webSocketClient(port);
+
+        webSocketClient.start();
+
+        // Wait for all chat instance to be ready
+        while (true)
+        {
+            if (webSocketClient.isReady()) break;
+            ix::msleep(10);
+        }
+
+        REQUIRE(server.getClients().size() == 1);
+
+        ix::msleep(100);
+
+        webSocketClient.stop();
+
+        ix::msleep(200);
+
+        // ensure client close is the same as values given
+        REQUIRE(webSocketClient.getCloseCode() == 1000);
+        REQUIRE(webSocketClient.getCloseReason() == "Normal closure");
+        REQUIRE(webSocketClient.getCloseRemote() == false);
+
+        {
+            std::lock_guard<std::mutex> lck(mutexWrite);
+            
+            // Here we read the code/reason received by the server, and ensure that remote is true
+            REQUIRE(serverReceivedCloseCode == 1000);
+            REQUIRE(serverReceivedCloseReason == "Normal closure");
+            REQUIRE(serverReceivedCloseRemote == true);
+        }
+
+        // Give us 1000ms for the server to notice that clients went away
+        ix::msleep(1000);
+        REQUIRE(server.getClients().size() == 0);
+
+        ix::reportWebSocketTraffic();
+    }
+}
+
+TEST_CASE("Websocket_client_close_params_given", "[close]")
+{
+    SECTION("Make sure that close code and reason was used and sent to server.")
+    {
+        ix::setupWebSocketTrafficTrackerCallback();
+
+        int port = getFreePort();
+        ix::WebSocketServer server(port);
+        
+        uint16_t serverReceivedCloseCode(0);
+        bool serverReceivedCloseRemote(false);
+        std::string serverReceivedCloseReason("");
+        std::mutex mutexWrite;
+
+        REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
+
+        std::string session = ix::generateSessionId();
+        WebSocketClient webSocketClient(port);
+
+        webSocketClient.start();
+
+        // Wait for all chat instance to be ready
+        while (true)
+        {
+            if (webSocketClient.isReady()) break;
+            ix::msleep(10);
+        }
+
+        REQUIRE(server.getClients().size() == 1);
+
+        ix::msleep(100);
+
+        webSocketClient.stop(4000, "My reason");
+
+        ix::msleep(500);
+
+        // ensure client close is the same as values given
+        REQUIRE(webSocketClient.getCloseCode() == 4000);
+        REQUIRE(webSocketClient.getCloseReason() == "My reason");
+        REQUIRE(webSocketClient.getCloseRemote() == false);
+
+        {
+            std::lock_guard<std::mutex> lck(mutexWrite);
+            
+            // Here we read the code/reason received by the server, and ensure that remote is true
+            REQUIRE(serverReceivedCloseCode == 4000);
+            REQUIRE(serverReceivedCloseReason == "My reason");
+            REQUIRE(serverReceivedCloseRemote == true);
+        }
+
+        // Give us 1000ms for the server to notice that clients went away
+        ix::msleep(1000);
+        REQUIRE(server.getClients().size() == 0);
+
+        ix::reportWebSocketTraffic();
+    }
+}
+
+TEST_CASE("Websocket_server_close", "[close]")
+{
+    SECTION("Make sure that close code and reason was read from server.")
+    {
+        ix::setupWebSocketTrafficTrackerCallback();
+
+        int port = getFreePort();
+        ix::WebSocketServer server(port);
+        
+        uint16_t serverReceivedCloseCode(0);
+        bool serverReceivedCloseRemote(false);
+        std::string serverReceivedCloseReason("");
+        std::mutex mutexWrite;
+
+        REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
+
+        std::string session = ix::generateSessionId();
+        WebSocketClient webSocketClient(port);
+
+        webSocketClient.start();
+
+        // Wait for all chat instance to be ready
+        while (true)
+        {
+            if (webSocketClient.isReady()) break;
+            ix::msleep(10);
+        }
+
+        REQUIRE(server.getClients().size() == 1);
+
+        ix::msleep(200);
+
+        server.stop();
+
+        ix::msleep(500);
+
+        // ensure client close is the same as values given
+        REQUIRE(webSocketClient.getCloseCode() == 1000);
+        REQUIRE(webSocketClient.getCloseReason() == "Normal closure");
+        REQUIRE(webSocketClient.getCloseRemote() == true);
+
+        {
+            std::lock_guard<std::mutex> lck(mutexWrite);
+            
+            // Here we read the code/reason received by the server, and ensure that remote is true
+            REQUIRE(serverReceivedCloseCode == 1000);
+            REQUIRE(serverReceivedCloseReason == "Normal closure");
+            REQUIRE(serverReceivedCloseRemote == false);
+        }
+
+        // Give us 1000ms for the server to notice that clients went away
+        ix::msleep(1000);
+        REQUIRE(server.getClients().size() == 0);
+
+        ix::reportWebSocketTraffic();
+    }
+}

test/run.py

@@ -251,8 +251,8 @@ def executeJob(job):
     sys.stderr.write('.')
     # print('Executing ' + job['cmd'] + '...')
 
-    # 2 minutes of timeout for a single test
+    # 10 minutes of timeout for a single test, cf PR #42
-    timeout = 2 * 60
+    timeout = 10 * 60
     command = Command(job['cmd'])
     timedout, ret = command.run(timeout)
 

ws/ws.cpp

@@ -58,7 +58,6 @@ int main(int argc, char** argv)
     bool compress = false;
     bool strict = false;
     bool stress = false;
-    bool disableAutomaticReconnection = false;
     int port = 8080;
     int redisPort = 6379;
     int statsdPort = 8125;
@@ -88,7 +87,6 @@ int main(int argc, char** argv)
 
     CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
     connectApp->add_option("url", url, "Connection url")->required();
-    connectApp->add_flag("-d", disableAutomaticReconnection, "Disable Automatic Reconnection");
 
     CLI::App* chatApp = app.add_subcommand("chat", "Group chat");
     chatApp->add_option("url", url, "Connection url")->required();
@@ -220,7 +218,7 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("connect"))
     {
-        ret = ix::ws_connect_main(url, disableAutomaticReconnection);
+        ret = ix::ws_connect_main(url);
     }
     else if (app.got_subcommand("chat"))
     {

ws/ws.h

@@ -31,7 +31,7 @@ namespace ix
     int ws_chat_main(const std::string& url,
                      const std::string& user);
 
-    int ws_connect_main(const std::string& url, bool disableAutomaticReconnection);
+    int ws_connect_main(const std::string& url);
 
     int ws_receive_main(const std::string& url,
                         bool enablePerMessageDeflate,

ws/ws_connect.cpp

@@ -14,8 +14,7 @@ namespace ix
     class WebSocketConnect
     {
         public:
-            WebSocketConnect(const std::string& _url,
+            WebSocketConnect(const std::string& _url);
-                             bool disableAutomaticReconnection);
 
             void subscribe(const std::string& channel);
             void start();
@@ -30,17 +29,10 @@ namespace ix
             void log(const std::string& msg);
     };
 
-    WebSocketConnect::WebSocketConnect(const std::string& url,
+    WebSocketConnect::WebSocketConnect(const std::string& url) :
-                                       bool disableAutomaticReconnection) :
         _url(url)
     {
-        if (disableAutomaticReconnection)
+        ;
-        {
-            std::cout << "Disabling automatic reconnection with "
-                         "_webSocket.disableAutomaticReconnection()"
-                         " not supported yet" << std::endl;
-            _webSocket.disableAutomaticReconnection();
-        }
     }
 
     void WebSocketConnect::log(const std::string& msg)
@@ -121,10 +113,10 @@ namespace ix
         _webSocket.send(text);
     }
 
-    int ws_connect_main(const std::string& url, bool disableAutomaticReconnection)
+    void interactiveMain(const std::string& url)
     {
         std::cout << "Type Ctrl-D to exit prompt..." << std::endl;
-        WebSocketConnect webSocketChat(url, disableAutomaticReconnection);
+        WebSocketConnect webSocketChat(url);
         webSocketChat.start();
 
         while (true)
@@ -157,7 +149,11 @@ namespace ix
 
         std::cout << std::endl;
         webSocketChat.stop();
+    }
 
+    int ws_connect_main(const std::string& url)
+    {
+        interactiveMain(url);
         return 0;
     }
 }