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

per message deflate support (with zlib)

Browse Source
This commit is contained in:
Benjamin Sergeant
2018-11-09 18:23:49 -08:00
parent de8bcd36e8
commit 4fed156b90
32 changed files with 1003 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
ixwebsocket/IXWebSocket.cpp
View File

@ -41,12 +41,29 @@ namespace ix {
stop();
}
void WebSocket::configure(const std::string& url)
void WebSocket::setUrl(const std::string& url)
{
std::lock_guard<std::mutex> lock(_urlMutex);
std::lock_guard<std::mutex> lock(_configMutex);
_url = url;
}
const std::string& WebSocket::getUrl() const
{
std::lock_guard<std::mutex> lock(_configMutex);
return _url;
}
void WebSocket::setPerMessageDeflateOptions(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions)
{
_perMessageDeflateOptions = perMessageDeflateOptions;
}
const WebSocketPerMessageDeflateOptions& WebSocket::getPerMessageDeflateOptions() const
{
std::lock_guard<std::mutex> lock(_configMutex);
return _perMessageDeflateOptions;
}
void WebSocket::start()
{
if (_thread.joinable()) return; // we've already been started
@ -76,15 +93,17 @@ namespace ix {
WebSocketInitResult WebSocket::connect()
{
{
std::lock_guard<std::mutex> lock(_urlMutex);
_ws.configure(_url);
std::lock_guard<std::mutex> lock(_configMutex);
_ws.configure(_url, _perMessageDeflateOptions);
}
_ws.setOnCloseCallback(
[this](uint16_t code, const std::string& reason)
[this](uint16_t code, const std::string& reason, size_t wireSize)
{
_onMessageCallback(WebSocket_MessageType_Close, "",
WebSocketErrorInfo(), CloseInfo(code, reason));
_onMessageCallback(WebSocket_MessageType_Close, "", wireSize,
WebSocketErrorInfo(),
WebSocketCloseInfo(code, reason),
WebSocketHttpHeaders());
}
);
@ -94,8 +113,9 @@ namespace ix {
return status;
}
_onMessageCallback(WebSocket_MessageType_Open, "",
WebSocketErrorInfo(), CloseInfo());
_onMessageCallback(WebSocket_MessageType_Open, "", 0,
WebSocketErrorInfo(), WebSocketCloseInfo(),
status.headers);
return status;
}
@ -139,8 +159,9 @@ namespace ix {
connectErr.wait_time = duration.count();
connectErr.reason = status.errorStr;
connectErr.http_status = status.http_status;
_onMessageCallback(WebSocket_MessageType_Error, "",
connectErr, CloseInfo());
_onMessageCallback(WebSocket_MessageType_Error, "", 0,
connectErr, WebSocketCloseInfo(),
WebSocketHttpHeaders());
std::this_thread::sleep_for(duration);
}
@ -166,6 +187,7 @@ namespace ix {
// 3. Dispatch the incoming messages
_ws.dispatch(
[this](const std::string& msg,
size_t wireSize,
WebSocketTransport::MessageKind messageKind)
{
WebSocketMessageType webSocketMessageType;
@ -187,8 +209,9 @@ namespace ix {
} break;
}
_onMessageCallback(webSocketMessageType, msg,
WebSocketErrorInfo(), CloseInfo());
_onMessageCallback(webSocketMessageType, msg, wireSize,
WebSocketErrorInfo(), WebSocketCloseInfo(),
WebSocketHttpHeaders());
WebSocket::invokeTrafficTrackerCallback(msg.size(), true);
});
@ -218,23 +241,23 @@ namespace ix {
}
}
bool WebSocket::send(const std::string& text)
WebSocketSendInfo WebSocket::send(const std::string& text)
{
return sendMessage(text, false);
}
bool WebSocket::ping(const std::string& text)
WebSocketSendInfo WebSocket::ping(const std::string& text)
{
// Standard limit ping message size
constexpr size_t pingMaxPayloadSize = 125;
if (text.size() > pingMaxPayloadSize) return false;
if (text.size() > pingMaxPayloadSize) return WebSocketSendInfo(false);
return sendMessage(text, true);
}
bool WebSocket::sendMessage(const std::string& text, bool ping)
WebSocketSendInfo WebSocket::sendMessage(const std::string& text, bool ping)
{
if (!isConnected()) return false;
if (!isConnected()) return WebSocketSendInfo(false);
//
// It is OK to read and write on the same socket in 2 different threads.
@ -246,19 +269,20 @@ namespace ix {
// incoming messages are arriving / there's data to be received.
//
std::lock_guard<std::mutex> lock(_writeMutex);
WebSocketSendInfo webSocketSendInfo;
if (ping)
{
_ws.sendPing(text);
webSocketSendInfo = _ws.sendPing(text);
}
else
{
_ws.sendBinary(text);
webSocketSendInfo = _ws.sendBinary(text);
}
WebSocket::invokeTrafficTrackerCallback(text.size(), false);
WebSocket::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize, false);
return true;
return webSocketSendInfo;
}
ReadyState WebSocket::getReadyState() const
@ -282,10 +306,4 @@ namespace ix {
case WebSocket_ReadyState_Closed: return "CLOSED";
}
}
const std::string& WebSocket::getUrl() const
{
std::lock_guard<std::mutex> lock(_urlMutex);
return _url;
}
}

