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 BEAST_WEBSOCKET_ASYNC_ECHO_SERVER_HPP
9 #define BEAST_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 template<class DynamicBuffer, std::size_t N>
291 match(DynamicBuffer& db, char const(&s)[N])
293 using boost::asio::buffer;
294 using boost::asio::buffer_copy;
297 beast::static_string<N-1> t;
299 buffer_copy(buffer(t.data(), t.size()),
307 void operator()(error_code ec, std::size_t)
312 void operator()(error_code ec)
314 using boost::asio::buffer;
315 using boost::asio::buffer_copy;
322 return fail("async_accept", ec);
327 return fail("async_handshake", ec);
328 d.db.consume(d.db.size());
331 d.ws.async_read(d.op, d.db,
332 d.strand.wrap(std::move(*this)));
337 if(ec == beast::websocket::error::closed)
340 return fail("async_read", ec);
341 if(match(d.db, "RAW"))
344 boost::asio::async_write(d.ws.next_layer(),
345 d.db.data(), d.strand.wrap(std::move(*this)));
348 else if(match(d.db, "TEXT"))
352 beast::websocket::message_type{
353 beast::websocket::opcode::text});
355 d.db.data(), d.strand.wrap(std::move(*this)));
358 else if(match(d.db, "PING"))
360 beast::websocket::ping_data payload;
361 d.db.consume(buffer_copy(
362 buffer(payload.data(), payload.size()),
365 d.ws.async_ping(payload,
366 d.strand.wrap(std::move(*this)));
369 else if(match(d.db, "CLOSE"))
373 d.strand.wrap(std::move(*this)));
379 beast::websocket::message_type(d.op));
380 d.ws.async_write(d.db.data(),
381 d.strand.wrap(std::move(*this)));
388 fail(std::string what, error_code ec)
392 if(ec != beast::websocket::error::closed)
393 d.server.fail("[#" + std::to_string(d.id) +
394 " " + boost::lexical_cast<std::string>(d.ep) +
400 fail(std::string what, error_code ec)
405 std::lock_guard<std::mutex> lock{m};
406 (*log_) << what << ": " <<
407 ec.message() << std::endl;
412 on_accept(error_code ec)
414 if(! acceptor_.is_open())
416 if(ec == boost::asio::error::operation_aborted)
420 peer{*this, ep_, std::move(sock_)};
421 acceptor_.async_accept(sock_, ep_,
422 std::bind(&async_echo_server::on_accept, this,
423 beast::asio::placeholders::error));