2 // Copyright (c) 2013-2017 Vinnie Falco (vinnie dot falco at gmail dot com)
4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8 #ifndef WEBSOCKET_ASYNC_ECHO_SERVER_HPP
9 #define WEBSOCKET_ASYNC_ECHO_SERVER_HPP
11 #include <beast/core/placeholders.hpp>
12 #include <beast/core/streambuf.hpp>
13 #include <beast/websocket/stream.hpp>
14 #include <boost/lexical_cast.hpp>
15 #include <boost/optional.hpp>
23 #include <type_traits>
25 #include <unordered_map>
30 /** Asynchronous WebSocket echo client/server
32 class async_echo_server
35 using error_code = beast::error_code;
36 using address_type = boost::asio::ip::address;
37 using socket_type = boost::asio::ip::tcp::socket;
38 using endpoint_type = boost::asio::ip::tcp::endpoint;
43 template<class Body, class Fields>
45 operator()(beast::http::message<
46 true, Body, Fields>& req) const
48 req.fields.replace("User-Agent", "async_echo_client");
51 template<class Body, class Fields>
53 operator()(beast::http::message<
54 false, Body, Fields>& resp) const
56 resp.fields.replace("Server", "async_echo_server");
60 /** A container of type-erased option setters.
62 template<class NextLayer>
65 // workaround for std::function bug in msvc
68 virtual ~callable() = default;
69 virtual void operator()(
70 beast::websocket::stream<NextLayer>&) = 0;
74 class callable_impl : public callable
81 : t_(std::forward<U>(u))
86 operator()(beast::websocket::stream<NextLayer>& ws)
98 lambda(lambda&&) = default;
99 lambda(lambda const&) = default;
101 lambda(Opt const& opt)
107 operator()(beast::websocket::stream<NextLayer>& ws) const
113 std::unordered_map<std::type_index,
114 std::unique_ptr<callable>> list_;
119 set_option(Opt const& opt)
121 std::unique_ptr<callable> p;
122 p.reset(new callable_impl<lambda<Opt>>{opt});
123 list_[std::type_index{
124 typeid(Opt)}] = std::move(p);
128 set_options(beast::websocket::stream<NextLayer>& ws)
130 for(auto const& op : list_)
136 boost::asio::io_service ios_;
139 boost::asio::ip::tcp::acceptor acceptor_;
140 std::vector<std::thread> thread_;
141 boost::optional<boost::asio::io_service::work> work_;
142 options_set<socket_type> opts_;
145 async_echo_server(async_echo_server const&) = delete;
146 async_echo_server& operator=(async_echo_server const&) = delete;
150 @param log A pointer to a stream to log to, or `nullptr`
153 @param threads The number of threads in the io_service.
155 async_echo_server(std::ostream* log,
163 beast::websocket::decorate(identity{}));
164 thread_.reserve(threads);
165 for(std::size_t i = 0; i < threads; ++i)
166 thread_.emplace_back(
177 [&]{ acceptor_.close(ec); });
178 for(auto& t : thread_)
182 /** Return the listening endpoint.
185 local_endpoint() const
187 return acceptor_.local_endpoint();
190 /** Set a websocket option.
192 The option will be applied to all new connections.
194 @param opt The option to apply.
198 set_option(Opt const& opt)
200 opts_.set_option(opt);
203 /** Open a listening port.
205 @param ep The address and port to bind to.
207 @param ec Set to the error, if any occurred.
210 open(endpoint_type const& ep, error_code& ec)
212 acceptor_.open(ep.protocol(), ec);
214 return fail("open", ec);
215 acceptor_.set_option(
216 boost::asio::socket_base::reuse_address{true});
217 acceptor_.bind(ep, ec);
219 return fail("bind", ec);
221 boost::asio::socket_base::max_connections, ec);
223 return fail("listen", ec);
224 acceptor_.async_accept(sock_, ep_,
225 std::bind(&async_echo_server::on_accept, this,
226 beast::asio::placeholders::error));
234 async_echo_server& server;
237 beast::websocket::stream<socket_type> ws;
238 boost::asio::io_service::strand strand;
239 beast::websocket::opcode op;
243 data(async_echo_server& server_,
244 endpoint_type const& ep_,
248 , ws(std::move(sock_))
249 , strand(ws.get_io_service())
252 static std::atomic<std::size_t> n{0};
259 // VFALCO This could be unique_ptr in [Net.TS]
260 std::shared_ptr<data> d_;
263 peer(peer&&) = default;
264 peer(peer const&) = default;
265 peer& operator=(peer&&) = delete;
266 peer& operator=(peer const&) = delete;
268 template<class... Args>
270 peer(async_echo_server& server,
271 endpoint_type const& ep, socket_type&& sock,
273 : d_(std::make_shared<data>(server, ep,
274 std::forward<socket_type>(sock),
275 std::forward<Args>(args)...))
278 d.server.opts_.set_options(d.ws);
285 d.ws.async_accept(std::move(*this));
288 void operator()(error_code ec, std::size_t)
293 void operator()(error_code ec)
295 using boost::asio::buffer;
296 using boost::asio::buffer_copy;
303 return fail("async_accept", ec);
308 return fail("async_handshake", ec);
309 d.db.consume(d.db.size());
312 d.ws.async_read(d.op, d.db,
313 d.strand.wrap(std::move(*this)));
318 if(ec == beast::websocket::error::closed)
321 return fail("async_read", ec);
325 beast::websocket::message_type(d.op));
326 d.ws.async_write(d.db.data(),
327 d.strand.wrap(std::move(*this)));
334 fail(std::string what, error_code ec)
338 if(ec != beast::websocket::error::closed)
339 d.server.fail("[#" + std::to_string(d.id) +
340 " " + boost::lexical_cast<std::string>(d.ep) +
346 fail(std::string what, error_code ec)
351 std::lock_guard<std::mutex> lock{m};
352 (*log_) << what << ": " <<
353 ec.message() << std::endl;
358 on_accept(error_code ec)
360 if(! acceptor_.is_open())
362 if(ec == boost::asio::error::operation_aborted)
366 peer{*this, ep_, std::move(sock_)};
367 acceptor_.async_accept(sock_, ep_,
368 std::bind(&async_echo_server::on_accept, this,
369 beast::asio::placeholders::error));