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
parent 8d819053ff
commit a0a53ab986
72 changed files with 9116 additions and 259 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CMakeLists.txt
View File

@ -41,6 +41,7 @@ set( IXWEBSOCKET_HEADERS
ixwebsocket/IXSetThreadName.h
ixwebsocket/IXDNSLookup.h
ixwebsocket/IXCancellationRequest.h
ixwebsocket/IXProgressCallback.h
ixwebsocket/IXWebSocket.h
ixwebsocket/IXWebSocketServer.h
ixwebsocket/IXWebSocketTransport.h

2
examples/chat/cmd_websocket_chat.cpp
View File

@ -115,7 +115,7 @@ namespace
// store text
_receivedQueue.push(result.second);
ss << std::endl
ss << std::endl
<< result.first << " > " << result.second
<< std::endl
<< _user << " > ";

22
examples/cobra_publisher/IXCobraConnection.cpp
View File

@ -59,8 +59,8 @@ namespace ix
}
void CobraConnection::invokeEventCallback(ix::CobraConnectionEventType eventType,
const std::string& errorMsg,
const WebSocketHttpHeaders& headers)
const std::string& errorMsg,
const WebSocketHttpHeaders& headers)
{
std::lock_guard<std::mutex> lock(_eventCallbackMutex);
if (_eventCallback)
@ -176,10 +176,10 @@ namespace ix
}
void CobraConnection::configure(const std::string& appkey,
const std::string& endpoint,
const std::string& rolename,
const std::string& rolesecret,
WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
const std::string& endpoint,
const std::string& rolename,
const std::string& rolesecret,
WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
{
_appkey = appkey;
_endpoint = endpoint;
@ -229,7 +229,7 @@ namespace ix
return _webSocket.send(serializedJson).success;
}
//
//
// Extract the nonce from the handshake response
// use it to compute a hash during authentication
//
@ -297,7 +297,7 @@ namespace ix
if (!pdu.isMember("body")) return false;
Json::Value body = pdu["body"];
// Identify subscription_id, so that we can find
// Identify subscription_id, so that we can find
// which callback to execute
if (!body.isMember("subscription_id")) return false;
Json::Value subscriptionId = body["subscription_id"];
@ -339,7 +339,7 @@ namespace ix
// publish is not thread safe as we are trying to reuse some Json objects.
//
bool CobraConnection::publish(const Json::Value& channels,
const Json::Value& msg)
const Json::Value& msg)
{
_body["channels"] = channels;
_body["message"] = msg;
@ -371,7 +371,7 @@ namespace ix
}
void CobraConnection::subscribe(const std::string& channel,
SubscriptionCallback cb)
SubscriptionCallback cb)
{
// Create and send a subscribe pdu
Json::Value body;
@ -471,5 +471,5 @@ namespace ix
{
connect();
}
} // namespace ix

10
examples/cobra_publisher/IXCobraConnection.h
View File

@ -84,7 +84,7 @@ namespace ix
/// Returns true only if we're connected
bool isConnected() const;
/// Flush the publish queue
bool flushQueue();
@ -118,7 +118,7 @@ namespace ix
///
/// Member variables
///
///
WebSocket _webSocket;
/// Configuration data
@ -148,10 +148,10 @@ namespace ix
std::unordered_map<std::string, SubscriptionCallback> _cbs;
mutable std::mutex _cbsMutex;
// Message Queue can be touched on control+background thread,
// Message Queue can be touched on control+background thread,
// protecting with a mutex.
//
// Message queue is used when there are problems sending messages so
// Message queue is used when there are problems sending messages so
// that sending can be retried later.
std::deque<std::string> _messageQueue;
mutable std::mutex _queueMutex;
@ -159,5 +159,5 @@ namespace ix
// Cap the queue size (100 elems so far -> ~100k)
static constexpr size_t kQueueMaxSize = 256;
};
} // namespace ix

91
examples/cobra_publisher/ixcrypto/IXBase64.cpp
View File

