mirror of
https://github.com/XRPLF/rippled.git
synced 2025-11-21 11:35:53 +00:00
Merge commit 'c652cf066d0b43c7c5bc10b4d56ff99a867e7873' into develop
This commit is contained in:
Vinnie Falco
326
src/beast/examples/websocket_sync_echo_server.hpp
Normal file
326
src/beast/examples/websocket_sync_echo_server.hpp
Normal file
@@ -0,0 +1,326 @@
|
||||
//
|
||||
// Copyright (c) 2013-2016 Vinnie Falco (vinnie dot falco at gmail dot com)
|
||||
//
|
||||
// Distributed under the Boost Software License, Version 1.0. (See accompanying
|
||||
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
|
||||
//
|
||||
|
||||
#ifndef WEBSOCKET_SYNC_ECHO_SERVER_HPP
|
||||
#define WEBSOCKET_SYNC_ECHO_SERVER_HPP
|
||||
|
||||
#include <beast/core/placeholders.hpp>
|
||||
#include <beast/core/streambuf.hpp>
|
||||
#include <beast/websocket.hpp>
|
||||
#include <boost/lexical_cast.hpp>
|
||||
#include <boost/optional.hpp>
|
||||
#include <atomic>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <ostream>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <type_traits>
|
||||
#include <typeindex>
|
||||
#include <unordered_map>
|
||||
#include <utility>
|
||||
|
||||
namespace websocket {
|
||||
|
||||
/** Synchronous WebSocket echo client/server
|
||||
*/
|
||||
class sync_echo_server
|
||||
{
|
||||
public:
|
||||
using error_code = beast::error_code;
|
||||
using endpoint_type = boost::asio::ip::tcp::endpoint;
|
||||
using address_type = boost::asio::ip::address;
|
||||
using socket_type = boost::asio::ip::tcp::socket;
|
||||
|
||||
private:
|
||||
struct identity
|
||||
{
|
||||
template<class Body, class Fields>
|
||||
void
|
||||
operator()(beast::http::message<
|
||||
true, Body, Fields>& req) const
|
||||
{
|
||||
req.fields.replace("User-Agent", "sync_echo_client");
|
||||
}
|
||||
|
||||
template<class Body, class Fields>
|
||||
void
|
||||
operator()(beast::http::message<
|
||||
false, Body, Fields>& resp) const
|
||||
{
|
||||
resp.fields.replace("Server", "sync_echo_server");
|
||||
}
|
||||
};
|
||||
|
||||
/** A container of type-erased option setters.
|
||||
*/
|
||||
template<class NextLayer>
|
||||
class options_set
|
||||
{
|
||||
// workaround for std::function bug in msvc
|
||||
struct callable
|
||||
{
|
||||
virtual ~callable() = default;
|
||||
virtual void operator()(
|
||||
beast::websocket::stream<NextLayer>&) = 0;
|
||||
};
|
||||
|
||||
template<class T>
|
||||
class callable_impl : public callable
|
||||
{
|
||||
T t_;
|
||||
|
||||
public:
|
||||
template<class U>
|
||||
callable_impl(U&& u)
|
||||
: t_(std::forward<U>(u))
|
||||
{
|
||||
}
|
||||
|
||||
void
|
||||
operator()(beast::websocket::stream<NextLayer>& ws)
|
||||
{
|
||||
t_(ws);
|
||||
}
|
||||
};
|
||||
|
||||
template<class Opt>
|
||||
class lambda
|
||||
{
|
||||
Opt opt_;
|
||||
|
||||
public:
|
||||
lambda(lambda&&) = default;
|
||||
lambda(lambda const&) = default;
|
||||
|
||||
lambda(Opt const& opt)
|
||||
: opt_(opt)
|
||||
{
|
||||
}
|
||||
|
||||
void
|
||||
operator()(beast::websocket::stream<NextLayer>& ws) const
|
||||
{
|
||||
ws.set_option(opt_);
|
||||
}
|
||||
};
|
||||
|
||||
std::unordered_map<std::type_index,
|
||||
std::unique_ptr<callable>> list_;
|
||||
|
||||
public:
|
||||
template<class Opt>
|
||||
void
|
||||
set_option(Opt const& opt)
|
||||
{
|
||||
std::unique_ptr<callable> p;
|
||||
p.reset(new callable_impl<lambda<Opt>>{opt});
|
||||
list_[std::type_index{
|
||||
typeid(Opt)}] = std::move(p);
|
||||
}
|
||||
|
||||
void
|
||||
set_options(beast::websocket::stream<NextLayer>& ws)
|
||||
{
|
||||
for(auto const& op : list_)
|
||||
(*op.second)(ws);
|
||||
}
|
||||
};
|
||||
|
||||
std::ostream* log_;
|
||||
boost::asio::io_service ios_;
|
||||
socket_type sock_;
|
||||
endpoint_type ep_;
|
||||
boost::asio::ip::tcp::acceptor acceptor_;
|
||||
std::thread thread_;
|
||||
options_set<socket_type> opts_;
|
||||
|
||||
public:
|
||||
/** Constructor.
|
||||
|
||||
@param log A pointer to a stream to log to, or `nullptr`
|
||||
to disable logging.
|
||||
*/
|
||||
sync_echo_server(std::ostream* log)
|
||||
: log_(log)
|
||||
, sock_(ios_)
|
||||
, acceptor_(ios_)
|
||||
{
|
||||
opts_.set_option(
|
||||
beast::websocket::decorate(identity{}));
|
||||
}
|
||||
|
||||
/** Destructor.
|
||||
*/
|
||||
~sync_echo_server()
|
||||
{
|
||||
if(thread_.joinable())
|
||||
{
|
||||
error_code ec;
|
||||
ios_.dispatch(
|
||||
[&]{ acceptor_.close(ec); });
|
||||
thread_.join();
|
||||
}
|
||||
}
|
||||
|
||||
/** Return the listening endpoint.
|
||||
*/
|
||||
endpoint_type
|
||||
local_endpoint() const
|
||||
{
|
||||
return acceptor_.local_endpoint();
|
||||
}
|
||||
|
||||
/** Set a websocket option.
|
||||
|
||||
The option will be applied to all new connections.
|
||||
|
||||
@param opt The option to apply.
|
||||
*/
|
||||
template<class Opt>
|
||||
void
|
||||
set_option(Opt const& opt)
|
||||
{
|
||||
opts_.set_option(opt);
|
||||
}
|
||||
|
||||
/** Open a listening port.
|
||||
|
||||
@param ep The address and port to bind to.
|
||||
|
||||
@param ec Set to the error, if any occurred.
|
||||
*/
|
||||
void
|
||||
open(endpoint_type const& ep, error_code& ec)
|
||||
{
|
||||
acceptor_.open(ep.protocol(), ec);
|
||||
if(ec)
|
||||
return fail("open", ec);
|
||||
acceptor_.set_option(
|
||||
boost::asio::socket_base::reuse_address{true});
|
||||
acceptor_.bind(ep, ec);
|
||||
if(ec)
|
||||
return fail("bind", ec);
|
||||
acceptor_.listen(
|
||||
boost::asio::socket_base::max_connections, ec);
|
||||
if(ec)
|
||||
return fail("listen", ec);
|
||||
acceptor_.async_accept(sock_, ep_,
|
||||
std::bind(&sync_echo_server::on_accept, this,
|
||||
beast::asio::placeholders::error));
|
||||
thread_ = std::thread{[&]{ ios_.run(); }};
|
||||
}
|
||||
|
||||
private:
|
||||
void
|
||||
fail(std::string what, error_code ec)
|
||||
{
|
||||
if(log_)
|
||||
{
|
||||
static std::mutex m;
|
||||
std::lock_guard<std::mutex> lock{m};
|
||||
(*log_) << what << ": " <<
|
||||
ec.message() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
fail(std::string what, error_code ec,
|
||||
int id, endpoint_type const& ep)
|
||||
{
|
||||
if(log_)
|
||||
if(ec != beast::websocket::error::closed)
|
||||
fail("[#" + std::to_string(id) + " " +
|
||||
boost::lexical_cast<std::string>(ep) +
|
||||
"] " + what, ec);
|
||||
}
|
||||
|
||||
void
|
||||
on_accept(error_code ec)
|
||||
{
|
||||
if(ec == boost::asio::error::operation_aborted)
|
||||
return;
|
||||
if(ec)
|
||||
return fail("accept", ec);
|
||||
struct lambda
|
||||
{
|
||||
std::size_t id;
|
||||
endpoint_type ep;
|
||||
sync_echo_server& self;
|
||||
boost::asio::io_service::work work;
|
||||
// Must be destroyed before work otherwise the
|
||||
// io_service could be destroyed before the socket.
|
||||
socket_type sock;
|
||||
|
||||
lambda(sync_echo_server& self_,
|
||||
endpoint_type const& ep_,
|
||||
socket_type&& sock_)
|
||||
: id([]
|
||||
{
|
||||
static std::atomic<std::size_t> n{0};
|
||||
return ++n;
|
||||
}())
|
||||
, ep(ep_)
|
||||
, self(self_)
|
||||
, work(sock_.get_io_service())
|
||||
, sock(std::move(sock_))
|
||||
{
|
||||
}
|
||||
|
||||
void operator()()
|
||||
{
|
||||
self.do_peer(id, ep, std::move(sock));
|
||||
}
|
||||
};
|
||||
std::thread{lambda{*this, ep_, std::move(sock_)}}.detach();
|
||||
acceptor_.async_accept(sock_, ep_,
|
||||
std::bind(&sync_echo_server::on_accept, this,
|
||||
beast::asio::placeholders::error));
|
||||
}
|
||||
|
||||
void
|
||||
do_peer(std::size_t id,
|
||||
endpoint_type const& ep, socket_type&& sock)
|
||||
{
|
||||
using boost::asio::buffer;
|
||||
using boost::asio::buffer_copy;
|
||||
beast::websocket::stream<
|
||||
socket_type> ws{std::move(sock)};
|
||||
opts_.set_options(ws);
|
||||
error_code ec;
|
||||
ws.accept(ec);
|
||||
if(ec)
|
||||
{
|
||||
fail("accept", ec, id, ep);
|
||||
return;
|
||||
}
|
||||
for(;;)
|
||||
{
|
||||
beast::websocket::opcode op;
|
||||
beast::streambuf sb;
|
||||
ws.read(op, sb, ec);
|
||||
if(ec)
|
||||
{
|
||||
auto const s = ec.message();
|
||||
break;
|
||||
}
|
||||
ws.set_option(beast::websocket::message_type{op});
|
||||
ws.write(sb.data(), ec);
|
||||
if(ec)
|
||||
break;
|
||||
}
|
||||
if(ec && ec != beast::websocket::error::closed)
|
||||
{
|
||||
fail("read", ec, id, ep);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
} // websocket
|
||||
|
||||
#endif
|
||||
Reference in New Issue
Block a user