timur/ADC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AdcNetServiceASIO/AdcNetServiceASIOTLS: new implementation

Browse Source
This commit is contained in:
Timur A. Fatkhullin 2024-10-26 20:56:05 +03:00
parent 062c26537d
commit a496758ca4
4 changed files with 1071 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
net/adc_net_concepts.h
View File

@ -113,15 +113,15 @@ concept adc_netservice_c = requires(SRVT srv, const SRVT srv_const) {
// acceptor type // acceptor type
requires std::is_class_v<typename SRVT::acceptor_t>; // requires std::is_class_v<typename SRVT::acceptor_t>;
requires adc_async_callback_t<typename SRVT::acceptor_t::async_accept_callback_t>; // requires adc_async_callback_t<typename SRVT::acceptor_t::async_accept_callback_t>;
requires requires(typename SRVT::acceptor_t acc) { // requires requires(typename SRVT::acceptor_t acc) {
acc.asyncAccept(std::declval<typename SRVT::acceptor_t::async_accept_callback_t>(), // acc.asyncAccept(std::declval<typename SRVT::acceptor_t::async_accept_callback_t>(),
std::declval<const typename SRVT::timeout_t&>()); // std::declval<const typename SRVT::timeout_t&>());
// { acc.accept(std::declval<const typename SRVT::timeout_t&>()) } -> std::same_as<SRVT>; // // { acc.accept(std::declval<const typename SRVT::timeout_t&>()) } -> std::same_as<SRVT>;
acc.accept(std::declval<const typename SRVT::timeout_t&>()); // acc.accept(std::declval<const typename SRVT::timeout_t&>());
}; // };
// netservice_ident_t ident() const // netservice_ident_t ident() const

554
net/asio/adc_netservice_asio.h
View File

@ -1,5 +1,6 @@
#pragma once #pragma once
#include <asio/any_completion_handler.hpp>
#include <asio/basic_datagram_socket.hpp> #include <asio/basic_datagram_socket.hpp>
#include <asio/basic_seq_packet_socket.hpp> #include <asio/basic_seq_packet_socket.hpp>
#include <asio/basic_stream_socket.hpp> #include <asio/basic_stream_socket.hpp>
@ -33,7 +34,6 @@
#include "../../common/adc_traits.h" #include "../../common/adc_traits.h"
#include "../adc_net_concepts.h" #include "../adc_net_concepts.h"
#include "../adc_netproto.h"
namespace adc::traits namespace adc::traits
{ {
@ -66,15 +66,10 @@ struct adc_func_traits<asio::deferred_t> {
} // namespace adc::traits } // namespace adc::traits
namespace adc::impl namespace adc::impl
{ {
// typedef for ASIO streambuf iterators
using asio_streambuff_iter_t = asio::buffers_iterator<asio::streambuf::const_buffers_type>;
// ASIO match condition result typedef
using asio_matchcond_result_t = std::pair<asio_streambuff_iter_t, bool>;
template <typename T> template <typename T>
concept adc_asio_transport_proto_c = concept adc_asio_transport_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::ip::udp> || std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::ip::udp> ||
@ -112,16 +107,27 @@ concept adc_asio_special_comp_token_c =
template <adc_asio_transport_proto_c TRANSPORT_PROTOT, template <
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, bool USE_TLS, // true - use of asio::ssl::stream<TRANSPORT_PROTOT>
traits::adc_output_char_range RMSGT = adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
std::vector<char>> // used only for inner storing of message byte sequence interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
class AdcNetServiceASIOBase : public SESSION_PROTOT traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
class AdcBaseNetServiceASIO : public SESSION_PROTOT
{ {
public: public:
#ifdef USE_OPENSSL_WITH_ASIO
static_assert(!(USE_TLS && adc_asio_tls_transport_proto_c<TRANSPORT_PROTOT>),
"INVALID 'TRANSPORT_PROTOT' TEMPLATE ARGUMENT!");
static constexpr bool isTLS = USE_TLS;
#else // ignore USE_TLS
static constexpr bool isTLS = false;
#endif
// typedefs to satisfy 'adc_netservice_c' concept // typedefs to satisfy 'adc_netservice_c' concept
typedef std::string_view netservice_ident_t; typedef std::string_view netservice_ident_t;
typedef std::vector<char> send_msg_t; // in general, only one of several possible typedef std::vector<char> send_msg_t; // in general, only one of several possible
typedef RMSGT recv_msg_t; // in general, only one of several possible (see class template arguments declaration) typedef RMSGT recv_msg_t; // in general, only one of several possible (see class template arguments declaration)
typedef traits::adc_common_duration_t timeout_t; typedef traits::adc_common_duration_t timeout_t;
@ -134,103 +140,102 @@ public:
// typedefs from transport protocol // typedefs from transport protocol
using socket_t = typename TRANSPORT_PROTOT::socket; using socket_t = typename TRANSPORT_PROTOT::socket; // low-level socket type
using tls_stream_t = std::nullptr_t; #ifdef USE_OPENSSL_WITH_ASIO
using session_level_socket_t = std::conditional_t<isTLS, asio::ssl::stream<socket_t>, std::nullptr_t>;
// TLS certificate attributes comparison function:
// 'serial' - as returned by OpenSSL BN_bn2hex
// 'fingerprint' - as returned by OpenSSL X509_digest
// 'depth' - depth in chain
// the function must return 0 - if comparison failed; otherwise - something != 0
typedef std::function<int(const std::string& serial, const std::vector<unsigned char>& fingerprint, int depth)>
cert_comp_func_t;
#else
using session_level_socket_t = socket_t;
#endif
struct acceptor_t {
using srv_acceptor_t = std::conditional_t<
std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>,
std::nullptr_t, // there is no acceptor
typename TRANSPORT_PROTOT::acceptor>;
// acceptor
class acceptor_t
{
public:
static constexpr std::chrono::duration DEFAULT_ACCEPT_TIMEOUT = std::chrono::seconds::max(); static constexpr std::chrono::duration DEFAULT_ACCEPT_TIMEOUT = std::chrono::seconds::max();
acceptor_t(asio::io_context& io_ctx) typedef AdcBaseNetServiceASIO netservice_t;
: _ioContext(io_ctx), _endpoint(), _socket(_ioContext), _acceptor(_ioContext)
{
}
acceptor_t(asio::io_context& io_ctx, const AdcNetServiceASIOBase::endpoint_t& endpoint)
: _ioContext(io_ctx), _endpoint(endpoint), _socket(_ioContext), _acceptor(_ioContext, endpoint)
{
}
virtual ~acceptor_t() = default;
typedef AdcNetServiceASIOBase netservice_t;
typedef std::shared_ptr<netservice_t> sptr_netservice_t;
typedef std::function<void(std::error_code, netservice_t)> async_accept_callback_t; typedef std::function<void(std::error_code, netservice_t)> async_accept_callback_t;
// typedef std::function<void(std::error_code, sptr_netservice_t)> async_accept_callback_t;
// template <asio::completion_token_for<void(std::error_code, sptr_netservice_t)> TokenT,
acceptor_t(asio::io_context& io_context) : _ioContext(io_context), _acceptor(io_context) {}
acceptor_t(asio::io_context& io_context, const netservice_t::endpoint_t& endpoint)
: _ioContext(io_context), _acceptor(io_context, endpoint)
{
}
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT, template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)> traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(TokenT&& token, const DT& timeout = DEFAULT_ACCEPT_TIMEOUT) auto asyncAccept(TokenT&& token, const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{ {
enum { start_state, handshake_state, stop_state }; enum { sock_accept, handshake, done };
// no acceptor for UDP-sockets auto timer = netservice_t::getDeadlineTimer(_acceptor, timeout);
if constexpr (std::is_null_pointer_v<_acceptor_t>) {
static_assert(false, "INVALID TRANSPORT PROTOCOL TYPE!");
}
netservice_t srv{_ioContext}; auto srv = std::make_unique<netservice_t>(_ioContext);
_socket = AdcNetServiceASIOBase::socket_t{_ioContext};
auto timer = getDeadlineTimer(_acceptor, timeout);
// return asio::async_compose<TokenT, void(std::error_code, sptr_netservice_t)>(
return asio::async_compose<TokenT, void(std::error_code, netservice_t)>( return asio::async_compose<TokenT, void(std::error_code, netservice_t)>(
[timer = std::move(timer), srv = std::move(srv), state = start_state, this]( [timer = std::move(timer), srv = std::move(srv), state = sock_accept, this](
auto& self, std::error_code ec = {}) mutable { auto& self, std::error_code ec = {}) mutable {
if (!ec) { if (!ec) {
switch (state) { switch (state) {
case start_state: case sock_accept:
state = handshake_state; if constexpr (netservice_t::isTLS) {
try { state = handshake;
if (!_acceptor.is_open() || (_acceptor.local_endpoint() != _endpoint)) { } else {
_acceptor = _acceptor_t(_ioContext, _endpoint); state = done;
}
} catch (std::system_error err) {
timer->cancel();
self.complete(err.code(), netservice_t{_ioContext});
// self.complete(err.code(), std::make_shared<netservice_t>(_ioContext));
return;
} }
return _acceptor.async_accept(srv->_socket, std::move(self));
return _acceptor.async_accept(_socket, std::move(self));
break; break;
case handshake_state: case handshake:
state = stop_state; state = done;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (netservice_t::isTLS) {
srv->_sessSocket =
netservice_t::session_level_socket_t(srv->_socket, srv->_tlsContext);
return srv->_sessSocket.async_handshake(asio::ssl::stream_base::server,
std::move(self));
}
#endif
break; break;
case stop_state: case done:
finalize();
break; break;
default: default:
break; break;
} }
} }
if (isTimeout(timer, ec)) { if (netservice_t::isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out); ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_accept } else { // an error occured in async_accept od async_handshake
timer->cancel(); timer->cancel();
} }
self.complete(ec, netservice_t(std::move(_socket))); self.complete(ec, std::move(*srv));
// self.complete(ec, std::make_shared<netservice_t>(std::move(_socket)));
srv.reset();
}, },
token, _ioContext); token, _ioContext);
} }
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT, template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)> traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(const AdcNetServiceASIOBase::endpoint_t& endpoint, auto asyncAccept(const netservice_t::endpoint_t& endpoint,
TokenT&& token, TokenT&& token,
const DT& timeout = DEFAULT_ACCEPT_TIMEOUT) const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{ {
_endpoint = endpoint; _acceptor = srv_acceptor_t(_ioContext, endpoint);
return asyncAccept(std::forward<TokenT>(token), timeout); return asyncAccept(std::forward<TokenT>(token), timeout);
} }
@ -248,35 +253,26 @@ public:
return accept(endpoint, timeout); return accept(endpoint, timeout);
} }
protected: private:
asio::io_context& _ioContext; asio::io_context& _ioContext;
AdcNetServiceASIOBase::endpoint_t _endpoint; srv_acceptor_t _acceptor;
AdcNetServiceASIOBase::socket_t _socket;
using _acceptor_t = std::conditional_t<
std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>,
std::nullptr_t, // there is no acceptor
typename TRANSPORT_PROTOT::acceptor>;
_acceptor_t _acceptor;
virtual void handshake() {}
virtual void finalize() {}
}; };
static constexpr std::chrono::duration DEFAULT_CONNECT_TIMEOUT = std::chrono::seconds(10); static constexpr std::chrono::duration DEFAULT_CONNECT_TIMEOUT = std::chrono::seconds(10);
static constexpr std::chrono::duration DEFAULT_SEND_TIMEOUT = std::chrono::seconds(5); static constexpr std::chrono::duration DEFAULT_SEND_TIMEOUT = std::chrono::seconds(5);
static constexpr std::chrono::duration DEFAULT_RECEIVE_TIMEOUT = std::chrono::seconds(5); static constexpr std::chrono::duration DEFAULT_RECEIVE_TIMEOUT = std::chrono::seconds(5);
AdcNetServiceASIOBase(asio::io_context& ctx) /* CONSTRUCTORS AND DESTRUCTOR */
AdcBaseNetServiceASIO(asio::io_context& ctx)
requires(!isTLS)
: SESSION_PROTOT(), _ioContext(ctx), _receiveStrand(_ioContext), _socket(_ioContext), _receiveQueue() : SESSION_PROTOT(), _ioContext(ctx), _receiveStrand(_ioContext), _socket(_ioContext), _receiveQueue()
{ {
} }
AdcNetServiceASIOBase(socket_t socket) AdcBaseNetServiceASIO(socket_t socket)
: SESSION_PROTOT(), : SESSION_PROTOT(),
_ioContext(static_cast<asio::io_context&>(socket.get_executor().context())), _ioContext(static_cast<asio::io_context&>(socket.get_executor().context())),
_socket(std::move(socket)), _socket(std::move(socket)),
@ -285,13 +281,28 @@ public:
{ {
} }
#ifdef USE_OPENSSL_WITH_ASIO
AdcBaseNetServiceASIO(asio::io_context& ctx,
const asio::ssl::context& tls_context,
const asio::ssl::verify_mode& tls_peer_verify_mode = asio::ssl::verify_peer)
requires isTLS
: SESSION_PROTOT(),
_ioContext(ctx),
_receiveStrand(_ioContext),
_socket(_ioContext),
_receiveQueue(),
_tlsContext(tls_context),
_tlsPeerVerifyMode(tls_peer_verify_mode)
{
}
#endif
// NOTE: CANNOT MOVE asio::streambuf CORRECTLY?!!!
// AdcNetServiceASIOBase(AdcNetServiceASIOBase&& other) = default; AdcBaseNetServiceASIO(AdcBaseNetServiceASIO&& other)
AdcNetServiceASIOBase(AdcNetServiceASIOBase&& other)
: _ioContext(other._ioContext), : _ioContext(other._ioContext),
_receiveStrand(std::move(other._receiveStrand)), _receiveStrand(std::move(other._receiveStrand)),
_socket(std::move(other._socket)), _socket(std::move(other._socket)),
_sessSocket(std::move(other._sessSocket)),
_streamBuffer(), _streamBuffer(),
_receiveQueue(std::move(other._receiveQueue)) _receiveQueue(std::move(other._receiveQueue))
@ -301,17 +312,14 @@ public:
} }
// AdcNetServiceASIOBase(AdcNetServiceASIOBase&& other) = delete; AdcBaseNetServiceASIO(const AdcBaseNetServiceASIO&) = delete; // no copy constructor!
AdcNetServiceASIOBase(const AdcNetServiceASIOBase&) = delete; // no copy constructor!
virtual ~AdcNetServiceASIOBase() {} virtual ~AdcBaseNetServiceASIO() {}
AdcNetServiceASIOBase& operator=(const AdcNetServiceASIOBase&) = delete; AdcBaseNetServiceASIO& operator=(const AdcBaseNetServiceASIO&) = delete;
// AdcNetServiceASIOBase& operator=(AdcNetServiceASIOBase&& other) = delete; AdcBaseNetServiceASIO& operator=(AdcBaseNetServiceASIO&& other)
// AdcNetServiceASIOBase& operator=(AdcNetServiceASIOBase&& other) = default;
AdcNetServiceASIOBase& operator=(AdcNetServiceASIOBase&& other)
{ {
_ioContext = other._ioContext; _ioContext = other._ioContext;
_receiveStrand = std::move(other._receiveStrand); _receiveStrand = std::move(other._receiveStrand);
@ -335,25 +343,47 @@ public:
/* asynchronuos methods */ /* asynchronuos methods */
template <asio::completion_token_for<void(std::error_code)> TokenT, template <asio::completion_token_for<void(std::error_code)> TokenT,
traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_CONNECT_TIMEOUT)> traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_CONNECT_TIMEOUT)>
auto asyncConnect(const endpoint_t& endpoint, TokenT&& token, const TimeoutT& timeout = DEFAULT_CONNECT_TIMEOUT) auto asyncConnect(const endpoint_t& endpoint, TokenT&& token, const TimeoutT& timeout = DEFAULT_CONNECT_TIMEOUT)
{ {
enum { sock_connect, handshake, done };
auto timer = getDeadlineTimer(_socket, timeout); auto timer = getDeadlineTimer(_socket, timeout);
return asio::async_compose<TokenT, void(std::error_code)>( return asio::async_compose<TokenT, void(std::error_code)>(
[start = true, endpoint, timer = std::move(timer), this](auto& self, std::error_code ec = {}) mutable { [state = sock_connect, endpoint, timer = std::move(timer), this](auto& self,
std::error_code ec = {}) mutable {
if (!ec) { if (!ec) {
if (start) { switch (state) {
start = false; case sock_connect:
return _socket.async_connect(endpoint, std::move(self)); if constexpr (isTLS) {
state = handshake;
} else {
state = done;
}
return _socket.async_connect(endpoint, std::move(self));
break;
case handshake:
state = done;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (isTLS) {
_sessSocket = session_level_socket_t(_socket, _tlsContext);
return _sessSocket.async_handshake(session_level_socket_t::client, std::move(self));
}
#endif
break;
case done:
break;
default:
break;
} }
} }
if (isTimeout(timer, ec)) { if (isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out); ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_connect } else { // an error occured in async_connect or async_handshake
timer->cancel(); timer->cancel();
} }
@ -382,17 +412,21 @@ public:
if (!ec) { if (!ec) {
if (start) { if (start) {
start = false; start = false;
if constexpr (std::derived_from<socket_t, if constexpr (isTLS) {
asio::basic_stream_socket<typename socket_t::protocol_type>>) { return asio::async_write(_sessSocket, buff_seq, std::move(self));
return asio::async_write(_socket, buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else { } else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!"); if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_write(_socket, buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
} }
} }
} }
@ -455,22 +489,30 @@ public:
} }
auto n_avail = _socket.available(); auto n_avail = _socket.available();
if (!n_avail) {
return _socket.async_wait(asio::ip::tcp::socket::wait_read, std::move(self));
}
auto buff = _streamBuffer.prepare(n_avail ? n_avail : 1); auto buff = _streamBuffer.prepare(n_avail ? n_avail : 1);
if constexpr (std::derived_from<socket_t, if constexpr (isTLS) {
asio::basic_stream_socket<typename socket_t::protocol_type>>) { return asio::async_read(_sessSocket, std::move(buff), asio::transfer_at_least(1),
return asio::async_read(_socket, std::move(buff), asio::transfer_at_least(1),
std::move(self)); std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), *out_flags, std::move(self));
} else { } else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!"); if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_read(_socket, std::move(buff), asio::transfer_at_least(1),
std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), *out_flags, std::move(self));
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
} }
} }
@ -573,6 +615,12 @@ public:
{ {
std::error_code ec; std::error_code ec;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (isTLS) {
_sessSocket.shutdown(_shutdownType, ec);
}
#endif
_socket.shutdown(_shutdownType, ec); _socket.shutdown(_shutdownType, ec);
if (!ec) { if (!ec) {
_socket.close(ec); _socket.close(ec);
@ -633,7 +681,6 @@ protected:
asio::io_context::strand _receiveStrand; asio::io_context::strand _receiveStrand;
socket_t _socket; socket_t _socket;
tls_stream_t _tlsStream;
asio::streambuf _streamBuffer; asio::streambuf _streamBuffer;
@ -641,195 +688,14 @@ protected:
asio::socket_base::shutdown_type _shutdownType = asio::socket_base::shutdown_both; asio::socket_base::shutdown_type _shutdownType = asio::socket_base::shutdown_both;
template <typename CancelableT, traits::adc_time_duration_c TimeoutT>
static std::unique_ptr<asio::steady_timer> getDeadlineTimer(CancelableT& obj,
const TimeoutT& timeout,
bool arm = true)
{
auto timer = std::make_unique<asio::steady_timer>(obj.get_executor());
// if (timeout == std::chrono::duration<typename TimeoutT::rep, typename TimeoutT::period>::max()) {
// return timer; // do not arm the timer if MAX duration are given
// }
if (arm) {
std::chrono::seconds max_d = std::chrono::duration_cast<std::chrono::seconds>(
std::chrono::steady_clock::time_point::max() - std::chrono::steady_clock::now() -
std::chrono::seconds(1));
timer->expires_after(timeout < max_d ? timeout : max_d); // to avoid overflow!
// timer->expires_after(timeout);
timer->async_wait([&obj](const std::error_code& ec) mutable {
if (!ec) {
obj.cancel();
}
});
}
return timer;
}
template <typename TimerT>
static bool isTimeout(const std::unique_ptr<TimerT>& timer, const std::error_code& ec)
{
auto exp_time = timer->expiry();
return (exp_time < std::chrono::steady_clock::now()) && (ec == asio::error::operation_aborted);
}
};
#ifdef USE_OPENSSL_WITH_ASIO #ifdef USE_OPENSSL_WITH_ASIO
session_level_socket_t _sessSocket;
/* std::conditional_t<isTLS, asio::ssl::context, std::nullptr_t> _tlsContext;
template <adc_asio_tls_transport_proto_c TRANSPORT_PROTOT,
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT,
traits::adc_output_char_range RMSGT =
std::vector<char>> // used only for inner storing of message byte sequence
class AdcNetServiceASIOTLS : public AdcNetServiceASIOBase<TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>
{
typedef AdcNetServiceASIOBase<TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT> service_base_t;
public:
using typename service_base_t::socket_t;
typedef asio::ssl::stream<socket_t> tls_stream_t;
// TLS certificate attributes comparison function:
// 'serial' - as returned by OpenSSL BN_bn2hex
// 'fingerprint' - as returned by OpenSSL X509_digest
// 'depth' - depth in chain
// the function must return 0 - if comparison failed; otherwise - something != 0
typedef std::function<int(const std::string& serial, const std::vector<unsigned char>& fingerprint, int depth)>
cert_comp_func_t;
// reimplement acceptor class
class acceptor_t
{
public:
static constexpr std::chrono::duration DEFAULT_ACCEPT_TIMEOUT = std::chrono::seconds::max();
typedef AdcNetServiceASIOTLS netservice_t;
typedef std::shared_ptr<netservice_t> sptr_netservice_t;
acceptor_t(asio::io_context& io_ctx, asio::ssl::context tls_context)
: _ioContext(io_ctx), _endpoint(), _socket(_ioContext), _acceptor(_ioContext)
{
}
acceptor_t(asio::io_context& io_ctx, const service_base_t::endpoint_t& endpoint, asio::ssl::context tls_context)
: _ioContext(io_ctx), _endpoint(endpoint), _socket(_ioContext), _acceptor(_ioContext, endpoint)
{
}
typedef std::function<void(std::error_code, netservice_t)> async_accept_callback_t;
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(TokenT&& token, const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
enum { start, handshake, stop };
this->_socket = AdcNetServiceASIOTLS::socket_t(this->_ioContext);
auto timer = getDeadlineTimer(this->_acceptor, timeout);
netservice_t srv(_ioContext);
return asio::async_compose<TokenT, void(std::error_code, netservice_t)>(
[timer = std::move(timer), srv = std::move(srv), state = start, this](auto& self,
std::error_code ec = {}) mutable {
if (!ec) {
switch (state) {
case start:
state = handshake;
try {
if (!_acceptor.is_open() || (_acceptor.local_endpoint() != _endpoint)) {
_acceptor = _acceptor_t(_ioContext, _endpoint);
}
} catch (std::system_error err) {
timer->cancel();
self.complete(err.code(), netservice_t{_ioContext});
// self.complete(err.code(), std::make_shared<netservice_t>(_ioContext));
return;
}
return _acceptor.async_accept(_socket, std::move(self));
break;
case handshake:
state = stop;
srv._socket = std::move(_socket);
srv._tlsStream = asio::ssl::stream(srv._socket, _tlsContext);
return srv._tlsStream.async_handshake(asio::ssl::stream_base::server, std::move(self));
default:
break;
}
}
if (isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_accept od async_handshake
timer->cancel();
}
self.complete(ec, std::move(srv));
},
token, this->_ioContext);
}
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(const AdcNetServiceASIOTLS::endpoint_t& endpoint,
TokenT&& token,
const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
this->_endpoint = endpoint;
return asyncAccept(std::forward<TokenT>(token), timeout);
}
protected:
asio::io_context& _ioContext;
AdcNetServiceASIOTLS::endpoint_t _endpoint;
AdcNetServiceASIOTLS::socket_t _socket;
asio::ssl::context& _tlsContext;
using _acceptor_t = std::conditional_t<
std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>,
std::nullptr_t, // there is no acceptor
typename TRANSPORT_PROTOT::acceptor>;
_acceptor_t _acceptor;
};
std::error_code close()
{
std::error_code ec;
this->_sock_stream.shutdown(ec); // shutdown OpenSSL stream
if (!ec) {
this->_sock_stream.lowest_layer().shutdown(this->_shutdownType, ec);
if (!ec) {
this->_sock_stream.lowest_layer().close(ec);
}
}
return ec;
}
protected:
tls_stream_t _tlsStream;
asio::ssl::context _tlsContext;
asio::ssl::verify_mode _tlsPeerVerifyMode; asio::ssl::verify_mode _tlsPeerVerifyMode;
std::string _tlsCertFingerprintDigest;
cert_comp_func_t _tlsCertCompFunc; cert_comp_func_t _tlsCertCompFunc;
std::string _tlsCertFingerprintDigest;
asio::streambuf _streamBuffer; // reference implementation fo certificate verification function
// reference implementation
virtual bool verifyCertificate(int preverified_ok, X509_STORE_CTX* store) virtual bool verifyCertificate(int preverified_ok, X509_STORE_CTX* store)
{ {
if (preverified_ok == 0) { if (preverified_ok == 0) {
@ -885,16 +751,72 @@ protected:
return preverified_ok; return preverified_ok;
} }
};
*/
#endif #endif
static_assert(adc::interfaces::adc_netservice_c<AdcNetServiceASIOBase<asio::ip::tcp, adc::AdcStopSeqSessionProto<>>>, template <typename CancelableT, traits::adc_time_duration_c TimeoutT>
""); static std::unique_ptr<asio::steady_timer> getDeadlineTimer(CancelableT& obj,
const TimeoutT& timeout,
bool arm = true)
{
auto timer = std::make_unique<asio::steady_timer>(obj.get_executor());
static_assert(adc::interfaces::adc_netsession_proto_c<adc::AdcStopSeqSessionProto<>>, ""); // if (timeout == std::chrono::duration<typename TimeoutT::rep, typename TimeoutT::period>::max()) {
// return timer; // do not arm the timer if MAX duration are given
// }
if (arm) {
std::chrono::seconds max_d = std::chrono::duration_cast<std::chrono::seconds>(
std::chrono::steady_clock::time_point::max() - std::chrono::steady_clock::now() -
std::chrono::seconds(1));
timer->expires_after(timeout < max_d ? timeout : max_d); // to avoid overflow!
// timer->expires_after(timeout);
timer->async_wait([&obj](const std::error_code& ec) mutable {
if (!ec) {
obj.cancel();
}
});
}
return timer;
}
template <typename TimerT>
static bool isTimeout(const std::unique_ptr<TimerT>& timer, const std::error_code& ec)
{
auto exp_time = timer->expiry();
return (exp_time < std::chrono::steady_clock::now()) && (ec == asio::error::operation_aborted);
}
template <typename TimerT>
static bool isTimeout(const std::shared_ptr<TimerT>& timer, const std::error_code& ec)
{
auto exp_time = timer->expiry();
return (exp_time < std::chrono::steady_clock::now()) && (ec == asio::error::operation_aborted);
}
};
/* */
template <
adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
using AdcNetServiceASIO = AdcBaseNetServiceASIO<false, TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>;
#ifdef USE_OPENSSL_WITH_ASIO
template <
adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
using AdcNetServiceASIOTLS = AdcBaseNetServiceASIO<true, TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>;
#endif
} // namespace adc::impl } // namespace adc::impl

822
net/asio/adc_netsrv1_asio.h Normal file
View File

@ -0,0 +1,822 @@
#pragma once
#include <asio/any_completion_handler.hpp>
#include <asio/basic_datagram_socket.hpp>
#include <asio/basic_seq_packet_socket.hpp>
#include <asio/basic_stream_socket.hpp>
#include <asio/bind_executor.hpp>
#include <asio/compose.hpp>
#include <asio/deferred.hpp>
#include <asio/ip/tcp.hpp>
#include <asio/ip/udp.hpp>
#include <asio/local/datagram_protocol.hpp>
#include <asio/local/seq_packet_protocol.hpp>
#include <asio/local/stream_protocol.hpp>
#include <asio/read.hpp>
#include <asio/read_until.hpp>
#include <asio/steady_timer.hpp>
#include <asio/strand.hpp>
#include <asio/streambuf.hpp>
#include <asio/use_awaitable.hpp>
#include <asio/use_future.hpp>
#include <asio/write.hpp>
#include <functional>
#include <queue>
#ifdef USE_OPENSSL_WITH_ASIO
#include <asio/ssl.hpp>
#include <asio/ssl/stream.hpp>
#endif
#include "../../common/adc_traits.h"
#include "../adc_net_concepts.h"
namespace adc::traits
{
// special ASIO-related template specializations
template <>
struct adc_func_traits<asio::use_future_t<>> {
using ret_t = std::nullptr_t;
using args_t = std::tuple<std::nullptr_t, std::nullptr_t>;
using arg1_t = std::nullptr_t;
static constexpr size_t arity = 0;
};
template <>
struct adc_func_traits<asio::use_awaitable_t<>> {
using ret_t = std::nullptr_t;
using args_t = std::tuple<std::nullptr_t, std::nullptr_t>;
using arg1_t = std::nullptr_t;
static constexpr size_t arity = 0;
};
template <>
struct adc_func_traits<asio::deferred_t> {
using ret_t = std::nullptr_t;
using args_t = std::tuple<std::nullptr_t, std::nullptr_t>;
using arg1_t = std::nullptr_t;
static constexpr size_t arity = 0;
};
} // namespace adc::traits
namespace adc::impl
{
template <typename T>
concept adc_asio_transport_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::ip::udp> ||
std::derived_from<T, asio::local::seq_packet_protocol> || std::derived_from<T, asio::local::stream_protocol> ||
std::derived_from<T, asio::local::datagram_protocol>;
template <typename T>
concept adc_asio_tls_transport_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::local::stream_protocol> ||
std::derived_from<T, asio::local::seq_packet_protocol>;
template <typename T>
concept adc_asio_stream_transport_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::local::stream_protocol>;
template <typename T>
concept adc_asio_is_future = requires {
// [](std::type_identity<asio::use_future_t<>>) {}(std::type_identity<std::remove_cvref_t<T>>());
[]<typename AllocatorT>(std::type_identity<asio::use_future_t<AllocatorT>>) {
}(std::type_identity<std::remove_cvref_t<T>>{});
};
template <typename T>
concept adc_asio_is_awaitable = requires {
[]<typename ExecutorT>(std::type_identity<asio::use_awaitable_t<ExecutorT>>) {
}(std::type_identity<std::remove_cvref_t<T>>{});
};
template <typename T>
concept adc_asio_special_comp_token_c =
adc_asio_is_future<T> || adc_asio_is_awaitable<T> || std::same_as<std::remove_cvref_t<T>, asio::deferred_t>;
template <
bool USE_TLS, // true - use of asio::ssl::stream<TRANSPORT_PROTOT>
adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
class AdcBaseNetServiceASIO : public SESSION_PROTOT
{
public:
#ifdef USE_OPENSSL_WITH_ASIO
static_assert(!(USE_TLS && adc_asio_tls_transport_proto_c<TRANSPORT_PROTOT>),
"INVALID 'TRANSPORT_PROTOT' TEMPLATE ARGUMENT!");
static constexpr bool isTLS = USE_TLS;
#else // ignore USE_TLS
static constexpr bool isTLS = false;
#endif
// typedefs to satisfy 'adc_netservice_c' concept
typedef std::string_view netservice_ident_t;
typedef std::vector<char> send_msg_t; // in general, only one of several possible
typedef RMSGT recv_msg_t; // in general, only one of several possible (see class template arguments declaration)
typedef traits::adc_common_duration_t timeout_t;
using endpoint_t = typename TRANSPORT_PROTOT::endpoint;
// typedefs for completion tokens (callbacks, required by 'adc_netservice_c' concept)
typedef std::function<void(std::error_code)> async_connect_callback_t;
typedef std::function<void(std::error_code)> async_send_callback_t;
typedef std::function<void(std::error_code, RMSGT)> async_receive_callback_t;
// typedefs from transport protocol
using socket_t = typename TRANSPORT_PROTOT::socket; // low-level socket type
#ifdef USE_OPENSSL_WITH_ASIO
using session_level_socket_t = std::conditional_t<isTLS, asio::ssl::stream<socket_t>, std::nullptr_t>;
// TLS certificate attributes comparison function:
// 'serial' - as returned by OpenSSL BN_bn2hex
// 'fingerprint' - as returned by OpenSSL X509_digest
// 'depth' - depth in chain
// the function must return 0 - if comparison failed; otherwise - something != 0
typedef std::function<int(const std::string& serial, const std::vector<unsigned char>& fingerprint, int depth)>
cert_comp_func_t;
#else
using session_level_socket_t = socket_t;
#endif
struct acceptor_t {
using srv_acceptor_t = std::conditional_t<
std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>,
std::nullptr_t, // there is no acceptor
typename TRANSPORT_PROTOT::acceptor>;
static constexpr std::chrono::duration DEFAULT_ACCEPT_TIMEOUT = std::chrono::seconds::max();
typedef AdcBaseNetServiceASIO netservice_t;
typedef std::function<void(std::error_code, netservice_t)> async_accept_callback_t;
acceptor_t(asio::io_context& io_context) : _ioContext(io_context), _acceptor(io_context) {}
acceptor_t(asio::io_context& io_context, const netservice_t::endpoint_t& endpoint)
: _ioContext(io_context), _acceptor(io_context, endpoint)
{
}
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(TokenT&& token, const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
enum { sock_accept, handshake, done };
auto timer = netservice_t::getDeadlineTimer(_acceptor, timeout);
auto srv = std::make_unique<netservice_t>(_ioContext);
return asio::async_compose<TokenT, void(std::error_code, netservice_t)>(
[timer = std::move(timer), srv = std::move(srv), state = sock_accept, this](
auto& self, std::error_code ec = {}) mutable {
if (!ec) {
switch (state) {
case sock_accept:
if constexpr (netservice_t::isTLS) {
state = handshake;
} else {
state = done;
}
return _acceptor.async_accept(srv->_socket, std::move(self));
break;
case handshake:
state = done;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (netservice_t::isTLS) {
srv->_sessSocket =
netservice_t::session_level_socket_t(srv->_socket, srv->_tlsContext);
return srv->_sessSocket.async_handshake(asio::ssl::stream_base::server,
std::move(self));
}
#endif
break;
case done:
break;
default:
break;
}
}
if (netservice_t::isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_accept od async_handshake
timer->cancel();
}
self.complete(ec, std::move(*srv));
srv.reset();
},
token, _ioContext);
}
template <asio::completion_token_for<void(std::error_code, netservice_t)> TokenT,
traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto asyncAccept(const netservice_t::endpoint_t& endpoint,
TokenT&& token,
const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
_acceptor = srv_acceptor_t(_ioContext, endpoint);
return asyncAccept(std::forward<TokenT>(token), timeout);
}
template <traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto accept(const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
auto f = asyncAccept(asio::use_future, timeout);
return f.get();
}
template <traits::adc_time_duration_c DT = decltype(DEFAULT_ACCEPT_TIMEOUT)>
auto accept(const endpoint_t& endpoint, const DT& timeout = DEFAULT_ACCEPT_TIMEOUT)
{
return accept(endpoint, timeout);
}
private:
asio::io_context& _ioContext;
srv_acceptor_t _acceptor;
};
static constexpr std::chrono::duration DEFAULT_CONNECT_TIMEOUT = std::chrono::seconds(10);
static constexpr std::chrono::duration DEFAULT_SEND_TIMEOUT = std::chrono::seconds(5);
static constexpr std::chrono::duration DEFAULT_RECEIVE_TIMEOUT = std::chrono::seconds(5);
/* CONSTRUCTORS AND DESTRUCTOR */
AdcBaseNetServiceASIO(asio::io_context& ctx)
requires(!isTLS)
: SESSION_PROTOT(), _ioContext(ctx), _receiveStrand(_ioContext), _socket(_ioContext), _receiveQueue()
{
}
AdcBaseNetServiceASIO(socket_t socket)
: SESSION_PROTOT(),
_ioContext(static_cast<asio::io_context&>(socket.get_executor().context())),
_socket(std::move(socket)),
_receiveStrand(_ioContext),
_receiveQueue()
{
}
#ifdef USE_OPENSSL_WITH_ASIO
AdcBaseNetServiceASIO(asio::io_context& ctx,
const asio::ssl::context& tls_context,
const asio::ssl::verify_mode& tls_peer_verify_mode = asio::ssl::verify_peer)
requires isTLS
: SESSION_PROTOT(),
_ioContext(ctx),
_receiveStrand(_ioContext),
_socket(_ioContext),
_receiveQueue(),
_tlsContext(tls_context),
_tlsPeerVerifyMode(tls_peer_verify_mode)
{
}
#endif
AdcBaseNetServiceASIO(AdcBaseNetServiceASIO&& other)
: _ioContext(other._ioContext),
_receiveStrand(std::move(other._receiveStrand)),
_socket(std::move(other._socket)),
_sessSocket(std::move(other._sessSocket)),
_streamBuffer(),
_receiveQueue(std::move(other._receiveQueue))
{
auto bytes = asio::buffer_copy(_streamBuffer.prepare(other._streamBuffer.size()), other._streamBuffer.data());
_streamBuffer.commit(bytes);
}
AdcBaseNetServiceASIO(const AdcBaseNetServiceASIO&) = delete; // no copy constructor!
virtual ~AdcBaseNetServiceASIO() {}
AdcBaseNetServiceASIO& operator=(const AdcBaseNetServiceASIO&) = delete;
AdcBaseNetServiceASIO& operator=(AdcBaseNetServiceASIO&& other)
{
_ioContext = other._ioContext;
_receiveStrand = std::move(other._receiveStrand);
_receiveQueue = std::move(other._receiveQueue);
_socket = std::move(other._socket);
_streamBuffer.consume(_streamBuffer.size());
auto bytes = asio::buffer_copy(_streamBuffer.prepare(other._streamBuffer.size()), other._streamBuffer.data());
_streamBuffer.commit(bytes);
return *this;
};
constexpr netservice_ident_t ident() const
{
return _ident;
}
/* asynchronuos methods */
template <asio::completion_token_for<void(std::error_code)> TokenT,
traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_CONNECT_TIMEOUT)>
auto asyncConnect(const endpoint_t& endpoint, TokenT&& token, const TimeoutT& timeout = DEFAULT_CONNECT_TIMEOUT)
{
enum { sock_connect, handshake, done };
auto timer = getDeadlineTimer(_socket, timeout);
return asio::async_compose<TokenT, void(std::error_code)>(
[state = sock_connect, endpoint, timer = std::move(timer), this](auto& self,
std::error_code ec = {}) mutable {
if (!ec) {
switch (state) {
case sock_connect:
if constexpr (isTLS) {
state = handshake;
} else {
state = done;
}
return _socket.async_connect(endpoint, std::move(self));
break;
case handshake:
state = done;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (isTLS) {
_sessSocket = session_level_socket_t(_socket, _tlsContext);
return _sessSocket.async_handshake(session_level_socket_t::client, std::move(self));
}
#endif
break;
case done:
break;
default:
break;
}
}
if (isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_connect or async_handshake
timer->cancel();
}
self.complete(ec);
},
token, _socket);
}
template <traits::adc_input_char_range MessageT,
asio::completion_token_for<void(std::error_code ec)> TokenT,
traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_SEND_TIMEOUT)>
auto asyncSend(const MessageT& msg, TokenT&& token, const TimeoutT& timeout = DEFAULT_SEND_TIMEOUT)
{
// create buffer sequence of sending session protocol representation of the input message
std::vector<asio::const_buffer> buff_seq;
// std::ranges::for_each(this->toProto(msg), [&buff_seq](const auto& el) { buff_seq.emplace_back(el); });
std::ranges::for_each(this->toProto(msg),
[&buff_seq](const auto& el) { buff_seq.emplace_back(el.data(), el.size()); });
auto timer = getDeadlineTimer(_socket, timeout);
return asio::async_compose<TokenT, void(std::error_code)>(
[start = true, buff_seq = std::move(buff_seq), timer = std::move(timer), this](
auto& self, std::error_code ec = {}, size_t = 0) mutable {
if (!ec) {
if (start) {
start = false;
if constexpr (isTLS) {
return asio::async_write(_sessSocket, buff_seq, std::move(self));
} else {
if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_write(_socket, buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff_seq, std::move(self));
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
}
}
}
if (isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_write/async_send
timer->cancel();
}
self.complete(ec);
},
token, _socket);
}
template <typename TokenT, traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_RECEIVE_TIMEOUT)>
auto asyncReceive(TokenT&& token, const TimeoutT& timeout = DEFAULT_RECEIVE_TIMEOUT)
{
// static asio::streambuf _streamBuffer;
// check completion token signature and deduce message type
// if constexpr (!adc_asio_special_comp_token_c<TokenT> && !is_async_ctx_t) {
if constexpr (!adc_asio_special_comp_token_c<TokenT>) {
static_assert(traits::adc_func_traits<TokenT>::arity == 2, "INVALID COMPLETION TOKEN SIGNATURE!");
static_assert(std::is_same_v<std::remove_cvref_t<traits::adc_func_arg1_t<TokenT>>, std::error_code>,
"INVALID COMPLETION TOKEN SIGNATURE!");
static_assert(traits::adc_output_char_range<
std::tuple_element_t<1, typename traits::adc_func_traits<TokenT>::args_t>>,
"INVALID COMPLETION TOKEN SIGNATURE!");
}
using msg_t = std::conditional_t<
// adc_asio_special_comp_token_c<TokenT> || is_async_ctx_t, RMSGT,
adc_asio_special_comp_token_c<TokenT>, RMSGT,
std::remove_cvref_t<std::tuple_element_t<1, typename traits::adc_func_traits<TokenT>::args_t>>>;
auto out_flags = std::make_shared<asio::socket_base::message_flags>();
auto timer = getDeadlineTimer(_socket, timeout);
return asio::async_compose<TokenT, void(std::error_code, msg_t)>(
[out_flags, do_read = true, timer = std::move(timer), this](auto& self, std::error_code ec = {},
size_t nbytes = 0) mutable {
msg_t msg;
if (!ec) {
if (do_read) {
do_read = false;
if (_receiveQueue.size()) { // return message from queue
timer->cancel();
auto imsg = _receiveQueue.front();
_receiveQueue.pop();
if constexpr (std::is_same_v<msg_t, RMSGT>) {
self.complete(std::error_code(), std::move(imsg));
} else {
self.complete(std::error_code(), {imsg.begin(), imsg.end()});
}
return;
}
auto n_avail = _socket.available();
if (!n_avail) {
return _socket.async_wait(asio::ip::tcp::socket::wait_read, std::move(self));
}
auto buff = _streamBuffer.prepare(n_avail ? n_avail : 1);
if constexpr (isTLS) {
return asio::async_read(_sessSocket, std::move(buff), asio::transfer_at_least(1),
std::move(self));
} else {
if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_read(_socket, std::move(buff), asio::transfer_at_least(1),
std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
// datagram, so it should be received at once
return _socket.async_receive(std::move(buff), *out_flags, std::move(self));
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
}
}
// zero-length message for SEQ_PACK sockets is EOF
if constexpr (std::derived_from<socket_t,
asio::basic_seq_packet_socket<typename socket_t::protocol_type>>) {
if (!nbytes) {
timer->cancel();
self.complete(std::error_code(asio::error::misc_errors::eof), std::move(msg));
return;
}
}
_streamBuffer.commit(nbytes);
// if (!nbytes) {
// do_read = true;
// asio::post(std::move(self)); // initiate consequence socket's read operation
// return;
// }
auto start_ptr = static_cast<const char*>(_streamBuffer.data().data());
auto net_pack = this->search(std::span(start_ptr, _streamBuffer.size()));
if (net_pack.empty()) {
do_read = true;
asio::post(std::move(self)); // initiate consequence socket's read operation
return;
}
timer->cancel(); // there were no errors in the asynchronous read-operation, so stop timer
// here one has at least a single message
std::ranges::copy(this->fromProto(net_pack), std::back_inserter(msg));
_streamBuffer.consume(net_pack.size());
while (_streamBuffer.size()) { // search for possible additional session protocol packets
start_ptr = static_cast<const char*>(_streamBuffer.data().data());
net_pack = this->search(std::span(start_ptr, _streamBuffer.size()));
if (!net_pack.empty()) {
_receiveQueue.emplace();
std::ranges::copy(this->fromProto(net_pack), std::back_inserter(_receiveQueue.back()));
_streamBuffer.consume(net_pack.size());
} else {
break; // exit and hold remaining bytes in stream buffer
}
}
}
if (isTimeout(timer, ec)) {
ec = std::make_error_code(std::errc::timed_out);
} else { // an error occured in async_*
timer->cancel();
}
if constexpr (std::is_same_v<msg_t, RMSGT>) {
self.complete(ec, std::move(msg));
} else {
self.complete(ec, {msg.begin(), msg.end()});
}
// if constexpr (adc_asio_special_comp_token_c<TokenT>) {
// self.complete(ec, std::move(msg));
// } else { // may be of non-RMSGT type
// self.complete(ec, {msg.begin(), msg.end()});
// }
},
token, _receiveStrand);
}
/* blocking methods */
template <traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_CONNECT_TIMEOUT)>
auto connect(const endpoint_t& endpoint, const TimeoutT& timeout = DEFAULT_CONNECT_TIMEOUT)
{
std::future<void> ftr = asyncConnect(endpoint, asio::use_future, timeout);
ftr.get();
}
template <traits::adc_input_char_range R, traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_SEND_TIMEOUT)>
auto send(const R& msg, const TimeoutT& timeout = DEFAULT_SEND_TIMEOUT)
{
std::future<void> ftr = asyncSend(msg, asio::use_future, timeout);
ftr.get();
}
template <traits::adc_time_duration_c TimeoutT = decltype(DEFAULT_RECEIVE_TIMEOUT)>
auto receive(const TimeoutT& timeout = DEFAULT_RECEIVE_TIMEOUT)
{
std::future<RMSGT> ftr = asyncReceive(asio::use_future, timeout);
return ftr.get();
}
// one still may receive messages from queue!
std::error_code close()
{
std::error_code ec;
#ifdef USE_OPENSSL_WITH_ASIO
if constexpr (isTLS) {
_sessSocket.shutdown(_shutdownType, ec);
}
#endif
_socket.shutdown(_shutdownType, ec);
if (!ec) {
_socket.close(ec);
}
return ec;
}
/* additional ASIO-related methods */
void clearRcvQueue()
{
// clear receiving messages queue
// NOTE: there is no racing condition here since using asio::strand!
asio::post(_receiveStrand, [this]() {
//
_receiveQueue = {};
});
}
void clearRcvBuff()
{
asio::post(_receiveStrand, [this]() {
//
_streamBuffer.consume(_streamBuffer.size());
});
}
void clearRcvData()
{
asio::post(_receiveStrand, [this]() {
_receiveQueue = {};
_streamBuffer.consume(_streamBuffer.size());
});
}
void setShutdownType(asio::socket_base::shutdown_type shutdown_type)
{
_shutdownType = shutdown_type;
}
asio::socket_base::shutdown_type getShutdownType() const
{
return _shutdownType;
}
protected:
static constexpr netservice_ident_t _ident =
std::derived_from<socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>
? "STREAM-SOCKET NETWORK SERVICE"
: std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>
? "DATAGRAM-SOCKET NETWORK SERVICE"
: std::derived_from<socket_t, asio::basic_seq_packet_socket<typename socket_t::protocol_type>>
? "SEQPACKET-SOCKET NETWORK SERVICE"
: "UNKNOWN";
asio::io_context& _ioContext;
asio::io_context::strand _receiveStrand;
socket_t _socket;
asio::streambuf _streamBuffer;
std::queue<std::vector<char>> _receiveQueue;
asio::socket_base::shutdown_type _shutdownType = asio::socket_base::shutdown_both;
#ifdef USE_OPENSSL_WITH_ASIO
session_level_socket_t _sessSocket;
std::conditional_t<isTLS, asio::ssl::context, std::nullptr_t> _tlsContext;
asio::ssl::verify_mode _tlsPeerVerifyMode;
cert_comp_func_t _tlsCertCompFunc;
std::string _tlsCertFingerprintDigest;
// reference implementation fo certificate verification function
virtual bool verifyCertificate(int preverified_ok, X509_STORE_CTX* store)
{
if (preverified_ok == 0) {
int err = X509_STORE_CTX_get_error(store);
auto err_str = X509_verify_cert_error_string(err);
// log_error("TLS certificate verification error: {}", err_str);
return preverified_ok;
}
char subject_name[256];
int depth;
ASN1_INTEGER* serial;
BIGNUM* bnser;
X509* cert = X509_STORE_CTX_get_current_cert(store);
if (cert != NULL) {
depth = X509_STORE_CTX_get_error_depth(store);
X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
serial = X509_get_serialNumber(cert); // IT IS INTERNAL POINTER SO IT MUST NOT BE FREED UP!!!
bnser = ASN1_INTEGER_to_BN(serial, NULL);
auto serial_hex = BN_bn2hex(bnser);
// log_debug("Received TLS certificate: SUBJECT = {}, SERIAL = {}, DEPTH = {}", subject_name, serial_hex,
// depth);
// if no compare function then do not compute fingerprint
if (_tlsCertCompFunc) {
// compute certificate fingerprint
unsigned char digest_buff[EVP_MAX_MD_SIZE];
const EVP_MD* digest = EVP_get_digestbyname(_tlsCertFingerprintDigest.c_str());
unsigned int N;
if (X509_digest(cert, digest, digest_buff, &N)) {
preverified_ok = _tlsCertCompFunc(std::string(serial_hex),
std::vector<unsigned char>(digest_buff, digest_buff + N), depth);
} else {
// log_error("Cannot compute client certificate fingerprint! Cannot verify the certificate!");
preverified_ok = 0;
}
}
BN_free(bnser);
OPENSSL_free(serial_hex);
} else {
// log_error("OpenSSL error: cannot get current certificate");
preverified_ok = 0;
}
return preverified_ok;
}
#endif
template <typename CancelableT, traits::adc_time_duration_c TimeoutT>
static std::unique_ptr<asio::steady_timer> getDeadlineTimer(CancelableT& obj,
const TimeoutT& timeout,
bool arm = true)
{
auto timer = std::make_unique<asio::steady_timer>(obj.get_executor());
// if (timeout == std::chrono::duration<typename TimeoutT::rep, typename TimeoutT::period>::max()) {
// return timer; // do not arm the timer if MAX duration are given
// }
if (arm) {
std::chrono::seconds max_d = std::chrono::duration_cast<std::chrono::seconds>(
std::chrono::steady_clock::time_point::max() - std::chrono::steady_clock::now() -
std::chrono::seconds(1));
timer->expires_after(timeout < max_d ? timeout : max_d); // to avoid overflow!
// timer->expires_after(timeout);
timer->async_wait([&obj](const std::error_code& ec) mutable {
if (!ec) {
obj.cancel();
}
});
}
return timer;
}
template <typename TimerT>
static bool isTimeout(const std::unique_ptr<TimerT>& timer, const std::error_code& ec)
{
auto exp_time = timer->expiry();
return (exp_time < std::chrono::steady_clock::now()) && (ec == asio::error::operation_aborted);
}
template <typename TimerT>
static bool isTimeout(const std::shared_ptr<TimerT>& timer, const std::error_code& ec)
{
auto exp_time = timer->expiry();
return (exp_time < std::chrono::steady_clock::now()) && (ec == asio::error::operation_aborted);
}
};
/* */
template <
adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
using AdcNetServiceASIO = AdcBaseNetServiceASIO<false, TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>;
#ifdef USE_OPENSSL_WITH_ASIO
template <
adc_asio_transport_proto_c TRANSPORT_PROTOT, // transport-level proto (e.g. asio::ip::tcp)
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT, // session-level proto (see ../adc_netproto.h)
traits::adc_output_char_range RMSGT = std::vector<char>> // used only for inner storing of message byte sequence
using AdcNetServiceASIOTLS = AdcBaseNetServiceASIO<true, TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>;
#endif
} // namespace adc::impl