@ -1,39 +1,39 @@
/*
base64.cpp and base64.h
Copyright (C) 2004-2008 René Nyffenegger
This source code is provided 'as-is', without any express or implied
warranty. In no event will the author be held liable for any damages
arising from the use of this software.
Permission is granted to anyone to use this software for any purpose,
including commercial applications, and to alter it and redistribute it
freely, subject to the following restrictions:
1. The origin of this source code must not be misrepresented; you must not
claim that you wrote the original source code. If you use this source code
in a product, an acknowledgment in the product documentation would be
appreciated but is not required.
2. Altered source versions must be plainly marked as such, and must not be
misrepresented as being the original source code.
3. This notice may not be removed or altered from any source distribution.
René Nyffenegger rene.nyffenegger@adp-gmbh.ch
*/
#include "IXBase64.h"
namespace ix
{
static const std::string base64_chars =
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789+/";
std::string base64_encode(const std::string& data, size_t len)
{
std::string ret;
@ -41,9 +41,9 @@ namespace ix
int j = 0;
unsigned char char_array_3[3];
unsigned char char_array_4[4];
const char* bytes_to_encode = data.c_str();
while(len--)
{
char_array_3[i++] = *(bytes_to_encode++);
@ -53,32 +53,83 @@ namespace ix
char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
char_array_4[3] = char_array_3[2] & 0x3f;
for(i = 0; (i <4) ; i++)
ret += base64_chars[char_array_4[i]];
i = 0;
}
}
if(i)
{
for(j = i; j < 3; j++)
char_array_3[j] = '\0';
char_array_4[0] = (char_array_3[0] & 0xfc) >> 2;
char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
char_array_4[3] = char_array_3[2] & 0x3f;
for(j = 0; (j < i + 1); j++)
ret += base64_chars[char_array_4[j]];
while((i++ < 3))
ret += '=';
}
return ret;
}
static inline bool is_base64(unsigned char c)
{
return (isalnum(c) || (c == '+') || (c == '/'));
}
std::string base64_decode(const std::string& encoded_string)
{
int in_len = (int)encoded_string.size();
int i = 0;
int j = 0;
int in_ = 0;
unsigned char char_array_4[4], char_array_3[3];
std::string ret;
while(in_len-- && ( encoded_string[in_] != '=') && is_base64(encoded_string[in_]))
{
char_array_4[i++] = encoded_string[in_]; in_++;
if(i ==4)
{
for(i = 0; i <4; i++)
char_array_4[i] = base64_chars.find(char_array_4[i]);
char_array_3[0] = (char_array_4[0] << 2) + ((char_array_4[1] & 0x30) >> 4);
char_array_3[1] = ((char_array_4[1] & 0xf) << 4) + ((char_array_4[2] & 0x3c) >> 2);
char_array_3[2] = ((char_array_4[2] & 0x3) << 6) + char_array_4[3];
for(i = 0; (i < 3); i++)
ret += char_array_3[i];
i = 0;
}
}
if(i)
{
for(j = i; j <4; j++)
char_array_4[j] = 0;
for(j = 0; j <4; j++)
char_array_4[j] = base64_chars.find(char_array_4[j]);
char_array_3[0] = (char_array_4[0] << 2) + ((char_array_4[1] & 0x30) >> 4);
char_array_3[1] = ((char_array_4[1] & 0xf) << 4) + ((char_array_4[2] & 0x3c) >> 2);
char_array_3[2] = ((char_array_4[2] & 0x3) << 6) + char_array_4[3];
for(j = 0; (j < i - 1); j++) ret += char_array_3[j];
}
return ret;
}
}

1
examples/cobra_publisher/ixcrypto/IXBase64.h
View File

@ -11,4 +11,5 @@
namespace ix
{
std::string base64_encode(const std::string& data, size_t len);
std::string base64_decode(const std::string& encoded_string);
}

22
examples/cobra_publisher/ixcrypto/IXHash.cpp Normal file
View File

@ -0,0 +1,22 @@
/*
* IXHash.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
#include <string>
namespace ix
{
uint64_t djb2Hash(const std::string& data)
{
uint64_t hashAddress = 5381;
for (auto& c : data)
{
hashAddress = ((hashAddress << 5) + hashAddress) + c;
}
return hashAddress;
}
}

15
examples/cobra_publisher/ixcrypto/IXHash.h Normal file
View File

@ -0,0 +1,15 @@
/*
* IXHash.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
uint64_t djb2Hash(const std::string& data);
}

75
examples/cobra_publisher/ixcrypto/IXUuid.cpp Normal file
View File

@ -0,0 +1,75 @@
/*
* IXUuid.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
/**
* Generate a random uuid similar to the uuid python module
*
* >>> import uuid
* >>> uuid.uuid4().hex
* 'bec08155b37d4050a1f3c3fa0276bf12'
*
* Code adapted from https://github.com/r-lyeh-archived/sole
*/
#include "IXUuid.h"
#include <sstream>
#include <string>
#include <iomanip>
#include <random>
namespace ix
{
class Uuid
{
public:
Uuid();
std::string toString() const;
private:
uint64_t _ab;
uint64_t _cd;
};
Uuid::Uuid()
{
static std::random_device rd;
static std::uniform_int_distribution<uint64_t> dist(0, (uint64_t)(~0));
_ab = dist(rd);
_cd = dist(rd);
_ab = (_ab & 0xFFFFFFFFFFFF0FFFULL) | 0x0000000000004000ULL;
_cd = (_cd & 0x3FFFFFFFFFFFFFFFULL) | 0x8000000000000000ULL;
}
std::string Uuid::toString() const
{
std::stringstream ss;
ss << std::hex << std::nouppercase << std::setfill('0');
uint32_t a = (_ab >> 32);
uint32_t b = (_ab & 0xFFFFFFFF);
uint32_t c = (_cd >> 32);
uint32_t d = (_cd & 0xFFFFFFFF);
ss << std::setw(8) << (a);
ss << std::setw(4) << (b >> 16);
ss << std::setw(4) << (b & 0xFFFF);
ss << std::setw(4) << (c >> 16 );
ss << std::setw(4) << (c & 0xFFFF);
ss << std::setw(8) << d;
return ss.str();
}
std::string uuid4()
{
Uuid id;
return id.toString();
}
}

17
examples/cobra_publisher/ixcrypto/IXUuid.h Normal file
View File

@ -0,0 +1,17 @@
/*
* IXUuid.h
* Author: Benjamin Sergeant
* Copyright (c) 2017 Machine Zone. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
/**
* Generate a random uuid
*/
std::string uuid4();
}

14
examples/cobra_publisher/jsoncpp/json/json-forwards.h
View File

