Line data Source code
1 : #include "unix_socket.hpp"
2 : #include "../with_timeout.hpp"
3 : #include "../../util/logger.hpp"
4 : #include <boost/asio/read.hpp>
5 : #include <boost/asio/write.hpp>
6 : #include <boost/asio/read_until.hpp>
7 : #include <fcntl.h>
8 :
9 : namespace thinger::asio {
10 :
11 1218 : unix_socket::unix_socket(const std::string &context, boost::asio::io_context &io_context)
12 1218 : : socket(context, io_context), socket_(io_context) {
13 1218 : }
14 :
15 16767 : unix_socket::unix_socket(const std::string &context, boost::asio::local::stream_protocol::socket&& sock)
16 16767 : : socket(context, static_cast<boost::asio::io_context&>(sock.get_executor().context()))
17 33534 : , socket_(std::move(sock)) {
18 16767 : }
19 :
20 17985 : unix_socket::~unix_socket() {
21 17985 : unix_socket::close();
22 17985 : }
23 :
24 27069 : void unix_socket::close() {
25 27069 : if (socket_.is_open()) {
26 17985 : boost::system::error_code ec;
27 17985 : socket_.close(ec);
28 : }
29 27069 : }
30 :
31 1873 : void unix_socket::cancel() {
32 1873 : socket_.cancel();
33 1870 : }
34 :
35 243 : bool unix_socket::shutdown_send() {
36 243 : boost::system::error_code ec;
37 243 : socket_.shutdown(boost::asio::local::stream_protocol::socket::shutdown_send, ec);
38 486 : return !ec;
39 : }
40 :
41 1218 : awaitable<boost::system::error_code> unix_socket::connect(const std::string &path, std::chrono::seconds timeout) {
42 : close();
43 :
44 : // Race between connect and timeout: a connect not done in time is cancelled
45 : auto connected = co_await with_timeout(
46 : socket_.async_connect(boost::asio::local::stream_protocol::endpoint(path), use_nothrow_awaitable),
47 : timeout);
48 : if (!connected) {
49 : close();
50 : co_return boost::asio::error::timed_out;
51 : }
52 : auto connect_ec = std::get<0>(*connected);
53 : // Not inherited by the programs the process runs (fork and exec): asio opens sockets
54 : // without SOCK_CLOEXEC
55 : if (!connect_ec) ::fcntl(socket_.native_handle(), F_SETFD, FD_CLOEXEC);
56 : co_return connect_ec;
57 2436 : }
58 :
59 0 : awaitable<boost::system::error_code> unix_socket::connect(
60 : const std::string &host,
61 : const std::string &port,
62 : std::chrono::seconds timeout)
63 : {
64 : LOG_WARNING("calling connect to a unix socket over host/port");
65 : co_return co_await connect(host, timeout);
66 0 : }
67 :
68 231 : std::string unix_socket::get_remote_ip() const {
69 231 : boost::system::error_code ec;
70 231 : auto remote_ep = socket_.remote_endpoint(ec);
71 231 : if (!ec) {
72 231 : return remote_ep.path();
73 : }
74 0 : return "";
75 : }
76 :
77 231 : std::string unix_socket::get_local_port() const {
78 462 : return "0";
79 : }
80 :
81 231 : std::string unix_socket::get_remote_port() const {
82 462 : return "0";
83 : }
84 :
85 275466 : awaitable<io_result> unix_socket::read_some(uint8_t buffer[], size_t max_size) {
86 : co_return co_await socket_.async_read_some(
87 : boost::asio::buffer(buffer, max_size),
88 : use_nothrow_awaitable);
89 550932 : }
90 :
91 5275 : awaitable<io_result> unix_socket::read_some_cancellable(uint8_t buffer[], size_t max_size,
92 : boost::asio::cancellation_slot slot) {
93 : co_return co_await detail::read_some_on_slot(socket_, buffer, max_size, slot);
94 10550 : }
95 :
96 603 : awaitable<io_result> unix_socket::read(uint8_t buffer[], size_t size) {
97 : co_return co_await boost::asio::async_read(
98 : socket_,
99 : boost::asio::buffer(buffer, size),
100 : boost::asio::transfer_exactly(size),
101 : use_nothrow_awaitable);
102 1206 : }
103 :
104 0 : awaitable<io_result> unix_socket::read(boost::asio::streambuf& buffer, size_t size) {
105 : co_return co_await boost::asio::async_read(
106 : socket_,
107 : buffer,
108 : boost::asio::transfer_exactly(size),
109 : use_nothrow_awaitable);
110 0 : }
111 :
112 3 : awaitable<io_result> unix_socket::read_until(boost::asio::streambuf& buffer, std::string_view delim) {
113 : co_return co_await boost::asio::async_read_until(
114 : socket_,
115 : buffer,
116 : std::string(delim),
117 : use_nothrow_awaitable);
118 6 : }
119 :
120 595 : awaitable<io_result> unix_socket::write(const uint8_t buffer[], size_t size) {
121 : co_return co_await boost::asio::async_write(
122 : socket_,
123 : boost::asio::buffer(buffer, size),
124 : use_nothrow_awaitable);
125 1190 : }
126 :
127 2115 : awaitable<io_result> unix_socket::write(std::string_view str) {
128 : co_return co_await boost::asio::async_write(
129 : socket_,
130 : boost::asio::buffer(str.data(), str.size()),
131 : use_nothrow_awaitable);
132 4230 : }
133 :
134 15586 : awaitable<io_result> unix_socket::write(const std::vector<boost::asio::const_buffer>& buffers) {
135 : co_return co_await boost::asio::async_write(
136 : socket_,
137 : buffers,
138 : use_nothrow_awaitable);
139 31172 : }
140 :
141 340 : awaitable<boost::system::error_code> unix_socket::wait(boost::asio::socket_base::wait_type type) {
142 : auto [ec] = co_await socket_.async_wait(type, use_nothrow_awaitable);
143 : co_return ec;
144 680 : }
145 :
146 39 : awaitable<boost::system::error_code> unix_socket::peer_closed() {
147 : co_return peek_closed(socket_.native_handle());
148 78 : }
149 :
150 34198 : bool unix_socket::is_open() const {
151 34198 : return socket_.is_open();
152 : }
153 :
154 9759 : bool unix_socket::is_secure() const {
155 9759 : return false;
156 : }
157 :
158 198 : bool unix_socket::readable_now() const {
159 396 : return socket_.is_open() &&
160 396 : poll_readable(const_cast<boost::asio::local::stream_protocol::socket&>(socket_).native_handle());
161 : }
162 :
163 30 : size_t unix_socket::send_backlog() const {
164 30 : return socket_.is_open()
165 30 : ? output_queue(const_cast<boost::asio::local::stream_protocol::socket&>(socket_).native_handle())
166 30 : : 0;
167 : }
168 :
169 112 : size_t unix_socket::available() const {
170 112 : boost::system::error_code ec;
171 112 : auto size = socket_.available(ec);
172 112 : if (ec) {
173 0 : LOG_ERROR("error while getting socket available bytes ({}): {}", size, ec.message());
174 : }
175 112 : return size;
176 : }
177 :
178 : }
|
