Compare commits

...

8 commits

Author SHA1 Message Date
pravin-battu
3f4a4468e0 static variant update 2020-10-16 12:12:57 -03:00
Bobinson K B
fb27454cdf Merge branch 'ws-fc-updates' into 'latest-fc'
ws-updates

See merge request PBSA/PeerplaysIO/tools-libs/peerplays-fc!18
2020-08-12 13:00:39 +00:00
blockc p
917e4d348b ws-updates 2020-08-12 13:00:39 +00:00
Bobinson K B
8dd2fbe1b6 Merge branch 'issue-13-fix-websocket' into 'latest-fc'
Issue-13-fix (websocket shutdown)

See merge request PBSA/PeerplaysIO/tools-libs/peerplays-fc!17
2020-08-11 14:53:56 +00:00
Roshan Syed
b99585ca69 Added .gitlab-ci.yml 2020-07-09 13:11:09 +00:00
pravin-battu
c1e362a8dd websocket issue 2020-07-05 22:23:29 -03:00
blockc p
13b7cfaec7 Revert "websocket fix, issue 13"
This reverts commit 0cf6f461b6
2020-07-02 17:07:12 +00:00
pravin-battu
0cf6f461b6 websocket fix, issue 13 2020-07-02 13:34:29 -03:00
18 changed files with 348 additions and 244 deletions

10
.gitlab-ci.yml Normal file
View file

@ -0,0 +1,10 @@
include:
- template: Jobs/Code-Quality.gitlab-ci.yml
stages:
- test
trigger_build_peerplays:
stage: test
script:
- "curl -X POST -F token=$CI_JOB_TOKEN -F ref=$REF_NAME https://gitlab.com/api/v4/projects/10735589/trigger/pipeline"

View file