@ -7,28 +7,28 @@
// //////////////////////////////////////////////////////////////////////
/*
The JsonCpp library's source code, including accompanying documentation,
The JsonCpp library's source code, including accompanying documentation,
tests and demonstration applications, are licensed under the following
conditions...
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
this software is released into the Public Domain.
In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
The JsonCpp Authors, and is released under the terms of the MIT License (see below).
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
Public Domain/MIT License conditions described here, as they choose.
The MIT License is about as close to Public Domain as a license can get, and is
described in clear, concise terms at:
http://en.wikipedia.org/wiki/MIT_License
The full text of the MIT License follows:
========================================================================

16
examples/cobra_publisher/jsoncpp/json/json.h
View File

@ -6,28 +6,28 @@
// //////////////////////////////////////////////////////////////////////
/*
The JsonCpp library's source code, including accompanying documentation,
The JsonCpp library's source code, including accompanying documentation,
tests and demonstration applications, are licensed under the following
conditions...
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
this software is released into the Public Domain.
In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
The JsonCpp Authors, and is released under the terms of the MIT License (see below).
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
Public Domain/MIT License conditions described here, as they choose.
The MIT License is about as close to Public Domain as a license can get, and is
described in clear, concise terms at:
http://en.wikipedia.org/wiki/MIT_License
The full text of the MIT License follows:
========================================================================
@ -1673,7 +1673,7 @@ public:
- `"rejectDupKeys": false or true`
- If true, `parse()` returns false when a key is duplicated within an object.
- `"allowSpecialFloats": false or true`
- If true, special float values (NaNs and infinities) are allowed
- If true, special float values (NaNs and infinities) are allowed
and their values are lossfree restorable.
You can examine 'settings_` yourself

22
examples/cobra_publisher/jsoncpp/jsoncpp.cpp
View File

@ -6,28 +6,28 @@
// //////////////////////////////////////////////////////////////////////
/*
The JsonCpp library's source code, including accompanying documentation,
The JsonCpp library's source code, including accompanying documentation,
tests and demonstration applications, are licensed under the following
conditions...
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
this software is released into the Public Domain.
In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
The JsonCpp Authors, and is released under the terms of the MIT License (see below).
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
Public Domain/MIT License conditions described here, as they choose.
The MIT License is about as close to Public Domain as a license can get, and is
described in clear, concise terms at:
http://en.wikipedia.org/wiki/MIT_License
The full text of the MIT License follows:
========================================================================
@ -238,7 +238,7 @@ static inline void fixNumericLocaleInput(char* begin, char* end) {
#include <limits>
#if defined(_MSC_VER)
#if !defined(WINCE) && defined(__STDC_SECURE_LIB__) && _MSC_VER >= 1500 // VC++ 9.0 and above
#if !defined(WINCE) && defined(__STDC_SECURE_LIB__) && _MSC_VER >= 1500 // VC++ 9.0 and above
#define snprintf sprintf_s
#elif _MSC_VER >= 1900 // VC++ 14.0 and above
#define snprintf std::snprintf
@ -383,7 +383,7 @@ bool Reader::parse(const char* beginDoc,
bool Reader::readValue() {
// readValue() may call itself only if it calls readObject() or ReadArray().
// These methods execute nodes_.push() just before and nodes_.pop)() just after calling readValue().
// These methods execute nodes_.push() just before and nodes_.pop)() just after calling readValue().
// parse() executes one nodes_.push(), so > instead of >=.
if (nodes_.size() > stackLimit_g) throwRuntimeError("Exceeded stackLimit in readValue().");
@ -4215,7 +4215,7 @@ Value& Path::make(Value& root) const {
#endif
#endif
#if defined(__BORLANDC__)
#if defined(__BORLANDC__)
#include <float.h>
#define isfinite _finite
#define snprintf _snprintf
@ -5290,7 +5290,7 @@ StreamWriter* StreamWriterBuilder::newStreamWriter() const
JSONCPP_STRING cs_str = settings_["commentStyle"].asString();
bool eyc = settings_["enableYAMLCompatibility"].asBool();
bool dnp = settings_["dropNullPlaceholders"].asBool();
bool usf = settings_["useSpecialFloats"].asBool();
bool usf = settings_["useSpecialFloats"].asBool();
unsigned int pre = settings_["precision"].asUInt();
CommentStyle::Enum cs = CommentStyle::All;
if (cs_str == "All") {

1
examples/ws_receive/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
build

30
examples/ws_receive/CMakeLists.txt Normal file
View File

@ -0,0 +1,30 @@
#
# Author: Benjamin Sergeant
# Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
#
cmake_minimum_required (VERSION 3.4.1)
project (ws_receive)
# There's -Weverything too for clang
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
set (CMAKE_CXX_STANDARD 14)
option(USE_TLS "Add TLS support" ON)
add_subdirectory(${PROJECT_SOURCE_DIR}/../.. ixwebsocket)
include_directories(ws_receive .)
add_executable(ws_receive
jsoncpp/jsoncpp.cpp
ixcrypto/IXBase64.cpp
ixcrypto/IXHash.cpp
ws_receive.cpp)
if (APPLE AND USE_TLS)
target_link_libraries(ws_receive "-framework foundation" "-framework security")
endif()
target_link_libraries(ws_receive ixwebsocket)

1
examples/ws_receive/README.md Normal file
View File

@ -0,0 +1 @@
ws_receive is a simple server upload program. It needs to be used in conjonction with ws_send.

1
examples/ws_receive/ixcrypto Symbolic link
View File

@ -0,0 +1 @@
../cobra_publisher/ixcrypto

333
examples/ws_receive/jsoncpp/json/json-forwards.h Normal file
View File

@ -0,0 +1,333 @@
/// Json-cpp amalgated forward header (http://jsoncpp.sourceforge.net/).
/// It is intended to be used with #include "json/json-forwards.h"
/// This header provides forward declaration for all JsonCpp types.
// //////////////////////////////////////////////////////////////////////
// Beginning of content of file: LICENSE
// //////////////////////////////////////////////////////////////////////
/*
The JsonCpp library's source code, including accompanying documentation,
tests and demonstration applications, are licensed under the following
conditions...
Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
jurisdictions which recognize such a disclaimer. In such jurisdictions,
this software is released into the Public Domain.
In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
The JsonCpp Authors, and is released under the terms of the MIT License (see below).
In jurisdictions which recognize Public Domain property, the user of this
software may choose to accept it either as 1) Public Domain, 2) under the
conditions of the MIT License (see below), or 3) under the terms of dual
Public Domain/MIT License conditions described here, as they choose.
The MIT License is about as close to Public Domain as a license can get, and is
described in clear, concise terms at:
http://en.wikipedia.org/wiki/MIT_License
The full text of the MIT License follows:
========================================================================
Copyright (c) 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
Permission is hereby granted, free of charge, to any person
obtaining a copy of this software and associated documentation
files (the "Software"), to deal in the Software without
restriction, including without limitation the rights to use, copy,
modify, merge, publish, distribute, sublicense, and/or sell copies
of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
========================================================================
(END LICENSE TEXT)
The MIT license is compatible with both the GPL and commercial
software, affording one all of the rights of Public Domain with the
minor nuisance of being required to keep the above copyright notice
and license text in the source code. Note also that by accepting the
Public Domain "license" you can re-license your copy using whatever
license you like.
*/
// //////////////////////////////////////////////////////////////////////
// End of content of file: LICENSE
// //////////////////////////////////////////////////////////////////////
#ifndef JSON_FORWARD_AMALGATED_H_INCLUDED
# define JSON_FORWARD_AMALGATED_H_INCLUDED
/// If defined, indicates that the source file is amalgated
/// to prevent private header inclusion.
#define JSON_IS_AMALGAMATION
// //////////////////////////////////////////////////////////////////////
// Beginning of content of file: include/json/config.h
// //////////////////////////////////////////////////////////////////////
// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
// Distributed under MIT license, or public domain if desired and
// recognized in your jurisdiction.
// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
#ifndef JSON_CONFIG_H_INCLUDED
#define JSON_CONFIG_H_INCLUDED
#include <stddef.h>
#include <string> //typedef String
#include <stdint.h> //typedef int64_t, uint64_t
/// If defined, indicates that json library is embedded in CppTL library.
//# define JSON_IN_CPPTL 1
/// If defined, indicates that json may leverage CppTL library
//# define JSON_USE_CPPTL 1
/// If defined, indicates that cpptl vector based map should be used instead of
/// std::map
/// as Value container.
//# define JSON_USE_CPPTL_SMALLMAP 1
// If non-zero, the library uses exceptions to report bad input instead of C
// assertion macros. The default is to use exceptions.
#ifndef JSON_USE_EXCEPTION
#define JSON_USE_EXCEPTION 1
#endif
/// If defined, indicates that the source file is amalgated
/// to prevent private header inclusion.
/// Remarks: it is automatically defined in the generated amalgated header.
// #define JSON_IS_AMALGAMATION
#ifdef JSON_IN_CPPTL
#include <cpptl/config.h>
#ifndef JSON_USE_CPPTL
#define JSON_USE_CPPTL 1
#endif
#endif
#ifdef JSON_IN_CPPTL
#define JSON_API CPPTL_API
#elif defined(JSON_DLL_BUILD)
#if defined(_MSC_VER) || defined(__MINGW32__)
#define JSON_API __declspec(dllexport)
#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
#endif // if defined(_MSC_VER)
#elif defined(JSON_DLL)
#if defined(_MSC_VER) || defined(__MINGW32__)
#define JSON_API __declspec(dllimport)
#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
#endif // if defined(_MSC_VER)
#endif // ifdef JSON_IN_CPPTL
#if !defined(JSON_API)
#define JSON_API
#endif
// If JSON_NO_INT64 is defined, then Json only support C++ "int" type for
// integer
// Storages, and 64 bits integer support is disabled.
// #define JSON_NO_INT64 1
#if defined(_MSC_VER) // MSVC
# if _MSC_VER <= 1200 // MSVC 6
// Microsoft Visual Studio 6 only support conversion from __int64 to double
// (no conversion from unsigned __int64).
# define JSON_USE_INT64_DOUBLE_CONVERSION 1
// Disable warning 4786 for VS6 caused by STL (identifier was truncated to '255'
// characters in the debug information)
// All projects I've ever seen with VS6 were using this globally (not bothering
// with pragma push/pop).
# pragma warning(disable : 4786)
# endif // MSVC 6
# if _MSC_VER >= 1500 // MSVC 2008
/// Indicates that the following function is deprecated.
# define JSONCPP_DEPRECATED(message) __declspec(deprecated(message))
# endif
#endif // defined(_MSC_VER)
// In c++11 the override keyword allows you to explicity define that a function
// is intended to override the base-class version. This makes the code more
// managable and fixes a set of common hard-to-find bugs.
#if __cplusplus >= 201103L
# define JSONCPP_OVERRIDE override
# define JSONCPP_NOEXCEPT noexcept
#elif defined(_MSC_VER) && _MSC_VER > 1600 && _MSC_VER < 1900
# define JSONCPP_OVERRIDE override
# define JSONCPP_NOEXCEPT throw()
#elif defined(_MSC_VER) && _MSC_VER >= 1900
# define JSONCPP_OVERRIDE override
# define JSONCPP_NOEXCEPT noexcept
#else
# define JSONCPP_OVERRIDE
# define JSONCPP_NOEXCEPT throw()
#endif
#ifndef JSON_HAS_RVALUE_REFERENCES
#if defined(_MSC_VER) && _MSC_VER >= 1600 // MSVC >= 2010
#define JSON_HAS_RVALUE_REFERENCES 1
#endif // MSVC >= 2010
#ifdef __clang__
#if __has_feature(cxx_rvalue_references)
#define JSON_HAS_RVALUE_REFERENCES 1
#endif // has_feature
#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
#if defined(__GXX_EXPERIMENTAL_CXX0X__) || (__cplusplus >= 201103L)
#define JSON_HAS_RVALUE_REFERENCES 1
#endif // GXX_EXPERIMENTAL
#endif // __clang__ || __GNUC__
#endif // not defined JSON_HAS_RVALUE_REFERENCES
#ifndef JSON_HAS_RVALUE_REFERENCES
#define JSON_HAS_RVALUE_REFERENCES 0
#endif
#ifdef __clang__
# if __has_extension(attribute_deprecated_with_message)
# define JSONCPP_DEPRECATED(message) __attribute__ ((deprecated(message)))
# endif
#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
# if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 5))
# define JSONCPP_DEPRECATED(message) __attribute__ ((deprecated(message)))
# elif (__GNUC__ > 3 || (__GNUC__ == 3 && __GNUC_MINOR__ >= 1))
# define JSONCPP_DEPRECATED(message) __attribute__((__deprecated__))
# endif // GNUC version
#endif // __clang__ || __GNUC__
#if !defined(JSONCPP_DEPRECATED)
#define JSONCPP_DEPRECATED(message)
#endif // if !defined(JSONCPP_DEPRECATED)
#if __GNUC__ >= 6
# define JSON_USE_INT64_DOUBLE_CONVERSION 1
#endif
#if !defined(JSON_IS_AMALGAMATION)
# include "version.h"
# if JSONCPP_USING_SECURE_MEMORY
# include "allocator.h" //typedef Allocator
# endif
#endif // if !defined(JSON_IS_AMALGAMATION)
namespace Json {
typedef int Int;
typedef unsigned int UInt;
#if defined(JSON_NO_INT64)
typedef int LargestInt;
typedef unsigned int LargestUInt;
#undef JSON_HAS_INT64
#else // if defined(JSON_NO_INT64)
// For Microsoft Visual use specific types as long long is not supported
#if defined(_MSC_VER) // Microsoft Visual Studio
typedef __int64 Int64;
typedef unsigned __int64 UInt64;
#else // if defined(_MSC_VER) // Other platforms, use long long
typedef int64_t Int64;
typedef uint64_t UInt64;
#endif // if defined(_MSC_VER)
typedef Int64 LargestInt;
typedef UInt64 LargestUInt;
#define JSON_HAS_INT64
#endif // if defined(JSON_NO_INT64)
#if JSONCPP_USING_SECURE_MEMORY
#define JSONCPP_STRING std::basic_string<char, std::char_traits<char>, Json::SecureAllocator<char> >
#define JSONCPP_OSTRINGSTREAM std::basic_ostringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
#define JSONCPP_OSTREAM std::basic_ostream<char, std::char_traits<char>>
#define JSONCPP_ISTRINGSTREAM std::basic_istringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
#define JSONCPP_ISTREAM std::istream
#else
#define JSONCPP_STRING std::string
#define JSONCPP_OSTRINGSTREAM std::ostringstream
#define JSONCPP_OSTREAM std::ostream
#define JSONCPP_ISTRINGSTREAM std::istringstream
#define JSONCPP_ISTREAM std::istream
#endif // if JSONCPP_USING_SECURE_MEMORY
} // end namespace Json
#endif // JSON_CONFIG_H_INCLUDED
// //////////////////////////////////////////////////////////////////////
// End of content of file: include/json/config.h
// //////////////////////////////////////////////////////////////////////
// //////////////////////////////////////////////////////////////////////
// Beginning of content of file: include/json/forwards.h
// //////////////////////////////////////////////////////////////////////
// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
// Distributed under MIT license, or public domain if desired and
// recognized in your jurisdiction.
// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
#ifndef JSON_FORWARDS_H_INCLUDED
#define JSON_FORWARDS_H_INCLUDED
#if !defined(JSON_IS_AMALGAMATION)
#include "config.h"
#endif // if !defined(JSON_IS_AMALGAMATION)
namespace Json {
// writer.h
class FastWriter;
class StyledWriter;
// reader.h
class Reader;
// features.h
class Features;
// value.h
typedef unsigned int ArrayIndex;
class StaticString;
class Path;
class PathArgument;
class Value;
class ValueIteratorBase;
class ValueIterator;
class ValueConstIterator;
} // namespace Json
#endif // JSON_FORWARDS_H_INCLUDED
// //////////////////////////////////////////////////////////////////////
// End of content of file: include/json/forwards.h
// //////////////////////////////////////////////////////////////////////
#endif //ifndef JSON_FORWARD_AMALGATED_H_INCLUDED

