blob: 941bb1d8fb68d45a286841eb3833cf53e76c9aa2 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
#pragma once
#ifndef ASYNC_WRITER_H_HQIRH28I
#define ASYNC_WRITER_H_HQIRH28I
#include <boost/asio.hpp>
#include "rpc/msgpack.hpp"
#include <condition_variable>
#include <deque>
#include <memory>
#include <thread>
namespace rpc {
class client;
namespace detail {
//! \brief Common logic for classes that have a write queue with async writing.
class async_writer : public std::enable_shared_from_this<async_writer> {
public:
async_writer(boost::asio::io_service *io,
boost::asio::ip::tcp::socket socket)
: socket_(std::move(socket)), write_strand_(*io), exit_(false) {}
void do_write() {
if (exit_) {
return;
}
auto self(shared_from_this());
auto &item = write_queue_.front();
// the data in item remains valid until the handler is called
// since it will still be in the queue physically until then.
boost::asio::async_write(
socket_, boost::asio::buffer(item.data(), item.size()),
write_strand_.wrap(
[this, self](boost::system::error_code ec, std::size_t transferred) {
(void)transferred;
if (!ec) {
write_queue_.pop_front();
if (write_queue_.size() > 0) {
if (!exit_) {
do_write();
}
}
} else {
LOG_ERROR("Error while writing to socket: {}", ec);
}
if (exit_) {
LOG_INFO("Closing socket");
socket_.shutdown(
boost::asio::ip::tcp::socket::shutdown_both);
socket_.close();
}
}));
}
void write(RPCLIB_MSGPACK::sbuffer &&data) {
write_queue_.push_back(std::move(data));
if (write_queue_.size() > 1) {
return; // there is an ongoing write chain so don't start another
}
do_write();
}
friend class rpc::client;
protected:
boost::asio::ip::tcp::socket socket_;
boost::asio::strand write_strand_;
std::atomic_bool exit_{false};
bool exited_ = false;
std::mutex m_exit_;
std::condition_variable cv_exit_;
private:
std::deque<RPCLIB_MSGPACK::sbuffer> write_queue_;
RPCLIB_CREATE_LOG_CHANNEL(async_writer)
};
} /* detail */
} /* rpc */
#endif /* end of include guard: ASYNC_WRITER_H_HQIRH28I */
|
