Non-blocking sockets support (TLS, WebSocket)

Net/include/Poco/Net/WebSocket.h

@@ -49,6 +49,11 @@ class Net_API WebSocket: public StreamSocket
 	/// Note that special frames like PING must be handled at
 	/// application level. In the case of a PING, a PONG message
 	/// must be returned.
+	///
+	/// Once connected, a WebSocket can be put into non-blocking
+	/// mode, by calling setBlocking(false). 
+	/// Please refer to the sendFrame() and receiveFrame() documentation
+	/// for non-blocking behavior.
 {
 public:
 	enum Mode
@@ -221,15 +226,21 @@ class Net_API WebSocket: public StreamSocket
 
 #endif //POCO_NEW_STATE_ON_MOVE
 
-	void shutdown();
+	int shutdown();
 		/// Sends a Close control frame to the server end of
 		/// the connection to initiate an orderly shutdown
 		/// of the connection.
+		///
+		/// Returns the number of bytes sent or -1 if the socket
+		/// is non-blocking and the frame cannot be sent at this time.
 
-	void shutdown(Poco::UInt16 statusCode, const std::string& statusMessage = "");
+	int shutdown(Poco::UInt16 statusCode, const std::string& statusMessage = "");
 		/// Sends a Close control frame to the server end of
 		/// the connection to initiate an orderly shutdown
 		/// of the connection.
+		///
+		/// Returns the number of bytes sent or -1 if the socket
+		/// is non-blocking and the frame cannot be sent at this time.
 
 	int sendFrame(const void* buffer, int length, int flags = FRAME_TEXT);
 		/// Sends the contents of the given buffer through
@@ -238,11 +249,15 @@ class Net_API WebSocket: public StreamSocket
 		/// Values from the FrameFlags, FrameOpcodes and SendFlags enumerations
 		/// can be specified in flags.
 		///
-		/// Returns the number of bytes sent, which may be
-		/// less than the number of bytes specified.
+		/// Returns the number of bytes sent.
 		///
-		/// Certain socket implementations may also return a negative
-		/// value denoting a certain condition.
+		/// If the WebSocket is non-blocking and the frame could not
+		/// be sent in full, returns -1. In this case, the call
+		/// to sendFrame() must be repeated with exactly the same
+		/// parameters as soon as the socket becomes writable again 
+		/// (see select() or poll()). 
+		/// The value of length is returned after the complete
+		/// frame has been sent.
 
 	int receiveFrame(void* buffer, int length, int& flags);
 		/// Receives a frame from the socket and stores it
@@ -272,6 +287,14 @@ class Net_API WebSocket: public StreamSocket
 		///
 		/// The frame flags and opcode (FrameFlags and FrameOpcodes)
 		/// is stored in flags.
+		///
+		/// In case of a non-blocking socket, may return -1, even
+		/// if a partial frame has been received. 
+		/// In this case, receiveFrame() should be called again as 
+		/// soon as more data becomes available (see select() or poll()).
+		/// Eventually, receiveFrame() will return the complete frame.
+		/// The given buffer will not be modified until the full frame has 
+		/// been received.
 
 	int receiveFrame(Poco::Buffer<char>& buffer, int& flags);
 		/// Receives a frame from the socket and stores it
@@ -314,6 +337,14 @@ class Net_API WebSocket: public StreamSocket
 		/// called on the buffer beforehand, if the expectation is that
 		/// the received data is stored starting at the beginning of the
 		/// buffer.
+		///
+		/// In case of a non-blocking socket, may return -1, even
+		/// if a partial frame has been received. 
+		/// In this case, receiveFrame() should be called again as 
+		/// soon as more data becomes available (see select() or poll()).
+		/// Eventually, receiveFrame() will return the complete frame.
+		/// The given buffer will not be modified until the full frame has 
+		/// been received.
 
 	Mode mode() const;
 		/// Returns WS_SERVER if the WebSocket is a server-side

Net/include/Poco/Net/WebSocketImpl.h

@@ -42,12 +42,21 @@ class Net_API WebSocketImpl: public StreamSocketImpl
 	// StreamSocketImpl
 	virtual int sendBytes(const void* buffer, int length, int flags);
 		/// Sends a WebSocket protocol frame.
+		///
+		/// See WebSocket::sendFrame() for more information, including
+		/// behavior if set to non-blocking.
 
 	virtual int receiveBytes(void* buffer, int length, int flags);
 		/// Receives a WebSocket protocol frame.
+		///
+		/// See WebSocket::receiveFrame() for more information, including
+		/// behavior if set to non-blocking.
 
 	virtual int receiveBytes(Poco::Buffer<char>& buffer, int flags = 0, const Poco::Timespan& span = 0);
 		/// Receives a WebSocket protocol frame.
+		///
+		/// See WebSocket::receiveFrame() for more information, including
+		/// behavior if set to non-blocking.
 
 	virtual SocketImpl* acceptConnection(SocketAddress& clientAddr);
 	virtual void connect(const SocketAddress& address);
@@ -67,10 +76,16 @@ class Net_API WebSocketImpl: public StreamSocketImpl
 	virtual void sendUrgent(unsigned char data);
 	virtual int available();
 	virtual bool secure() const;
+	virtual void setSendBufferSize(int size);
+	virtual int getSendBufferSize();
+	virtual void setReceiveBufferSize(int size);
+	virtual int getReceiveBufferSize();
 	virtual void setSendTimeout(const Poco::Timespan& timeout);
 	virtual Poco::Timespan getSendTimeout();
 	virtual void setReceiveTimeout(const Poco::Timespan& timeout);
 	virtual Poco::Timespan getReceiveTimeout();
+	virtual void setBlocking(bool flag);
+	virtual bool getBlocking() const;
 
 	// Internal
 	int frameFlags() const;
@@ -93,13 +108,35 @@ class Net_API WebSocketImpl: public StreamSocketImpl
 	enum
 	{
 		FRAME_FLAG_MASK   = 0x80,
-		MAX_HEADER_LENGTH = 14
+		MAX_HEADER_LENGTH = 14,
+		MASK_LENGTH       = 4
+	};
+
+	struct ReceiveState
+	{
+		int frameFlags = 0;
+		bool useMask = false;
+		char mask[MASK_LENGTH];
+		int headerLength = 0;
+		int payloadLength = 0;
+		int remainingPayloadLength = 0;
+		Poco::Buffer<char> payload{0};
+	};
+
+	struct SendState
+	{
+		int length = 0;
+		int remainingPayloadOffset = 0;
+		int remainingPayloadLength = 0;
+		Poco::Buffer<char> payload{0};
 	};
 
-	int receiveHeader(char mask[4], bool& useMask);
-	int receivePayload(char *buffer, int payloadLength, char mask[4], bool useMask);
-	int receiveNBytes(void* buffer, int bytes);
-	int receiveSomeBytes(char* buffer, int bytes);
+	int peekHeader(ReceiveState& receiveState);
+	void skipHeader(int headerLength);
+	int receivePayload(char *buffer, int payloadLength, char mask[MASK_LENGTH], bool useMask);
+	int receiveNBytes(void* buffer, int length);
+	int receiveSomeBytes(char* buffer, int length);
+	int peekSomeBytes(char* buffer, int length);
 	virtual ~WebSocketImpl();
 
 private:
@@ -109,8 +146,9 @@ class Net_API WebSocketImpl: public StreamSocketImpl
 	int _maxPayloadSize;
 	Poco::Buffer<char> _buffer;
 	int _bufferOffset;
-	int _frameFlags;
 	bool _mustMaskPayload;
+	ReceiveState _receiveState;
+	SendState _sendState;
 	Poco::Random _rnd;
 };
 