2186
examples/ws_receive/jsoncpp/json/json.h Normal file
View File

File diff suppressed because it is too large Load Diff

5386
examples/ws_receive/jsoncpp/jsoncpp.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

29
examples/ws_receive/package-lock.json generated Normal file
View File

@ -0,0 +1,29 @@
{
"requires": true,
"lockfileVersion": 1,
"dependencies": {
"async-limiter": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/async-limiter/-/async-limiter-1.0.0.tgz",
"integrity": "sha512-jp/uFnooOiO+L211eZOoSyzpOITMXx1rBITauYykG3BRYPu8h0UcxsPNB04RR5vo4Tyz3+ay17tR6JVf9qzYWg=="
},
"base-64": {
"version": "0.1.0",
"resolved": "https://registry.npmjs.org/base-64/-/base-64-0.1.0.tgz",
"integrity": "sha1-eAqZyE59YAJgNhURxId2E78k9rs="
},
"djb2": {
"version": "0.0.2",
"resolved": "https://registry.npmjs.org/djb2/-/djb2-0.0.2.tgz",
"integrity": "sha1-RAs4kao6uBQrVNRpsXe66p6W5O8="
},
"ws": {
"version": "6.1.4",
"resolved": "https://registry.npmjs.org/ws/-/ws-6.1.4.tgz",
"integrity": "sha512-eqZfL+NE/YQc1/ZynhojeV8q+H050oR8AZ2uIev7RU10svA9ZnJUddHcOUZTJLinZ9yEfdA2kSATS2qZK5fhJA==",
"requires": {
"async-limiter": "1.0.0"
}
}
}
}