33
ixwebsocket/IXWebSocket.h
View File

@ -15,8 +15,11 @@
#include <atomic>
#include "IXWebSocketTransport.h"
#include "IXWebSocketSendInfo.h"
#include "IXWebSocketPerMessageDeflateOptions.h"
#include "IXWebSocketHttpHeaders.h"
namespace ix
namespace ix
{
// https://developer.mozilla.org/en-US/docs/Web/API/WebSocket#Ready_state_constants
enum ReadyState
@ -45,18 +48,18 @@ namespace ix
std::string reason;
};
struct CloseInfo
struct WebSocketCloseInfo
{
uint16_t code;
std::string reason;
CloseInfo(uint64_t c, const std::string& r)
WebSocketCloseInfo(uint64_t c, const std::string& r)
{
code = c;
reason = r;
}
CloseInfo()
WebSocketCloseInfo()
{
code = 0;
reason = "";
@ -65,8 +68,10 @@ namespace ix
using OnMessageCallback = std::function<void(WebSocketMessageType,
const std::string&,
const WebSocketErrorInfo,
const CloseInfo)>;
size_t wireSize,
const WebSocketErrorInfo&,
const WebSocketCloseInfo&,
const WebSocketHttpHeaders&)>;
using OnTrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
class WebSocket
@ -75,24 +80,27 @@ namespace ix
WebSocket();
~WebSocket();
void configure(const std::string& url);
void setUrl(const std::string& url);
void setPerMessageDeflateOptions(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
void start();
void stop();
bool send(const std::string& text);
bool ping(const std::string& text);
WebSocketSendInfo send(const std::string& text);
WebSocketSendInfo ping(const std::string& text);
void close();
void setOnMessageCallback(const OnMessageCallback& callback);
static void setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback);
static void resetTrafficTrackerCallback();
const std::string& getUrl() const;
ReadyState getReadyState() const;
const std::string& getUrl() const;
const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
private:
void run();
bool sendMessage(const std::string& text, bool ping);
WebSocketSendInfo sendMessage(const std::string& text, bool ping);
WebSocketInitResult connect();
bool isConnected() const;
@ -104,7 +112,8 @@ namespace ix
WebSocketTransport _ws;
std::string _url;
mutable std::mutex _urlMutex;
WebSocketPerMessageDeflateOptions _perMessageDeflateOptions;
mutable std::mutex _configMutex; // protect all config variables access
OnMessageCallback _onMessageCallback;
static OnTrafficTrackerCallback _onTrafficTrackerCallback;

15
ixwebsocket/IXWebSocketHttpHeaders.h Normal file
View File

@ -0,0 +1,15 @@
/*
* IXWebSocketHttpHeaders.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <string>
#include <unordered_map>
namespace ix
{
using WebSocketHttpHeaders = std::unordered_map<std::string, std::string>;
}

222
ixwebsocket/IXWebSocketPerMessageDeflate.cpp Normal file
View File

@ -0,0 +1,222 @@
/*
* IXWebSocketPerMessageDeflate.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*
* Per message Deflate RFC: https://tools.ietf.org/html/rfc7692
*
* Chrome websocket -> https://github.com/chromium/chromium/tree/2ca8c5037021c9d2ecc00b787d58a31ed8fc8bcb/net/websockets
*/
#include "IXWebSocketPerMessageDeflate.h"
#include "IXWebSocketPerMessageDeflateOptions.h"
#include <iostream>
#include <cassert>
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
// look like an empty string.
const std::string kEmptyUncompressedBlock = std::string("\x00\x00\xff\xff", 4);
const int kBufferSize = 1 << 14;
}
namespace ix
{
//
// Compressor
//
WebSocketPerMessageDeflateCompressor::WebSocketPerMessageDeflateCompressor()
: _compressBufferSize(kBufferSize)
{
_deflateState.zalloc = Z_NULL;
_deflateState.zfree = Z_NULL;
_deflateState.opaque = Z_NULL;
}
WebSocketPerMessageDeflateCompressor::~WebSocketPerMessageDeflateCompressor()
{
deflateEnd(&_deflateState);
}
bool WebSocketPerMessageDeflateCompressor::init(uint8_t deflateBits,
bool client_no_context_takeover)
{
int ret = deflateInit2(
&_deflateState,
Z_DEFAULT_COMPRESSION,
Z_DEFLATED,
-1*deflateBits,
4, // memory level 1-9
Z_DEFAULT_STRATEGY
);
if (ret != Z_OK) return false;
_compressBuffer.reset(new unsigned char[_compressBufferSize]);
if (client_no_context_takeover)
{
_flush = Z_FULL_FLUSH;
}
else
{
_flush = Z_SYNC_FLUSH;
}
return true;
}
bool WebSocketPerMessageDeflateCompressor::endsWith(const std::string& value,
const std::string& ending)
{
if (ending.size() > value.size()) return false;
return std::equal(ending.rbegin(), ending.rend(), value.rbegin());
}
bool WebSocketPerMessageDeflateCompressor::compress(const std::string& in,
std::string& out)
{
size_t output;
if (in.empty())
{
uint8_t buf[6] = {0x02, 0x00, 0x00, 0x00, 0xff, 0xff};
out.append((char *)(buf), 6);
return true;
}
_deflateState.avail_in = (uInt) in.size();
_deflateState.next_in = (Bytef*) in.data();
do
{
// Output to local buffer
_deflateState.avail_out = (uInt) _compressBufferSize;
_deflateState.next_out = _compressBuffer.get();
deflate(&_deflateState, _flush);
output = _compressBufferSize - _deflateState.avail_out;
out.append((char *)(_compressBuffer.get()),output);
} while (_deflateState.avail_out == 0);
if (endsWith(out, kEmptyUncompressedBlock))
{
out.resize(out.size() - 4);
}
return true;
}
//
// Decompressor
//
WebSocketPerMessageDeflateDecompressor::WebSocketPerMessageDeflateDecompressor()
: _compressBufferSize(kBufferSize)
{
_inflateState.zalloc = Z_NULL;
_inflateState.zfree = Z_NULL;
_inflateState.opaque = Z_NULL;
_inflateState.avail_in = 0;
_inflateState.next_in = Z_NULL;
}
WebSocketPerMessageDeflateDecompressor::~WebSocketPerMessageDeflateDecompressor()
{
inflateEnd(&_inflateState);
}
bool WebSocketPerMessageDeflateDecompressor::init(uint8_t inflateBits,
bool client_no_context_takeover)
{
int ret = inflateInit2(
&_inflateState,
-1*inflateBits
);
if (ret != Z_OK) return false;
_compressBuffer.reset(new unsigned char[_compressBufferSize]);
if (client_no_context_takeover)
{
_flush = Z_FULL_FLUSH;
}
else
{
_flush = Z_SYNC_FLUSH;
}
return true;
}
bool WebSocketPerMessageDeflateDecompressor::decompress(const std::string& in,
std::string& out)
{
std::string inFixed(in);
inFixed += kEmptyUncompressedBlock;
_inflateState.avail_in = (uInt) inFixed.size();
_inflateState.next_in = (unsigned char *)(const_cast<char *>(inFixed.data()));
do
{
_inflateState.avail_out = (uInt) _compressBufferSize;
_inflateState.next_out = _compressBuffer.get();
int ret = inflate(&_inflateState, Z_SYNC_FLUSH);
if (ret == Z_NEED_DICT || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR)
{
return false; // zlib error
}
out.append(
reinterpret_cast<char *>(_compressBuffer.get()),
_compressBufferSize - _inflateState.avail_out
);
} while (_inflateState.avail_out == 0);
return true;
}
WebSocketPerMessageDeflate::WebSocketPerMessageDeflate()
{
_compressor.reset(new WebSocketPerMessageDeflateCompressor());
_decompressor.reset(new WebSocketPerMessageDeflateDecompressor());
}
WebSocketPerMessageDeflate::~WebSocketPerMessageDeflate()
{
_compressor.reset();
_decompressor.reset();
}
bool WebSocketPerMessageDeflate::init(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions)
{
bool clientNoContextTakeover =
perMessageDeflateOptions.getClientNoContextTakeover();
uint8_t deflateBits = perMessageDeflateOptions.getClientMaxWindowBits();
uint8_t inflateBits = perMessageDeflateOptions.getServerMaxWindowBits();
return _compressor->init(deflateBits, clientNoContextTakeover) &&
_decompressor->init(inflateBits, clientNoContextTakeover);
}
bool WebSocketPerMessageDeflate::compress(const std::string& in,
std::string& out)
{
return _compressor->compress(in, out);
}
bool WebSocketPerMessageDeflate::decompress(const std::string& in,
std::string &out)
{
return _decompressor->decompress(in, out);
}
}

