diff --git a/test_client/async_client.hpp b/test_client/async_client.hpp index 3b55aca..8640c64 100644 --- a/test_client/async_client.hpp +++ b/test_client/async_client.hpp @@ -1,6 +1,8 @@ #pragma once #include +#include #include +#include using boost::asio::ip::tcp; #include "../codec.h" using namespace rest_rpc; @@ -8,7 +10,7 @@ using namespace rest_rpc::rpc_service; class async_client : private boost::noncopyable { public: - async_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_), + async_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_), strand_(ios_), deadline_(ios_), host_(host), port_(port) { thd_ = std::make_shared([this] { ios_.run(); @@ -20,7 +22,7 @@ public: } void set_connect_timeout(size_t seconds) { - timeout_ = seconds; + connect_timeout_ = seconds; } void set_reconnect_count(int reconnect_count) { @@ -28,7 +30,7 @@ public: } void connect() { - reset_connect_timer(); + reset_deadline_timer(connect_timeout_); auto addr = boost::asio::ip::make_address_v4(host_); socket_.async_connect({ addr, port_ }, [this](const boost::system::error_code& ec) { if (ec) { @@ -53,6 +55,19 @@ public: }); } + void write(std::string&& message) + { + strand_.post([this, msg = std::move(message)] { + outbox_.emplace_back(std::move(msg)); + if (outbox_.size() > 1) { + // outstanding async_write + return; + } + + this->write(); + }); + } + bool has_connected() const { return has_connected_; } @@ -66,8 +81,8 @@ public: } private: - void reset_connect_timer() { - deadline_.expires_from_now(boost::posix_time::seconds((long)timeout_)); + void reset_deadline_timer(size_t timeout) { + deadline_.expires_from_now(boost::posix_time::seconds((long)timeout)); deadline_.async_wait([this](const boost::system::error_code& ec) { if (!ec) { socket_.close(); @@ -75,16 +90,39 @@ private: }); } + void write() + { + const std::string& message = outbox_[0]; + boost::asio::async_write(socket_,boost::asio::buffer(message.c_str(), message.size()), + strand_.wrap([this](const boost::system::error_code& ec, const size_t bytesTransferred) { + outbox_.pop_front(); + + if (ec) { + std::cerr << "could not write: " << ec.message() << std::endl; + return; + } + + if (!outbox_.empty()) { + // more messages to send + this->write(); + } + }) + ); + } + bool has_connected_ = false; boost::asio::io_service ios_; tcp::socket socket_; boost::asio::io_service::work work_; + boost::asio::io_service::strand strand_; std::shared_ptr thd_ = nullptr; std::string host_; unsigned short port_; - int timeout_=1;//s + size_t connect_timeout_ = 1;//s int reconnect_cnt_ = -1; boost::asio::deadline_timer deadline_; + + std::deque outbox_; }; \ No newline at end of file