@@ -120,7 +158,7 @@ class Net_API WebSocketImpl: public StreamSocketImpl
 //
 inline int WebSocketImpl::frameFlags() const
 {
-	return _frameFlags;
+	return _receiveState.frameFlags;
 }
 
 

Net/src/SocketImpl.cpp

@@ -361,24 +361,37 @@ void SocketImpl::checkBrokenTimeout(SelectMode mode)
 
 int SocketImpl::sendBytes(const void* buffer, int length, int flags)
 {
-	checkBrokenTimeout(SELECT_WRITE);
-
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_WRITE);
+	}
 	int rc;
 	do
 	{
 		if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
 		rc = ::send(_sockfd, reinterpret_cast<const char*>(buffer), length, flags);
 	}
 	while (_blocking && rc < 0 && lastError() == POCO_EINTR);
-	if (rc < 0) error();
+	if (rc < 0)
+	{
+		int err = lastError();
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
+			;
+		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
+			throw TimeoutException(err);
+		else
+			error(err);
+	}
 	return rc;
 }
 
 
 int SocketImpl::sendBytes(const SocketBufVec& buffers, int flags)
 {
-	checkBrokenTimeout(SELECT_WRITE);
-
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_WRITE);
+	}
 	int rc = 0;
 	do
 	{
@@ -395,15 +408,26 @@ int SocketImpl::sendBytes(const SocketBufVec& buffers, int flags)
 #endif
 	}
 	while (_blocking && rc < 0 && lastError() == POCO_EINTR);
