This commit is contained in:
Timur A. Fatkhullin 2024-09-30 18:24:11 +03:00
parent 8aef1a7c25
commit 497b28f83e
2 changed files with 136 additions and 49 deletions

View File

@ -25,51 +25,20 @@ namespace adc
{
class AdcNetServer
/* SOME USEFULL PRIVITIVES */
// Ageneric implementation pf POSIX OS daemon
class AdcPosixGenericDaemon
{
protected:
public:
typedef std::string server_ident_t;
virtual ~AdcNetServer() = default;
virtual server_ident_t serverIdent() const
{
return _serverIdent;
}
template <interfaces::adc_netsession_c SessionT, typename... NetsrvCtorArgTs>
void start(const typename SessionT::netservice_t::endpoint_t& endpoint,
const typename SessionT::netsession_ident_t& id,
typename SessionT::netsession_ctx_t&& sess_ctx,
NetsrvCtorArgTs&&... ctor_args)
{
typename SessionT::netservice_t netservice(std::forward<NetsrvCtorArgTs>(ctor_args)...);
netservice.asyncAccept(endpoint, [&endpoint, &id, sess_ctx, this](auto ec, auto...) {
if (!ec) {
auto sess = std::make_shared<SessionT>(id, std::forward<typename SessionT::netsession_ctx_t>(sess_ctx));
startSession(sess);
start(endpoint, id, sess_ctx);
}
});
};
virtual void stop()
{
stopAllSessions();
};
virtual ~AdcPosixGenericDaemon() = default;
// run server as daemon (still only on POSIX OSes)
virtual void daemonize()
void daemonize()
{
daemonizePrepare();
// reference implementation of forking for POSIX OSes
// reference implementation of forking for POSIX OSes
#ifdef FORK_EXISTS
// get TEMP directory in OS
@ -108,23 +77,38 @@ public:
close(1);
close(2);
// _ioContext.notify_fork(asio::io_context::fork_child);
// _ioContext.notify_fork(asio::io_context::fork_child);
#endif
daemonizeFinalize();
}
protected:
virtual void daemonizePrepare() = 0;
virtual void daemonizeFinalize() = 0;
};
// a basic network session manager (basic start and stop functionality)
class AdcNetSessionManager
{
public:
virtual ~AdcNetSessionManager() = default;
protected:
server_ident_t _serverIdent;
template <interfaces::adc_netsession_c SessionT>
constexpr static bool anySessionPredicate(const typename SessionT::netsession_ident_t&)
{
return true;
}
// started sessions weak pointers
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;
static std::unordered_map<const AdcNetSessionManager*, std::set<std::weak_ptr<SessionT>>> _serverSessions;
std::vector<std::function<bool()>> _stopSessionFunc;
template <interfaces::adc_netsession_c SessionT>
void startSession(const typename SessionT::shared_ptr_t& sess_ptr)
void startSession(std::shared_ptr<SessionT>& sess_ptr)
{
auto res = _serverSessions<SessionT>[this].emplace(sess_ptr);
if (res.second) {
@ -134,22 +118,99 @@ protected:
if (!res.first.expired()) { // session is still existing
auto sess = res.first.lock();
sess->stop();
return true;
} else {
return false;
}
});
}
}
void stopAllSessions()
template <interfaces::adc_netsession_c SessionT,
std::predicate<typename SessionT::netsession_ident_t> PredT = decltype(anySessionPredicate<SessionT>)>
size_t stopSessions(PredT&& comp_func = anySessionPredicate<SessionT>())
{
for (auto& func : _stopSessionFunc) {
func();
size_t N = 0;
for (auto& wptr : _serverSessions<SessionT>[this]) {
if (std::shared_ptr<SessionT> sptr = wptr.lock()) {
if constexpr (std::same_as<PredT, decltype(anySessionPredicate<SessionT>)>) {
sptr->stop();
++N;
} else {
if (std::forward<PredT>(comp_func)(sptr->ident())) {
sptr->stop();
++N;
}
}
}
}
return N;
}
virtual void daemonizePrepare() = 0;
virtual void daemonizeFinalize() = 0;
size_t stopAllSessions()
{
size_t N = 0;
for (auto& func : _stopSessionFunc) {
func() ? ++N : 0;
}
return N;
}
};
/* very generic network server */
class AdcGenericNetServer : public AdcPosixGenericDaemon, public AdcNetSessionManager
{
protected:
public:
typedef std::string server_ident_t;
virtual ~AdcGenericNetServer() = default;
virtual server_ident_t ident() const
{
return _serverIdent;
}
template <interfaces::adc_netsession_c SessionT, typename... NetsrvCtorArgTs>
void start(const typename SessionT::netservice_t::endpoint_t& endpoint,
const typename SessionT::netsession_ident_t& id,
typename SessionT::netsession_ctx_t&& sess_ctx,
NetsrvCtorArgTs&&... ctor_args)
{
typename SessionT::netservice_t netservice(std::forward<NetsrvCtorArgTs>(ctor_args)...);
netservice.asyncAccept(endpoint, [&endpoint, &id, sess_ctx, this](auto ec, auto...) {
if (!ec) {
auto sess = std::make_shared<SessionT>(id, std::forward<typename SessionT::netsession_ctx_t>(sess_ctx));
startSession(sess);
start(endpoint, id, sess_ctx);
}
});
};
virtual void start() = 0;
virtual void stop()
{
stopAllSessions();
};
protected:
server_ident_t _serverIdent;
};

View File

@ -219,6 +219,9 @@ public:
_socket(std::move(other._socket)),
_streamBuffer()
{
if (*this == other)
return;
auto bytes = asio::buffer_copy(_streamBuffer.prepare(other._streamBuffer.size()), other._streamBuffer.data());
_streamBuffer.commit(bytes);
};
@ -228,6 +231,29 @@ public:
virtual ~AdcNetServiceASIOBase() {}
AdcNetServiceASIOBase& operator=(const AdcNetServiceASIOBase&) = delete;
AdcNetServiceASIOBase& operator=(AdcNetServiceASIOBase&& other)
{
if (*this != other) {
close();
_streamBuffer.consume(_streamBuffer.size());
auto bytes =
asio::buffer_copy(_streamBuffer.prepare(other._streamBuffer.size()), other._streamBuffer.data());
_streamBuffer.commit(bytes);
_ioContext = other._ioContext;
_receiveStrand = std::move(other._receiveStrand), _socket = std::move(other._socket);
_acceptor = std::move(other._acceptor);
_receiveQueue = other._receiveQueue;
}
return *this;
}
constexpr netservice_ident_t ident() const
{
return _ident;