6
tests/adc_netservice_test.cpp
View File

@ -2,8 +2,7 @@
#include <iostream> #include <iostream>
#include "../net/adc_netproto.h" #include "../net/adc_netproto.h"
// #include "../net/asio/adc_netservice_asio.h" #include "../net/asio/adc_netservice_asio.h"
#include "../net/asio/adc_netsrv_asio.h"
template <typename T> template <typename T>
void receive(T srv) void receive(T srv)
@ -35,9 +34,10 @@ int main()
asio::io_context ctx; asio::io_context ctx;
// using srv_t = adc::impl::AdcNetServiceASIOBase<tr_p_t, adc::AdcStopSeqSessionProto<>>; // using srv_t = adc::impl::AdcNetServiceASIOBase<tr_p_t, adc::AdcStopSeqSessionProto<>>;
using srv_t = adc::impl::AdcNetServiceASIOBase<tr_p_t, adc::AdcStopSeqSessionProto<>>; using srv_t = adc::impl::AdcNetServiceASIO<tr_p_t, adc::AdcStopSeqSessionProto<>>;
typename srv_t::acceptor_t acc(ctx, ept_c); typename srv_t::acceptor_t acc(ctx, ept_c);
// typename adc::impl::AdcNetserviceAcceptor<srv_t> acc(ctx, ept_c);
acc.asyncAccept([](std::error_code ec, auto srv) { acc.asyncAccept([](std::error_code ec, auto srv) {
if (!ec) { if (!ec) {