diff options
author | 2016-10-07 14:13:19 +0200 | |
---|---|---|
committer | 2016-10-07 14:13:19 +0200 | |
commit | ff01b716711b97c2fcaa709ea97f7650f106aa10 (patch) | |
tree | 50dd4d687f38f50c4e136af030c02c267c769f3a /3rdparty/asio/src/examples/cpp03/timeouts | |
parent | 2a138159c30457aecd9e9679be5159704db0f954 (diff) |
Added ASIO networking library (nw)
Diffstat (limited to '3rdparty/asio/src/examples/cpp03/timeouts')
5 files changed, 1170 insertions, 0 deletions
diff --git a/3rdparty/asio/src/examples/cpp03/timeouts/.gitignore b/3rdparty/asio/src/examples/cpp03/timeouts/.gitignore new file mode 100644 index 00000000000..dda7f39c3c2 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp03/timeouts/.gitignore @@ -0,0 +1,11 @@ +.deps +.dirstamp +*.o +*.obj +*.exe +*_client +server +*.ilk +*.manifest +*.pdb +*.tds diff --git a/3rdparty/asio/src/examples/cpp03/timeouts/async_tcp_client.cpp b/3rdparty/asio/src/examples/cpp03/timeouts/async_tcp_client.cpp new file mode 100644 index 00000000000..98b90e9a1b3 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp03/timeouts/async_tcp_client.cpp @@ -0,0 +1,308 @@ +// +// async_tcp_client.cpp +// ~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff 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) +// + +#include "asio/deadline_timer.hpp" +#include "asio/io_context.hpp" +#include "asio/ip/tcp.hpp" +#include "asio/read_until.hpp" +#include "asio/streambuf.hpp" +#include "asio/write.hpp" +#include <boost/bind.hpp> +#include <iostream> + +using asio::deadline_timer; +using asio::ip::tcp; + +// +// This class manages socket timeouts by applying the concept of a deadline. +// Some asynchronous operations are given deadlines by which they must complete. +// Deadlines are enforced by an "actor" that persists for the lifetime of the +// client object: +// +// +----------------+ +// | | +// | check_deadline |<---+ +// | | | +// +----------------+ | async_wait() +// | | +// +---------+ +// +// If the deadline actor determines that the deadline has expired, the socket +// is closed and any outstanding operations are consequently cancelled. +// +// Connection establishment involves trying each endpoint in turn until a +// connection is successful, or the available endpoints are exhausted. If the +// deadline actor closes the socket, the connect actor is woken up and moves to +// the next endpoint. +// +// +---------------+ +// | | +// | start_connect |<---+ +// | | | +// +---------------+ | +// | | +// async_- | +----------------+ +// connect() | | | +// +--->| handle_connect | +// | | +// +----------------+ +// : +// Once a connection is : +// made, the connect : +// actor forks in two - : +// : +// an actor for reading : and an actor for +// inbound messages: : sending heartbeats: +// : +// +------------+ : +-------------+ +// | |<- - - - -+- - - - ->| | +// | start_read | | start_write |<---+ +// | |<---+ | | | +// +------------+ | +-------------+ | async_wait() +// | | | | +// async_- | +-------------+ async_- | +--------------+ +// read_- | | | write() | | | +// until() +--->| handle_read | +--->| handle_write | +// | | | | +// +-------------+ +--------------+ +// +// The input actor reads messages from the socket, where messages are delimited +// by the newline character. The deadline for a complete message is 30 seconds. +// +// The heartbeat actor sends a heartbeat (a message that consists of a single +// newline character) every 10 seconds. In this example, no deadline is applied +// message sending. +// +class client +{ +public: + client(asio::io_context& io_context) + : stopped_(false), + socket_(io_context), + deadline_(io_context), + heartbeat_timer_(io_context) + { + } + + // Called by the user of the client class to initiate the connection process. + // The endpoints will have been obtained using a tcp::resolver. + void start(tcp::resolver::results_type endpoints) + { + // Start the connect actor. + endpoints_ = endpoints; + start_connect(endpoints_.begin()); + + // Start the deadline actor. You will note that we're not setting any + // particular deadline here. Instead, the connect and input actors will + // update the deadline prior to each asynchronous operation. + deadline_.async_wait(boost::bind(&client::check_deadline, this)); + } + + // This function terminates all the actors to shut down the connection. It + // may be called by the user of the client class, or by the class itself in + // response to graceful termination or an unrecoverable error. + void stop() + { + stopped_ = true; + asio::error_code ignored_ec; + socket_.close(ignored_ec); + deadline_.cancel(); + heartbeat_timer_.cancel(); + } + +private: + void start_connect(tcp::resolver::results_type::iterator endpoint_iter) + { + if (endpoint_iter != endpoints_.end()) + { + std::cout << "Trying " << endpoint_iter->endpoint() << "...\n"; + + // Set a deadline for the connect operation. + deadline_.expires_from_now(boost::posix_time::seconds(60)); + + // Start the asynchronous connect operation. + socket_.async_connect(endpoint_iter->endpoint(), + boost::bind(&client::handle_connect, + this, _1, endpoint_iter)); + } + else + { + // There are no more endpoints to try. Shut down the client. + stop(); + } + } + + void handle_connect(const asio::error_code& ec, + tcp::resolver::results_type::iterator endpoint_iter) + { + if (stopped_) + return; + + // The async_connect() function automatically opens the socket at the start + // of the asynchronous operation. If the socket is closed at this time then + // the timeout handler must have run first. + if (!socket_.is_open()) + { + std::cout << "Connect timed out\n"; + + // Try the next available endpoint. + start_connect(++endpoint_iter); + } + + // Check if the connect operation failed before the deadline expired. + else if (ec) + { + std::cout << "Connect error: " << ec.message() << "\n"; + + // We need to close the socket used in the previous connection attempt + // before starting a new one. + socket_.close(); + + // Try the next available endpoint. + start_connect(++endpoint_iter); + } + + // Otherwise we have successfully established a connection. + else + { + std::cout << "Connected to " << endpoint_iter->endpoint() << "\n"; + + // Start the input actor. + start_read(); + + // Start the heartbeat actor. + start_write(); + } + } + + void start_read() + { + // Set a deadline for the read operation. + deadline_.expires_from_now(boost::posix_time::seconds(30)); + + // Start an asynchronous operation to read a newline-delimited message. + asio::async_read_until(socket_, input_buffer_, '\n', + boost::bind(&client::handle_read, this, _1)); + } + + void handle_read(const asio::error_code& ec) + { + if (stopped_) + return; + + if (!ec) + { + // Extract the newline-delimited message from the buffer. + std::string line; + std::istream is(&input_buffer_); + std::getline(is, line); + + // Empty messages are heartbeats and so ignored. + if (!line.empty()) + { + std::cout << "Received: " << line << "\n"; + } + + start_read(); + } + else + { + std::cout << "Error on receive: " << ec.message() << "\n"; + + stop(); + } + } + + void start_write() + { + if (stopped_) + return; + + // Start an asynchronous operation to send a heartbeat message. + asio::async_write(socket_, asio::buffer("\n", 1), + boost::bind(&client::handle_write, this, _1)); + } + + void handle_write(const asio::error_code& ec) + { + if (stopped_) + return; + + if (!ec) + { + // Wait 10 seconds before sending the next heartbeat. + heartbeat_timer_.expires_from_now(boost::posix_time::seconds(10)); + heartbeat_timer_.async_wait(boost::bind(&client::start_write, this)); + } + else + { + std::cout << "Error on heartbeat: " << ec.message() << "\n"; + + stop(); + } + } + + void check_deadline() + { + if (stopped_) + return; + + // Check whether the deadline has passed. We compare the deadline against + // the current time since a new asynchronous operation may have moved the + // deadline before this actor had a chance to run. + if (deadline_.expires_at() <= deadline_timer::traits_type::now()) + { + // The deadline has passed. The socket is closed so that any outstanding + // asynchronous operations are cancelled. + socket_.close(); + + // There is no longer an active deadline. The expiry is set to positive + // infinity so that the actor takes no action until a new deadline is set. + deadline_.expires_at(boost::posix_time::pos_infin); + } + + // Put the actor back to sleep. + deadline_.async_wait(boost::bind(&client::check_deadline, this)); + } + +private: + bool stopped_; + tcp::resolver::results_type endpoints_; + tcp::socket socket_; + asio::streambuf input_buffer_; + deadline_timer deadline_; + deadline_timer heartbeat_timer_; +}; + +int main(int argc, char* argv[]) +{ + try + { + if (argc != 3) + { + std::cerr << "Usage: client <host> <port>\n"; + return 1; + } + + asio::io_context io_context; + tcp::resolver r(io_context); + client c(io_context); + + c.start(r.resolve(argv[1], argv[2])); + + io_context.run(); + } + catch (std::exception& e) + { + std::cerr << "Exception: " << e.what() << "\n"; + } + + return 0; +} diff --git a/3rdparty/asio/src/examples/cpp03/timeouts/blocking_tcp_client.cpp b/3rdparty/asio/src/examples/cpp03/timeouts/blocking_tcp_client.cpp new file mode 100644 index 00000000000..69d467af1bb --- /dev/null +++ b/3rdparty/asio/src/examples/cpp03/timeouts/blocking_tcp_client.cpp @@ -0,0 +1,242 @@ +// +// blocking_tcp_client.cpp +// ~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff 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) +// + +#include "asio/connect.hpp" +#include "asio/deadline_timer.hpp" +#include "asio/io_context.hpp" +#include "asio/ip/tcp.hpp" +#include "asio/read_until.hpp" +#include "asio/streambuf.hpp" +#include "asio/system_error.hpp" +#include "asio/write.hpp" +#include "asio/thread.hpp" +#include <cstdlib> +#include <iostream> +#include <string> +#include <boost/lambda/bind.hpp> +#include <boost/lambda/lambda.hpp> + +using asio::deadline_timer; +using asio::ip::tcp; +using boost::lambda::bind; +using boost::lambda::var; +using boost::lambda::_1; + +//---------------------------------------------------------------------- + +// +// This class manages socket timeouts by applying the concept of a deadline. +// Each asynchronous operation is given a deadline by which it must complete. +// Deadlines are enforced by an "actor" that persists for the lifetime of the +// client object: +// +// +----------------+ +// | | +// | check_deadline |<---+ +// | | | +// +----------------+ | async_wait() +// | | +// +---------+ +// +// If the actor determines that the deadline has expired, the socket is closed +// and any outstanding operations are consequently cancelled. The socket +// operations themselves use boost::lambda function objects as completion +// handlers. For a given socket operation, the client object runs the +// io_context to block thread execution until the actor completes. +// +class client +{ +public: + client() + : socket_(io_context_), + deadline_(io_context_) + { + // No deadline is required until the first socket operation is started. We + // set the deadline to positive infinity so that the actor takes no action + // until a specific deadline is set. + deadline_.expires_at(boost::posix_time::pos_infin); + + // Start the persistent actor that checks for deadline expiry. + check_deadline(); + } + + void connect(const std::string& host, const std::string& service, + boost::posix_time::time_duration timeout) + { + // Resolve the host name and service to a list of endpoints. + tcp::resolver::results_type endpoints = + tcp::resolver(io_context_).resolve(host, service); + + // Set a deadline for the asynchronous operation. As a host name may + // resolve to multiple endpoints, this function uses the composed operation + // async_connect. The deadline applies to the entire operation, rather than + // individual connection attempts. + deadline_.expires_from_now(timeout); + + // Set up the variable that receives the result of the asynchronous + // operation. The error code is set to would_block to signal that the + // operation is incomplete. Asio guarantees that its asynchronous + // operations will never fail with would_block, so any other value in + // ec indicates completion. + asio::error_code ec = asio::error::would_block; + + // Start the asynchronous operation itself. The boost::lambda function + // object is used as a callback and will update the ec variable when the + // operation completes. The blocking_udp_client.cpp example shows how you + // can use boost::bind rather than boost::lambda. + asio::async_connect(socket_, endpoints, var(ec) = _1); + + // Block until the asynchronous operation has completed. + do io_context_.run_one(); while (ec == asio::error::would_block); + + // Determine whether a connection was successfully established. The + // deadline actor may have had a chance to run and close our socket, even + // though the connect operation notionally succeeded. Therefore we must + // check whether the socket is still open before deciding if we succeeded + // or failed. + if (ec || !socket_.is_open()) + throw asio::system_error( + ec ? ec : asio::error::operation_aborted); + } + + std::string read_line(boost::posix_time::time_duration timeout) + { + // Set a deadline for the asynchronous operation. Since this function uses + // a composed operation (async_read_until), the deadline applies to the + // entire operation, rather than individual reads from the socket. + deadline_.expires_from_now(timeout); + + // Set up the variable that receives the result of the asynchronous + // operation. The error code is set to would_block to signal that the + // operation is incomplete. Asio guarantees that its asynchronous + // operations will never fail with would_block, so any other value in + // ec indicates completion. + asio::error_code ec = asio::error::would_block; + + // Start the asynchronous operation itself. The boost::lambda function + // object is used as a callback and will update the ec variable when the + // operation completes. The blocking_udp_client.cpp example shows how you + // can use boost::bind rather than boost::lambda. + asio::async_read_until(socket_, input_buffer_, '\n', var(ec) = _1); + + // Block until the asynchronous operation has completed. + do io_context_.run_one(); while (ec == asio::error::would_block); + + if (ec) + throw asio::system_error(ec); + + std::string line; + std::istream is(&input_buffer_); + std::getline(is, line); + return line; + } + + void write_line(const std::string& line, + boost::posix_time::time_duration timeout) + { + std::string data = line + "\n"; + + // Set a deadline for the asynchronous operation. Since this function uses + // a composed operation (async_write), the deadline applies to the entire + // operation, rather than individual writes to the socket. + deadline_.expires_from_now(timeout); + + // Set up the variable that receives the result of the asynchronous + // operation. The error code is set to would_block to signal that the + // operation is incomplete. Asio guarantees that its asynchronous + // operations will never fail with would_block, so any other value in + // ec indicates completion. + asio::error_code ec = asio::error::would_block; + + // Start the asynchronous operation itself. The boost::lambda function + // object is used as a callback and will update the ec variable when the + // operation completes. The blocking_udp_client.cpp example shows how you + // can use boost::bind rather than boost::lambda. + asio::async_write(socket_, asio::buffer(data), var(ec) = _1); + + // Block until the asynchronous operation has completed. + do io_context_.run_one(); while (ec == asio::error::would_block); + + if (ec) + throw asio::system_error(ec); + } + +private: + void check_deadline() + { + // Check whether the deadline has passed. We compare the deadline against + // the current time since a new asynchronous operation may have moved the + // deadline before this actor had a chance to run. + if (deadline_.expires_at() <= deadline_timer::traits_type::now()) + { + // The deadline has passed. The socket is closed so that any outstanding + // asynchronous operations are cancelled. This allows the blocked + // connect(), read_line() or write_line() functions to return. + asio::error_code ignored_ec; + socket_.close(ignored_ec); + + // There is no longer an active deadline. The expiry is set to positive + // infinity so that the actor takes no action until a new deadline is set. + deadline_.expires_at(boost::posix_time::pos_infin); + } + + // Put the actor back to sleep. + deadline_.async_wait(bind(&client::check_deadline, this)); + } + + asio::io_context io_context_; + tcp::socket socket_; + deadline_timer deadline_; + asio::streambuf input_buffer_; +}; + +//---------------------------------------------------------------------- + +int main(int argc, char* argv[]) +{ + try + { + if (argc != 4) + { + std::cerr << "Usage: blocking_tcp <host> <port> <message>\n"; + return 1; + } + + client c; + c.connect(argv[1], argv[2], boost::posix_time::seconds(10)); + + boost::posix_time::ptime time_sent = + boost::posix_time::microsec_clock::universal_time(); + + c.write_line(argv[3], boost::posix_time::seconds(10)); + + for (;;) + { + std::string line = c.read_line(boost::posix_time::seconds(10)); + + // Keep going until we get back the line that was sent. + if (line == argv[3]) + break; + } + + boost::posix_time::ptime time_received = + boost::posix_time::microsec_clock::universal_time(); + + std::cout << "Round trip time: "; + std::cout << (time_received - time_sent).total_microseconds(); + std::cout << " microseconds\n"; + } + catch (std::exception& e) + { + std::cerr << "Exception: " << e.what() << "\n"; + } + + return 0; +} diff --git a/3rdparty/asio/src/examples/cpp03/timeouts/blocking_udp_client.cpp b/3rdparty/asio/src/examples/cpp03/timeouts/blocking_udp_client.cpp new file mode 100644 index 00000000000..fd2da7640b8 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp03/timeouts/blocking_udp_client.cpp @@ -0,0 +1,182 @@ +// +// blocking_udp_client.cpp +// ~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff 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) +// + +#include "asio/deadline_timer.hpp" +#include "asio/io_context.hpp" +#include "asio/ip/udp.hpp" +#include <cstdlib> +#include <boost/bind.hpp> +#include <boost/date_time/posix_time/posix_time_types.hpp> +#include <iostream> + +using asio::deadline_timer; +using asio::ip::udp; + +//---------------------------------------------------------------------- + +// +// This class manages socket timeouts by applying the concept of a deadline. +// Each asynchronous operation is given a deadline by which it must complete. +// Deadlines are enforced by an "actor" that persists for the lifetime of the +// client object: +// +// +----------------+ +// | | +// | check_deadline |<---+ +// | | | +// +----------------+ | async_wait() +// | | +// +---------+ +// +// If the actor determines that the deadline has expired, any outstanding +// socket operations are cancelled. The socket operations themselves are +// implemented as transient actors: +// +// +---------------+ +// | | +// | receive | +// | | +// +---------------+ +// | +// async_- | +----------------+ +// receive() | | | +// +--->| handle_receive | +// | | +// +----------------+ +// +// The client object runs the io_context to block thread execution until the +// actor completes. +// +class client +{ +public: + client(const udp::endpoint& listen_endpoint) + : socket_(io_context_, listen_endpoint), + deadline_(io_context_) + { + // No deadline is required until the first socket operation is started. We + // set the deadline to positive infinity so that the actor takes no action + // until a specific deadline is set. + deadline_.expires_at(boost::posix_time::pos_infin); + + // Start the persistent actor that checks for deadline expiry. + check_deadline(); + } + + std::size_t receive(const asio::mutable_buffer& buffer, + boost::posix_time::time_duration timeout, asio::error_code& ec) + { + // Set a deadline for the asynchronous operation. + deadline_.expires_from_now(timeout); + + // Set up the variables that receive the result of the asynchronous + // operation. The error code is set to would_block to signal that the + // operation is incomplete. Asio guarantees that its asynchronous + // operations will never fail with would_block, so any other value in + // ec indicates completion. + ec = asio::error::would_block; + std::size_t length = 0; + + // Start the asynchronous operation itself. The handle_receive function + // used as a callback will update the ec and length variables. + socket_.async_receive(asio::buffer(buffer), + boost::bind(&client::handle_receive, _1, _2, &ec, &length)); + + // Block until the asynchronous operation has completed. + do io_context_.run_one(); while (ec == asio::error::would_block); + + return length; + } + +private: + void check_deadline() + { + // Check whether the deadline has passed. We compare the deadline against + // the current time since a new asynchronous operation may have moved the + // deadline before this actor had a chance to run. + if (deadline_.expires_at() <= deadline_timer::traits_type::now()) + { + // The deadline has passed. The outstanding asynchronous operation needs + // to be cancelled so that the blocked receive() function will return. + // + // Please note that cancel() has portability issues on some versions of + // Microsoft Windows, and it may be necessary to use close() instead. + // Consult the documentation for cancel() for further information. + socket_.cancel(); + + // There is no longer an active deadline. The expiry is set to positive + // infinity so that the actor takes no action until a new deadline is set. + deadline_.expires_at(boost::posix_time::pos_infin); + } + + // Put the actor back to sleep. + deadline_.async_wait(boost::bind(&client::check_deadline, this)); + } + + static void handle_receive( + const asio::error_code& ec, std::size_t length, + asio::error_code* out_ec, std::size_t* out_length) + { + *out_ec = ec; + *out_length = length; + } + +private: + asio::io_context io_context_; + udp::socket socket_; + deadline_timer deadline_; +}; + +//---------------------------------------------------------------------- + +int main(int argc, char* argv[]) +{ + try + { + using namespace std; // For atoi. + + if (argc != 3) + { + std::cerr << "Usage: blocking_udp_timeout <listen_addr> <listen_port>\n"; + return 1; + } + + udp::endpoint listen_endpoint( + asio::ip::make_address(argv[1]), + std::atoi(argv[2])); + + client c(listen_endpoint); + + for (;;) + { + char data[1024]; + asio::error_code ec; + std::size_t n = c.receive(asio::buffer(data), + boost::posix_time::seconds(10), ec); + + if (ec) + { + std::cout << "Receive error: " << ec.message() << "\n"; + } + else + { + std::cout << "Received: "; + std::cout.write(data, n); + std::cout << "\n"; + } + } + } + catch (std::exception& e) + { + std::cerr << "Exception: " << e.what() << "\n"; + } + + return 0; +} diff --git a/3rdparty/asio/src/examples/cpp03/timeouts/server.cpp b/3rdparty/asio/src/examples/cpp03/timeouts/server.cpp new file mode 100644 index 00000000000..212cd90d820 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp03/timeouts/server.cpp @@ -0,0 +1,427 @@ +// +// server.cpp +// ~~~~~~~~~~ +// +// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff 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) +// + +#include <algorithm> +#include <cstdlib> +#include <deque> +#include <iostream> +#include <set> +#include <boost/bind.hpp> +#include <boost/shared_ptr.hpp> +#include <boost/enable_shared_from_this.hpp> +#include "asio/deadline_timer.hpp" +#include "asio/io_context.hpp" +#include "asio/ip/tcp.hpp" +#include "asio/ip/udp.hpp" +#include "asio/read_until.hpp" +#include "asio/streambuf.hpp" +#include "asio/write.hpp" + +using asio::deadline_timer; +using asio::ip::tcp; +using asio::ip::udp; + +//---------------------------------------------------------------------- + +class subscriber +{ +public: + virtual ~subscriber() {} + virtual void deliver(const std::string& msg) = 0; +}; + +typedef boost::shared_ptr<subscriber> subscriber_ptr; + +//---------------------------------------------------------------------- + +class channel +{ +public: + void join(subscriber_ptr subscriber) + { + subscribers_.insert(subscriber); + } + + void leave(subscriber_ptr subscriber) + { + subscribers_.erase(subscriber); + } + + void deliver(const std::string& msg) + { + std::for_each(subscribers_.begin(), subscribers_.end(), + boost::bind(&subscriber::deliver, _1, boost::ref(msg))); + } + +private: + std::set<subscriber_ptr> subscribers_; +}; + +//---------------------------------------------------------------------- + +// +// This class manages socket timeouts by applying the concept of a deadline. +// Some asynchronous operations are given deadlines by which they must complete. +// Deadlines are enforced by two "actors" that persist for the lifetime of the +// session object, one for input and one for output: +// +// +----------------+ +----------------+ +// | | | | +// | check_deadline |<---+ | check_deadline |<---+ +// | | | async_wait() | | | async_wait() +// +----------------+ | on input +----------------+ | on output +// | | deadline | | deadline +// +---------+ +---------+ +// +// If either deadline actor determines that the corresponding deadline has +// expired, the socket is closed and any outstanding operations are cancelled. +// +// The input actor reads messages from the socket, where messages are delimited +// by the newline character: +// +// +------------+ +// | | +// | start_read |<---+ +// | | | +// +------------+ | +// | | +// async_- | +-------------+ +// read_- | | | +// until() +--->| handle_read | +// | | +// +-------------+ +// +// The deadline for receiving a complete message is 30 seconds. If a non-empty +// message is received, it is delivered to all subscribers. If a heartbeat (a +// message that consists of a single newline character) is received, a heartbeat +// is enqueued for the client, provided there are no other messages waiting to +// be sent. +// +// The output actor is responsible for sending messages to the client: +// +// +--------------+ +// | |<---------------------+ +// | await_output | | +// | |<---+ | +// +--------------+ | | +// | | | async_wait() | +// | +--------+ | +// V | +// +-------------+ +--------------+ +// | | async_write() | | +// | start_write |-------------->| handle_write | +// | | | | +// +-------------+ +--------------+ +// +// The output actor first waits for an output message to be enqueued. It does +// this by using a deadline_timer as an asynchronous condition variable. The +// deadline_timer will be signalled whenever the output queue is non-empty. +// +// Once a message is available, it is sent to the client. The deadline for +// sending a complete message is 30 seconds. After the message is successfully +// sent, the output actor again waits for the output queue to become non-empty. +// +class tcp_session + : public subscriber, + public boost::enable_shared_from_this<tcp_session> +{ +public: + tcp_session(asio::io_context& io_context, channel& ch) + : channel_(ch), + socket_(io_context), + input_deadline_(io_context), + non_empty_output_queue_(io_context), + output_deadline_(io_context) + { + input_deadline_.expires_at(boost::posix_time::pos_infin); + output_deadline_.expires_at(boost::posix_time::pos_infin); + + // The non_empty_output_queue_ deadline_timer is set to pos_infin whenever + // the output queue is empty. This ensures that the output actor stays + // asleep until a message is put into the queue. + non_empty_output_queue_.expires_at(boost::posix_time::pos_infin); + } + + tcp::socket& socket() + { + return socket_; + } + + // Called by the server object to initiate the four actors. + void start() + { + channel_.join(shared_from_this()); + + start_read(); + + input_deadline_.async_wait( + boost::bind(&tcp_session::check_deadline, + shared_from_this(), &input_deadline_)); + + await_output(); + + output_deadline_.async_wait( + boost::bind(&tcp_session::check_deadline, + shared_from_this(), &output_deadline_)); + } + +private: + void stop() + { + channel_.leave(shared_from_this()); + + asio::error_code ignored_ec; + socket_.close(ignored_ec); + input_deadline_.cancel(); + non_empty_output_queue_.cancel(); + output_deadline_.cancel(); + } + + bool stopped() const + { + return !socket_.is_open(); + } + + void deliver(const std::string& msg) + { + output_queue_.push_back(msg + "\n"); + + // Signal that the output queue contains messages. Modifying the expiry + // will wake the output actor, if it is waiting on the timer. + non_empty_output_queue_.expires_at(boost::posix_time::neg_infin); + } + + void start_read() + { + // Set a deadline for the read operation. + input_deadline_.expires_from_now(boost::posix_time::seconds(30)); + + // Start an asynchronous operation to read a newline-delimited message. + asio::async_read_until(socket_, input_buffer_, '\n', + boost::bind(&tcp_session::handle_read, shared_from_this(), _1)); + } + + void handle_read(const asio::error_code& ec) + { + if (stopped()) + return; + + if (!ec) + { + // Extract the newline-delimited message from the buffer. + std::string msg; + std::istream is(&input_buffer_); + std::getline(is, msg); + + if (!msg.empty()) + { + channel_.deliver(msg); + } + else + { + // We received a heartbeat message from the client. If there's nothing + // else being sent or ready to be sent, send a heartbeat right back. + if (output_queue_.empty()) + { + output_queue_.push_back("\n"); + + // Signal that the output queue contains messages. Modifying the + // expiry will wake the output actor, if it is waiting on the timer. + non_empty_output_queue_.expires_at(boost::posix_time::neg_infin); + } + } + + start_read(); + } + else + { + stop(); + } + } + + void await_output() + { + if (stopped()) + return; + + if (output_queue_.empty()) + { + // There are no messages that are ready to be sent. The actor goes to + // sleep by waiting on the non_empty_output_queue_ timer. When a new + // message is added, the timer will be modified and the actor will wake. + non_empty_output_queue_.expires_at(boost::posix_time::pos_infin); + non_empty_output_queue_.async_wait( + boost::bind(&tcp_session::await_output, shared_from_this())); + } + else + { + start_write(); + } + } + + void start_write() + { + // Set a deadline for the write operation. + output_deadline_.expires_from_now(boost::posix_time::seconds(30)); + + // Start an asynchronous operation to send a message. + asio::async_write(socket_, + asio::buffer(output_queue_.front()), + boost::bind(&tcp_session::handle_write, shared_from_this(), _1)); + } + + void handle_write(const asio::error_code& ec) + { + if (stopped()) + return; + + if (!ec) + { + output_queue_.pop_front(); + + await_output(); + } + else + { + stop(); + } + } + + void check_deadline(deadline_timer* deadline) + { + if (stopped()) + return; + + // Check whether the deadline has passed. We compare the deadline against + // the current time since a new asynchronous operation may have moved the + // deadline before this actor had a chance to run. + if (deadline->expires_at() <= deadline_timer::traits_type::now()) + { + // The deadline has passed. Stop the session. The other actors will + // terminate as soon as possible. + stop(); + } + else + { + // Put the actor back to sleep. + deadline->async_wait( + boost::bind(&tcp_session::check_deadline, + shared_from_this(), deadline)); + } + } + + channel& channel_; + tcp::socket socket_; + asio::streambuf input_buffer_; + deadline_timer input_deadline_; + std::deque<std::string> output_queue_; + deadline_timer non_empty_output_queue_; + deadline_timer output_deadline_; +}; + +typedef boost::shared_ptr<tcp_session> tcp_session_ptr; + +//---------------------------------------------------------------------- + +class udp_broadcaster + : public subscriber +{ +public: + udp_broadcaster(asio::io_context& io_context, + const udp::endpoint& broadcast_endpoint) + : socket_(io_context) + { + socket_.connect(broadcast_endpoint); + } + +private: + void deliver(const std::string& msg) + { + asio::error_code ignored_ec; + socket_.send(asio::buffer(msg), 0, ignored_ec); + } + + udp::socket socket_; +}; + +//---------------------------------------------------------------------- + +class server +{ +public: + server(asio::io_context& io_context, + const tcp::endpoint& listen_endpoint, + const udp::endpoint& broadcast_endpoint) + : io_context_(io_context), + acceptor_(io_context, listen_endpoint) + { + subscriber_ptr bc(new udp_broadcaster(io_context_, broadcast_endpoint)); + channel_.join(bc); + + start_accept(); + } + + void start_accept() + { + tcp_session_ptr new_session(new tcp_session(io_context_, channel_)); + + acceptor_.async_accept(new_session->socket(), + boost::bind(&server::handle_accept, this, new_session, _1)); + } + + void handle_accept(tcp_session_ptr session, + const asio::error_code& ec) + { + if (!ec) + { + session->start(); + } + + start_accept(); + } + +private: + asio::io_context& io_context_; + tcp::acceptor acceptor_; + channel channel_; +}; + +//---------------------------------------------------------------------- + +int main(int argc, char* argv[]) +{ + try + { + using namespace std; // For atoi. + + if (argc != 4) + { + std::cerr << "Usage: server <listen_port> <bcast_address> <bcast_port>\n"; + return 1; + } + + asio::io_context io_context; + + tcp::endpoint listen_endpoint(tcp::v4(), atoi(argv[1])); + + udp::endpoint broadcast_endpoint( + asio::ip::make_address(argv[2]), atoi(argv[3])); + + server s(io_context, listen_endpoint, broadcast_endpoint); + + io_context.run(); + } + catch (std::exception& e) + { + std::cerr << "Exception: " << e.what() << "\n"; + } + + return 0; +} |