@ -168,6 +168,7 @@ set( fc_sources
src/variant.cpp
src/exception.cpp
src/variant_object.cpp
src/static_variant.cpp
src/thread/thread.cpp
src/thread/thread_specific.cpp
src/thread/future.cpp

View file

@ -2,7 +2,7 @@
#include <functional>
#include <memory>
#include <string>
#include <fc/any.hpp>
#include <boost/any.hpp>
#include <fc/network/ip.hpp>
#include <fc/signals.hpp>
@ -26,14 +26,14 @@ namespace fc { namespace http {
void on_message_handler( const std::function<void(const std::string&)>& h ) { _on_message = h; }
void on_http_handler( const std::function<std::string(const std::string&)>& h ) { _on_http = h; }
void set_session_data( fc::any d ){ _session_data = std::move(d); }
fc::any& get_session_data() { return _session_data; }
void set_session_data( boost::any d ){ _session_data = std::move(d); }
boost::any& get_session_data() { return _session_data; }
virtual std::string get_request_header(const std::string& key) = 0;
fc::signal<void()> closed;
private:
fc::any _session_data;
boost::any _session_data;
std::function<void(const std::string&)> _on_message;
std::function<string(const std::string&)> _on_http;
};
@ -53,6 +53,8 @@ namespace fc { namespace http {
uint16_t get_listening_port();
void start_accept();
void stop_listening();
void close();
private:
friend class detail::websocket_server_impl;
std::unique_ptr<detail::websocket_server_impl> my;
@ -84,6 +86,10 @@ namespace fc { namespace http {
websocket_connection_ptr connect( const std::string& uri );
websocket_connection_ptr secure_connect( const std::string& uri );
void close();
void synchronous_close();
private:
std::unique_ptr<detail::websocket_client_impl> my;
std::unique_ptr<detail::websocket_tls_client_impl> smy;

View file

@ -10,7 +10,7 @@ namespace fc { namespace rpc {
class websocket_api_connection : public api_connection
{
public:
websocket_api_connection( fc::http::websocket_connection& c, uint32_t max_conversion_depth );
websocket_api_connection(const std::shared_ptr<fc::http::websocket_connection> &c, uint32_t max_conversion_depth );
~websocket_api_connection();
virtual variant send_call(
@ -29,7 +29,7 @@ namespace fc { namespace rpc {
const std::string& message,
bool send_message = true );
fc::http::websocket_connection& _connection;
std::shared_ptr<fc::http::websocket_connection> _connection;
fc::rpc::state _rpc_state;
};

View file

@ -1,4 +1,5 @@
/** This source adapted from https://github.com/kmicklas/variadic-static_variant
/** This source adapted from https://github.com/kmicklas/variadic-static_variant Now available at https://github.com/kmicklas/variadic-variant.
*
* Copyright (C) 2013 Kenneth Micklas
*
@ -13,21 +14,15 @@
#include <stdexcept>
#include <typeinfo>
#include <fc/exception/exception.hpp>
namespace fc {
// Implementation details, the user should not import this:
namespace impl {
template<int N, typename... Ts>
struct storage_ops;
template<typename X, typename... Ts>
struct position;
template<typename... Ts>
struct type_info;
template<typename StaticVariant>
struct copy_construct
{
@ -40,7 +35,6 @@ struct copy_construct
sv.init(v);
}
};
template<typename StaticVariant>
struct move_construct
{
@ -53,25 +47,19 @@ struct move_construct
sv.init( std::move(v) );
}
};
template<int N, typename T, typename... Ts>
struct storage_ops<N, T&, Ts...> {
static void del(int n, void *data) {}
static void con(int n, void *data) {}
template<typename visitor>
static typename visitor::result_type apply(int n, void *data, visitor& v) {}
template<typename visitor>
static typename visitor::result_type apply(int n, void *data, const visitor& v) {}
template<typename visitor>
static typename visitor::result_type apply(int n, const void *data, visitor& v) {}
template<typename visitor>
static typename visitor::result_type apply(int n, const void *data, const visitor& v) {}
};
template<int N, typename T, typename... Ts>
struct storage_ops<N, T, Ts...> {
static void del(int n, void *data) {
@ -82,32 +70,27 @@ struct storage_ops<N, T, Ts...> {
if(n == N) new(reinterpret_cast<T*>(data)) T();
else storage_ops<N + 1, Ts...>::con(n, data);
}
template<typename visitor>
static typename visitor::result_type apply(int n, void *data, visitor& v) {
if(n == N) return v(*reinterpret_cast<T*>(data));
else return storage_ops<N + 1, Ts...>::apply(n, data, v);
}
template<typename visitor>
static typename visitor::result_type apply(int n, void *data, const visitor& v) {
if(n == N) return v(*reinterpret_cast<T*>(data));
else return storage_ops<N + 1, Ts...>::apply(n, data, v);
}
template<typename visitor>
static typename visitor::result_type apply(int n, const void *data, visitor& v) {
if(n == N) return v(*reinterpret_cast<const T*>(data));
else return storage_ops<N + 1, Ts...>::apply(n, data, v);
}
template<typename visitor>
static typename visitor::result_type apply(int n, const void *data, const visitor& v) {
if(n == N) return v(*reinterpret_cast<const T*>(data));
else return storage_ops<N + 1, Ts...>::apply(n, data, v);
}
};
template<int N>
struct storage_ops<N> {
static void del(int n, void *data) {
@ -116,7 +99,6 @@ struct storage_ops<N> {
static void con(int n, void *data) {
FC_THROW_EXCEPTION( fc::assert_exception, "Internal error: static_variant tag is invalid." );
}
template<typename visitor>
static typename visitor::result_type apply(int n, void *data, visitor& v) {
FC_THROW_EXCEPTION( fc::assert_exception, "Internal error: static_variant tag is invalid." );
@ -134,22 +116,18 @@ struct storage_ops<N> {
FC_THROW_EXCEPTION( fc::assert_exception, "Internal error: static_variant tag is invalid." );
}
};
template<typename X>
struct position<X> {
static const int pos = -1;
};
template<typename X, typename... Ts>
struct position<X, X, Ts...> {
static const int pos = 0;
};
template<typename X, typename T, typename... Ts>
struct position<X, T, Ts...> {
static const int pos = position<X, Ts...>::pos != -1 ? position<X, Ts...>::pos + 1 : -1;
};
template<typename T, typename... Ts>
struct type_info<T&, Ts...> {
static const bool no_reference_types = false;
@ -157,7 +135,6 @@ struct type_info<T&, Ts...> {
static const size_t size = type_info<Ts...>::size > sizeof(T&) ? type_info<Ts...>::size : sizeof(T&);
static const size_t count = 1 + type_info<Ts...>::count;
};
template<typename T, typename... Ts>
struct type_info<T, Ts...> {
static const bool no_reference_types = type_info<Ts...>::no_reference_types;
@ -165,7 +142,6 @@ struct type_info<T, Ts...> {
static const size_t size = type_info<Ts...>::size > sizeof(T) ? type_info<Ts...>::size : sizeof(T&);
static const size_t count = 1 + type_info<Ts...>::count;
};
template<>
struct type_info<> {
static const bool no_reference_types = true;
@ -174,48 +150,94 @@ struct type_info<> {
static const size_t size = 0;
};
} // namespace impl
template<typename TTag>
size_t size( TTag )
{
return 0;
}
template<typename TTag, typename A, typename...Ts>
size_t size( TTag tag )
{
if (tag <= 0)
{
return sizeof(A);
}
return size<TTag, Ts...>( --tag );
}
class dynamic_storage
{
char* storage;
public:
dynamic_storage();
~dynamic_storage();
void* data() const;
void alloc( size_t size );
void release();
};
} // namespace impl
template<typename... Types>
class static_variant {
using tag_type = int64_t;
static_assert(impl::type_info<Types...>::no_reference_types, "Reference types are not permitted in static_variant.");
static_assert(impl::type_info<Types...>::no_duplicates, "static_variant type arguments contain duplicate types.");
int _tag;
char storage[impl::type_info<Types...>::size];
template<typename X>
using type_in_typelist = typename std::enable_if<impl::position<X, Types...>::pos != -1, X>::type; // type is in typelist of static_variant.
int _tag;
impl::dynamic_storage storage;
template<typename X, typename = type_in_typelist<X>>
void init(const X& x) {
_tag = impl::position<X, Types...>::pos;
new(storage) X(x);
storage.alloc( sizeof(X) );
new(storage.data()) X(x);
}
template<typename X>
template<typename X, typename = type_in_typelist<X>>
void init(X&& x) {
_tag = impl::position<X, Types...>::pos;
new(storage) X( std::move(x) );
storage.alloc( sizeof(X) );
new(storage.data()) X( std::move(x) );
}
void init(tag_type tag)
{
FC_ASSERT( tag >= 0 );
FC_ASSERT( tag < count() );
_tag = tag;
storage.alloc( impl::size<tag_type, Types...>( tag ) );
impl::storage_ops<0, Types...>::con(_tag, storage.data());
}
void clean()
{
impl::storage_ops<0, Types...>::del(_tag, storage.data() );
storage.release();
}
template<typename StaticVariant>
friend struct impl::copy_construct;
template<typename StaticVariant>
friend struct impl::move_construct;
public:
template<typename X>
template<typename X, typename = type_in_typelist<X>>
struct tag
{
static_assert(
impl::position<X, Types...>::pos != -1,
"Type not in static_variant."
);
static const int value = impl::position<X, Types...>::pos;
};
static_variant()
{
_tag = 0;
impl::storage_ops<0, Types...>::con(0, storage);
}
init(0);
}
template<typename... Other>
static_variant( const static_variant<Other...>& cpy )
{
@ -225,46 +247,34 @@ public:
{
cpy.visit( impl::copy_construct<static_variant>(*this) );
}
static_variant( static_variant&& mv )
{
mv.visit( impl::move_construct<static_variant>(*this) );
}
template<typename X>
template<typename X, typename = type_in_typelist<X>>
static_variant(const X& v) {
static_assert(
impl::position<X, Types...>::pos != -1,
"Type not in static_variant."
);
init(v);
}
~static_variant() {
impl::storage_ops<0, Types...>::del(_tag, storage);
clean();
}
template<typename X>
template<typename X, typename = type_in_typelist<X>>
static_variant& operator=(const X& v) {
static_assert(
impl::position<X, Types...>::pos != -1,
"Type not in static_variant."
);
this->~static_variant();
clean();
init(v);
return *this;
}
static_variant& operator=( const static_variant& v )
{
if( this == &v ) return *this;
this->~static_variant();
clean();
v.visit( impl::copy_construct<static_variant>(*this) );
return *this;
}
static_variant& operator=( static_variant&& v )
{
if( this == &v ) return *this;
this->~static_variant();
clean();
v.visit( impl::move_construct<static_variant>(*this) );
return *this;
}
@ -276,98 +286,73 @@ public:
{
return a.which() < b.which();
}
template<typename X>
template<typename X, typename = type_in_typelist<X>>
X& get() {
static_assert(
impl::position<X, Types...>::pos != -1,
"Type not in static_variant."
);
if(_tag == impl::position<X, Types...>::pos) {
void* tmp(storage);
return *reinterpret_cast<X*>(tmp);
return *reinterpret_cast<X*>(storage.data());
} else {
FC_THROW_EXCEPTION( fc::assert_exception, "static_variant does not contain a value of type ${t}", ("t",fc::get_typename<X>::name()) );
// std::string("static_variant does not contain value of type ") + typeid(X).name()
// );
}
}
template<typename X>
template<typename X, typename = type_in_typelist<X>>
const X& get() const {
static_assert(
impl::position<X, Types...>::pos != -1,
"Type not in static_variant."
);
if(_tag == impl::position<X, Types...>::pos) {
const void* tmp(storage);
return *reinterpret_cast<const X*>(tmp);
return *reinterpret_cast<const X*>(storage.data());
} else {
FC_THROW_EXCEPTION( fc::assert_exception, "static_variant does not contain a value of type ${t}", ("t",fc::get_typename<X>::name()) );
}
}
template<typename visitor>
typename visitor::result_type visit(visitor& v) {
return impl::storage_ops<0, Types...>::apply(_tag, storage, v);
return impl::storage_ops<0, Types...>::apply(_tag, storage.data(), v);
}
template<typename visitor>
typename visitor::result_type visit(const visitor& v) {
return impl::storage_ops<0, Types...>::apply(_tag, storage, v);
return impl::storage_ops<0, Types...>::apply(_tag, storage.data(), v);
}
template<typename visitor>
typename visitor::result_type visit(visitor& v)const {
return impl::storage_ops<0, Types...>::apply(_tag, storage, v);
return impl::storage_ops<0, Types...>::apply(_tag, storage.data(), v);
}
template<typename visitor>
typename visitor::result_type visit(const visitor& v)const {
return impl::storage_ops<0, Types...>::apply(_tag, storage, v);
return impl::storage_ops<0, Types...>::apply(_tag, storage.data(), v);
}
static int count() { return impl::type_info<Types...>::count; }
void set_which( int w ) {
FC_ASSERT( w >= 0 );
FC_ASSERT( w < count() );
this->~static_variant();
_tag = w;
impl::storage_ops<0, Types...>::con(_tag, storage);
clean();
init(w);
}
int which() const {return _tag;}
};
template<typename Result>
struct visitor {
typedef Result result_type;
};
struct from_static_variant
{
variant& var;
const uint32_t _max_depth;
from_static_variant( variant& dv, uint32_t max_depth ):var(dv),_max_depth(max_depth){}
typedef void result_type;
template<typename T> void operator()( const T& v )const
{
to_variant( v, var, _max_depth );
}
};
struct to_static_variant
{
const variant& var;
const uint32_t _max_depth;
to_static_variant( const variant& dv, uint32_t max_depth ):var(dv),_max_depth(max_depth){}
typedef void result_type;
template<typename T> void operator()( T& v )const
{
from_variant( var, v, _max_depth );
}
};
template<typename... T> void to_variant( const fc::static_variant<T...>& s, fc::variant& v, uint32_t max_depth )
{
FC_ASSERT( max_depth > 0 );
@ -384,6 +369,5 @@ struct visitor {
s.set_which( ar[0].as_uint64() );
s.visit( to_static_variant(ar[1], max_depth - 1) );
}
template<typename... T> struct get_typename { static const char* name() { return typeid(static_variant<T...>).name(); } };
} // namespace fc
} // namespace fc

View file

@ -132,14 +132,10 @@ namespace fc { namespace http {
typedef websocketpp::transport::asio::tls_socket::endpoint socket_type;
};
typedef websocketpp::transport::asio::endpoint<transport_config>
transport_type;
typedef websocketpp::transport::asio::endpoint<transport_config> transport_type;
};
using websocketpp::connection_hdl;
typedef websocketpp::server<asio_with_stub_log> websocket_server_type;
typedef websocketpp::server<asio_tls_stub_log> websocket_tls_server_type;
@ -152,7 +148,7 @@ namespace fc { namespace http {
:_ws_connection(con){
}
~websocket_connection_impl()
virtual ~websocket_connection_impl()
{
}
@ -399,35 +395,26 @@ namespace fc { namespace http {
};
typedef websocketpp::client<asio_with_stub_log> websocket_client_type;
typedef websocketpp::client<asio_tls_stub_log> websocket_tls_client_type;
typedef websocket_client_type::connection_ptr websocket_client_connection_type;
typedef websocket_tls_client_type::connection_ptr websocket_tls_client_connection_type;
class websocket_client_impl
using websocketpp::connection_hdl;
template<typename T>
class generic_websocket_client_impl
{
public:
typedef websocket_client_type::message_ptr message_ptr;
websocket_client_impl()
generic_websocket_client_impl()
:_client_thread( fc::thread::current() )
{
_client.clear_access_channels( websocketpp::log::alevel::all );
_client.set_message_handler( [&]( connection_hdl hdl, message_ptr msg ){
_client.set_message_handler( [&]( connection_hdl hdl,
typename websocketpp::client<T>::message_ptr msg ){
_client_thread.async( [&](){
idump((msg->get_payload()));
wdump((msg->get_payload()));
//std::cerr<<"recv: "<<msg->get_payload()<<"\n";
auto received = msg->get_payload();
fc::async( [=](){
@ -453,73 +440,40 @@ namespace fc { namespace http {
_client.init_asio( &fc::asio::default_io_service() );
}
~websocket_client_impl()
virtual ~generic_websocket_client_impl()
{
if(_connection )
if( _connection )
{
_connection->close(0, "client closed");
_connection.reset();
_closed->wait();
}
if( _closed )
_closed->wait();
}
fc::promise<void>::ptr _connected;
fc::promise<void>::ptr _closed;
fc::thread& _client_thread;
websocket_client_type _client;
websocketpp::client<T> _client;
websocket_connection_ptr _connection;
std::string _uri;
fc::optional<connection_hdl> _hdl;
};
class websocket_client_impl : public generic_websocket_client_impl<asio_with_stub_log>
{};
class websocket_tls_client_impl
class websocket_tls_client_impl : public generic_websocket_client_impl<asio_tls_stub_log>
{
public:
typedef websocket_tls_client_type::message_ptr message_ptr;
websocket_tls_client_impl( const std::string& ca_filename )
:_client_thread( fc::thread::current() )
: generic_websocket_client_impl()
{
// ca_filename has special values:
// "_none" disables cert checking (potentially insecure!)
// "_default" uses default CA's provided by OS
_client.clear_access_channels( websocketpp::log::alevel::all );
_client.set_message_handler( [&]( connection_hdl hdl, message_ptr msg ){
_client_thread.async( [&](){
idump((msg->get_payload()));
_connection->on_message( msg->get_payload() );
}).wait();
});
_client.set_close_handler( [=]( connection_hdl hdl ){
if( _connection )
{
try {
_client_thread.async( [&](){
ilog(". ${p}", ("p",uint64_t(_connection.get())));
if( !_shutting_down && !_closed && _connection )
_connection->closed();
_connection.reset();
} ).wait();
} catch ( const fc::exception& e )
{
if( _closed ) _closed->set_exception( e.dynamic_copy_exception() );
}
if( _closed ) _closed->set_value();
}
});
_client.set_fail_handler( [=]( connection_hdl hdl ){
elog( "." );
auto con = _client.get_con_from_hdl(hdl);
auto message = con->get_ec().message();
if( _connection )
_client_thread.async( [&](){ if( _connection ) _connection->closed(); _connection.reset(); } ).wait();
if( _connected && !_connected->ready() )
_connected->set_exception( exception_ptr( new FC_EXCEPTION( exception, "${message}", ("message",message)) ) );
if( _closed )
_closed->set_value();
});
//
// We need ca_filename to be copied into the closure, as the referenced object might be destroyed by the caller by the time
// tls_init_handler() is called. According to [1], capture-by-value results in the desired behavior (i.e. creation of
@ -552,18 +506,8 @@ namespace fc { namespace http {
return ctx;
});
_client.init_asio( &fc::asio::default_io_service() );
}
~websocket_tls_client_impl()
{
if(_connection )
{
ilog(".");
_shutting_down = true;
_connection->close(0, "client closed");
_closed->wait();
}
}
virtual ~websocket_tls_client_impl() {}
std::string get_host()const
{
@ -576,20 +520,19 @@ namespace fc { namespace http {
return;
ctx->set_verify_mode( boost::asio::ssl::verify_peer );
if( ca_filename == "_default" )
{
#if WIN32
add_windows_root_certs( *ctx );
#else
ctx->set_default_verify_paths();
#endif
}
else
ctx->load_verify_file( ca_filename );
ctx->set_verify_depth(10);
ctx->set_verify_callback( boost::asio::ssl::rfc2818_verification( get_host() ) );
}
bool _shutting_down = false;
fc::promise<void>::ptr _connected;
fc::promise<void>::ptr _closed;
fc::thread& _client_thread;
websocket_tls_client_type _client;
websocket_connection_ptr _connection;
std::string _uri;
};
@ -622,8 +565,16 @@ namespace fc { namespace http {
my->_server.start_accept();
}
void websocket_server::stop_listening()
{
my->_server.stop_listening();
}
void websocket_server::close()
{
for (auto& connection : my->_connections)
my->_server.close(connection.first, websocketpp::close::status::normal, "Goodbye");
}
websocket_tls_server::websocket_tls_server( const string& server_pem, const string& ssl_password ):my( new detail::websocket_tls_server_impl(server_pem, ssl_password) ) {}
websocket_tls_server::~websocket_tls_server(){}
@ -646,7 +597,6 @@ namespace fc { namespace http {
my->_server.start_accept();
}
websocket_tls_client::websocket_tls_client( const std::string& ca_filename ):my( new detail::websocket_tls_client_impl( ca_filename ) ) {}
websocket_tls_client::~websocket_tls_client(){ }
@ -659,7 +609,7 @@ namespace fc { namespace http {
{ try {
if( uri.substr(0,4) == "wss:" )
return secure_connect(uri);
FC_ASSERT( uri.substr(0,3) == "ws:" );
FC_ASSERT( uri.substr(0,4) == "wss:" || uri.substr(0,3) == "ws:", "Unsupported protocol" );
// wlog( "connecting to ${uri}", ("uri",uri));
websocketpp::lib::error_code ec;
@ -687,7 +637,8 @@ namespace fc { namespace http {
{ try {
if( uri.substr(0,3) == "ws:" )
return connect(uri);
FC_ASSERT( uri.substr(0,4) == "wss:" );
FC_ASSERT( uri.substr(0,4) == "wss:" || uri.substr(0,3) == "ws:", "Unsupported protocol" );
// wlog( "connecting to ${uri}", ("uri",uri));
websocketpp::lib::error_code ec;
@ -709,6 +660,19 @@ namespace fc { namespace http {
return smy->_connection;
} FC_CAPTURE_AND_RETHROW( (uri) ) }
void websocket_client::close()
{
if (my->_hdl)
my->_client.close(*my->_hdl, websocketpp::close::status::normal, "Goodbye");
}
void websocket_client::synchronous_close()
{
close();
if (my->_closed)
my->_closed->wait();
}
websocket_connection_ptr websocket_tls_client::connect( const std::string& uri )
{ try {
// wlog( "connecting to ${uri}", ("uri",uri));

View file

@ -7,7 +7,7 @@ websocket_api_connection::~websocket_api_connection()
{
}
websocket_api_connection::websocket_api_connection( fc::http::websocket_connection& c, uint32_t max_depth )
websocket_api_connection::websocket_api_connection(const std::shared_ptr<fc::http::websocket_connection>& c, uint32_t max_depth )
: api_connection(max_depth),_connection(c)
{
_rpc_state.add_method( "call", [this]( const variants& args ) -> variant
@ -47,9 +47,12 @@ websocket_api_connection::websocket_api_connection( fc::http::websocket_connecti
return this->receive_call( 0, method_name, args );
} );
_connection.on_message_handler( [&]( const std::string& msg ){ on_message(msg,true); } );
_connection.on_http_handler( [&]( const std::string& msg ){ return on_message(msg,false); } );
_connection.closed.connect( [this](){ closed(); } );
_connection->on_message_handler( [&]( const std::string& msg ){ on_message(msg,true); } );
_connection->on_http_handler( [&]( const std::string& msg ){ return on_message(msg,false); } );
_connection->closed.connect( [this](){
closed();
_connection = nullptr;
} );
}
variant websocket_api_connection::send_call(
@ -58,7 +61,7 @@ variant websocket_api_connection::send_call(
variants args /* = variants() */ )
{
auto request = _rpc_state.start_remote_call( "call", {api_id, std::move(method_name), std::move(args) } );
_connection.send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth),
_connection->send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth),
fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) );
return _rpc_state.wait_for_response( *request.id );
}
@ -68,7 +71,7 @@ variant websocket_api_connection::send_callback(
variants args /* = variants() */ )
{
auto request = _rpc_state.start_remote_call( "callback", {callback_id, std::move(args) } );
_connection.send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth),
_connection->send_message( fc::json::to_string(fc::variant(request, _max_conversion_depth),
fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) );
return _rpc_state.wait_for_response( *request.id );
}
@ -78,7 +81,7 @@ void websocket_api_connection::send_notice(
variants args /* = variants() */ )
{
fc::rpc::request req{ optional<uint64_t>(), "notice", {callback_id, std::move(args)}};
_connection.send_message( fc::json::to_string(fc::variant(req, _max_conversion_depth),
_connection->send_message( fc::json::to_string(fc::variant(req, _max_conversion_depth),
fc::json::stringify_large_ints_and_doubles, _max_conversion_depth ) );
}
@ -118,7 +121,7 @@ std::string websocket_api_connection::on_message(
{
auto reply = fc::json::to_string( response( *call.id, result, "2.0" ), fc::json::stringify_large_ints_and_doubles, _max_conversion_depth );
if( send_message )
_connection.send_message( reply );
_connection->send_message( reply );
return reply;
}
}
@ -136,7 +139,7 @@ std::string websocket_api_connection::on_message(
auto reply = fc::json::to_string( variant(response( *call.id, error_object{ 1, optexcept->to_string(), fc::variant(*optexcept, _max_conversion_depth)}, "2.0" ), _max_conversion_depth ),
fc::json::stringify_large_ints_and_doubles, _max_conversion_depth );
if( send_message )
_connection.send_message( reply );
_connection->send_message( reply );
return reply;
}

31
src/static_variant.cpp Normal file
View file

@ -0,0 +1,31 @@
#include <fc/static_variant.hpp>
namespace fc { namespace impl {
dynamic_storage::dynamic_storage() : storage(nullptr) {};
dynamic_storage::~dynamic_storage()
{
release();
}
void* dynamic_storage::data() const
{
FC_ASSERT( storage != nullptr );
return (void*)storage;
}
void dynamic_storage::alloc( size_t size )
{
release();
storage = new char[size];
}
void dynamic_storage::release()
{
delete [] storage;
storage = nullptr;
}
}}

View file

@ -61,7 +61,7 @@ int main( int argc, char** argv )
fc::http::websocket_server server;
server.on_connection([&]( const websocket_connection_ptr& c ){
auto wsc = std::make_shared<websocket_api_connection>(*c, MAX_DEPTH);
auto wsc = std::make_shared<websocket_api_connection>(c, MAX_DEPTH);
auto login = std::make_shared<login_api>();
login->calc = calc_api;
wsc->register_api(fc::api<login_api>(login));
@ -76,7 +76,7 @@ int main( int argc, char** argv )
try {
fc::http::websocket_client client;
auto con = client.connect( "ws://localhost:8090" );
auto apic = std::make_shared<websocket_api_connection>(*con, MAX_DEPTH);
auto apic = std::make_shared<websocket_api_connection>(con, MAX_DEPTH);
auto remote_login_api = apic->get_remote_api<login_api>();
auto remote_calc = remote_login_api->get_calc();
remote_calc->on_result( []( uint32_t r ) { elog( "callback result ${r}", ("r",r) ); } );

View file

@ -1,4 +1,7 @@
HEAD
- Bug: Change default listen backlog from 0 to socket_base::max_connections.
#549. Thank you derwassi and zwelab for reporting and na1pir for providing
access to hardware to debug the issue.
0.7.0 - 2016-02-22
- MINOR BREAKING SOCKET POLICY CHANGE: Asio transport socket policy method

View file

@ -30,6 +30,8 @@
#include "connection_tu2.hpp"
#include <algorithm>
// Include special debugging transport
//#include <websocketpp/config/minimal_client.hpp>
#include <websocketpp/transport/debug/endpoint.hpp>
@ -172,6 +174,20 @@ void http_func(server* s, websocketpp::connection_hdl hdl) {
BOOST_CHECK_EQUAL(con->get_response_msg(), status_code::get_string(status_code::ok));
}
void http_func_with_move(server* s, websocketpp::connection_hdl hdl) {
using namespace websocketpp::http;
server::connection_ptr con = s->get_con_from_hdl(hdl);
std::string res = con->get_resource();
con->set_body( std::move(res) );
con->set_status(status_code::ok);
BOOST_CHECK_EQUAL(con->get_response_code(), status_code::ok);
BOOST_CHECK_EQUAL(con->get_response_msg(), status_code::get_string(status_code::ok));
}
void defer_http_func(server* s, bool * deferred, websocketpp::connection_hdl hdl) {
*deferred = true;
@ -237,6 +253,18 @@ BOOST_AUTO_TEST_CASE( http_request ) {
BOOST_CHECK_EQUAL(run_server_test(s,input), output);
}
BOOST_AUTO_TEST_CASE( http_request_with_move ) {
std::string input = "GET /foo/bar HTTP/1.1\r\nHost: www.example.com\r\nOrigin: http://www.example.com\r\n\r\n";
std::string output = "HTTP/1.1 200 OK\r\nContent-Length: 8\r\nServer: ";
output+=websocketpp::user_agent;
output+="\r\n\r\n/foo/bar";
server s;
s.set_http_handler(bind(&http_func_with_move,&s,::_1));
BOOST_CHECK_EQUAL(run_server_test(s,input), output);
}
BOOST_AUTO_TEST_CASE( deferred_http_request ) {
std::string input = "GET /foo/bar HTTP/1.1\r\nHost: www.example.com\r\nOrigin: http://www.example.com\r\n\r\n";
std::string output = "HTTP/1.1 200 OK\r\nContent-Length: 8\r\nServer: ";

View file

@ -1041,6 +1041,7 @@ public:
* @see websocketpp::http::response::set_body
*/
void set_body(std::string const & value);
void set_body( std::string&& value );
/// Append a header
/**

View file

@ -32,6 +32,7 @@
#include <websocketpp/common/cpp11.hpp>
#include <websocketpp/common/memory.hpp>
#include <websocketpp/common/platforms.hpp>
#include <websocketpp/common/stdint.hpp>
#include <websocketpp/common/system_error.hpp>
#include <websocketpp/error.hpp>
@ -46,7 +47,7 @@
namespace websocketpp {
namespace extensions {
/// Implementation of the draft permessage-deflate WebSocket extension
/// Implementation of RFC 7692, the permessage-deflate WebSocket extension
/**
* ### permessage-deflate interface
*
@ -174,18 +175,30 @@ namespace websocketpp {
namespace extensions {
namespace permessage_deflate {
/// Default value for server_max_window_bits as defined by draft 17
/// Default value for server_max_window_bits as defined by RFC 7692
static uint8_t const default_server_max_window_bits = 15;
/// Minimum value for server_max_window_bits as defined by draft 17
/// Minimum value for server_max_window_bits as defined by RFC 7692
/**
* NOTE: A value of 8 is not actually supported by zlib, the deflate
* library that WebSocket++ uses. To preserve backwards compatibility
* with RFC 7692 and previous versions of the library a value of 8
* is accepted by the library but will always be negotiated as 9.
*/
static uint8_t const min_server_max_window_bits = 8;
/// Maximum value for server_max_window_bits as defined by draft 17
/// Maximum value for server_max_window_bits as defined by RFC 7692
static uint8_t const max_server_max_window_bits = 15;
/// Default value for client_max_window_bits as defined by draft 17
/// Default value for client_max_window_bits as defined by RFC 7692
static uint8_t const default_client_max_window_bits = 15;
/// Minimum value for client_max_window_bits as defined by draft 17
/// Minimum value for client_max_window_bits as defined by RFC 7692
/**
* NOTE: A value of 8 is not actually supported by zlib, the deflate
* library that WebSocket++ uses. To preserve backwards compatibility
* with RFC 7692 and previous versions of the library a value of 8
* is accepted by the library but will always be negotiated as 9.
*/
static uint8_t const min_client_max_window_bits = 8;
/// Maximum value for client_max_window_bits as defined by draft 17
/// Maximum value for client_max_window_bits as defined by RFC 7692
static uint8_t const max_client_max_window_bits = 15;
namespace mode {
@ -213,7 +226,7 @@ public:
, m_server_max_window_bits_mode(mode::accept)
, m_client_max_window_bits_mode(mode::accept)
, m_initialized(false)
, m_compress_buffer_size(16384)
, m_compress_buffer_size(8192)
{
m_dstate.zalloc = Z_NULL;
m_dstate.zfree = Z_NULL;
@ -292,6 +305,7 @@ public:
}
m_compress_buffer.reset(new unsigned char[m_compress_buffer_size]);
m_decompress_buffer.reset(new unsigned char[m_compress_buffer_size]);
if ((m_server_no_context_takeover && is_server) ||
(m_client_no_context_takeover && !is_server))
{
@ -372,7 +386,7 @@ public:
/**
* The bits setting is the base 2 logarithm of the maximum window size that
* the server must use to compress outgoing messages. The permitted range
* is 8 to 15 inclusive. 8 represents a 256 byte window and 15 a 32KiB
* is 9 to 15 inclusive. 9 represents a 512 byte window and 15 a 32KiB
* window. The default setting is 15.
*
* Mode Options:
@ -386,6 +400,14 @@ public:
* adjusted by the server. A server may unilaterally set this value without
* client support.
*
* NOTE: The permessage-deflate spec specifies that a value of 8 is allowed.
* Prior to version 0.8.0 a value of 8 was also allowed by this library.
* zlib, the deflate compression library that WebSocket++ uses has always
* silently adjusted a value of 8 to 9. In recent versions of zlib (1.2.9
* and greater) a value of 8 is now explicitly rejected. WebSocket++ 0.8.0
* continues to perform the 8->9 conversion for backwards compatibility
* purposes but this should be considered deprecated functionality.
*
* @param bits The size to request for the outgoing window size
* @param mode The mode to use for negotiating this parameter
* @return A status code
@ -394,6 +416,12 @@ public:
if (bits < min_server_max_window_bits || bits > max_server_max_window_bits) {
return error::make_error_code(error::invalid_max_window_bits);
}
// See note in doc comment above about what is happening here
if (bits == 8) {
bits = 9;
}
m_server_max_window_bits = bits;
m_server_max_window_bits_mode = mode;
@ -403,8 +431,8 @@ public:
/// Limit client LZ77 sliding window size
/**
* The bits setting is the base 2 logarithm of the window size that the
* client must use to compress outgoing messages. The permitted range is 8
* to 15 inclusive. 8 represents a 256 byte window and 15 a 32KiB window.
* client must use to compress outgoing messages. The permitted range is 9
* to 15 inclusive. 9 represents a 512 byte window and 15 a 32KiB window.
* The default setting is 15.
*
* Mode Options:
@ -417,6 +445,14 @@ public:
* outgoing window size unilaterally. A server may only limit the client's
* window size if the remote client supports that feature.
*
* NOTE: The permessage-deflate spec specifies that a value of 8 is allowed.
* Prior to version 0.8.0 a value of 8 was also allowed by this library.
* zlib, the deflate compression library that WebSocket++ uses has always
* silently adjusted a value of 8 to 9. In recent versions of zlib (1.2.9
* and greater) a value of 8 is now explicitly rejected. WebSocket++ 0.8.0
* continues to perform the 8->9 conversion for backwards compatibility
* purposes but this should be considered deprecated functionality.
*
* @param bits The size to request for the outgoing window size
* @param mode The mode to use for negotiating this parameter
* @return A status code
@ -425,6 +461,12 @@ public:
if (bits < min_client_max_window_bits || bits > max_client_max_window_bits) {
return error::make_error_code(error::invalid_max_window_bits);
}
// See note in doc comment above about what is happening here
if (bits == 8) {
bits = 9;
}
m_client_max_window_bits = bits;
m_client_max_window_bits_mode = mode;
@ -555,7 +597,7 @@ public:
do {
m_istate.avail_out = m_compress_buffer_size;
m_istate.next_out = m_compress_buffer.get();
m_istate.next_out = m_decompress_buffer.get();
ret = inflate(&m_istate, Z_SYNC_FLUSH);
@ -564,7 +606,7 @@ public:
}
out.append(
reinterpret_cast<char *>(m_compress_buffer.get()),
reinterpret_cast<char *>(m_decompress_buffer.get()),
m_compress_buffer_size - m_istate.avail_out
);
} while (m_istate.avail_out == 0);
@ -642,11 +684,17 @@ private:
* client requested that we use.
*
* options:
* - decline (refuse to use the attribute)
* - accept (use whatever the client says)
* - largest (use largest possible value)
* - decline (ignore value, offer our default instead)
* - accept (use the value requested by the client)
* - largest (use largest value acceptable to both)
* - smallest (use smallest possible value)
*
* NOTE: As a value of 8 is no longer explicitly supported by zlib but might
* be requested for negotiation by an older client/server, if the result of
* the negotiation would be to send a value of 8, a value of 9 is offered
* instead. This ensures that WebSocket++ will only ever negotiate connections
* with compression settings explicitly supported by zlib.
*
* @param [in] value The value of the attribute from the offer
* @param [out] ec A reference to the error code to return errors via
*/
@ -678,6 +726,11 @@ private:
ec = make_error_code(error::invalid_mode);
m_server_max_window_bits = default_server_max_window_bits;
}
// See note in doc comment
if (m_server_max_window_bits == 8) {
m_server_max_window_bits = 9;
}
}
/// Negotiate client_max_window_bits attribute
@ -687,11 +740,17 @@ private:
* negotiation mode.
*
* options:
* - decline (refuse to use the attribute)
* - accept (use whatever the client says)
* - largest (use largest possible value)
* - decline (ignore value, offer our default instead)
* - accept (use the value requested by the client)
* - largest (use largest value acceptable to both)
* - smallest (use smallest possible value)
*
* NOTE: As a value of 8 is no longer explicitly supported by zlib but might
* be requested for negotiation by an older client/server, if the result of
* the negotiation would be to send a value of 8, a value of 9 is offered
* instead. This ensures that WebSocket++ will only ever negotiate connections
* with compression settings explicitly supported by zlib.
*
* @param [in] value The value of the attribute from the offer
* @param [out] ec A reference to the error code to return errors via
*/
@ -727,6 +786,11 @@ private:
ec = make_error_code(error::invalid_mode);
m_client_max_window_bits = default_client_max_window_bits;
}
// See note in doc comment
if (m_client_max_window_bits == 8) {
m_client_max_window_bits = 9;
}
}
bool m_enabled;
@ -741,6 +805,7 @@ private:
int m_flush;
size_t m_compress_buffer_size;
lib::unique_ptr_uchar_array m_compress_buffer;
lib::unique_ptr_uchar_array m_decompress_buffer;
z_stream m_dstate;
z_stream m_istate;
};
@ -750,3 +815,4 @@ private:
} // namespace websocketpp
#endif // WEBSOCKETPP_PROCESSOR_EXTENSION_PERMESSAGEDEFLATE_HPP

View file

@ -575,6 +575,17 @@ void connection<config>::set_body(std::string const & value) {
m_response.set_body(value);
}
template <typename config>
void connection<config>::set_body( std::string&& value )
{
if (m_internal_state != istate::PROCESS_HTTP_REQUEST) {
throw exception("Call to set_status from invalid state",
error::make_error_code(error::invalid_state));
}
m_response.set_body(std::move(value));
}
// TODO: EXCEPTION_FREE
template <typename config>
void connection<config>::append_header(std::string const & key,

View file

@ -311,10 +311,10 @@ public:
* needed.
*/
timer_ptr set_timer(long duration, timer_handler callback) {
timer_ptr new_timer = lib::make_shared<lib::asio::steady_timer>(
lib::ref(*m_io_service),
lib::asio::milliseconds(duration)
);
timer_ptr new_timer(
new lib::asio::steady_timer(
*m_io_service,
lib::asio::milliseconds(duration)));
if (config::enable_multithreading) {
new_timer->async_wait(m_strand->wrap(lib::bind(
@ -461,8 +461,7 @@ protected:
m_io_service = io_service;
if (config::enable_multithreading) {
m_strand = lib::make_shared<lib::asio::io_service::strand>(
lib::ref(*io_service));
m_strand.reset(new lib::asio::io_service::strand(*io_service));
}
lib::error_code ec = socket_con_type::init_asio(io_service, m_strand,

View file

@ -35,6 +35,7 @@
#include <websocketpp/uri.hpp>
#include <websocketpp/logger/levels.hpp>
#include <websocketpp/common/asio.hpp>
#include <websocketpp/common/functional.hpp>
#include <sstream>
@ -191,8 +192,7 @@ public:
m_io_service = ptr;
m_external_io_service = true;
m_acceptor = lib::make_shared<lib::asio::ip::tcp::acceptor>(
lib::ref(*m_io_service));
m_acceptor.reset(new lib::asio::ip::tcp::acceptor(*m_io_service));
m_state = READY;
ec = lib::error_code();
@ -314,8 +314,10 @@ public:
*
* New values affect future calls to listen only.
*
* A value of zero will use the operating system default. This is the
* default value.
* The default value is specified as *::asio::socket_base::max_connections
* which uses the operating system defined maximum queue length. Your OS
* may restrict or silently lower this value. A value of zero may cause
* all connections to be rejected.
*
* @since 0.3.0
*
@ -660,9 +662,7 @@ public:
* @since 0.3.0
*/
void start_perpetual() {
m_work = lib::make_shared<lib::asio::io_service::work>(
lib::ref(*m_io_service)
);
m_work.reset(new lib::asio::io_service::work(*m_io_service));
}
/// Clears the endpoint's perpetual flag, allowing it to exit when empty
@ -826,8 +826,7 @@ protected:
// Create a resolver
if (!m_resolver) {
m_resolver = lib::make_shared<lib::asio::ip::tcp::resolver>(
lib::ref(*m_io_service));
m_resolver.reset(new lib::asio::ip::tcp::resolver(*m_io_service));
}
tcon->set_uri(u);

View file

@ -168,8 +168,7 @@ protected:
return socket::make_error_code(socket::error::invalid_state);
}
m_socket = lib::make_shared<lib::asio::ip::tcp::socket>(
lib::ref(*service));
m_socket.reset(new lib::asio::ip::tcp::socket(*service));
m_state = READY;

View file

@ -193,8 +193,7 @@ protected:
if (!m_context) {
return socket::make_error_code(socket::error::invalid_tls_context);
}
m_socket = lib::make_shared<socket_type>(
_WEBSOCKETPP_REF(*service),lib::ref(*m_context));
m_socket.reset(new socket_type(*service, *m_context));
m_io_service = service;
m_strand = strand;