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

...

Browse Source
This commit is contained in:
Timur A. Fatkhullin 2024-09-26 18:12:21 +03:00
parent 78f2f1b8ed
commit 1c327f8fd3
4 changed files with 166 additions and 228 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
net/adc_netserver.h
View File

@ -18,203 +18,19 @@ ABSTRACT DEVICE COMPONENTS LIBRARY
#include <cerrno>
#endif
#include <memory>
#include "adc_net_concepts.h"
namespace adc
{
namespace traits
{
// network server session implementation concept
template <typename T>
concept adc_netserver_session_impl_c = requires(T t, const T t_const) {
typename T::session_ident_t;
{ t_const.sessionIdent() } -> std::same_as<typename T::session_ident_t>;
{ t.start() } -> std::same_as<void>;
{ t.stop() } -> std::same_as<void>;
};
} // namespace traits
/* Server session */
class AdcNetServerGenericSession
{
public:
struct Opt {
std::function<void()> startSess;
std::function<void()> stopSess;
std::function<void()> run;
};
AdcNetServerGenericSession(Opt&& opts) : _opts(std::move(opts)) {}
virtual void start()
{
_opts.startSess();
}
virtual void stop()
{
_opts.stopSess();
}
protected:
Opt _opts;
virtual void run()
{
_opts.run();
}
};
template <traits::adc_netserver_session_impl_c ImplT>
class AdcNetServerSession : std::enable_shared_from_this<AdcNetServerSession<ImplT>>
{
protected:
ImplT _impl;
public:
typedef ImplT session_impl_t;
typedef std::shared_ptr<AdcNetServerSession> shared_ptr_t;
typedef std::weak_ptr<AdcNetServerSession> weak_ptr_t;
using typename ImplT::session_ident_t;
template <typename... ImplCtorArgTs>
AdcNetServerSession(ImplCtorArgTs&&... ctor_args) : _impl(std::forward<ImplCtorArgTs>(ctor_args)...)
{
}
virtual ~AdcNetServerSession() = default;
virtual session_ident_t sessionIdent() const
{
//
return _impl._sessionIdent();
}
virtual void start()
{
//
_impl.start();
}
virtual void stop()
{
//
_impl.stop();
}
};
namespace traits
{
// network server session concept
template <typename T>
concept adc_netserver_session_c = requires {
typename T::session_impl_t;
std::derived_from<T, AdcNetServerSession<typename T::session_impl_t>>;
};
} // namespace traits
/* network server */
namespace traits
{
template <typename T>
concept adc_generic_netserver_impl_c = requires(T t, const T t_const) {
typename T::server_ident_t;
{ t_const.serverIdent() } -> std::same_as<typename T::server_ident_t>;
{ t.start() } -> std::same_as<void>;
{ t.stop() } -> std::same_as<void>;
};
template <typename T>
concept adc_netserver_impl_c = requires(T t, const T t_const) {
typename T::server_ident_t;
{ t_const.serverIdent() } -> std::same_as<typename T::server_ident_t>;
{ t.start() } -> std::same_as<void>;
{ t.stop() } -> std::same_as<void>;
{ t.daemonize() } -> std::same_as<void>;
{ t.daemonizePrepare() } -> std::same_as<void>;
{ t.daemonizeFinalize() } -> std::same_as<void>;
};
} // namespace traits
/* VERY GENERIC NETWORK SERVER INTERFACE */
template <traits::adc_generic_netserver_impl_c ImplT>
class AdcGenericNetServer
{
protected:
ImplT _impl;
public:
typedef ImplT server_impl_t;
using typename ImplT::server_ident_t;
template <typename... ImplCtorArgTs>
AdcGenericNetServer(ImplCtorArgTs&&... ctor_args) : _impl(std::make_unique<ImplT>(ctor_args)...)
{
}
virtual ~AdcGenericNetServer() = default;
virtual server_ident_t serverIdent() const
{
//
return _impl.serverIdent();
}
virtual void start()
{
//
_impl.start();
};
virtual void stop()
{
//
_impl.stop();
};
};
template <traits::adc_netserver_impl_c ImplT>
class AdcNetServer
{
protected:
ImplT _impl;
public:
typedef ImplT server_impl_t;
using typename ImplT::server_ident_t;
template <typename... ImplCtorArgTs>
AdcNetServer(ImplCtorArgTs&&... ctor_args) : _impl(std::make_unique<ImplT>(ctor_args)...)
{
}
typedef std::string server_ident_t;
virtual ~AdcNetServer() = default;
@ -222,21 +38,13 @@ public:
virtual server_ident_t serverIdent() const
{
//
return _impl.serverIdent();
return _serverIdent;
}
virtual void start()
{
//
_impl.start();
};
template <interfaces::adc_netservice_c SRVT>
void start(SRVT&& netservice, const typename SRVT::endpoint_t& endpoint) {};
virtual void stop()
{
//
_impl.stop();
};
virtual void stop() {};
// run server as daemon (still only on POSIX OSes)
@ -290,12 +98,15 @@ public:
protected:
server_ident_t _serverIdent;
// started sessions weak pointers
template <traits::adc_netserver_session_c SessionT>
template <interfaces::adc_netsession_c SessionT>
static std::unordered_map<const AdcNetServer*, std::set<typename SessionT::weak_ptr_t>> _serverSessions;
std::vector<std::function<void()>> _stopSessionFunc;
template <traits::adc_netserver_session_c SessionT>
template <interfaces::adc_netsession_c SessionT>
void startSession(const typename SessionT::shared_ptr_t& sess_ptr)
{
auto res = _serverSessions<SessionT>[this].emplace(sess_ptr);

32
net/asio/adc_netservice_asio.h
View File

@ -69,7 +69,7 @@ concept adc_asio_is_awaitable = requires {
template <typename T>
concept adc_asio_special_comp_token =
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>;
@ -198,6 +198,18 @@ public:
}
AdcNetServiceASIOBase(AdcNetServiceASIOBase&& other)
: _ioContext(other._ioContext),
_receiveStrand(std::move(other._receiveStrand)),
_receiveQueue(std::move(_receiveQueue)),
_acceptor(std::move(other._acceptor)),
_socket(std::move(other._socket)),
_streamBuffer()
{
auto bytes = asio::buffer_copy(_streamBuffer.prepare(other._streamBuffer.size()), other._streamBuffer.data());
_streamBuffer.commit(bytes);
};
AdcNetServiceASIOBase(const AdcNetServiceASIOBase&) = delete; // no copy constructor!
virtual ~AdcNetServiceASIOBase() {}
@ -336,7 +348,7 @@ public:
constexpr auto is_async_ctx_t = std::same_as<std::remove_cvref_t<TokenT>, async_call_ctx_t>;
// check completion token signature and deduce message type
if constexpr (!adc_asio_special_comp_token<TokenT> && !is_async_ctx_t) {
if constexpr (!adc_asio_special_comp_token_c<TokenT> && !is_async_ctx_t) {
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!");
@ -346,7 +358,7 @@ public:
}
using msg_t = std::conditional_t<
adc_asio_special_comp_token<TokenT> || is_async_ctx_t, RMSGT,
adc_asio_special_comp_token_c<TokenT> || is_async_ctx_t, RMSGT,
std::remove_cvref_t<std::tuple_element_t<1, typename traits::adc_func_traits<TokenT>::args_t>>>;
// auto s_res = std::make_shared<std::invoke_result_t<decltype(this->template search<RMSGT>), RMSGT>>();
@ -595,15 +607,19 @@ protected:
const TimeoutT& timeout,
bool arm = true)
{
// TODO: now()+timeout overflow!!!
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 (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) {
timer->expires_after(timeout);
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) {

114
net/asio/adc_netsession_asio.h Normal file
View File

@ -0,0 +1,114 @@
#pragma once
#include "adc_netservice_asio.h"
namespace adc::impl
{
class AdcNetSessionASIO
{
public:
typedef std::string netsession_ident_t;
template <traits::adc_input_char_range R,
adc_asio_transport_proto_c TRANSPORT_PROTOT,
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT,
traits::adc_output_char_range RMSGT = std::vector<char>,
traits::adc_is_callable RECV_MSG_TOKENT>
AdcNetSessionASIO(const R& id,
std::shared_ptr<AdcNetServiceASIOBase<TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>> netservice,
RECV_MSG_TOKENT&& recv_msg_token)
: _ident(id.begin(), id.end())
{
// check receive message completion token signature and deduce message type
if constexpr (!adc_asio_special_comp_token_c<RECV_MSG_TOKENT>) {
static_assert(traits::adc_func_traits<RECV_MSG_TOKENT>::arity == 2, "INVALID COMPLETION TOKEN SIGNATURE!");
static_assert(
std::is_same_v<std::remove_cvref_t<traits::adc_func_arg1_t<RECV_MSG_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<RECV_MSG_TOKENT>::args_t>>,
"INVALID COMPLETION TOKEN SIGNATURE!");
}
using msg_t = std::conditional_t<
adc_asio_special_comp_token_c<RECV_MSG_TOKENT>, RMSGT,
std::remove_cvref_t<std::tuple_element_t<1, typename traits::adc_func_traits<RECV_MSG_TOKENT>::args_t>>>;
_startFunc = [netservice, wrapper = traits::adc_pf_wrapper(std::forward<RECV_MSG_TOKENT>(recv_msg_token)),
this]() {
//
netservice->asyncReceive(std::get<0>(wrapper), _recvTimeout);
};
_stopFunc = [netservice]() {
// stop
netservice->close();
};
}
template <adc_asio_transport_proto_c TRANSPORT_PROTOT,
interfaces::adc_netsession_proto_c<std::string_view> SESSION_PROTOT,
traits::adc_output_char_range RMSGT = std::vector<char>,
traits::adc_is_callable RECV_MSG_TOKENT>
AdcNetSessionASIO(std::shared_ptr<AdcNetServiceASIOBase<TRANSPORT_PROTOT, SESSION_PROTOT, RMSGT>> netservice,
RECV_MSG_TOKENT&& recv_msg_token)
: AdcNetSessionASIO(std::derived_from<TRANSPORT_PROTOT, asio::ip::tcp> ? "TCP SESSION"
: std::derived_from<TRANSPORT_PROTOT, asio::ip::udp> ? "UDP SESSION"
: std::derived_from<TRANSPORT_PROTOT, asio::local::seq_packet_protocol>
? "UNIX SEQPACKET SESSION"
: std::derived_from<TRANSPORT_PROTOT, asio::local::stream_protocol> ? "UNIX STREAM SESSION"
: "UNKNOWN",
std::move(netservice),
std::forward<RECV_MSG_TOKENT>(recv_msg_token))
{
}
virtual ~AdcNetSessionASIO()
{
stop();
}
netsession_ident_t ident() const
{
return _ident;
}
void start()
{
_startFunc();
}
void stop()
{
_stopFunc();
}
template <traits::adc_time_duration_c TimeoutT>
AdcNetSessionASIO& setDefaultTimeouts(const TimeoutT& send_timeout, const TimeoutT& recv_timeout)
{
_sendTimeout = send_timeout;
_recvTimeout = recv_timeout;
return *this;
}
protected:
netsession_ident_t _ident;
std::function<void()> _startFunc;
std::function<void()> _stopFunc;
std::chrono::duration<std::chrono::seconds::rep, std::chrono::seconds::period> _recvTimeout =
std::chrono::seconds::max();
std::chrono::duration<std::chrono::seconds::rep, std::chrono::seconds::period> _sendTimeout =
std::chrono::seconds(5);
};
} // namespace adc::impl

37
tests/adc_netservice_test.cpp
View File

@ -47,27 +47,24 @@ int main()
// srv.asyncConnect(ept_c, s_ctx);
// auto res = srv.asyncConnect(ept_c, asio::use_awaitable);
srv.asyncAccept(
ept_c,
[&srv](std::error_code ec) {
if (!ec) {
std::cout << "New connection\n";
srv.asyncAccept(ept_c, [&srv](std::error_code ec) {
if (!ec) {
std::cout << "New connection\n";
// srv.asyncReceive(
// [](std::error_code ec, std::string msg) {
// if (!ec) {
// std::cout << "Received: [" << msg << "]\n";
// } else {
// std::cout << "Received: " << ec.message() << "\n";
// }
// },
// std::chrono::minutes(1));
receive(srv);
} else {
std::cout << "ACCEPT ERR: " << ec.message() << "\n";
}
},
std::chrono::minutes(3));
// srv.asyncReceive(
// [](std::error_code ec, std::string msg) {
// if (!ec) {
// std::cout << "Received: [" << msg << "]\n";
// } else {
// std::cout << "Received: " << ec.message() << "\n";
// }
// },
// std::chrono::minutes(1));
receive(srv);
} else {
std::cout << "ACCEPT ERR: " << ec.message() << "\n";
}
});
ctx.run();