153
examples/ws_receive/ws_receive.cpp Normal file
View File

@ -0,0 +1,153 @@
/*
* ws_receive.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <sstream>
#include <fstream>
#include <ixwebsocket/IXWebSocketServer.h>
#include <jsoncpp/json/json.h>
#include <ixcrypto/IXBase64.h>
#include <ixcrypto/IXHash.h>
namespace
{
// We should cleanup the file name and full path further to remove .. as well
std::string extractFilename(const std::string& path)
{
std::string filename("filename.conf");
std::string::size_type idx;
idx = path.rfind('/');
if (idx != std::string::npos)
{
std::string filename = path.substr(idx+1);
return filename;
}
else
{
return std::string();
}
}
}
namespace ix
{
void errorHandler(const std::string& errMsg,
const std::string& id,
std::shared_ptr<ix::WebSocket> webSocket)
{
Json::Value pdu;
pdu["kind"] = "error";
pdu["id"] = id;
pdu["message"] = errMsg;
webSocket->send(pdu.toStyledString());
}
void messageHandler(const std::string& str,
std::shared_ptr<ix::WebSocket> webSocket)
{
std::cerr << "Received message: " << str.size() << std::endl;
Json::Value data;
Json::Reader reader;
if (!reader.parse(str, data))
{
errorHandler("Invalid JSON", std::string(), webSocket);
return;
}
std::cout << "id: " << data["id"].asString() << std::endl;
std::string content = ix::base64_decode(data["content"].asString());
std::cout << "Content size: " << content.size() << std::endl;
// Validate checksum
uint64_t cksum = ix::djb2Hash(data["content"].asString());
uint64_t cksumRef = data["djb2_hash"].asUInt64();
std::cout << "Computed hash: " << cksum << std::endl;
std::cout << "Reference hash: " << cksumRef << std::endl;
if (cksum != cksumRef)
{
errorHandler("Hash mismatch.", std::string(), webSocket);
return;
}
std::string filename = data["filename"].asString();
filename = extractFilename(filename);
std::ofstream out(filename);
out << content;
out.close();
Json::Value pdu;
pdu["ack"] = true;
pdu["id"] = data["id"];
pdu["filename"] = data["filename"];
webSocket->send(pdu.toStyledString());
}
}
int main(int argc, char** argv)
{
int port = 8080;
if (argc == 2)
{
std::stringstream ss;
ss << argv[1];
ss >> port;
}
ix::WebSocketServer server(port);
server.setOnConnectionCallback(
[&server](std::shared_ptr<ix::WebSocket> webSocket)
{
webSocket->setOnMessageCallback(
[webSocket, &server](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)
{
std::cerr << "New connection" << std::endl;
std::cerr << "Uri: " << openInfo.uri << std::endl;
std::cerr << "Headers:" << std::endl;
for (auto it : openInfo.headers)
{
std::cerr << it.first << ": " << it.second << std::endl;
}
}
else if (messageType == ix::WebSocket_MessageType_Close)
{
std::cerr << "Closed connection" << std::endl;
}
else if (messageType == ix::WebSocket_MessageType_Message)
{
messageHandler(str, webSocket);
}
}
);
}
);
auto res = server.listen();
if (!res.first)
{
std::cerr << res.second << std::endl;
return 1;
}
server.start();
server.wait();
return 0;
}

43
examples/ws_receive/ws_receive.js Normal file
View File

@ -0,0 +1,43 @@
/*
* ws_receive.js
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
const WebSocket = require('ws')
const djb2 = require('djb2')
const fs = require('fs')
const wss = new WebSocket.Server({ port: 8080,
perMessageDeflate: false,
maxPayload: 1024 * 1024 * 1024 * 1024});
wss.on('connection', function connection(ws) {
ws.on('message', function incoming(data) {
console.log('Received message')
let str = data.toString()
let obj = JSON.parse(str)
console.log(obj.id)
console.log(obj.djb2_hash)
console.log(djb2(obj.content))
var content = Buffer.from(obj.content, 'base64')
// let bytes = base64.decode(obj.content)
let path = obj.filename
fs.writeFile(path, content, function(err) {
if (err) {
throw err
} else {
console.log('wrote data to disk')
}
});
let response = {
id: obj.id
}
ws.send(JSON.stringify(response))
});
});

1
examples/ws_send/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
build

31
examples/ws_send/CMakeLists.txt Normal file
View File

@ -0,0 +1,31 @@
#
# Author: Benjamin Sergeant
# Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
#
cmake_minimum_required (VERSION 3.4.1)
project (ws_send)
# There's -Weverything too for clang
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
set (CMAKE_CXX_STANDARD 14)
option(USE_TLS "Add TLS support" ON)
add_subdirectory(${PROJECT_SOURCE_DIR}/../.. ixwebsocket)
include_directories(ws_send .)
add_executable(ws_send
jsoncpp/jsoncpp.cpp
ixcrypto/IXBase64.cpp
ixcrypto/IXUuid.cpp
ixcrypto/IXHash.cpp
ws_send.cpp)
if (APPLE AND USE_TLS)
target_link_libraries(ws_send "-framework foundation" "-framework security")
endif()
target_link_libraries(ws_send ixwebsocket)

1
examples/ws_send/README.md Normal file
View File

@ -0,0 +1 @@
ws_send is a simple upload program. It needs to be used in conjonction with ws_receive.

1
examples/ws_send/ixcrypto Symbolic link
View File

@ -0,0 +1 @@
../cobra_publisher/ixcrypto

1
examples/ws_send/jsoncpp Symbolic link
View File

@ -0,0 +1 @@
../cobra_publisher/jsoncpp

306
examples/ws_send/ws_send.cpp Normal file
View File

@ -0,0 +1,306 @@
/*
* ws_send.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <vector>
#include <condition_variable>
#include <mutex>
#include <chrono>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include <ixcrypto/IXUuid.h>
#include <ixcrypto/IXBase64.h>
#include <ixcrypto/IXHash.h>
#include <jsoncpp/json/json.h>
using namespace ix;
namespace
{
void log(const std::string& msg)
{
std::cout << msg << std::endl;
}
class WebSocketSender
{
public:
WebSocketSender(const std::string& _url,
bool enablePerMessageDeflate);
void subscribe(const std::string& channel);
void start();
void stop();
void waitForConnection();
void waitForAck();
void sendMessage(const std::string& filename, bool throttle);
private:
std::string _url;
std::string _id;
ix::WebSocket _webSocket;
bool _enablePerMessageDeflate;
std::mutex _conditionVariableMutex;
std::condition_variable _condition;
};
WebSocketSender::WebSocketSender(const std::string& url,
bool enablePerMessageDeflate) :
_url(url),
_enablePerMessageDeflate(enablePerMessageDeflate)
{
;
}
void WebSocketSender::stop()
{
_webSocket.stop();
}
void WebSocketSender::waitForConnection()
{
std::cout << "Connecting..." << std::endl;
std::unique_lock<std::mutex> lock(_conditionVariableMutex);
_condition.wait(lock);
}
void WebSocketSender::waitForAck()
{
std::cout << "Waiting for ack..." << std::endl;
std::unique_lock<std::mutex> lock(_conditionVariableMutex);
_condition.wait(lock);
}
std::string load(const std::string& path)
{
// std::vector<uint8_t> memblock;
std::string str;
std::ifstream file(path);
if (!file.is_open()) return std::string();
file.seekg(0, file.end);
std::streamoff size = file.tellg();
file.seekg(0, file.beg);
str.resize(size);
file.read((char*)&str.front(), static_cast<std::streamsize>(size));
return str;
}
void WebSocketSender::start()
{
_webSocket.setUrl(_url);
ix::WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions(
_enablePerMessageDeflate, false, false, 15, 15);
_webSocket.setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
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)
{
_condition.notify_one();
log("ws_send: connected");
std::cout << "Uri: " << openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
}
else if (messageType == ix::WebSocket_MessageType_Close)
{
ss << "ws_send: connection closed:";
ss << " code " << closeInfo.code;
ss << " reason " << closeInfo.reason << std::endl;
log(ss.str());
}
else if (messageType == ix::WebSocket_MessageType_Message)
{
_condition.notify_one();
ss << "ws_send: received message: "
<< str;
log(ss.str());
Json::Value data;
Json::Reader reader;
if (!reader.parse(str, data))
{
std::cerr << "Invalid JSON response" << std::endl;
return;
}
std::string id = data["id"].asString();
if (_id != id)
{
std::cerr << "Invalid id" << std::endl;
}
}
else if (messageType == ix::WebSocket_MessageType_Error)
{
ss << "Connection error: " << error.reason << std::endl;
ss << "#retries: " << error.retries << std::endl;
ss << "Wait time(ms): " << error.wait_time << std::endl;
ss << "HTTP Status: " << error.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
class Bench
{
public:
Bench(const std::string& description) :
_description(description),
_start(std::chrono::system_clock::now()),
_reported(false)
{
;
}
~Bench()
{
if (!_reported)
{
report();
}
}
void report()
{
auto now = std::chrono::system_clock::now();
auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - _start);
_ms = milliseconds.count();
std::cout << _description << " completed in "
<< _ms << "ms" << std::endl;
_reported = true;
}
uint64_t getDuration() const
{
return _ms;
}
private:
std::string _description;
std::chrono::time_point<std::chrono::system_clock> _start;
uint64_t _ms;
bool _reported;
};
void WebSocketSender::sendMessage(const std::string& filename,
bool throttle)
{
std::string content;
{
Bench bench("load file from disk");
content = load(filename);
}
_id = uuid4();
std::string b64Content;
{
Bench bench("base 64 encode file");
b64Content = base64_encode(content, content.size());
}
Json::Value pdu;
pdu["kind"] = "send";
pdu["id"] = _id;
pdu["content"] = b64Content;
pdu["djb2_hash"] = djb2Hash(b64Content);
pdu["filename"] = filename;
Bench bench("Sending file through websocket");
_webSocket.send(pdu.toStyledString(),
[throttle](int current, int total) -> bool
{
std::cout << "Step " << current << " out of " << total << std::endl;
if (throttle)
{
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
}
return true;
});
bench.report();
auto duration = bench.getDuration();
auto transferRate = 1000 * b64Content.size() / duration;
transferRate /= (1024 * 1024);
std::cout << "Send transfer rate: " << transferRate << "MB/s" << std::endl;
}
void wsSend(const std::string& url,
const std::string& path,
bool enablePerMessageDeflate,
bool throttle)
{
WebSocketSender webSocketSender(url, enablePerMessageDeflate);
webSocketSender.start();
webSocketSender.waitForConnection();
std::cout << "Sending..." << std::endl;
webSocketSender.sendMessage(path, throttle);
webSocketSender.waitForAck();
std::cout << "Done !" << std::endl;
webSocketSender.stop();
}
}
int main(int argc, char** argv)
{
if (argc != 3)
{
std::cerr << "Usage: ws_send <url> <path>" << std::endl;
return 1;
}
std::string url = argv[1];
std::string path = argv[2];
bool throttle = false;
bool enablePerMessageDeflate = false;
Socket::init();
wsSend(url, path, enablePerMessageDeflate, throttle);
return 0;
}

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
{

12
ixwebsocket/IXWebSocketHandshake.cpp
View File

@ -17,7 +17,7 @@
#include <algorithm>
namespace ix
namespace ix
{
WebSocketHandshake::WebSocketHandshake(std::atomic<bool>& requestInitCancellation,
std::shared_ptr<Socket> socket,
@ -88,7 +88,7 @@ namespace ix
std::string WebSocketHandshake::genRandomString(const int len)
{
std::string alphanum =
std::string alphanum =
"0123456789"
"ABCDEFGH"
"abcdefgh";
@ -139,7 +139,7 @@ namespace ix
{
_requestInitCancellation = false;
auto isCancellationRequested =
auto isCancellationRequested =
makeCancellationRequestWithTimeout(timeoutSecs, _requestInitCancellation);
std::string errMsg;
@ -234,7 +234,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"))
{
@ -280,7 +280,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);
@ -294,7 +294,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

@ -13,7 +13,7 @@
#include <memory>
#include <algorithm>
namespace ix
namespace ix
{
class Socket;

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

@ -30,12 +30,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),
@ -46,7 +49,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;
};
}

4
makefile
View File

@ -10,6 +10,10 @@ docker:
run: docker
docker run --cap-add sys_ptrace -it ws_connect:latest bash
# this is helpful to remove trailing whitespaces
trail:
sh third_party/remote_trailing_whitespaces.sh
build:
(cd examples/satori_publisher ; mkdir -p build ; cd build ; cmake .. ; make)
(cd examples/chat ; mkdir -p build ; cd build ; cmake .. ; make)

2
test/IXSocketTest.cpp
View File

@ -16,7 +16,7 @@
# endif
#endif
#include "IXTest.h"
#include "IXTest.h"
#include "catch.hpp"
using namespace ix;

4
test/IXTest.cpp
View File

@ -57,7 +57,7 @@ namespace ix
std::string generateSessionId()
{
auto now = std::chrono::system_clock::now();
auto seconds =
auto seconds =
std::chrono::duration_cast<std::chrono::seconds>(
now.time_since_epoch()).count();
@ -73,7 +73,7 @@ namespace ix
{
int defaultPort = 8090;
int sockfd;
int sockfd;
if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
{
log("Cannot compute a free port. socket error.");

6
test/IXWebSocketHeartBeatTest.cpp
View File

@ -34,7 +34,7 @@ namespace
int _port;
};
WebSocketClient::WebSocketClient(int port)
WebSocketClient::WebSocketClient(int port)
: _port(port)
{
;
@ -56,7 +56,7 @@ namespace
{
std::stringstream ss;
ss << "ws://localhost:"
<< _port
<< _port
<< "/";
url = ss.str();
@ -64,7 +64,7 @@ namespace
_webSocket.setUrl(url);
// The important bit for this test.
// The important bit for this test.
// Set a 1 second hearbeat ; if no traffic is present on the connection for 1 second
// a ping message will be sent by the client.
_webSocket.setHeartBeatPeriod(1);

15
test/cmd_websocket_chat.cpp
View File

@ -79,7 +79,7 @@ namespace
return _receivedMessages;
}
void WebSocketChat::appendMessage(const std::string& message)
void WebSocketChat::appendMessage(const std::string& message)
{
std::lock_guard<std::mutex> lock(_mutex);
_receivedMessages.push_back(message);
@ -101,7 +101,7 @@ namespace
{
std::stringstream ss;
ss << "ws://localhost:"
<< _port
<< _port
<< "/"
<< _user;
@ -150,10 +150,10 @@ namespace
std::string payload = result.second;
if (payload.size() > 2000)
{
payload = "<message too large>";
payload = "<message too large>";
}
ss << std::endl
ss << std::endl
<< result.first << " > " << payload
<< std::endl
<< _user << " > ";
@ -293,12 +293,13 @@ TEST_CASE("Websocket_chat", "[websocket_chat]")
chatB.sendMessage("from B1");
chatB.sendMessage("from B2");
// FIXME: cannot handle large message, we need to break them down
// into small one at the websocket layer (using CONTINUATION opcode)
size_t size = 512 * 1000; // 512K is OK, larger is not !!
// Test large messages that needs to be broken into small fragments
size_t size = 1 * 1024 * 1024; // ~1Mb
std::string bigMessage(size, 'a');
chatB.sendMessage(bigMessage);
log("Sent all messages");
// Wait until all messages are received. 10s timeout
int attempts = 0;
while (chatA.getReceivedMessagesCount() != 3 ||

2
test/run.py
View File

@ -20,7 +20,7 @@ if osName == 'Windows':
else:
generator = ''
make = 'make -j6'
testBinary ='./ixwebsocket_unittest'
testBinary ='./ixwebsocket_unittest'
sanitizersFlags = {
'asan': '-DSANITIZE_ADDRESS=On',

2
test/test_runner.cpp
View File

@ -9,7 +9,7 @@
#include <ixwebsocket/IXSocket.h>
int main(int argc, char* argv[])
int main(int argc, char* argv[])
{
ix::Socket::init(); // for Windows