Peerplays_Blockchain/peerplays-fc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #201 from bitshares/ws-refactory

Browse source 
Websocket code refactory and bugfix
This commit is contained in:
Abit 2020-06-22 16:27:42 +02:00 committed by GitHub
commit 12777b8358
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 352 additions and 348 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
include/fc/network/http/websocket.hpp
View file

@ -48,6 +48,8 @@ namespace fc { namespace http {
typedef std::function<void(const websocket_connection_ptr&)> on_connection_handler; typedef std::function<void(const websocket_connection_ptr&)> on_connection_handler;
// TODO websocket_tls_server and websocket_server have almost the same interface and implementation,
// better refactor to remove duplicate code and to avoid undesired or unnecessary differences
class websocket_server class websocket_server
{ {
public: public:
@ -69,6 +71,8 @@ namespace fc { namespace http {
}; };
// TODO websocket_tls_server and websocket_server have almost the same interface and implementation,
// better refactor to remove duplicate code and to avoid undesired or unnecessary differences
class websocket_tls_server class websocket_tls_server
{ {
public: public:
@ -80,7 +84,12 @@ namespace fc { namespace http {
void on_connection( const on_connection_handler& handler); void on_connection( const on_connection_handler& handler);
void listen( uint16_t port ); void listen( uint16_t port );
void listen( const fc::ip::endpoint& ep ); void listen( const fc::ip::endpoint& ep );
uint16_t get_listening_port();
void start_accept(); void start_accept();
void stop_listening();
void close();
private: private:
friend class detail::websocket_tls_server_impl; friend class detail::websocket_tls_server_impl;
std::unique_ptr<detail::websocket_tls_server_impl> my; std::unique_ptr<detail::websocket_tls_server_impl> my;
@ -101,7 +110,7 @@ namespace fc { namespace http {
private: private:
std::unique_ptr<detail::websocket_client_impl> my; std::unique_ptr<detail::websocket_client_impl> my;
std::unique_ptr<detail::websocket_tls_client_impl> smy; std::unique_ptr<detail::websocket_tls_client_impl> smy;
std::vector<std::pair<std::string,std::string>> headers; fc::http::headers _headers;
}; };
} } } }

689
src/network/http/websocket.cpp
View file

@ -84,23 +84,20 @@ namespace fc { namespace http {
typedef type::elog_type elog_type; typedef type::elog_type elog_type;
typedef type::request_type request_type; typedef type::request_type request_type;
typedef type::response_type response_type; typedef type::response_type response_type;
typedef websocketpp::transport::asio::basic_socket::endpoint typedef websocketpp::transport::asio::basic_socket::endpoint socket_type;
socket_type;
}; };
typedef websocketpp::transport::asio::endpoint<transport_config> typedef websocketpp::transport::asio::endpoint<transport_config> transport_type;
transport_type;
static const long timeout_open_handshake = 0; // permessage_compress extension
struct permessage_deflate_config {};
// permessage_compress extension
struct permessage_deflate_config {};
#ifdef HAS_ZLIB #ifdef HAS_ZLIB
typedef websocketpp::extensions::permessage_deflate::enabled <permessage_deflate_config> permessage_deflate_type; typedef websocketpp::extensions::permessage_deflate::enabled <permessage_deflate_config>
permessage_deflate_type;
#else #else
typedef websocketpp::extensions::permessage_deflate::disabled <permessage_deflate_config> permessage_deflate_type; typedef websocketpp::extensions::permessage_deflate::disabled <permessage_deflate_config>
permessage_deflate_type;
#endif #endif
}; };
struct asio_tls_with_stub_log : public websocketpp::config::asio_tls { struct asio_tls_with_stub_log : public websocketpp::config::asio_tls {
@ -130,10 +127,17 @@ namespace fc { namespace http {
typedef websocketpp::transport::asio::tls_socket::endpoint socket_type; typedef websocketpp::transport::asio::tls_socket::endpoint socket_type;
}; };
typedef websocketpp::transport::asio::endpoint<transport_config> typedef websocketpp::transport::asio::endpoint<transport_config> transport_type;
transport_type;
static const long timeout_open_handshake = 0; // permessage_compress extension
struct permessage_deflate_config {};
#ifdef HAS_ZLIB
typedef websocketpp::extensions::permessage_deflate::enabled <permessage_deflate_config>
permessage_deflate_type;
#else
typedef websocketpp::extensions::permessage_deflate::disabled <permessage_deflate_config>
permessage_deflate_type;
#endif
}; };
struct asio_tls_stub_log : public websocketpp::config::asio_tls { struct asio_tls_stub_log : public websocketpp::config::asio_tls {
typedef asio_tls_stub_log type; typedef asio_tls_stub_log type;
@ -154,21 +158,26 @@ namespace fc { namespace http {
typedef base::rng_type rng_type; typedef base::rng_type rng_type;
struct transport_config : public base::transport_config { struct transport_config : public base::transport_config {
typedef type::concurrency_type concurrency_type; typedef type::concurrency_type concurrency_type;
typedef type::alog_type alog_type; typedef type::alog_type alog_type;
typedef type::elog_type elog_type; typedef type::elog_type elog_type;
typedef type::request_type request_type; typedef type::request_type request_type;
typedef type::response_type response_type; typedef type::response_type response_type;
typedef websocketpp::transport::asio::tls_socket::endpoint socket_type; typedef websocketpp::transport::asio::tls_socket::endpoint socket_type;
}; };
typedef websocketpp::transport::asio::endpoint<transport_config> typedef websocketpp::transport::asio::endpoint<transport_config> transport_type;
transport_type;
};
using websocketpp::connection_hdl; // permessage_compress extension
typedef websocketpp::server<asio_with_stub_log> websocket_server_type; struct permessage_deflate_config {};
typedef websocketpp::server<asio_tls_stub_log> websocket_tls_server_type; #ifdef HAS_ZLIB
typedef websocketpp::extensions::permessage_deflate::enabled <permessage_deflate_config>
permessage_deflate_type;
#else
typedef websocketpp::extensions::permessage_deflate::disabled <permessage_deflate_config>
permessage_deflate_type;
#endif
};
template<typename T> template<typename T>
class websocket_connection_impl : public websocket_connection class websocket_connection_impl : public websocket_connection
@ -227,257 +236,212 @@ namespace fc { namespace http {
typedef websocketpp::lib::shared_ptr<boost::asio::ssl::context> context_ptr; typedef websocketpp::lib::shared_ptr<boost::asio::ssl::context> context_ptr;
class websocket_server_impl using websocketpp::connection_hdl;
template<typename T>
class generic_websocket_server_impl
{
public:
generic_websocket_server_impl( const std::string& forward_header_key )
:_server_thread( fc::thread::current() ), _forward_header_key( forward_header_key )
{
_server.clear_access_channels( websocketpp::log::alevel::all );
_server.init_asio( &fc::asio::default_io_service() );
_server.set_reuse_addr( true );
_server.set_open_handler( [this]( connection_hdl hdl ){
_server_thread.async( [this, hdl](){
auto new_con = std::make_shared<possibly_proxied_websocket_connection<
typename websocketpp::server<T>::connection_ptr>>( _server.get_con_from_hdl(hdl),
_forward_header_key );
_on_connection( _connections[hdl] = new_con );
}).wait();
});
_server.set_message_handler( [this]( connection_hdl hdl,
typename websocketpp::server<T>::message_ptr msg ){
_server_thread.async( [this,hdl,msg](){
auto current_con = _connections.find(hdl);
if( current_con == _connections.end() )
return;
auto payload = msg->get_payload();
std::shared_ptr<websocket_connection> con = current_con->second;
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",con->get_remote_endpoint_string()) ("msg",payload) );
++_pending_messages;
auto f = fc::async([this,con,payload](){
if( _pending_messages )
--_pending_messages;
con->on_message( payload );
});
if( _pending_messages > 100 )
f.wait(); // Note: this is a bit strange, because it forces the server to process all
// 100 pending messages (assuming this message is the last one) before
// trying to accept a new message.
// Ideally the `wait` should be canceled immediately when the number of
// pending messages falls below 100. That said, wait on the whole queue,
// but not wait on one message.
}).wait();
});
_server.set_socket_init_handler( [this]( websocketpp::connection_hdl hdl,
typename websocketpp::server<T>::connection_type::socket_type& s ) {
boost::asio::ip::tcp::no_delay option(true);
s.lowest_layer().set_option(option);
} );
_server.set_http_handler( [this]( connection_hdl hdl ){
_server_thread.async( [this,hdl](){
auto con = _server.get_con_from_hdl(hdl);
auto current_con = std::make_shared<possibly_proxied_websocket_connection<
typename websocketpp::server<T>::connection_ptr>>( con, _forward_header_key );
_on_connection( current_con );
con->defer_http_response(); // Note: this can tie up resources if send_http_response() is not
// called quickly enough
std::string remote_endpoint = current_con->get_remote_endpoint_string();
std::string request_body = con->get_request_body();
wlog( "[HTTP-IN] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",request_body) );
fc::async([current_con, request_body, con, remote_endpoint] {
fc::http::reply response = current_con->on_http(request_body);
ilog( "[HTTP-OUT] ${remote_endpoint} ${status} ${msg}",
("remote_endpoint",remote_endpoint)
("status",response.status)
("msg",response.body_as_string) );
con->set_body( std::move( response.body_as_string ) );
con->set_status( websocketpp::http::status_code::value(response.status) );
con->send_http_response();
current_con->closed();
}, "call on_http");
}).wait();
});
_server.set_close_handler( [this]( connection_hdl hdl ){
_server_thread.async( [this,hdl](){
if( _connections.find(hdl) != _connections.end() )
{
_connections[hdl]->closed();
_connections.erase( hdl );
if( _connections.empty() && _all_connections_closed )
_all_connections_closed->set_value();
}
else
{
wlog( "unknown connection closed" );
}
}).wait();
});
_server.set_fail_handler( [this]( connection_hdl hdl ){
_server_thread.async( [this,hdl](){
if( _connections.find(hdl) != _connections.end() )
{
_connections[hdl]->closed();
_connections.erase( hdl );
if( _connections.empty() && _all_connections_closed )
_all_connections_closed->set_value();
}
else
{
// if the server is shutting down, assume this hdl is the server socket
if( _server_socket_closed )
_server_socket_closed->set_value();
else
wlog( "unknown connection failed" );
}
}).wait();
});
}
virtual ~generic_websocket_server_impl()
{
if( _server.is_listening() )
{
// _server.stop_listening() may trigger the on_fail callback function (the lambda function set by
// _server.set_fail_handler(...) ) for the listening server socket (note: the connection handle
// associated with the server socket is not in our connection map),
// the on_fail callback function may fire (async) a new task which may run really late
// and it will try to access the member variables of this server object,
// so we need to wait for it before destructing this object.
_server_socket_closed = promise<void>::create();
_server.stop_listening();
}
// Note: since _connections can be modified by lambda functions in set_*_handler, which are running
// in other tasks, perhaps we need to wait for them (especially the one in set_open_handler)
// being processed. Otherwise `_all_connections_closed.wait()` may hang.
if( !_connections.empty() )
{
_all_connections_closed = promise<void>::create();
auto cpy_con = _connections;
websocketpp::lib::error_code ec;
for( auto& item : cpy_con )
_server.close( item.first, 0, "server exit", ec );
_all_connections_closed->wait();
}
if( _server_socket_closed )
_server_socket_closed->wait();
}
typedef std::map<connection_hdl, websocket_connection_ptr, std::owner_less<connection_hdl> > con_map;
// Note: std::map is not thread-safe nor task-safe, we may need
// to use a mutex or similar to avoid concurrent access.
con_map _connections; ///< Holds accepted connections
fc::thread& _server_thread; ///< The thread that runs the server
websocketpp::server<T> _server; ///< The server
on_connection_handler _on_connection; ///< A handler to be called when a new connection is accepted
fc::promise<void>::ptr _all_connections_closed; ///< Promise to wait for all connections to be closed
fc::promise<void>::ptr _server_socket_closed; ///< Promise to wait for the server socket to be closed
uint32_t _pending_messages = 0; ///< Number of messages not processed, for rate limiting
std::string _forward_header_key; ///< A header like "X-Forwarded-For" (XFF) with data IP:port
};
class websocket_server_impl : public generic_websocket_server_impl<asio_with_stub_log>
{ {
public: public:
websocket_server_impl( const std::string& forward_header_key ) websocket_server_impl( const std::string& forward_header_key )
:_server_thread( fc::thread::current() ), _forward_header_key(forward_header_key) : generic_websocket_server_impl( forward_header_key )
{ {}
_server.clear_access_channels( websocketpp::log::alevel::all );
_server.init_asio(&fc::asio::default_io_service());
_server.set_reuse_addr(true);
_server.set_open_handler( [&]( connection_hdl hdl ){
_server_thread.async( [this, hdl](){
auto new_con = std::make_shared<possibly_proxied_websocket_connection<
websocket_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl),
_forward_header_key );
_on_connection( _connections[hdl] = new_con );
}).wait();
});
_server.set_message_handler( [&]( connection_hdl hdl, websocket_server_type::message_ptr msg ){
_server_thread.async( [&](){
auto current_con = _connections.find(hdl);
assert( current_con != _connections.end() );
auto payload = msg->get_payload();
std::shared_ptr<websocket_connection> con = current_con->second;
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",con->get_remote_endpoint_string()) ("msg",payload) );
++_pending_messages;
auto f = fc::async([this,con,payload](){
if( _pending_messages )
--_pending_messages;
con->on_message( payload );
});
if( _pending_messages > 100 )
f.wait();
}).wait();
});
_server.set_socket_init_handler( [&](websocketpp::connection_hdl hdl, boost::asio::ip::tcp::socket& s ) { virtual ~websocket_server_impl() {}
boost::asio::ip::tcp::no_delay option(true);
s.lowest_layer().set_option(option);
} );
_server.set_http_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto con = _server.get_con_from_hdl(hdl);
auto current_con = std::make_shared<possibly_proxied_websocket_connection<
websocket_server_type::connection_ptr>>( con, _forward_header_key );
_on_connection( current_con );
con->defer_http_response();
std::string remote_endpoint = current_con->get_remote_endpoint_string();
std::string request_body = con->get_request_body();
wlog( "[HTTP-IN] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",request_body) );
fc::async([current_con, request_body, con, remote_endpoint] {
fc::http::reply response = current_con->on_http(request_body);
ilog( "[HTTP-OUT] ${remote_endpoint} ${status} ${msg}",
("remote_endpoint",remote_endpoint)
("status",response.status)
("msg",response.body_as_string) );
con->set_body( std::move( response.body_as_string ) );
con->set_status( websocketpp::http::status_code::value(response.status) );
con->send_http_response();
current_con->closed();
}, "call on_http");
}).wait();
});
_server.set_close_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
if( _connections.find(hdl) != _connections.end() )
{
_connections[hdl]->closed();
_connections.erase( hdl );
}
else
{
wlog( "unknown connection closed" );
}
if( _connections.empty() && _closed )
_closed->set_value();
}).wait();
});
_server.set_fail_handler( [&]( connection_hdl hdl ){
if( _server.is_listening() )
{
_server_thread.async( [&](){
if( _connections.find(hdl) != _connections.end() )
{
_connections[hdl]->closed();
_connections.erase( hdl );
}
else
{
wlog( "unknown connection failed" );
}
if( _connections.empty() && _closed )
_closed->set_value();
}).wait();
}
});
}
~websocket_server_impl()
{
if( _server.is_listening() )
_server.stop_listening();
if( _connections.size() )
_closed = promise<void>::create();
auto cpy_con = _connections;
for( auto item : cpy_con )
_server.close( item.first, 0, "server exit" );
if( _closed ) _closed->wait();
}
typedef std::map<connection_hdl, websocket_connection_ptr,std::owner_less<connection_hdl> > con_map;
con_map _connections;
fc::thread& _server_thread;
websocket_server_type _server;
on_connection_handler _on_connection;
fc::promise<void>::ptr _closed;
uint32_t _pending_messages = 0;
std::string _forward_header_key; // A header like "X-Forwarded-For" (XFF), with data IP:port
}; };
class websocket_tls_server_impl class websocket_tls_server_impl : public generic_websocket_server_impl<asio_tls_stub_log>
{ {
public: public:
websocket_tls_server_impl( const string& server_pem, const string& ssl_password, websocket_tls_server_impl( const string& server_pem, const string& ssl_password,
const std::string& forward_header_key ) const std::string& forward_header_key )
:_server_thread( fc::thread::current() ), _forward_header_key(forward_header_key) : generic_websocket_server_impl( forward_header_key )
{ {
{ _server.set_tls_init_handler( [this,server_pem,ssl_password]( websocketpp::connection_hdl hdl ) {
_server.set_tls_init_handler( [=]( websocketpp::connection_hdl hdl ) -> context_ptr { context_ptr ctx = websocketpp::lib::make_shared<boost::asio::ssl::context>(
context_ptr ctx = websocketpp::lib::make_shared<boost::asio::ssl::context>(boost::asio::ssl::context::tlsv1); boost::asio::ssl::context::tlsv1 );
try { try {
ctx->set_options(boost::asio::ssl::context::default_workarounds | ctx->set_options( boost::asio::ssl::context::default_workarounds |
boost::asio::ssl::context::no_sslv2 | boost::asio::ssl::context::no_sslv2 |
boost::asio::ssl::context::no_sslv3 | boost::asio::ssl::context::no_sslv3 |
boost::asio::ssl::context::single_dh_use); boost::asio::ssl::context::single_dh_use );
ctx->set_password_callback([=](std::size_t max_length, boost::asio::ssl::context::password_purpose){ return ssl_password;}); ctx->set_password_callback(
ctx->use_certificate_chain_file(server_pem); [ssl_password](std::size_t max_length, boost::asio::ssl::context::password_purpose){
ctx->use_private_key_file(server_pem, boost::asio::ssl::context::pem); return ssl_password;
} catch (std::exception& e) { });
std::cout << e.what() << std::endl; ctx->use_certificate_chain_file(server_pem);
} ctx->use_private_key_file(server_pem, boost::asio::ssl::context::pem);
return ctx; } catch (std::exception& e) {
}); std::cout << e.what() << std::endl;
} }
return ctx;
_server.clear_access_channels( websocketpp::log::alevel::all );
_server.init_asio(&fc::asio::default_io_service());
_server.set_reuse_addr(true);
_server.set_open_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto new_con = std::make_shared<possibly_proxied_websocket_connection<
websocket_tls_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl),
_forward_header_key );
_on_connection( _connections[hdl] = new_con );
}).wait();
});
_server.set_message_handler( [&]( connection_hdl hdl, websocket_server_type::message_ptr msg ){
_server_thread.async( [&](){
auto current_con = _connections.find(hdl);
assert( current_con != _connections.end() );
auto received = msg->get_payload();
std::shared_ptr<websocket_connection> con = current_con->second;
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",con->get_remote_endpoint_string()) ("msg",received) );
fc::async([con,received](){ con->on_message( received ); });
}).wait();
});
_server.set_http_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto con = _server.get_con_from_hdl(hdl);
auto current_con = std::make_shared<possibly_proxied_websocket_connection<
websocket_tls_server_type::connection_ptr>>( con, _forward_header_key );
try{
_on_connection( current_con );
std::string remote_endpoint = current_con->get_remote_endpoint_string();
std::string request_body = con->get_request_body();
wlog( "[HTTP-IN] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",request_body) );
auto response = current_con->on_http( request_body );
ilog( "[HTTP-OUT] ${remote_endpoint} ${status} ${msg}",
("remote_endpoint",remote_endpoint)
("status",response.status)
("msg",response.body_as_string) );
con->set_body( std::move( response.body_as_string ) );
con->set_status( websocketpp::http::status_code::value( response.status ) );
} catch ( const fc::exception& e )
{
edump((e.to_detail_string()));
}
current_con->closed();
}).wait();
});
_server.set_close_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
_connections[hdl]->closed();
_connections.erase( hdl );
}).wait();
});
_server.set_fail_handler( [&]( connection_hdl hdl ){
if( _server.is_listening() )
{
_server_thread.async( [&](){
if( _connections.find(hdl) != _connections.end() )
{
_connections[hdl]->closed();
_connections.erase( hdl );
}
}).wait();
}
}); });
} }
~websocket_tls_server_impl() virtual ~websocket_tls_server_impl() {}
{
if( _server.is_listening() )
_server.stop_listening();
auto cpy_con = _connections;
for( auto item : cpy_con )
_server.close( item.first, 0, "server exit" );
}
typedef std::map<connection_hdl, websocket_connection_ptr,std::owner_less<connection_hdl> > con_map;
con_map _connections;
fc::thread& _server_thread;
websocket_tls_server_type _server;
on_connection_handler _on_connection;
fc::promise<void>::ptr _closed;
std::string _forward_header_key; // A header like "X-Forwarded-For" (XFF), with data IP:port
}; };
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;
template<typename T> template<typename T>
class generic_websocket_client_impl class generic_websocket_client_impl
@ -487,34 +451,51 @@ namespace fc { namespace http {
:_client_thread( fc::thread::current() ) :_client_thread( fc::thread::current() )
{ {
_client.clear_access_channels( websocketpp::log::alevel::all ); _client.clear_access_channels( websocketpp::log::alevel::all );
_client.set_message_handler( [&]( connection_hdl hdl, _client.set_message_handler( [this]( connection_hdl hdl,
typename websocketpp::client<T>::message_ptr msg ){ typename websocketpp::client<T>::message_ptr msg ){
_client_thread.async( [&](){ _client_thread.async( [this,msg](){
wdump((msg->get_payload())); wdump((msg->get_payload()));
auto received = msg->get_payload(); auto received = msg->get_payload();
fc::async( [=](){ fc::async( [this,received](){
if( _connection ) if( _connection )
_connection->on_message(received); _connection->on_message(received);
}); });
}).wait(); }).wait();
}); });
_client.set_close_handler( [=]( connection_hdl hdl ){ _client.set_close_handler( [this]( connection_hdl hdl ){
_client_thread.async( [&](){ if( _connection ) {_connection->closed(); _connection.reset();} } ).wait(); _client_thread.async( [this](){
if( _closed ) _closed->set_value(); if( _connection ) {
_connection->closed();
_connection.reset();
}
} ).wait();
if( _closed )
_closed->set_value();
}); });
_client.set_fail_handler( [=]( connection_hdl hdl ){ _client.set_fail_handler( [this]( connection_hdl hdl ){
auto con = _client.get_con_from_hdl(hdl); auto con = _client.get_con_from_hdl(hdl);
auto message = con->get_ec().message(); auto message = con->get_ec().message();
if( _connection ) if( _connection )
_client_thread.async( [&](){ if( _connection ) _connection->closed(); _connection.reset(); } ).wait(); {
_client_thread.async( [this](){
if( _connection ) {
_connection->closed();
_connection.reset();
}
} ).wait();
}
if( _connected && !_connected->ready() ) if( _connected && !_connected->ready() )
_connected->set_exception( exception_ptr( new FC_EXCEPTION( exception, "${message}", ("message",message)) ) ); {
_connected->set_exception( exception_ptr(
new FC_EXCEPTION( exception, "${message}", ("message",message)) ) );
}
if( _closed ) if( _closed )
_closed->set_value(); _closed->set_value();
}); });
_client.init_asio( &fc::asio::default_io_service() ); _client.init_asio( &fc::asio::default_io_service() );
} }
virtual ~generic_websocket_client_impl() virtual ~generic_websocket_client_impl()
{ {
if( _connection ) if( _connection )
@ -525,6 +506,37 @@ namespace fc { namespace http {
if( _closed ) if( _closed )
_closed->wait(); _closed->wait();
} }
websocket_connection_ptr connect( const std::string& uri, const fc::http::headers& headers )
{
websocketpp::lib::error_code ec;
_uri = uri;
_connected = promise<void>::create("websocket::connect");
_client.set_open_handler( [this]( websocketpp::connection_hdl hdl ){
_hdl = hdl;
auto con = _client.get_con_from_hdl(hdl);
_connection = std::make_shared<websocket_connection_impl<
typename websocketpp::client<T>::connection_ptr>>( con );
_closed = promise<void>::create("websocket::closed");
_connected->set_value();
});
auto con = _client.get_connection( uri, ec );
for( const fc::http::header& h : headers )
{
con->append_header( h.key, h.val );
}
FC_ASSERT( !ec, "error: ${e}", ("e",ec.message()) );
_client.connect(con);
_connected->wait();
return _connection;
}
fc::promise<void>::ptr _connected; fc::promise<void>::ptr _connected;
fc::promise<void>::ptr _closed; fc::promise<void>::ptr _closed;
fc::thread& _client_thread; fc::thread& _client_thread;
@ -548,13 +560,18 @@ namespace fc { namespace http {
// "_default" uses default CA's provided by OS // "_default" uses default CA's provided by OS
// //
// We need ca_filename to be copied into the closure, as the referenced object might be destroyed by the caller by the time // We need ca_filename to be copied into the closure, as the
// tls_init_handler() is called. According to [1], capture-by-value results in the desired behavior (i.e. creation of // referenced object might be destroyed by the caller by the time
// a copy which is stored in the closure) on standards compliant compilers, but some compilers on some optimization levels // tls_init_handler() is called. According to [1], capture-by-value
// are buggy and are not standards compliant in this situation. Also, keep in mind this is the opinion of a single forum // results in the desired behavior (i.e. creation of
// a copy which is stored in the closure) on standards compliant compilers,
// but some compilers on some optimization levels
// are buggy and are not standards compliant in this situation.
// Also, keep in mind this is the opinion of a single forum
// poster and might be wrong. // poster and might be wrong.
// //
// To be safe, the following line explicitly creates a non-reference string which is captured by value, which should have the // To be safe, the following line explicitly creates a non-reference string
// which is captured by value, which should have the
// correct behavior on all compilers. // correct behavior on all compilers.
// //
// [1] http://www.cplusplus.com/forum/general/142165/ // [1] http://www.cplusplus.com/forum/general/142165/
@ -563,13 +580,14 @@ namespace fc { namespace http {
std::string ca_filename_copy = ca_filename; std::string ca_filename_copy = ca_filename;
_client.set_tls_init_handler( [=](websocketpp::connection_hdl) { _client.set_tls_init_handler( [this,ca_filename_copy](websocketpp::connection_hdl) {
context_ptr ctx = websocketpp::lib::make_shared<boost::asio::ssl::context>(boost::asio::ssl::context::tlsv1); context_ptr ctx = websocketpp::lib::make_shared<boost::asio::ssl::context>(
boost::asio::ssl::context::tlsv1);
try { try {
ctx->set_options(boost::asio::ssl::context::default_workarounds | ctx->set_options( boost::asio::ssl::context::default_workarounds |
boost::asio::ssl::context::no_sslv2 | boost::asio::ssl::context::no_sslv2 |
boost::asio::ssl::context::no_sslv3 | boost::asio::ssl::context::no_sslv3 |
boost::asio::ssl::context::single_dh_use); boost::asio::ssl::context::single_dh_use );
setup_peer_verify( ctx, ca_filename_copy ); setup_peer_verify( ctx, ca_filename_copy );
} catch (std::exception& e) { } catch (std::exception& e) {
@ -626,28 +644,30 @@ namespace fc { namespace http {
} }
void websocket_server::listen( const fc::ip::endpoint& ep ) void websocket_server::listen( const fc::ip::endpoint& ep )
{ {
my->_server.listen( boost::asio::ip::tcp::endpoint( boost::asio::ip::address_v4(uint32_t(ep.get_address())),ep.port()) ); my->_server.listen( boost::asio::ip::tcp::endpoint(
boost::asio::ip::address_v4(uint32_t(ep.get_address())),ep.port()) );
} }
uint16_t websocket_server::get_listening_port() uint16_t websocket_server::get_listening_port()
{ {
websocketpp::lib::asio::error_code ec; websocketpp::lib::asio::error_code ec;
return my->_server.get_local_endpoint(ec).port(); return my->_server.get_local_endpoint(ec).port();
} }
void websocket_server::start_accept() { void websocket_server::start_accept() {
my->_server.start_accept(); my->_server.start_accept();
} }
void websocket_server::stop_listening() void websocket_server::stop_listening()
{ {
my->_server.stop_listening(); my->_server.stop_listening();
} }
void websocket_server::close() void websocket_server::close()
{ {
for (auto& connection : my->_connections) websocketpp::lib::error_code ec;
my->_server.close(connection.first, websocketpp::close::status::normal, "Goodbye"); for( auto& connection : my->_connections )
my->_server.close( connection.first, websocketpp::close::status::normal, "Goodbye", ec );
} }
websocket_tls_server::websocket_tls_server( const string& server_pem, const string& ssl_password, websocket_tls_server::websocket_tls_server( const string& server_pem, const string& ssl_password,
@ -668,13 +688,32 @@ namespace fc { namespace http {
} }
void websocket_tls_server::listen( const fc::ip::endpoint& ep ) void websocket_tls_server::listen( const fc::ip::endpoint& ep )
{ {
my->_server.listen( boost::asio::ip::tcp::endpoint( boost::asio::ip::address_v4(uint32_t(ep.get_address())),ep.port()) ); my->_server.listen( boost::asio::ip::tcp::endpoint(
boost::asio::ip::address_v4(uint32_t(ep.get_address())),ep.port()) );
}
uint16_t websocket_tls_server::get_listening_port()
{
websocketpp::lib::asio::error_code ec;
return my->_server.get_local_endpoint(ec).port();
} }
void websocket_tls_server::start_accept() { void websocket_tls_server::start_accept() {
my->_server.start_accept(); my->_server.start_accept();
} }
void websocket_tls_server::stop_listening()
{
my->_server.stop_listening();
}
void websocket_tls_server::close()
{
websocketpp::lib::error_code ec;
for( auto& connection : my->_connections )
my->_server.close( connection.first, websocketpp::close::status::normal, "Goodbye", ec );
}
websocket_client::websocket_client( const std::string& ca_filename ) websocket_client::websocket_client( const std::string& ca_filename )
:my( new detail::websocket_client_impl() ), :my( new detail::websocket_client_impl() ),
@ -685,83 +724,39 @@ namespace fc { namespace http {
websocket_connection_ptr websocket_client::connect( const std::string& uri ) websocket_connection_ptr websocket_client::connect( const std::string& uri )
{ try { { try {
if( uri.substr(0,4) == "wss:" )
return secure_connect(uri);
FC_ASSERT( uri.substr(0,3) == "ws:" );
websocketpp::lib::error_code ec; FC_ASSERT( uri.substr(0,4) == "wss:" || uri.substr(0,3) == "ws:", "Unsupported protocol" );
my->_uri = uri; // WSS
my->_connected = promise<void>::create("websocket::connect"); if( uri.substr(0,4) == "wss:" )
return smy->connect( uri, _headers );
my->_client.set_open_handler( [=]( websocketpp::connection_hdl hdl ){ // WS
my->_hdl = hdl; return my->connect( uri, _headers );
auto con = my->_client.get_con_from_hdl(hdl);
my->_connection = std::make_shared<detail::websocket_connection_impl<
detail::websocket_client_connection_type>>( con );
my->_closed = promise<void>::create("websocket::closed");
my->_connected->set_value();
});
auto con = my->_client.get_connection( uri, ec );
std::for_each(headers.begin(), headers.end(), [con](std::pair<std::string, std::string> in) {
con->append_header(in.first, in.second);
});
if( ec ) FC_ASSERT( !ec, "error: ${e}", ("e",ec.message()) );
my->_client.connect(con);
my->_connected->wait();
return my->_connection;
} FC_CAPTURE_AND_RETHROW( (uri) ) } } FC_CAPTURE_AND_RETHROW( (uri) ) }
// TODO most code in this function is same as ::connect, best refactor
websocket_connection_ptr websocket_client::secure_connect( const std::string& uri ) websocket_connection_ptr websocket_client::secure_connect( const std::string& uri )
{ try { {
if( uri.substr(0,3) == "ws:" ) return connect( uri );
return connect(uri); }
FC_ASSERT( uri.substr(0,4) == "wss:" );
websocketpp::lib::error_code ec;
smy->_uri = uri;
smy->_connected = promise<void>::create("websocket::connect");
smy->_client.set_open_handler( [=]( websocketpp::connection_hdl hdl ){
smy->_hdl = hdl;
auto con = smy->_client.get_con_from_hdl(hdl);
smy->_connection = std::make_shared<detail::websocket_connection_impl<
detail::websocket_tls_client_connection_type>>( con );
smy->_closed = promise<void>::create("websocket::closed");
smy->_connected->set_value();
});
auto con = smy->_client.get_connection( uri, ec );
std::for_each(headers.begin(), headers.end(), [con](std::pair<std::string, std::string> in) {
con->append_header(in.first, in.second);
});
if( ec )
FC_ASSERT( !ec, "error: ${e}", ("e",ec.message()) );
smy->_client.connect(con);
smy->_connected->wait();
return smy->_connection;
} FC_CAPTURE_AND_RETHROW( (uri) ) }
void websocket_client::close() void websocket_client::close()
{ {
if (my->_hdl) if( my->_hdl )
my->_client.close(*my->_hdl, websocketpp::close::status::normal, "Goodbye"); my->_client.close( *my->_hdl, websocketpp::close::status::normal, "Goodbye" );
} }
void websocket_client::synchronous_close() void websocket_client::synchronous_close()
{ {
close(); close();
if (my->_closed) if( my->_closed )
my->_closed->wait(); my->_closed->wait();
} }
void websocket_client::append_header(const std::string& key, const std::string& value) void websocket_client::append_header(const std::string& key, const std::string& value)
{ {
headers.push_back( std::make_pair(key, value) ); _headers.emplace_back( key, value );
} }
} } // fc::http } } // fc::http