-	if (rc < 0) error();
+	if (rc < 0)
+	{
+		int err = lastError();
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
+			;
+		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
+			throw TimeoutException(err);
+		else
+			error(err);
+	}
 	return rc;
 }
 
 
 int SocketImpl::receiveBytes(void* buffer, int length, int flags)
 {
-	checkBrokenTimeout(SELECT_READ);
-
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_READ);
+	}
 	int rc;
 	do
 	{
@@ -414,7 +438,7 @@ int SocketImpl::receiveBytes(void* buffer, int length, int flags)
 	if (rc < 0)
 	{
 		int err = lastError();
-		if (err == POCO_EAGAIN && !_blocking)
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
 			;
 		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
 			throw TimeoutException(err);
@@ -427,8 +451,10 @@ int SocketImpl::receiveBytes(void* buffer, int length, int flags)
 
 int SocketImpl::receiveBytes(SocketBufVec& buffers, int flags)
 {
-	checkBrokenTimeout(SELECT_READ);
-
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_READ);
+	}
 	int rc = 0;
 	do
 	{
@@ -448,7 +474,7 @@ int SocketImpl::receiveBytes(SocketBufVec& buffers, int flags)
 	if (rc < 0)
 	{
 		int err = lastError();
-		if (err == POCO_EAGAIN && !_blocking)
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
 			;
 		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
 			throw TimeoutException(err);
@@ -476,7 +502,7 @@ int SocketImpl::receiveBytes(Poco::Buffer<char>& buffer, int flags, const Poco::
 		if (rc < 0)
 		{
 			int err = lastError();
-			if (err == POCO_EAGAIN && !_blocking)
+			if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
 				;
 			else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
 				throw TimeoutException(err);
@@ -502,7 +528,16 @@ int SocketImpl::sendTo(const void* buffer, int length, const SocketAddress& addr
 #endif
 	}
 	while (_blocking && rc < 0 && lastError() == POCO_EINTR);
-	if (rc < 0) error();
+	if (rc < 0)
+	{
+		int err = lastError();
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
+			;
+		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
+			throw TimeoutException(err);
+		else
+			error(err);
+	}
 	return rc;
 }
 
@@ -534,7 +569,16 @@ int SocketImpl::sendTo(const SocketBufVec& buffers, const SocketAddress& address
 #endif
 	}
 	while (_blocking && rc < 0 && lastError() == POCO_EINTR);
-	if (rc < 0) error();
+	if (rc < 0)
+	{
+		int err = lastError();
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
+			;
+		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
+			throw TimeoutException(err);
+		else
+			error(err);
+	}
 	return rc;
 }
 
@@ -556,7 +600,10 @@ int SocketImpl::receiveFrom(void* buffer, int length, SocketAddress& address, in
 
 int SocketImpl::receiveFrom(void* buffer, int length, struct sockaddr** ppSA, poco_socklen_t** ppSALen, int flags)
 {
-	checkBrokenTimeout(SELECT_READ);
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_READ);
+	}
 	int rc;
 	do
 	{
@@ -567,7 +614,7 @@ int SocketImpl::receiveFrom(void* buffer, int length, struct sockaddr** ppSA, po
 	if (rc < 0)
 	{
 		int err = lastError();
-		if (err == POCO_EAGAIN && !_blocking)
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
 			;
 		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
 			throw TimeoutException(err);
@@ -595,7 +642,10 @@ int SocketImpl::receiveFrom(SocketBufVec& buffers, SocketAddress& address, int f
 
 int SocketImpl::receiveFrom(SocketBufVec& buffers, struct sockaddr** pSA, poco_socklen_t** ppSALen, int flags)
 {
-	checkBrokenTimeout(SELECT_READ);
+	if (_blocking)
+	{
+		checkBrokenTimeout(SELECT_READ);
+	}
 	int rc = 0;
 	do
 	{
@@ -624,7 +674,7 @@ int SocketImpl::receiveFrom(SocketBufVec& buffers, struct sockaddr** pSA, poco_s
 	if (rc < 0)
 	{
 		int err = lastError();
-		if (err == POCO_EAGAIN && !_blocking)
+		if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
 			;
 		else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
 			throw TimeoutException(err);

Net/src/StreamSocketImpl.cpp

@@ -61,7 +61,7 @@ int StreamSocketImpl::sendBytes(const void* buffer, int length, int flags)
 	while (remaining > 0)
 	{
 		int n = SocketImpl::sendBytes(p, remaining, flags);
-		poco_assert_dbg (n >= 0);
+		poco_assert_dbg (!blocking || n >= 0);
 		p += n;
 		sent += n;
 		remaining -= n;