kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Feature/send large message (#14)

Browse Source 
* introduce send fragment

* can pass a fin frame

* can send messages which are a perfect multiple of the chunk size

* set fin only for last fragment

* cleanup

* last fragment should be of type CONTINUATION

* Add simple send and receive programs

* speedups receiving + better way to wait for thing

* receive speedup by using linked list of chunks instead of large array

* document bug

* use chunks to receive data

* trailing spaces
This commit is contained in:
Benjamin Sergeant
2019-02-20 18:59:07 -08:00
committed by GitHub
parent dd4e29542c
commit 932bb732e0
72 changed files with 9117 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ixwebsocket/IXCancellationRequest.cpp
View File

@ -8,7 +8,7 @@
#include <chrono>
namespace ix
namespace ix
{
CancellationRequest makeCancellationRequestWithTimeout(int secs,
std::atomic<bool>& requestInitCancellation)
@ -20,7 +20,7 @@ namespace ix
{
// Was an explicit cancellation requested ?
if (requestInitCancellation) return true;
auto now = std::chrono::system_clock::now();
if ((now - start) > timeout) return true;

2
ixwebsocket/IXCancellationRequest.h
View File

@ -9,7 +9,7 @@
#include <functional>
#include <atomic>
namespace ix
namespace ix
{
using CancellationRequest = std::function<bool()>;

12
ixwebsocket/IXDNSLookup.cpp
View File

@ -10,7 +10,7 @@
#include <string.h>
#include <chrono>
namespace ix
namespace ix
{
const int64_t DNSLookup::kDefaultWait = 10; // ms
@ -26,7 +26,7 @@ namespace ix
_done(false),
_id(_nextId++)
{
}
DNSLookup::~DNSLookup()
@ -36,7 +36,7 @@ namespace ix
_activeJobs.erase(_id);
}
struct addrinfo* DNSLookup::getAddrInfo(const std::string& hostname,
struct addrinfo* DNSLookup::getAddrInfo(const std::string& hostname,
int port,
std::string& errMsg)
{
@ -49,7 +49,7 @@ namespace ix
std::string sport = std::to_string(port);
struct addrinfo* res;
int getaddrinfo_result = getaddrinfo(hostname.c_str(), sport.c_str(),
int getaddrinfo_result = getaddrinfo(hostname.c_str(), sport.c_str(),
&hints, &res);
if (getaddrinfo_result)
{
@ -101,7 +101,7 @@ namespace ix
_activeJobs.insert(_id);
}
//
//
// Good resource on thread forced termination
// https://www.bo-yang.net/2017/11/19/cpp-kill-detached-thread
//
@ -141,7 +141,7 @@ namespace ix
void DNSLookup::run(uint64_t id, const std::string& hostname, int port) // thread runner
{
// We don't want to read or write into members variables of an object that could be
// gone, so we use temporary variables (res) or we pass in by copy everything that
// gone, so we use temporary variables (res) or we pass in by copy everything that
// getAddrInfo needs to work.
std::string errMsg;
struct addrinfo* res = getAddrInfo(hostname, port, errMsg);

6
ixwebsocket/IXDNSLookup.h
View File

@ -3,7 +3,7 @@
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*
* Resolve a hostname+port to a struct addrinfo obtained with getaddrinfo
* Resolve a hostname+port to a struct addrinfo obtained with getaddrinfo
* Does this in a background thread so that it can be cancelled, since
* getaddrinfo is a blocking call, and we don't want to block the main thread on Mobile.
*/
@ -20,7 +20,7 @@
struct addrinfo;
namespace ix
namespace ix
{
class DNSLookup {
public:
@ -39,7 +39,7 @@ namespace ix
struct addrinfo* resolveBlocking(std::string& errMsg,
const CancellationRequest& isCancellationRequested);
static struct addrinfo* getAddrInfo(const std::string& hostname,
static struct addrinfo* getAddrInfo(const std::string& hostname,
int port,
std::string& errMsg);

8
ixwebsocket/IXEventFd.cpp
View File

@ -14,7 +14,7 @@
// eventfd was added in Linux kernel 2.x, and our oldest Android (Kitkat 4.4)
// is on Kernel 3.x
//
// cf Android/Kernel table here
// cf Android/Kernel table here
// https://android.stackexchange.com/questions/51651/which-android-runs-which-linux-kernel
//
@ -28,9 +28,9 @@
#include <unistd.h> // for write
#endif
namespace ix
namespace ix
{
EventFd::EventFd() :
EventFd::EventFd() :
_eventfd(-1)
{
#ifdef __linux__
@ -65,7 +65,7 @@ namespace ix
#if defined(__linux__)
if (_eventfd == -1) return false;
// 0 is a special value ; select will not wake up
// 0 is a special value ; select will not wake up
uint64_t value = 0;
// we should write 8 bytes for an uint64_t

2
ixwebsocket/IXEventFd.h
View File

@ -6,7 +6,7 @@
#pragma once
namespace ix
namespace ix
{
class EventFd {
public:

14
ixwebsocket/IXProgressCallback.h Normal file
View File

@ -0,0 +1,14 @@
/*
* IXProgressCallback.h
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <functional>
namespace ix
{
using OnProgressCallback = std::function<bool(int current, int total)>;
}

33
ixwebsocket/IXSocket.cpp
View File

@ -15,17 +15,16 @@
#include <stdint.h>
#include <fcntl.h>
#include <sys/types.h>
#include <poll.h>
#include <algorithm>
#include <iostream>
namespace ix
namespace ix
{
const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
Socket::Socket(int fd) :
Socket::Socket(int fd) :
_sockfd(fd)
{
@ -44,21 +43,22 @@ namespace ix
return;
}
#ifdef __linux__
constexpr int nfds = 2;
#else
constexpr int nfds = 1;
#endif
struct pollfd fds[nfds];
fds[0].fd = _sockfd;
fds[0].events = POLLIN;
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(_sockfd, &rfds);
#ifdef __linux__
fds[1].fd = _eventfd.getFd();
fds[1].events = POLLIN;
FD_SET(_eventfd.getFd(), &rfds);
#endif
int ret = ::poll(fds, nfds, timeoutSecs * 1000);
struct timeval timeout;
timeout.tv_sec = timeoutSecs;
timeout.tv_usec = 0;
int sockfd = _sockfd;
int nfds = (std::max)(sockfd, _eventfd.getFd());
int ret = select(nfds + 1, &rfds, nullptr, nullptr,
(timeoutSecs < 0) ? nullptr : &timeout);
PollResultType pollResult = PollResultType_ReadyForRead;
if (ret < 0)
@ -71,7 +71,6 @@ namespace ix
}
onPollCallback(pollResult);
}
void Socket::wakeUpFromPoll()
@ -151,7 +150,7 @@ namespace ix
#ifdef _WIN32
INT rc;
WSADATA wsaData;
rc = WSAStartup(MAKEWORD(2, 2), &wsaData);
return rc != 0;
#else

4
ixwebsocket/IXSocket.h
View File

@ -19,7 +19,7 @@ typedef SSIZE_T ssize_t;
#include "IXEventFd.h"
#include "IXCancellationRequest.h"
namespace ix
namespace ix
{
enum PollResultType
{
@ -42,7 +42,7 @@ namespace ix
virtual void wakeUpFromPoll();
// Virtual methods
virtual bool connect(const std::string& url,
virtual bool connect(const std::string& url,
int port,
std::string& errMsg,
const CancellationRequest& isCancellationRequested);

14
ixwebsocket/IXSocketAppleSSL.cpp
View File

@ -50,7 +50,7 @@ OSStatus read_from_socket(SSLConnectionRef connection, void *data, size_t *len)
else
return noErr;
}
else if (0 == status)
else if (0 == status)
{
*len = 0;
return errSSLClosedGraceful;
@ -102,7 +102,7 @@ OSStatus write_to_socket(SSLConnectionRef connection, const void *data, size_t *
else
{
*len = 0;
if (EAGAIN == errno)
if (EAGAIN == errno)
{
return errSSLWouldBlock;
}
@ -141,7 +141,7 @@ std::string getSSLErrorDescription(OSStatus status)
} // anonymous namespace
namespace ix
namespace ix
{
SocketAppleSSL::SocketAppleSSL(int fd) : Socket(fd),
_sslContext(nullptr)
@ -176,11 +176,11 @@ namespace ix
do {
status = SSLHandshake(_sslContext);
} while (errSSLWouldBlock == status ||
} while (errSSLWouldBlock == status ||
errSSLServerAuthCompleted == status);
}
if (noErr != status)
if (noErr != status)
{
errMsg = getSSLErrorDescription(status);
close();
@ -230,7 +230,7 @@ namespace ix
ssize_t SocketAppleSSL::recv(void* buf, size_t nbyte)
{
OSStatus status = errSSLWouldBlock;
while (errSSLWouldBlock == status)
while (errSSLWouldBlock == status)
{
size_t processed = 0;
std::lock_guard<std::mutex> lock(_mutex);
@ -239,7 +239,7 @@ namespace ix
if (processed > 0)
return (ssize_t) processed;
// The connection was reset, inform the caller that this
// The connection was reset, inform the caller that this
// Socket should close
if (status == errSSLClosedGraceful ||
status == errSSLClosedNoNotify ||

6
ixwebsocket/IXSocketAppleSSL.h
View File

@ -14,15 +14,15 @@
#include <mutex>
namespace ix
namespace ix
{
class SocketAppleSSL : public Socket
class SocketAppleSSL : public Socket
{
public:
SocketAppleSSL(int fd = -1);
~SocketAppleSSL();
virtual bool connect(const std::string& host,
virtual bool connect(const std::string& host,
int port,
std::string& errMsg,
const CancellationRequest& isCancellationRequested) final;

8
ixwebsocket/IXSocketConnect.cpp
View File

@ -30,7 +30,7 @@ namespace
}
}
namespace ix
namespace ix
{
//
// This function can be cancelled every 50 ms
@ -42,7 +42,7 @@ namespace ix
const CancellationRequest& isCancellationRequested)
{
errMsg = "no error";
int fd = socket(address->ai_family,
address->ai_socktype,
address->ai_protocol);
@ -72,7 +72,7 @@ namespace ix
errMsg = "Cancelled";
return -1;
}
// Use select to check the status of the new connection
struct timeval timeout;
timeout.tv_sec = 0;
@ -179,7 +179,7 @@ namespace ix
// 3. (apple) prevent SIGPIPE from being emitted when the remote end disconnect
#ifdef SO_NOSIGPIPE
int value = 1;
setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE,
setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE,
(void *)&value, sizeof(value));
#endif
}

2
ixwebsocket/IXSocketConnect.h
View File

@ -12,7 +12,7 @@
struct addrinfo;
namespace ix
namespace ix
{
class SocketConnect {
public:

18
ixwebsocket/IXSocketOpenSSL.cpp
View File

@ -18,12 +18,12 @@
#include <errno.h>
#define socketerrno errno
namespace ix
namespace ix
{
std::atomic<bool> SocketOpenSSL::_openSSLInitializationSuccessful(false);
SocketOpenSSL::SocketOpenSSL(int fd) : Socket(fd),
_ssl_connection(nullptr),
_ssl_connection(nullptr),
_ssl_context(nullptr)
{
std::call_once(_openSSLInitFlag, &SocketOpenSSL::openSSLInitialize, this);
@ -80,7 +80,7 @@ namespace ix
return "OpenSSL failed - underlying BIO reported an I/O error";
}
}
else if (err == SSL_ERROR_SSL)
else if (err == SSL_ERROR_SSL)
{
e = ERR_get_error();
std::string errMsg("OpenSSL failed - ");
@ -149,7 +149,7 @@ namespace ix
#if OPENSSL_VERSION_NUMBER < 0x10100000L
// Check server name
bool hostname_verifies_ok = false;
STACK_OF(GENERAL_NAME) *san_names =
STACK_OF(GENERAL_NAME) *san_names =
(STACK_OF(GENERAL_NAME)*) X509_get_ext_d2i((X509 *)server_cert,
NID_subject_alt_name, NULL, NULL);
if (san_names)
@ -160,8 +160,8 @@ namespace ix
if (sk_name->type == GEN_DNS)
{
char *name = (char *)ASN1_STRING_data(sk_name->d.dNSName);
if ((size_t)ASN1_STRING_length(sk_name->d.dNSName) == strlen(name) &&
checkHost(hostname, name))
if ((size_t)ASN1_STRING_length(sk_name->d.dNSName) == strlen(name) &&
checkHost(hostname, name))
{
hostname_verifies_ok = true;
break;
@ -185,8 +185,8 @@ namespace ix
ASN1_STRING *cn_asn1 = X509_NAME_ENTRY_get_data(cn_entry);
char *cn = (char *)ASN1_STRING_data(cn_asn1);
if ((size_t)ASN1_STRING_length(cn_asn1) == strlen(cn) &&
checkHost(hostname, cn))
if ((size_t)ASN1_STRING_length(cn_asn1) == strlen(cn) &&
checkHost(hostname, cn))
{
hostname_verifies_ok = true;
}
@ -205,7 +205,7 @@ namespace ix
return true;
}
bool SocketOpenSSL::openSSLHandshake(const std::string& host, std::string& errMsg)
bool SocketOpenSSL::openSSLHandshake(const std::string& host, std::string& errMsg)
{
while (true)
{

6
ixwebsocket/IXSocketOpenSSL.h
View File

@ -17,15 +17,15 @@
#include <mutex>
namespace ix
namespace ix
{
class SocketOpenSSL : public Socket
class SocketOpenSSL : public Socket
{
public:
SocketOpenSSL(int fd = -1);
~SocketOpenSSL();
virtual bool connect(const std::string& host,
virtual bool connect(const std::string& host,
int port,
std::string& errMsg,
const CancellationRequest& isCancellationRequested) final;

8
ixwebsocket/IXSocketSChannel.cpp
View File

@ -47,7 +47,7 @@
// link with ntdsapi.lib for DsMakeSpn function
#pragma comment(lib, "ntdsapi.lib")
// The following function assumes that Winsock
// The following function assumes that Winsock
// has already been initialized
@ -59,7 +59,7 @@
# error("This file should only be built on Windows")
#endif
namespace ix
namespace ix
{
SocketSChannel::SocketSChannel()
{
@ -68,7 +68,7 @@ namespace ix
SocketSChannel::~SocketSChannel()
{
}
bool SocketSChannel::connect(const std::string& host,
@ -78,7 +78,7 @@ namespace ix
return Socket::connect(host, port, errMsg);
}
void SocketSChannel::secureSocket()
{
// there will be a lot to do here ...

6
ixwebsocket/IXSocketSChannel.h
View File

@ -8,15 +8,15 @@
#include "IXSocket.h"
namespace ix
namespace ix
{
class SocketSChannel : public Socket
class SocketSChannel : public Socket
{
public:
SocketSChannel();
~SocketSChannel();
virtual bool connect(const std::string& host,
virtual bool connect(const std::string& host,
int port,
std::string& errMsg) final;
virtual void close() final;

10
ixwebsocket/IXSocketServer.cpp
View File

@ -14,7 +14,7 @@
#include <future>
#include <string.h>
namespace ix
namespace ix
{
const int SocketServer::kDefaultPort(8080);
const std::string SocketServer::kDefaultHost("127.0.0.1");
@ -42,13 +42,13 @@ namespace ix
void SocketServer::logError(const std::string& str)
{
std::lock_guard<std::mutex> lock(_logMutex);
fprintf(stderr, "%s\n", str.c_str());
std::cerr << str << std::endl;
}
void SocketServer::logInfo(const std::string& str)
{
std::lock_guard<std::mutex> lock(_logMutex);
fprintf(stderr, "%s\n", str.c_str());
std::cout << str << std::endl;
}
std::pair<bool, std::string> SocketServer::listen()
@ -83,7 +83,7 @@ namespace ix
server.sin_family = AF_INET;
server.sin_port = htons(_port);
// Using INADDR_ANY trigger a pop-up box as binding to any address is detected
// Using INADDR_ANY trigger a pop-up box as binding to any address is detected
// by the osx firewall. We need to codesign the binary with a self-signed cert
// to allow that, but this is a bit of a pain. (this is what node or python would do).
//
@ -216,7 +216,7 @@ namespace ix
// Launch the handleConnection work asynchronously in its own thread.
//
// the destructor of a future returned by std::async blocks,
// the destructor of a future returned by std::async blocks,
// so we need to declare it outside of this loop
f = std::async(std::launch::async,
&SocketServer::handleConnection,

2
ixwebsocket/IXSocketServer.h
View File

@ -16,7 +16,7 @@
#include <atomic>
#include <condition_variable>
namespace ix
namespace ix
{
class SocketServer {
public:

27
ixwebsocket/IXWebSocket.cpp
View File

@ -50,7 +50,7 @@ namespace ix
);
}
WebSocket::~WebSocket()
WebSocket::~WebSocket()
{
stop();
}
@ -135,7 +135,7 @@ namespace ix
}
_onMessageCallback(WebSocket_MessageType_Open, "", 0,
WebSocketErrorInfo(),
WebSocketErrorInfo(),
WebSocketOpenInfo(status.uri, status.headers),
WebSocketCloseInfo());
return status;
@ -155,7 +155,7 @@ namespace ix
}
_onMessageCallback(WebSocket_MessageType_Open, "", 0,
WebSocketErrorInfo(),
WebSocketErrorInfo(),
WebSocketOpenInfo(status.uri, status.headers),
WebSocketCloseInfo());
return status;
@ -184,7 +184,7 @@ namespace ix
using millis = std::chrono::duration<double, std::milli>;
millis duration;
while (true)
while (true)
{
if (isConnected() || isClosing() || _stop || !_automaticReconnection)
{
@ -214,7 +214,7 @@ namespace ix
{
setThreadName(_url);
while (true)
while (true)
{
if (_stop) return;
@ -223,7 +223,7 @@ namespace ix
if (_stop) return;
// 2. Poll to see if there's any new data available
// 2. Poll to see if there's any new data available
_ws.poll();
if (_stop) return;
@ -273,7 +273,7 @@ namespace ix
void WebSocket::setOnMessageCallback(const OnMessageCallback& callback)
{
_onMessageCallback = callback;
_onMessageCallback = callback;
}
void WebSocket::setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback)
@ -294,9 +294,10 @@ namespace ix
}
}
WebSocketSendInfo WebSocket::send(const std::string& text)
WebSocketSendInfo WebSocket::send(const std::string& text,
const OnProgressCallback& onProgressCallback)
{
return sendMessage(text, false);
return sendMessage(text, false, onProgressCallback);
}
WebSocketSendInfo WebSocket::ping(const std::string& text)
@ -308,7 +309,9 @@ namespace ix
return sendMessage(text, true);
}
WebSocketSendInfo WebSocket::sendMessage(const std::string& text, bool ping)
WebSocketSendInfo WebSocket::sendMessage(const std::string& text,
bool ping,
const OnProgressCallback& onProgressCallback)
{
if (!isConnected()) return WebSocketSendInfo(false);
@ -330,7 +333,7 @@ namespace ix
}
else
{
webSocketSendInfo = _ws.sendBinary(text);
webSocketSendInfo = _ws.sendBinary(text, onProgressCallback);
}
WebSocket::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize, false);
@ -340,7 +343,7 @@ namespace ix
ReadyState WebSocket::getReadyState() const
{
switch (_ws.getReadyState())
switch (_ws.getReadyState())
{
case ix::WebSocketTransport::OPEN: return WebSocket_ReadyState_Open;
case ix::WebSocketTransport::CONNECTING: return WebSocket_ReadyState_Connecting;

12
ixwebsocket/IXWebSocket.h
View File

@ -19,11 +19,12 @@
#include "IXWebSocketSendInfo.h"
#include "IXWebSocketPerMessageDeflateOptions.h"
#include "IXWebSocketHttpHeaders.h"
#include "IXProgressCallback.h"
namespace ix
{
// https://developer.mozilla.org/en-US/docs/Web/API/WebSocket#Ready_state_constants
enum ReadyState
enum ReadyState
{
WebSocket_ReadyState_Connecting = 0,
WebSocket_ReadyState_Open = 1,
@ -78,7 +79,7 @@ namespace ix
using OnTrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
class WebSocket
class WebSocket
{
public:
WebSocket();
@ -97,7 +98,8 @@ namespace ix
WebSocketInitResult connect(int timeoutSecs);
void run();
WebSocketSendInfo send(const std::string& text);
WebSocketSendInfo send(const std::string& text,
const OnProgressCallback& onProgressCallback = nullptr);
WebSocketSendInfo ping(const std::string& text);
void close();
@ -115,7 +117,9 @@ namespace ix
private:
WebSocketSendInfo sendMessage(const std::string& text, bool ping);
WebSocketSendInfo sendMessage(const std::string& text,
bool ping,
const OnProgressCallback& callback = nullptr);
bool isConnected() const;
bool isClosing() const;

2
ixwebsocket/IXWebSocketErrorInfo.h
View File

@ -8,7 +8,7 @@
#include <string>
namespace ix
namespace ix
{
struct WebSocketErrorInfo
{

14
ixwebsocket/IXWebSocketHandshake.cpp
View File

@ -16,7 +16,7 @@
#include <algorithm>
namespace ix
namespace ix
{
WebSocketHandshake::WebSocketHandshake(std::atomic<bool>& requestInitCancellation,
std::shared_ptr<Socket> socket,
@ -171,7 +171,7 @@ namespace ix
std::string WebSocketHandshake::genRandomString(const int len)
{
std::string alphanum =
std::string alphanum =
"0123456789"
"ABCDEFGH"
"abcdefgh";
@ -201,7 +201,7 @@ namespace ix
char line[256];
int i;
while (true)
while (true)
{
int colon = 0;
@ -277,7 +277,7 @@ namespace ix
{
_requestInitCancellation = false;
auto isCancellationRequested =
auto isCancellationRequested =
makeCancellationRequestWithTimeout(timeoutSecs, _requestInitCancellation);
std::string errMsg;
@ -372,7 +372,7 @@ namespace ix
}
// Check the value of the connection field
// Some websocket servers (Go/Gorilla?) send lowercase values for the
// Some websocket servers (Go/Gorilla?) send lowercase values for the
// connection header, so do a case insensitive comparison
if (!insensitiveStringCompare(headers["connection"], "Upgrade"))
{
@ -418,7 +418,7 @@ namespace ix
// Set the socket to non blocking mode + other tweaks
SocketConnect::configure(fd);
auto isCancellationRequested =
auto isCancellationRequested =
makeCancellationRequestWithTimeout(timeoutSecs, _requestInitCancellation);
std::string remote = std::string("remote fd ") + std::to_string(fd);
@ -432,7 +432,7 @@ namespace ix
{
return sendErrorResponse(400, "Error reading HTTP request line");
}
// Validate request line (GET /foo HTTP/1.1\r\n)
auto requestLine = parseRequestLine(line);
auto method = std::get<0>(requestLine);

2
ixwebsocket/IXWebSocketHandshake.h
View File

@ -18,7 +18,7 @@
#include <memory>
#include <tuple>
namespace ix
namespace ix
{
struct WebSocketInitResult
{

2
ixwebsocket/IXWebSocketHttpHeaders.h
View File

@ -9,7 +9,7 @@
#include <string>
#include <unordered_map>
namespace ix
namespace ix
{
using WebSocketHttpHeaders = std::unordered_map<std::string, std::string>;
}

6
ixwebsocket/IXWebSocketPerMessageDeflate.cpp
View File

@ -34,7 +34,7 @@
* - Reused zlib compression + decompression bits.
* - Refactored to have 2 class for compression and decompression, to allow multi-threading
* and make sure that _compressBuffer is not shared between threads.
* - Original code wasn't working for some reason, I had to add checks
* - Original code wasn't working for some reason, I had to add checks
* for the presence of the kEmptyUncompressedBlock at the end of buffer so that servers
* would start accepting receiving/decoding compressed messages. Original code was probably
* modifying the passed in buffers before processing in enabled.hpp ?
@ -65,13 +65,13 @@ namespace ix
bool WebSocketPerMessageDeflate::init(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions)
{
bool clientNoContextTakeover =
bool clientNoContextTakeover =
perMessageDeflateOptions.getClientNoContextTakeover();
uint8_t deflateBits = perMessageDeflateOptions.getClientMaxWindowBits();
uint8_t inflateBits = perMessageDeflateOptions.getServerMaxWindowBits();
return _compressor->init(deflateBits, clientNoContextTakeover) &&
return _compressor->init(deflateBits, clientNoContextTakeover) &&
_decompressor->init(inflateBits, clientNoContextTakeover);
}

2
ixwebsocket/IXWebSocketPerMessageDeflate.h
View File

@ -37,7 +37,7 @@
#include <string>
#include <memory>
namespace ix
namespace ix
{
class WebSocketPerMessageDeflateOptions;
class WebSocketPerMessageDeflateCompressor;

18
ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
View File

@ -14,7 +14,7 @@
namespace
{
// The passed in size (4) is important, without it the string litteral
// is treated as a char* and the null termination (\x00) makes it
// is treated as a char* and the null termination (\x00) makes it
// look like an empty string.
const std::string kEmptyUncompressedBlock = std::string("\x00\x00\xff\xff", 4);
@ -76,16 +76,16 @@ namespace ix
{
//
// 7.2.1. Compression
//
//
// An endpoint uses the following algorithm to compress a message.
//
//
// 1. Compress all the octets of the payload of the message using
// DEFLATE.
//
//
// 2. If the resulting data does not end with an empty DEFLATE block
// with no compression (the "BTYPE" bits are set to 00), append an
// empty DEFLATE block with no compression to the tail end.
//
//
// 3. Remove 4 octets (that are 0x00 0x00 0xff 0xff) from the tail end.
// After this step, the last octet of the compressed data contains
// (possibly part of) the DEFLATE header bits with the "BTYPE" bits
@ -168,14 +168,14 @@ namespace ix
{
//
// 7.2.2. Decompression
//
//
// An endpoint uses the following algorithm to decompress a message.
//
//
// 1. Append 4 octets of 0x00 0x00 0xff 0xff to the tail end of the
// payload of the message.
//
//
// 2. Decompress the resulting data using DEFLATE.
//
//
std::string inFixed(in);
inFixed += kEmptyUncompressedBlock;

2
ixwebsocket/IXWebSocketPerMessageDeflateCodec.h
View File

@ -10,7 +10,7 @@
#include <string>
#include <memory>
namespace ix
namespace ix
{
class WebSocketPerMessageDeflateCompressor
{

12
ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
View File

@ -36,7 +36,7 @@ namespace ix
_serverMaxWindowBits = serverMaxWindowBits;
}
//
//
// Four extension parameters are defined for "permessage-deflate" to
// help endpoints manage per-connection resource usage.
//
@ -88,9 +88,9 @@ namespace ix
int x;
ss >> x;
// Sanitize values to be in the proper range [8, 15] in
// Sanitize values to be in the proper range [8, 15] in
// case a server would give us bogus values
_serverMaxWindowBits =
_serverMaxWindowBits =
std::min(maxServerMaxWindowBits,
std::max(x, minServerMaxWindowBits));
}
@ -103,9 +103,9 @@ namespace ix
int x;
ss >> x;
// Sanitize values to be in the proper range [8, 15] in
// Sanitize values to be in the proper range [8, 15] in
// case a server would give us bogus values
_clientMaxWindowBits =
_clientMaxWindowBits =
std::min(maxClientMaxWindowBits,
std::max(x, minClientMaxWindowBits));
}
@ -162,7 +162,7 @@ namespace ix
std::string WebSocketPerMessageDeflateOptions::removeSpaces(const std::string& str)
{
std::string out(str);
out.erase(std::remove_if(out.begin(),
out.erase(std::remove_if(out.begin(),
out.end(),
[](unsigned char x){ return std::isspace(x); }),
out.end());

2
ixwebsocket/IXWebSocketPerMessageDeflateOptions.h
View File

@ -8,7 +8,7 @@
#include <string>
namespace ix
namespace ix
{
class WebSocketPerMessageDeflateOptions
{

2
ixwebsocket/IXWebSocketSendInfo.h
View File

@ -9,7 +9,7 @@
#include <string>
#include <iostream>
namespace ix
namespace ix
{
struct WebSocketSendInfo
{

4
ixwebsocket/IXWebSocketServer.cpp
View File

@ -14,7 +14,7 @@
#include <future>
#include <string.h>
namespace ix
namespace ix
{
const int WebSocketServer::kDefaultHandShakeTimeoutSecs(3); // 3 seconds
@ -65,7 +65,7 @@ namespace ix
auto status = webSocket->connectToSocket(fd, _handshakeTimeoutSecs);
if (status.success)
{
// Process incoming messages and execute callbacks
// Process incoming messages and execute callbacks
// until the connection is closed
webSocket->run();
}

2
ixwebsocket/IXWebSocketServer.h
View File

@ -18,7 +18,7 @@
#include "IXWebSocket.h"
#include "IXSocketServer.h"
namespace ix
namespace ix
{
using OnConnectionCallback = std::function<void(std::shared_ptr<WebSocket>)>;

210
ixwebsocket/IXWebSocketTransport.cpp
View File

@ -29,12 +29,15 @@
#include <cstdarg>
#include <iostream>
#include <sstream>
#include <chrono>
#include <thread>
namespace ix
{
const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::hearbeat");
const int WebSocketTransport::kDefaultHeartBeatPeriod(-1);
constexpr size_t WebSocketTransport::kChunkSize;
WebSocketTransport::WebSocketTransport() :
_readyState(CLOSED),
@ -45,7 +48,7 @@ namespace ix
_heartBeatPeriod(kDefaultHeartBeatPeriod),
_lastSendTimePoint(std::chrono::steady_clock::now())
{
_readbuf.resize(kChunkSize);
}
WebSocketTransport::~WebSocketTransport()
@ -129,7 +132,7 @@ namespace ix
return result;
}
WebSocketTransport::ReadyStateValues WebSocketTransport::getReadyState() const
WebSocketTransport::ReadyStateValues WebSocketTransport::getReadyState() const
{
return _readyState;
}
@ -153,7 +156,7 @@ namespace ix
void WebSocketTransport::setOnCloseCallback(const OnCloseCallback& onCloseCallback)
{
_onCloseCallback = onCloseCallback;
_onCloseCallback = onCloseCallback;
}
// Only consider send time points for that computation.
@ -173,7 +176,7 @@ namespace ix
// If (1) heartbeat is enabled, and (2) no data was received or
// send for a duration exceeding our heart-beat period, send a
// ping to the server.
if (pollResult == PollResultType_Timeout &&
if (pollResult == PollResultType_Timeout &&
heartBeatPeriodExceeded())
{
std::stringstream ss;
@ -182,33 +185,31 @@ namespace ix
return;
}
while (true)
while (true)
{
int N = (int) _rxbuf.size();
ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
_rxbuf.resize(N + 1500);
ssize_t ret = _socket->recv((char*)&_rxbuf[0] + N, 1500);
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
_socket->getErrno() == EAGAIN)) {
_rxbuf.resize(N);
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
_socket->getErrno() == EAGAIN))
{
break;
}
else if (ret <= 0)
else if (ret <= 0)
{
_rxbuf.resize(N);
_rxbuf.clear();
_socket->close();
setReadyState(CLOSED);
break;
}
else
else
{
_rxbuf.resize(N + ret);
_rxbuf.insert(_rxbuf.end(),
_readbuf.begin(),
_readbuf.begin() + ret);
}
}
if (isSendBufferEmpty() && _readyState == CLOSING)
if (isSendBufferEmpty() && _readyState == CLOSING)
{
_socket->close();
setReadyState(CLOSED);
@ -282,7 +283,7 @@ namespace ix
//
void WebSocketTransport::dispatch(const OnMessageCallback& onMessageCallback)
{
while (true)
while (true)
{
wsheader_type ws;
if (_rxbuf.size() < 2) return; /* Need at least 2 */
@ -294,7 +295,7 @@ namespace ix
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() */
//
// Calculate payload length:
// 0-125 mean the payload is that long.
@ -332,7 +333,7 @@ namespace ix
// invalid payload length according to the spec. bail out
return;
}
if (ws.mask)
{
ws.masking_key[0] = ((uint8_t) data[i+0]) << 0;
@ -355,22 +356,40 @@ namespace ix
// We got a whole message, now do something with it:
if (
ws.opcode == wsheader_type::TEXT_FRAME
ws.opcode == wsheader_type::TEXT_FRAME
|| ws.opcode == wsheader_type::BINARY_FRAME
|| ws.opcode == wsheader_type::CONTINUATION
) {
unmaskReceiveBuffer(ws);
_receivedData.insert(_receivedData.end(),
_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size+(size_t)ws.N);// just feed
if (ws.fin)
{
// fire callback with a string message
std::string stringMessage(_receivedData.begin(),
_receivedData.end());
emitMessage(MSG, stringMessage, ws, onMessageCallback);
_receivedData.clear();
//
// Usual case. Small unfragmented messages
//
if (ws.fin && _chunks.empty())
{
emitMessage(MSG,
std::string(_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size+(size_t) ws.N),
ws,
onMessageCallback);
}
else
{
//
// Add intermediary message to our chunk list.
// We use a chunk list instead of a big buffer because resizing
// large buffer can be very costly when we need to re-allocate
// the internal buffer which is slow and can let the internal OS
// receive buffer fill out.
//
_chunks.emplace_back(
std::vector<uint8_t>(_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size+(size_t)ws.N));
if (ws.fin)
{
emitMessage(MSG, getMergedChunks(), ws, onMessageCallback);
_chunks.clear();
}
}
}
else if (ws.opcode == wsheader_type::PING)
@ -420,12 +439,33 @@ namespace ix
close();
}
// Erase the message that has been processed from the input/read buffer
_rxbuf.erase(_rxbuf.begin(),
_rxbuf.begin() + ws.header_size + (size_t) ws.N);
}
}
void WebSocketTransport::emitMessage(MessageKind messageKind,
std::string WebSocketTransport::getMergedChunks() const
{
size_t length = 0;
for (auto&& chunk : _chunks)
{
length += chunk.size();
}
std::string msg;
msg.reserve(length);
for (auto&& chunk : _chunks)
{
std::string str(chunk.begin(), chunk.end());
msg += str;
}
return msg;
}
void WebSocketTransport::emitMessage(MessageKind messageKind,
const std::string& message,
const wsheader_type& ws,
const OnMessageCallback& onMessageCallback)
@ -448,15 +488,17 @@ namespace ix
unsigned WebSocketTransport::getRandomUnsigned()
{
auto now = std::chrono::system_clock::now();
auto seconds =
auto seconds =
std::chrono::duration_cast<std::chrono::seconds>(
now.time_since_epoch()).count();
return static_cast<unsigned>(seconds);
}
WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
const std::string& message,
bool compress)
WebSocketSendInfo WebSocketTransport::sendData(
wsheader_type::opcode_type type,
const std::string& message,
bool compress,
const OnProgressCallback& onProgressCallback)
{
if (_readyState == CLOSING || _readyState == CLOSED)
{
@ -473,15 +515,81 @@ namespace ix
if (compress)
{
bool success = _perMessageDeflate.compress(message, compressedMessage);
compressionError = !success;
if (!_perMessageDeflate.compress(message, compressedMessage))
{
bool success = false;
compressionError = true;
payloadSize = 0;
wireSize = 0;
return WebSocketSendInfo(success, compressionError, payloadSize, wireSize);
}
compressionError = false;
wireSize = compressedMessage.size();
message_begin = compressedMessage.begin();
message_end = compressedMessage.end();
}
uint64_t message_size = wireSize;
// Common case for most message. No fragmentation required.
if (wireSize < kChunkSize)
{
sendFragment(type, true, message_begin, message_end, compress);
}
else
{
//
// Large messages need to be fragmented
//
// Rules:
// First message needs to specify a proper type (BINARY or TEXT)
// Intermediary and last messages need to be of type CONTINUATION
// Last message must set the fin byte.
//
auto steps = wireSize / kChunkSize;
std::string::const_iterator begin = message_begin;
std::string::const_iterator end = message_end;
for (uint64_t i = 0 ; i < steps; ++i)
{
bool firstStep = i == 0;
bool lastStep = (i+1) == steps;
bool fin = lastStep;
end = begin + kChunkSize;
if (lastStep)
{
end = message_end;
}
auto opcodeType = type;
if (!firstStep)
{
opcodeType = wsheader_type::CONTINUATION;
}
// Send message
sendFragment(opcodeType, fin, begin, end, compress);
if (onProgressCallback && !onProgressCallback(i, steps))
{
break;
}
begin += kChunkSize;
}
}
return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
}
void WebSocketTransport::sendFragment(wsheader_type::opcode_type type,
bool fin,
std::string::const_iterator message_begin,
std::string::const_iterator message_end,
bool compress)
{
auto message_size = message_end - message_begin;
unsigned x = getRandomUnsigned();
uint8_t masking_key[4] = {};
@ -494,7 +602,13 @@ namespace ix
header.assign(2 +
(message_size >= 126 ? 2 : 0) +
(message_size >= 65536 ? 6 : 0) + 4, 0);
header[0] = 0x80 | type;
header[0] = type;
// The fin bit indicate that this is the last fragment. Fin is French for end.
if (fin)
{
header[0] |= 0x80;
}
// This bit indicate that the frame is compressed
if (compress)
@ -511,7 +625,7 @@ namespace ix
header[4] = masking_key[2];
header[5] = masking_key[3];
}
else if (message_size < 65536)
else if (message_size < 65536)
{
header[1] = 126 | 0x80;
header[2] = (message_size >> 8) & 0xff;
@ -546,8 +660,6 @@ namespace ix
// Now actually send this data
sendOnSocket();
return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
}
WebSocketSendInfo WebSocketTransport::sendPing(const std::string& message)
@ -556,9 +668,13 @@ namespace ix
return sendData(wsheader_type::PING, message, compress);
}
WebSocketSendInfo WebSocketTransport::sendBinary(const std::string& message)
WebSocketSendInfo WebSocketTransport::sendBinary(
const std::string& message,
const OnProgressCallback& onProgressCallback)
{
return sendData(wsheader_type::BINARY_FRAME, message, _enablePerMessageDeflate);
return sendData(wsheader_type::BINARY_FRAME, message,
_enablePerMessageDeflate, onProgressCallback);
}
void WebSocketTransport::sendOnSocket()
@ -569,7 +685,7 @@ namespace ix
{
ssize_t ret = _socket->send((char*)&_txbuf[0], _txbuf.size());
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
_socket->getErrno() == EAGAIN))
{
break;

46
ixwebsocket/IXWebSocketTransport.h
View File

@ -16,6 +16,7 @@
#include <memory>
#include <mutex>
#include <atomic>
#include <list>
#include "IXWebSocketSendInfo.h"
#include "IXWebSocketPerMessageDeflate.h"
@ -23,8 +24,9 @@
#include "IXWebSocketHttpHeaders.h"
#include "IXCancellationRequest.h"
#include "IXWebSocketHandshake.h"
#include "IXProgressCallback.h"
namespace ix
namespace ix
{
class Socket;
@ -66,7 +68,8 @@ namespace ix
int timeoutSecs);
void poll();
WebSocketSendInfo sendBinary(const std::string& message);
WebSocketSendInfo sendBinary(const std::string& message,
const OnProgressCallback& onProgressCallback);
WebSocketSendInfo sendPing(const std::string& message);
void close();
ReadyStateValues getReadyState() const;
@ -76,7 +79,6 @@ namespace ix
private:
std::string _url;
std::string _origin;
struct wsheader_type {
unsigned header_size;
@ -96,13 +98,31 @@ namespace ix
uint8_t masking_key[4];
};
// Buffer for reading from our socket. That buffer is never resized.
std::vector<uint8_t> _readbuf;
// Contains all messages that were fetched in the last socket read.
// This could be a mix of control messages (Close, Ping, etc...) and
// data messages. That buffer
std::vector<uint8_t> _rxbuf;
// Contains all messages that are waiting to be sent
std::vector<uint8_t> _txbuf;
mutable std::mutex _txbufMutex;
std::vector<uint8_t> _receivedData;
// Hold fragments for multi-fragments messages in a list. We support receiving very large
// messages (tested messages up to 700M) and we cannot put them in a single
// buffer that is resized, as this operation can be slow when a buffer has its
// size increased 2 fold, while appending to a list has a fixed cost.
std::list<std::vector<uint8_t>> _chunks;
// Fragments are 32K long
static constexpr size_t kChunkSize = 1 << 15;
// Underlying TCP socket
std::shared_ptr<Socket> _socket;
// Hold the state of the connection (OPEN, CLOSED, etc...)
std::atomic<ReadyStateValues> _readyState;
OnCloseCallback _onCloseCallback;
@ -111,13 +131,14 @@ namespace ix
size_t _closeWireSize;
mutable std::mutex _closeDataMutex;
// Data used for Per Message Deflate compression (with zlib)
WebSocketPerMessageDeflate _perMessageDeflate;
WebSocketPerMessageDeflateOptions _perMessageDeflateOptions;
std::atomic<bool> _enablePerMessageDeflate;
// Used to cancel dns lookup + socket connect + http upgrade
std::atomic<bool> _requestInitCancellation;
// Optional Heartbeat
int _heartBeatPeriod;
static const int kDefaultHeartBeatPeriod;
@ -129,11 +150,18 @@ namespace ix
bool heartBeatPeriodExceeded();
void sendOnSocket();
WebSocketSendInfo sendData(wsheader_type::opcode_type type,
WebSocketSendInfo sendData(wsheader_type::opcode_type type,
const std::string& message,
bool compress);
bool compress,
const OnProgressCallback& onProgressCallback = nullptr);
void emitMessage(MessageKind messageKind,
void sendFragment(wsheader_type::opcode_type type,
bool fin,
std::string::const_iterator begin,
std::string::const_iterator end,
bool compress);
void emitMessage(MessageKind messageKind,
const std::string& message,
const wsheader_type& ws,
const OnMessageCallback& onMessageCallback);
@ -148,5 +176,7 @@ namespace ix
unsigned getRandomUnsigned();
void unmaskReceiveBuffer(const wsheader_type& ws);
std::string getMergedChunks() const;
};
}