68
ixwebsocket/IXWebSocketPerMessageDeflate.h Normal file
View File

@ -0,0 +1,68 @@
/*
* IXWebSocketPerMessageDeflate.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include "zlib.h"
#include <string>
namespace ix
{
class WebSocketPerMessageDeflateOptions;
class WebSocketPerMessageDeflateCompressor
{
public:
WebSocketPerMessageDeflateCompressor();
~WebSocketPerMessageDeflateCompressor();
bool init(uint8_t deflate_bits, bool client_no_context_takeover);
bool compress(const std::string& in, std::string& out);
private:
static bool endsWith(const std::string& value, const std::string& ending);
int _flush;
size_t _compressBufferSize;
std::unique_ptr<unsigned char[]> _compressBuffer;
z_stream _deflateState;
};
class WebSocketPerMessageDeflateDecompressor
{
public:
WebSocketPerMessageDeflateDecompressor();
~WebSocketPerMessageDeflateDecompressor();
bool init(uint8_t inflate_bits, bool client_no_context_takeover);
bool decompress(const std::string& in, std::string& out);
private:
int _flush;
size_t _compressBufferSize;
std::unique_ptr<unsigned char[]> _compressBuffer;
z_stream _inflateState;
};
class WebSocketPerMessageDeflate
{
public:
WebSocketPerMessageDeflate();
virtual ~WebSocketPerMessageDeflate();
bool init(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
bool compress(const std::string& in, std::string& out);
bool decompress(const std::string& in, std::string& out);
private:
// mode::value m_server_max_window_bits_mode;
// mode::value m_client_max_window_bits_mode;
std::shared_ptr<WebSocketPerMessageDeflateCompressor> _compressor;
std::shared_ptr<WebSocketPerMessageDeflateDecompressor> _decompressor;
};
}

171
ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp Normal file
View File

@ -0,0 +1,171 @@
/*
* IXWebSocketPerMessageDeflateOptions.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#include "IXWebSocketPerMessageDeflateOptions.h"
#include <sstream>
#include <iostream>
#include <algorithm>
namespace ix
{
/// Default values as defined in the RFC
const uint8_t WebSocketPerMessageDeflateOptions::kDefaultServerMaxWindowBits = 15;
static const int minServerMaxWindowBits = 8;
static const int maxServerMaxWindowBits = 15;
const uint8_t WebSocketPerMessageDeflateOptions::kDefaultClientMaxWindowBits = 15;
static const int minClientMaxWindowBits = 8;
static const int maxClientMaxWindowBits = 15;
WebSocketPerMessageDeflateOptions::WebSocketPerMessageDeflateOptions(
bool enabled,
bool clientNoContextTakeover,
bool serverNoContextTakeover,
uint8_t clientMaxWindowBits,
uint8_t serverMaxWindowBits)
{
_enabled = enabled;
_clientNoContextTakeover = clientNoContextTakeover;
_serverNoContextTakeover = serverNoContextTakeover;
_clientMaxWindowBits = clientMaxWindowBits;
_serverMaxWindowBits = serverMaxWindowBits;
}
//
// Four extension parameters are defined for "permessage-deflate" to
// help endpoints manage per-connection resource usage.
//
// - "server_no_context_takeover"
// - "client_no_context_takeover"
// - "server_max_window_bits"
// - "client_max_window_bits"
//
// Server response could look like that:
//
// Sec-WebSocket-Extensions: permessage-deflate; client_no_context_takeover; server_no_context_takeover
//
WebSocketPerMessageDeflateOptions::WebSocketPerMessageDeflateOptions(std::string extension)
{
extension = removeSpaces(extension);
_enabled = false;
_clientNoContextTakeover = false;
_serverNoContextTakeover = false;
_clientMaxWindowBits = kDefaultClientMaxWindowBits;
_serverMaxWindowBits = kDefaultServerMaxWindowBits;
// Split by ;
std::string token;
std::stringstream tokenStream(extension);
while (std::getline(tokenStream, token, ';'))
{
if (token == "permessage-deflate")
{
_enabled = true;
}
if (token == "server_no_context_takeover")
{
_serverNoContextTakeover = true;
}
if (token == "client_no_context_takeover")
{
_clientNoContextTakeover = true;
}
if (startsWith(token, "server_max_window_bits="))
{
std::string val = token.substr(token.find_last_of("=") + 1);
std::stringstream ss;
ss << val;
int x;
ss >> x;
// Sanitize values to be in the proper range [8, 15] in
// case a server would give us bogus values
_serverMaxWindowBits =
std::min(maxServerMaxWindowBits,
std::max(x, minServerMaxWindowBits));
}
if (startsWith(token, "client_max_window_bits="))
{
std::string val = token.substr(token.find_last_of("=") + 1);
std::stringstream ss;
ss << val;
int x;
ss >> x;
// Sanitize values to be in the proper range [8, 15] in
// case a server would give us bogus values
_clientMaxWindowBits =
std::min(maxClientMaxWindowBits,
std::max(x, minClientMaxWindowBits));
}
}
}
std::string WebSocketPerMessageDeflateOptions::generateHeader()
{
std::stringstream ss;
ss << "Sec-WebSocket-Extensions: permessage-deflate";
if (_clientNoContextTakeover) ss << "; client_no_context_takeover";
if (_serverNoContextTakeover) ss << "; server_no_context_takeover";
ss << "; server_max_window_bits=" << _serverMaxWindowBits;
ss << "; client_max_window_bits=" << _clientMaxWindowBits;
ss << "\r\n";
return ss.str();
}
bool WebSocketPerMessageDeflateOptions::enabled() const
{
return _enabled;
}
bool WebSocketPerMessageDeflateOptions::getClientNoContextTakeover() const
{
return _clientNoContextTakeover;
}
bool WebSocketPerMessageDeflateOptions::getServerNoContextTakeover() const
{
return _serverNoContextTakeover;
}
uint8_t WebSocketPerMessageDeflateOptions::getClientMaxWindowBits() const
{
return _clientMaxWindowBits;
}
uint8_t WebSocketPerMessageDeflateOptions::getServerMaxWindowBits() const
{
return _serverMaxWindowBits;
}
bool WebSocketPerMessageDeflateOptions::startsWith(const std::string& str,
const std::string& start)
{
return str.compare(0, start.length(), start) == 0;
}
std::string WebSocketPerMessageDeflateOptions::removeSpaces(const std::string& str)
{
std::string out(str);
out.erase(std::remove_if(out.begin(),
out.end(),
[](unsigned char x){ return std::isspace(x); }),
out.end());
return out;
}
}

46
ixwebsocket/IXWebSocketPerMessageDeflateOptions.h Normal file
View File

@ -0,0 +1,46 @@
/*
* IXWebSocketPerMessageDeflateOptions.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
class WebSocketPerMessageDeflateOptions
{
public:
WebSocketPerMessageDeflateOptions(
bool enabled = true,
bool clientNoContextTakeover = false,
bool serverNoContextTakeover = false,
uint8_t clientMaxWindowBits = kDefaultClientMaxWindowBits,
uint8_t serverMaxWindowBits = kDefaultServerMaxWindowBits);
WebSocketPerMessageDeflateOptions(std::string extension);
std::string generateHeader();
std::string parseHeader();
bool enabled() const;
bool getClientNoContextTakeover() const;
bool getServerNoContextTakeover() const;
uint8_t getServerMaxWindowBits() const;
uint8_t getClientMaxWindowBits() const;
static bool startsWith(const std::string& str, const std::string& start);
static std::string removeSpaces(const std::string& str);
private:
bool _enabled;
bool _clientNoContextTakeover;
bool _serverNoContextTakeover;
int _clientMaxWindowBits;
int _serverMaxWindowBits;
static uint8_t const kDefaultClientMaxWindowBits;
static uint8_t const kDefaultServerMaxWindowBits;
};
}

26
ixwebsocket/IXWebSocketSendInfo.h Normal file
View File

@ -0,0 +1,26 @@
/*
* IXWebSocketSendInfo.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
struct WebSocketSendInfo
{
bool success;
size_t payloadSize;
size_t wireSize;
WebSocketSendInfo(bool s = false, size_t p = -1, size_t w = -1)
{
success = s;
payloadSize = p;
wireSize = w;
}
};
}

205
ixwebsocket/IXWebSocketTransport.cpp
View File

@ -9,6 +9,7 @@
//
#include "IXWebSocketTransport.h"
#include "IXWebSocketHttpHeaders.h"
#include "IXSocket.h"
#ifdef IXWEBSOCKET_USE_TLS
@ -31,18 +32,17 @@
#include <iostream>
#include <sstream>
#include <regex>
#include <unordered_map>
#include <random>
#include <algorithm>
namespace ix {
namespace ix
{
WebSocketTransport::WebSocketTransport() :
_readyState(CLOSED),
_enablePerMessageDeflate(false)
{
_perMessageDeflate.init();
}
WebSocketTransport::~WebSocketTransport()
@ -50,9 +50,12 @@ namespace ix {
;
}
void WebSocketTransport::configure(const std::string& url)
void WebSocketTransport::configure(const std::string& url,
const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions)
{
_url = url;
_perMessageDeflateOptions = perMessageDeflateOptions;
_enablePerMessageDeflate = _perMessageDeflateOptions.enabled();
}
bool WebSocketTransport::parseUrl(const std::string& url,
@ -135,21 +138,22 @@ namespace ix {
std::string WebSocketTransport::genRandomString(const int len)
{
static const char alphanum[] =
std::string alphanum =
"0123456789"
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz";
"ABCDEFGH"
"abcdefgh";
std::random_device r;
std::default_random_engine e1(r());
std::uniform_int_distribution<int> dist(0, sizeof(alphanum) - 1);
std::uniform_int_distribution<int> dist(0, (int) alphanum.size() - 1);
std::string s;
s.resize(len);
for (int i = 0; i < len; ++i)
{
s[i] += alphanum[dist(e1)];
int x = dist(e1);
s[i] = alphanum[x];
}
return s;
@ -206,35 +210,31 @@ namespace ix {
std::string secWebSocketKey = genRandomString(22);
secWebSocketKey += "==";
std::string extensions;
std::stringstream ss;
ss << "GET " << path << " HTTP/1.1\r\n";
ss << "Host: "<< host << ":" << port << "\r\n";
ss << "Upgrade: websocket\r\n";
ss << "Connection: Upgrade\r\n";
ss << "Sec-WebSocket-Version: 13\r\n";
ss << "Sec-WebSocket-Key: " << secWebSocketKey << "\r\n";
if (_enablePerMessageDeflate)
{
// extensions = "Sec-WebSocket-Extensions: permessage-deflate; client_no_context_takeover; server_no_context_takeover\r\n";
extensions = "Sec-WebSocket-Extensions: permessage-deflate\r\n";
ss << _perMessageDeflateOptions.generateHeader();
}
char line[512];
int status;
int i;
snprintf(line, 512,
"GET %s HTTP/1.1\r\n"
"Host: %s:%d\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Key: %s\r\n"
"Sec-WebSocket-Version: 13\r\n"
"%s"
"\r\n",
path.c_str(), host.c_str(), port,
secWebSocketKey.c_str(), extensions.c_str());
ss << "\r\n";
size_t lineSize = strlen(line);
if (_socket->send(line, lineSize) != lineSize)
std::string request = ss.str();
int requestSize = (int) request.size();
if (_socket->send(const_cast<char*>(request.c_str()), requestSize) != requestSize)
{
return WebSocketInitResult(false, 0, std::string("Failed sending GET request to ") + _url);
}
char line[512];
int i;
for (i = 0; i < 2 || (i < 255 && line[i-2] != '\r' && line[i-1] != '\n'); ++i)
{
if (_socket->recv(line+i, 1) == 0)
@ -248,6 +248,9 @@ namespace ix {
return WebSocketInitResult(false, 0, std::string("Got bad status line connecting to ") + _url);
}
// Validate status
int status;
// HTTP/1.0 is too old.
if (sscanf(line, "HTTP/1.0 %d", &status) == 1)
{
@ -268,7 +271,7 @@ namespace ix {
return WebSocketInitResult(false, status, ss.str());
}
std::unordered_map<std::string, std::string> headers;
WebSocketHttpHeaders headers;
while (true)
{
@ -310,7 +313,6 @@ namespace ix {
std::transform(name.begin(), name.end(), name.begin(), ::tolower);
headers[name] = value;
std::cout << name << " -> " << value << std::endl;
}
}
@ -322,10 +324,29 @@ namespace ix {
return WebSocketInitResult(false, status, errorMsg);
}
if (_enablePerMessageDeflate)
{
// Parse the server response. Does it support deflate ?
std::string header = headers["sec-websocket-extensions"];
WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions(header);
// If the server does not support that extension, disable it.
if (!webSocketPerMessageDeflateOptions.enabled())
{
_enablePerMessageDeflate = false;
}
if (!_perMessageDeflate.init(webSocketPerMessageDeflateOptions))
{
return WebSocketInitResult(
false, 0,"Failed to initialize per message deflate engine");
}
}
_socket->configure();
setReadyState(OPEN);
return WebSocketInitResult(true, status, "");
return WebSocketInitResult(true, status, "", headers);
}
WebSocketTransport::ReadyStateValues WebSocketTransport::getReadyState() const
@ -341,7 +362,7 @@ namespace ix {
if (readyStateValue == CLOSED)
{
std::lock_guard<std::mutex> lock(_closeDataMutex);
_onCloseCallback(_closeCode, _closeReason);
_onCloseCallback(_closeCode, _closeReason, _closeWireSize);
_closeCode = 0;
_closeReason = std::string();
}
@ -546,33 +567,7 @@ namespace ix {
std::string stringMessage(_receivedData.begin(),
_receivedData.end());
std::cout << "raw msg: " << stringMessage << std::endl;
std::cout << "raw msg size: " << stringMessage.size() << std::endl;
// ws.rsv1 means the message is compressed
// FIXME hack hack
std::string decompressedMessage;
if (_enablePerMessageDeflate && ws.rsv1)
{
if (_perMessageDeflate.decompress(stringMessage,
decompressedMessage))
{
std::cout << "decompressed msg: " << decompressedMessage << std::endl;
std::cout << "msg size: " << decompressedMessage.size() << std::endl;
onMessageCallback(decompressedMessage, MSG);
}
else
{
std::cout << "error decompressing msg !"<< std::endl;
}
}
else
{
onMessageCallback(stringMessage, MSG);
}
emitMessage(MSG, stringMessage, ws, onMessageCallback);
_receivedData.clear();
}
}
@ -583,10 +578,9 @@ namespace ix {
_rxbuf.begin()+ws.header_size + (size_t) ws.N);
// Reply back right away
sendData(wsheader_type::PONG, pingData.size(),
pingData.begin(), pingData.end());
sendData(wsheader_type::PONG, pingData);
onMessageCallback(pingData, PING);
emitMessage(PING, pingData, ws, onMessageCallback);
}
else if (ws.opcode == wsheader_type::PONG)
{
@ -594,7 +588,7 @@ namespace ix {
std::string pongData(_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size + (size_t) ws.N);
onMessageCallback(pongData, PONG);
emitMessage(PONG, pongData, ws, onMessageCallback);
}
else if (ws.opcode == wsheader_type::CLOSE)
{
@ -613,6 +607,7 @@ namespace ix {
std::lock_guard<std::mutex> lock(_closeDataMutex);
_closeCode = code;
_closeReason = reason;
_closeWireSize = _rxbuf.size();
}
close();
@ -627,6 +622,32 @@ namespace ix {
}
}
void WebSocketTransport::emitMessage(MessageKind messageKind,
const std::string& message,
const wsheader_type& ws,
const OnMessageCallback& onMessageCallback)
{
// ws.rsv1 means the message is compressed
std::string decompressedMessage;
if (_enablePerMessageDeflate && ws.rsv1)
{
if (_perMessageDeflate.decompress(message, decompressedMessage))
{
onMessageCallback(decompressedMessage, decompressedMessage.size(),
messageKind);
}
else
{
std::cerr << "error decompressing msg !"<< std::endl;
}
}
else
{
onMessageCallback(message, message.size(), messageKind);
}
}
unsigned WebSocketTransport::getRandomUnsigned()
{
auto now = std::chrono::system_clock::now();
@ -636,16 +657,32 @@ namespace ix {
return static_cast<unsigned>(seconds);
}
void WebSocketTransport::sendData(wsheader_type::opcode_type type,
uint64_t message_size,
std::string::const_iterator message_begin,
std::string::const_iterator message_end)
WebSocketSendInfo WebSocketTransport::sendData(wsheader_type::opcode_type type,
const std::string& message)
{
if (_readyState == CLOSING || _readyState == CLOSED)
{
return;
return WebSocketSendInfo();
}
size_t payloadSize = message.size();
size_t wireSize = message.size();
std::string compressedMessage;
std::string::const_iterator message_begin = message.begin();
std::string::const_iterator message_end = message.end();
if (_enablePerMessageDeflate)
{
_perMessageDeflate.compress(message, compressedMessage);
wireSize = compressedMessage.size();
message_begin = compressedMessage.begin();
message_end = compressedMessage.end();
}
uint64_t message_size = wireSize;
unsigned x = getRandomUnsigned();
uint8_t masking_key[4] = {};
masking_key[0] = (x >> 24);
@ -709,36 +746,18 @@ namespace ix {
// Now actually send this data
sendOnSocket();
return WebSocketSendInfo(true, payloadSize, wireSize);
}
void WebSocketTransport::sendPing(const std::string& message)
WebSocketSendInfo WebSocketTransport::sendPing(const std::string& message)
{
sendData(wsheader_type::PING, message.size(), message.begin(), message.end());
return sendData(wsheader_type::PING, message);
}
void WebSocketTransport::sendBinary(const std::string& message)
WebSocketSendInfo WebSocketTransport::sendBinary(const std::string& message)
{
if (_enablePerMessageDeflate)
{
// FIXME hack hack
std::string compressedMessage;
_perMessageDeflate.compress(message, compressedMessage);
std::cout << "uncompressedMessage " << message << std::endl;
std::cout << "uncompressedMessage.size() " << message.size() << std::endl;
std::cout << "compressedMessage.size() " << compressedMessage.size()
<< std::endl;
// sendData(wsheader_type::BINARY_FRAME, message.size(), message.begin(), message.end());
sendData(wsheader_type::BINARY_FRAME,
compressedMessage.size(),
compressedMessage.begin(),
compressedMessage.end());
}
else
{
sendData(wsheader_type::BINARY_FRAME, message.size(),
message.begin(), message.end());
}
return sendData(wsheader_type::BINARY_FRAME, message);
}
void WebSocketTransport::sendOnSocket()

41
ixwebsocket/IXWebSocketTransport.h
View File

@ -17,7 +17,10 @@
#include <mutex>
#include <atomic>
#include "IXWebSocketSendInfo.h"
#include "IXWebSocketPerMessageDeflate.h"
#include "IXWebSocketPerMessageDeflateOptions.h"
#include "IXWebSocketHttpHeaders.h"
namespace ix
{
@ -28,12 +31,17 @@ namespace ix
bool success;
int http_status;
std::string errorStr;
WebSocketHttpHeaders headers;
WebSocketInitResult(bool s, int h, std::string e)
WebSocketInitResult(bool s,
int status,
const std::string& e,
WebSocketHttpHeaders h = WebSocketHttpHeaders())
{
success = s;
http_status = h;
http_status = status;
errorStr = e;
headers = h;
}
// need to define a default
@ -42,6 +50,7 @@ namespace ix
success = false;
http_status = 0;
errorStr = "";
headers.clear();
}
};
@ -64,21 +73,22 @@ namespace ix
};
using OnMessageCallback = std::function<void(const std::string&,
size_t,
MessageKind)>;
using OnCloseCallback = std::function<void(uint16_t,
const std::string&)>;
const std::string&,
size_t)>;
WebSocketTransport();
~WebSocketTransport();
void configure(const std::string& url);
void configure(const std::string& url,
const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
WebSocketInitResult init();
void poll();
void send(const std::string& message);
void sendBinary(const std::string& message);
void sendBinary(const std::vector<uint8_t>& message);
void sendPing(const std::string& message);
WebSocketSendInfo sendBinary(const std::string& message);
WebSocketSendInfo sendPing(const std::string& message);
void close();
ReadyStateValues getReadyState() const;
void setReadyState(ReadyStateValues readyStateValue);
@ -123,20 +133,25 @@ namespace ix
std::shared_ptr<Socket> _socket;
std::atomic<ReadyStateValues> _readyState;
std::atomic<bool> _enablePerMessageDeflate;
OnCloseCallback _onCloseCallback;
uint16_t _closeCode;
std::string _closeReason;
size_t _closeWireSize;
mutable std::mutex _closeDataMutex;
WebSocketPerMessageDeflate _perMessageDeflate;
WebSocketPerMessageDeflateOptions _perMessageDeflateOptions;
std::atomic<bool> _enablePerMessageDeflate;
void sendOnSocket();
void sendData(wsheader_type::opcode_type type,
uint64_t message_size,
std::string::const_iterator message_begin,
std::string::const_iterator message_end);
WebSocketSendInfo sendData(wsheader_type::opcode_type type,
const std::string& message);
void emitMessage(MessageKind messageKind,
const std::string& message,
const wsheader_type& ws,
const OnMessageCallback& onMessageCallback);
bool isSendBufferEmpty() const;
void appendToSendBuffer(const std::vector<uint8_t>& header,

2
ixwebsocket/libwshandshake.hpp
View File

@ -32,7 +32,7 @@ class WebSocketHandshake {
template <typename T>
struct static_for<0, T> {
void operator()(uint32_t *a, uint32_t *hash) {}
void operator()(uint32_t * /*a*/, uint32_t * /*hash*/) {}
};
template <int state>