libjmmcg  release_579_6_g8cffd
A C++ library containing an eclectic mix of useful, advanced components.
socket_wrapper_asio_impl.hpp
Go to the documentation of this file.
1 /******************************************************************************
2 ** Copyright © 2015 by J.M.McGuiness, coder@hussar.me.uk
3 **
4 ** This library is free software; you can redistribute it and/or
5 ** modify it under the terms of the GNU Lesser General Public
6 ** License as published by the Free Software Foundation; either
7 ** version 2.1 of the License, or (at your option) any later version.
8 **
9 ** This library is distributed in the hope that it will be useful,
10 ** but WITHOUT ANY WARRANTY; without even the implied warranty of
11 ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 ** Lesser General Public License for more details.
13 **
14 ** You should have received a copy of the GNU Lesser General Public
15 ** License along with this library; if not, write to the Free Software
16 ** Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17 */
18 
19 namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace socket { namespace asio {
20 
21 template<class LkT>
22 inline void
23 socket_wrapper<LkT>::set_options(std::size_t /*min_message_size*/, std::size_t max_message_size, unsigned short timeout, socket_priority /*priority*/, std::size_t /*incoming_cpu*/) {
24 // TODO "protocol not available": socket_.set_option(boost::asio::socket_base::send_low_watermark(min_message_size));
25  socket_.set_option(boost::asio::socket_base::send_buffer_size(max_message_size));
26 // TODO "protocol not available": socket_.set_option(boost::asio::socket_base::receive_low_watermark(min_message_size));
27  socket_.set_option(boost::asio::socket_base::receive_buffer_size(max_message_size));
28  socket_.set_option(boost::asio::socket_base::linger(true, timeout));
29  socket_.set_option(boost::asio::ip::tcp::no_delay(true));
30 }
31 
32 template<class LkT>
33 inline
34 socket_wrapper<LkT>::socket_wrapper(boost::asio::io_context &io_context)
35 : socket_(io_context) {
36 }
37 
38 template<class LkT>
39 inline
40 socket_wrapper<LkT>::socket_wrapper(socket_t &&socket)
41 : socket_(std::move(socket)) {
42 }
43 
44 template<class LkT>
45 inline void
46 socket_wrapper<LkT>::connect(boost::asio::ip::tcp::endpoint const &endpoint) {
47  socket_.connect(endpoint);
48 }
49 
50 template<class LkT>
51 template<class MsgT> inline void
52 socket_wrapper<LkT>::write(MsgT const &message) noexcept(false) {
53  boost::system::error_code io_error;
54  const write_lock_t lk(mutex_);
55  if constexpr (MsgT::has_static_size) {
56  using raw_buff_t=char const [sizeof(MsgT)];
57  assert(message.length()==sizeof(MsgT));
58  [[maybe_unused]] const std::size_t bytes_written=boost::asio::write(socket_, boost::asio::buffer(reinterpret_cast<raw_buff_t &>(message)), io_error);
59  if (UNLIKELY(io_error==boost::asio::error::eof)) {
60  return; // Connection closed cleanly by peer.
61  } else if (UNLIKELY(io_error)) {
62  throw boost::system::system_error(io_error); // Some other error.
63  }
64  assert(bytes_written==sizeof(MsgT));
65  } else {
66  [[maybe_unused]] const std::size_t bytes_written=boost::asio::write(socket_, boost::asio::buffer(reinterpret_cast<std::uint8_t const *>(&message), message.length()), io_error);
67  if (UNLIKELY(io_error==boost::asio::error::eof)) {
68  return; // Connection closed cleanly by peer.
69  } else if (UNLIKELY(io_error)) {
70  throw boost::system::system_error(io_error); // Some other error.
71  }
72  assert(bytes_written<=sizeof(MsgT));
73  assert(bytes_written==message.length());
74  }
75 }
76 
77 template<class LkT>
78 template<class V, std::size_t N> inline void
79 socket_wrapper<LkT>::write(std::array<V, N> const &message) noexcept(false) {
80  boost::system::error_code io_error;
81  const write_lock_t lk(mutex_);
82  [[maybe_unused]] const std::size_t bytes_written=boost::asio::write(socket_, boost::asio::buffer(message), io_error);
83  if (UNLIKELY(io_error==boost::asio::error::eof)) {
84  return; // Connection closed cleanly by peer.
85  } else if (UNLIKELY(io_error)) {
86  throw boost::system::system_error(io_error); // Some other error.
87  }
88  assert(bytes_written==sizeof(V)*N);
89 }
90 
91 template<class LkT>
92 template<class MsgT> inline void
93 socket_wrapper<LkT>::read(MsgT &dest) noexcept(false) {
94  boost::system::error_code io_error;
95  if constexpr (MsgT::has_static_size) {
96  using raw_buff_t=char [sizeof(MsgT)];
97  [[maybe_unused]] const std::size_t bytes_read=boost::asio::read(socket_, boost::asio::buffer(reinterpret_cast<raw_buff_t &>(dest)), io_error);
98  if (UNLIKELY(io_error==boost::asio::error::eof)) {
99  return; // Connection closed cleanly by peer.
100  } else if (UNLIKELY(io_error)) {
101  throw boost::system::system_error(io_error); // Some other error.
102  }
103  assert(bytes_read>0);
104  assert(bytes_read==sizeof(MsgT));
105  assert(dest.length()<=sizeof(MsgT));
106  } else {
107  constexpr std::size_t header_t_sz=MsgT::header_t_size;
108  [[maybe_unused]] const std::size_t bytes_read=boost::asio::read(socket_, boost::asio::buffer(reinterpret_cast<std::uint8_t *>(&dest), header_t_sz), io_error);
109  if (UNLIKELY(io_error==boost::asio::error::eof)) {
110  return; // Connection closed cleanly by peer.
111  } else if (UNLIKELY(io_error)) {
112  throw boost::system::system_error(io_error); // Some other error.
113  }
114  assert(bytes_read>0);
115  assert(bytes_read==header_t_sz);
116  typename MsgT::Header_t const *hdr=reinterpret_cast<typename MsgT::Header_t const *>(&dest);
117  const std::size_t length=hdr->length();
118  if (length>=header_t_sz && length<=sizeof(MsgT)) {
119  const std::size_t body_size=length-header_t_sz;
120  [[maybe_unused]] const std::size_t bytes_read_body=boost::asio::read(socket_, boost::asio::buffer(reinterpret_cast<std::uint8_t *>(&dest)+header_t_sz, body_size), io_error);
121  if (UNLIKELY(io_error==boost::asio::error::eof)) {
122  return; // Connection closed cleanly by peer.
123  } else if (UNLIKELY(io_error)) {
124  throw boost::system::system_error(io_error); // Some other error.
125  }
126  assert(bytes_read_body==body_size);
127  } else {
128  return;
129  }
130  }
131  assert(dest.is_valid());
132 }
133 
134 template<class LkT>
135 template<class V, std::size_t SrcSz> inline void
136 socket_wrapper<LkT>::read(V (& dest)[SrcSz]) noexcept(false) {
137  boost::system::error_code io_error;
138  [[maybe_unused]] const std::size_t bytes_read=boost::asio::read(socket_, boost::asio::buffer(dest), io_error);
139  if (UNLIKELY(io_error==boost::asio::error::eof)) {
140  return; // Connection closed cleanly by peer.
141  } else if (UNLIKELY(io_error)) {
142  throw boost::system::system_error(io_error); // Some other error.
143  }
144  assert(bytes_read==sizeof(V)*SrcSz);
145 }
146 
147 template<class LkT>
148 template<class MsgDetails, class V, std::size_t N> inline bool
149 socket_wrapper<LkT>::read(std::array<V, N> &buff) noexcept(false) {
150  using msg_details_t=MsgDetails;
151 
152  boost::system::error_code io_error;
153  assert(socket_.is_open());
154  BOOST_MPL_ASSERT_RELATION(msg_details_t::max_msg_size, >=, msg_details_t::header_t_size);
155  BOOST_MPL_ASSERT_RELATION(msg_details_t::max_msg_size, <=, N*sizeof(V));
156  [[maybe_unused]] std::size_t bytes_read=boost::asio::read(socket_, boost::asio::buffer(buff, msg_details_t::header_t_size), io_error);
157  if (UNLIKELY(io_error==boost::asio::error::eof)) {
158  return true; // Connection closed cleanly by peer.
159  } else if (UNLIKELY(io_error)) {
160  throw boost::system::system_error(io_error); // Some other error.
161  }
162  assert(bytes_read<=N);
163  assert(bytes_read<=msg_details_t::max_msg_size);
164  assert(bytes_read==msg_details_t::header_t_size);
165  typename msg_details_t::Header_t const *hdr=reinterpret_cast<typename msg_details_t::Header_t const *>(buff.data());
166  const std::size_t length=hdr->length();
167  if (length>=msg_details_t::header_t_size && length<=msg_details_t::max_msg_size) {
168  const std::size_t body_size=length-msg_details_t::header_t_size;
169  if (body_size>0) {
170  bytes_read=boost::asio::read(socket_, boost::asio::buffer(&*std::next(buff.begin(), msg_details_t::header_t_size), body_size), io_error);
171  if (io_error==boost::asio::error::eof) {
172  return true; // Connection closed cleanly by peer.
173  } else if (io_error) {
174  throw boost::system::system_error(io_error); // Some other error.
175  }
176  assert(bytes_read<=msg_details_t::max_msg_size);
177  assert(bytes_read==body_size);
178  }
179  assert(hdr->is_valid());
180  return false;
181  } else {
182  return true;
183  }
184 }
185 
186 template<class LkT>
187 inline bool
188 socket_wrapper<LkT>::is_open() const {
189  return socket_.is_open();
190 }
191 
192 template<class LkT>
193 inline void
194 socket_wrapper<LkT>::close() {
195  socket_.set_option(boost::asio::socket_base::linger(true, 1));
196  boost::system::error_code ec;
197  socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec);
198 }
199 
200 template<class LkT>
201 inline std::string
202 socket_wrapper<LkT>::to_string() const noexcept(false) {
203  std::ostringstream ss;
204  ss
205  <<"socket_="<<const_cast<boost::asio::ip::tcp::socket &>(socket_).native_handle();
206  return ss.str();
207 }
208 
209 template<class LkT>
210 inline std::ostream &
211 operator<<(std::ostream &os, socket_wrapper<LkT> const &ec) noexcept(false) {
212  os<<ec.to_string();
213  return os;
214 }
215 
216 } } } }