Added interrupt() function to break out of poll() before timeout

easywsclient.cpp

@@ -1,5 +1,6 @@
 
 #ifdef _WIN32
+    #define NOMINMAX
     #if defined(_MSC_VER) && !defined(_CRT_SECURE_NO_WARNINGS)
         #define _CRT_SECURE_NO_WARNINGS // _CRT_SECURE_NO_WARNINGS for sscanf errors in MSVC2013 Express
     #endif
@@ -71,6 +72,7 @@
 
 #include <vector>
 #include <string>
+#include <algorithm>
 
 #include "easywsclient.hpp"
 
@@ -109,11 +111,81 @@ socket_t hostname_connect(const std::string& hostname, int port) {
     return sockfd;
 }
 
+#ifdef _WIN32
+/* dumb_socketpair
+ * Copyright 2007 by Nathan C. Myers <ncm@cantrip.org>; some rights reserved.
+ * This code is Free Software.  It may be copied freely, in original or
+ * modified form, subject only to the restrictions that (1) the author is
+ * relieved from all responsibilities for any use for any purpose, and (2)
+ * this copyright notice must be retained, unchanged, in its entirety.  If
+ * for any reason the author might be held responsible for any consequences
+ * of copying or use, license is withheld.
+ */
+int dumb_socketpair(SOCKET socks[2], int make_overlapped)
+{
+	union {
+		struct sockaddr_in inaddr;
+		struct sockaddr addr;
+	} a;
+	SOCKET listener;
+	int e;
+	socklen_t addrlen = sizeof(a.inaddr);
+	DWORD flags = (make_overlapped ? WSA_FLAG_OVERLAPPED : 0);
+	int reuse = 1;
+
+	if (socks == 0) {
+		WSASetLastError(WSAEINVAL);
+		return SOCKET_ERROR;
+	}
+
+	listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+	if (listener == INVALID_SOCKET)
+		return SOCKET_ERROR;
+
+	memset(&a, 0, sizeof(a));
+	a.inaddr.sin_family = AF_INET;
+	a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+	a.inaddr.sin_port = 0;
+
+	socks[0] = socks[1] = INVALID_SOCKET;
+	do {
+		if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
+			(char*)& reuse, (socklen_t) sizeof(reuse)) == -1)
+			break;
+		if (bind(listener, &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+			break;
+		if (getsockname(listener, &a.addr, &addrlen) == SOCKET_ERROR)
+			break;
+		if (listen(listener, 1) == SOCKET_ERROR)
+			break;
+		socks[0] = WSASocketW(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
+		if (socks[0] == INVALID_SOCKET)
+			break;
+		if (connect(socks[0], &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+			break;
+		socks[1] = accept(listener, NULL, NULL);
+		if (socks[1] == INVALID_SOCKET)
+			break;
+
+		closesocket(listener);
+		return 0;
+
+	} while (0);
+
+	e = WSAGetLastError();
+	closesocket(listener);
+	closesocket(socks[0]);
+	closesocket(socks[1]);
+	WSASetLastError(e);
+	return SOCKET_ERROR;
+}
+#endif
 
 class _DummyWebSocket : public easywsclient::WebSocket
 {
   public:
     void poll(int timeout) { }
+    void interrupt() { }
     void send(const std::string& message) { }
     void sendBinary(const std::string& message) { }
     void sendBinary(const std::vector<uint8_t>& message) { }
@@ -171,14 +243,51 @@ class _RealWebSocket : public easywsclient::WebSocket
 
     socket_t sockfd;
     readyStateValues readyState;
+    int interruptIn;
+    int interruptOut;
     bool useMask;
     bool isRxBad;
 
     _RealWebSocket(socket_t sockfd, bool useMask)
             : sockfd(sockfd)
             , readyState(OPEN)
+            , interruptIn(0)
+            , interruptOut(0)
             , useMask(useMask)
             , isRxBad(false) {
+       #ifdef _WIN32
+        SOCKET pipes[2] = {0, 0};
+        if (!dumb_socketpair(pipes, 0))
+        {
+            interruptIn  = pipes[0];
+            interruptOut = pipes[1];
+
+            u_long on = 1;
+            ioctlsocket(interruptIn, FIONBIO, &on);
+        }
+       #else
+                int fd[2] = {0, 0};
+        if (socketpair(PF_LOCAL, SOCK_STREAM, 0, fd) == 0)
+        {
+            interruptIn  = fd[0];
+            interruptOut = fd[1];
+
+            int flags = fcntl(interruptIn, F_GETFL, 0);
+            if (fcntl(interruptIn, F_SETFL, flags | O_NONBLOCK) == -1)
+            {
+                closesocket(interruptIn);
+                closesocket(interruptOut);
+                interruptIn  = 0;
+                interruptOut = 0;
+            }
+        }
+       #endif
+    }
+
+    ~_RealWebSocket()
+    {
+        if (interruptIn)  closesocket(interruptIn);
+        if (interruptOut) closesocket(interruptOut);
     }
 
     readyStateValues getReadyState() const {
@@ -200,9 +309,25 @@ class _RealWebSocket : public easywsclient::WebSocket
             FD_ZERO(&rfds);
             FD_ZERO(&wfds);
             FD_SET(sockfd, &rfds);
+
+            int maxSocket = sockfd;
+            if (interruptIn) {
+                FD_SET(interruptIn, &rfds);
+                maxSocket = std::max(maxSocket, interruptIn);
+            }
+
             if (txbuf.size()) { FD_SET(sockfd, &wfds); }
-            select(sockfd + 1, &rfds, &wfds, 0, timeout > 0 ? &tv : 0);
+
+            select(maxSocket + 1, &rfds, &wfds, nullptr, &tv);
+        }
+
+        while (true) {
+            char dummy[128] = {0};
+            ssize_t ret = recv(interruptIn, dummy, sizeof(dummy), 0);
+            if (ret <= 0)
+                break;
         }
+
         while (true) {
             // FD_ISSET(0, &rfds) will be true
             int N = rxbuf.size();
@@ -247,6 +372,12 @@ class _RealWebSocket : public easywsclient::WebSocket
         }
     }
 
+    void interrupt()
+    {
+        if (interruptOut)
+            ::send(interruptOut, "\0", 1, 0);
+    }
+
     // Callable must have signature: void(const std::string & message).
     // Should work with C functions, C++ functors, and C++11 std::function and
     // lambda:

easywsclient.hpp

@@ -19,7 +19,7 @@ struct BytesCallback_Imp { virtual void operator()(const std::vector<uint8_t>& m
 class WebSocket {
   public:
     typedef WebSocket * pointer;
-    typedef enum readyStateValues { CLOSING, CLOSED, CONNECTING, OPEN } readyStateValues;
+    typedef enum readyStateValues { CLOSING, CLOSED, OPEN } readyStateValues;
 
     // Factories:
     static pointer create_dummy();
@@ -29,6 +29,7 @@ class WebSocket {
     // Interfaces:
     virtual ~WebSocket() { }
     virtual void poll(int timeout = 0) = 0; // timeout in milliseconds
+    virtual void interrupt() = 0; // interrupt polling
     virtual void send(const std::string& message) = 0;
     virtual void sendBinary(const std::string& message) = 0;
     virtual void sendBinary(const std::vector<uint8_t>& message) = 0;