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-24 18:09:59 +03:00
parent eb44cd114d
commit 02dee70db9
3 changed files with 102 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
net/adc_netproto.h
View File

@ -36,16 +36,27 @@ struct AdcStopSeqSessionProto {
typedef std::string proto_ident_t; typedef std::string proto_ident_t;
proto_ident_t ident() const { return "STOP SEQUENCE PROTO"; } proto_ident_t ident() const
{
return "STOP SEQUENCE PROTO";
}
template <traits::adc_input_char_range R> template <traits::adc_input_char_range R>
auto search(const R& r) auto search(const R& r)
{ {
std::tuple<std::ranges::iterator_t<R>, std::ranges::iterator_t<R>, bool> res{r.begin(), r.end(), false}; std::tuple<std::ranges::iterator_t<R>, std::ranges::iterator_t<R>, bool> res{r.begin(), r.end(), false};
std::get<1>(res) = std::search(r.begin(), r.end(), STOP_SEQ.begin(), STOP_SEQ.end()); if (!r.size()) {
if (std::get<1>(res) != r.end()) { // move iterator to the one-past-the-end position return res;
std::advance(std::get<1>(res), STOP_SEQ_SIZE); }
// std::get<1>(res) = std::search(r.begin(), r.end(), STOP_SEQ.begin(), STOP_SEQ.end());
auto found = std::ranges::search(r, STOP_SEQ);
// if (std::get<1>(res) != r.end()) { // move iterator to the one-past-the-end position
if (!found.empty()) { // move iterator to the one-past-the-end position
// std::advance(std::get<1>(res), STOP_SEQ_SIZE);
// std::get<1>(res) = found.begin() + STOP_SEQ_SIZE;
std::get<1>(res) = found.end();
std::get<2>(res) = true; std::get<2>(res) = true;
} }
@ -53,6 +64,28 @@ struct AdcStopSeqSessionProto {
} }
template <std::input_iterator IT>
std::pair<IT, bool> search(IT begin, IT end)
{
auto res = std::make_pair(begin, false);
if (begin == end) {
return res;
}
res.first = std::search(begin, end, STOP_SEQ.begin(), STOP_SEQ.end());
if (res.first != end) {
std::advance(res.first, STOP_SEQ_SIZE); // move iterator to the one-past-the-end position
res.second = true;
} else {
// may be only a part of valid byte sequence was received,
// so start next matching from previous begin-iterator
res.first = begin;
}
return res;
}
template <traits::adc_input_char_range R> template <traits::adc_input_char_range R>
auto toProto(const R& r) auto toProto(const R& r)
{ {

62
net/asio/adc_netservice_asio.h
View File

@ -191,7 +191,12 @@ public:
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) AdcNetServiceASIOBase(asio::io_context& ctx)
: SESSION_PROTOT(), _ioContext(ctx), _receiveStrand(_ioContext), _receiveQueue(), _socket(_ioContext) : SESSION_PROTOT(),
_ioContext(ctx),
_receiveStrand(_ioContext),
_receiveQueue(),
_acceptor(_ioContext),
_socket(_ioContext)
{ {
} }
@ -228,26 +233,30 @@ public:
static_assert(false, "INVALID TRANSPORT PROTOCOL TYPE!"); static_assert(false, "INVALID TRANSPORT PROTOCOL TYPE!");
} }
auto acc = acceptor_t(_ioContext); // auto acc = acceptor_t(_ioContext);
auto timer = getDeadlineTimer(acc, timeout); // auto timer = getDeadlineTimer(acc, timeout);
auto timer = getDeadlineTimer(_acceptor, timeout);
return asio::async_compose<TokenT, void(std::error_code)>( return asio::async_compose<TokenT, void(std::error_code)>(
[acc = std::move(acc), timer = std::move(timer), start = true, &endpoint, this]( // [acc = std::move(acc), timer = std::move(timer), start = true, &endpoint, this](
auto& self, std::error_code ec = {}) mutable { [timer = std::move(timer), start = true, &endpoint, this](auto& self, std::error_code ec = {}) mutable {
if (!ec) { if (!ec) {
if (start) { if (start) {
start = false; start = false;
try { try {
acc = acceptor_t(_ioContext, endpoint); // acc = acceptor_t(_ioContext, endpoint);
// _socket.open(asio::ip::tcp::v4()); if (!_acceptor.is_open() || (_acceptor.local_endpoint() != endpoint)) {
_acceptor = acceptor_t(_ioContext, endpoint);
}
} catch (std::system_error err) { } catch (std::system_error err) {
timer->cancel(); timer->cancel();
self.complete(err.code()); self.complete(err.code());
return; return;
} }
return acc.async_accept(_socket, std::move(self)); // return acc.async_accept(_socket, std::move(self));
return _acceptor.async_accept(_socket, std::move(self));
} }
} }
@ -354,8 +363,10 @@ public:
std::remove_cvref_t<std::tuple_element_t<1, typename traits::adc_func_traits<TokenT>::args_t>>>; 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>>(); // auto s_res = std::make_shared<std::invoke_result_t<decltype(this->template search<RMSGT>), RMSGT>>();
auto tp = this->search(std::span<const char>()); auto tp = this->search(std::span<const char>());
auto s_res = std::make_shared<decltype(tp)>(); auto s_res = std::make_shared<decltype(tp)>();
// auto s_res = std::make_shared<std::tuple<asio_streambuff_iter_t, asio_streambuff_iter_t, bool>>();
auto out_flags = std::make_shared<asio::socket_base::message_flags>(); auto out_flags = std::make_shared<asio::socket_base::message_flags>();
@ -390,13 +401,14 @@ public:
// std::pair<IT, bool> res{std::get<1>(*s_res), std::get<2>(*s_res)}; // std::pair<IT, bool> res{std::get<1>(*s_res), std::get<2>(*s_res)};
// return res; // return res;
// }; // };
auto match_func = [s_res, this](asio_streambuff_iter_t begin, asio_streambuff_iter_t end) { // auto match_func = [s_res, this](asio_streambuff_iter_t begin, asio_streambuff_iter_t end)
*s_res = this->search(std::span(&*begin, &*end)); // {
// return std::make_pair(std::get<1>(*s_res), std::get<2>(*s_res)); // *s_res = this->search(std::span(&*begin, &*end));
auto N = std::distance(std::get<0>(*s_res), std::get<1>(*s_res)); // // return std::make_pair(std::get<1>(*s_res), std::get<2>(*s_res));
std::pair<asio_streambuff_iter_t, bool> res{begin + N, std::get<2>(*s_res)}; // auto N = std::distance(std::get<0>(*s_res), std::get<1>(*s_res));
return res; // std::pair<asio_streambuff_iter_t, bool> res{begin + N, std::get<2>(*s_res)};
}; // return res;
// };
// return asio::async_read_until(_socket, _streamBuffer, std::move(match_func), // return asio::async_read_until(_socket, _streamBuffer, std::move(match_func),
// std::move(self)); // std::move(self));
@ -540,7 +552,7 @@ protected:
socket_t _socket; socket_t _socket;
// acceptor_t _acceptor; acceptor_t _acceptor;
asio::streambuf _streamBuffer; asio::streambuf _streamBuffer;
@ -551,10 +563,22 @@ protected:
template <typename T> template <typename T>
auto MatchCondition(asio_streambuff_iter_t begin, asio_streambuff_iter_t end, T& s_res) auto MatchCondition(asio_streambuff_iter_t begin, asio_streambuff_iter_t end, T& s_res)
{ {
*s_res = this->search(std::span(&*begin, &*end)); if (begin == end) {
*s_res = this->search(std::span<const char>());
} else {
*s_res = this->search(std::span(&*begin, &*end));
// *s_res = this->search(begin, end);
}
// return std::make_pair(std::get<1>(*s_res), std::get<2>(*s_res)); // return std::make_pair(std::get<1>(*s_res), std::get<2>(*s_res));
auto N = std::distance(std::get<0>(*s_res), std::get<1>(*s_res));
std::pair<asio_streambuff_iter_t, bool> res{begin + N, std::get<2>(*s_res)}; std::pair<asio_streambuff_iter_t, bool> res{end, false};
typename std::iterator_traits<asio_streambuff_iter_t>::difference_type N = 0;
if (std::get<2>(*s_res)) {
N = std::distance(std::get<0>(*s_res), std::get<1>(*s_res));
res = std::make_pair(begin + N, true);
}
return res; return res;
}; };

38
tests/adc_netservice_test.cpp
View File

@ -7,8 +7,11 @@
int main() int main()
{ {
asio::ip::tcp::endpoint ept_s(asio::ip::tcp::v4(), 9999); asio::ip::tcp::endpoint ept_s(asio::ip::tcp::v4(), 9999);
// asio::ip::tcp::endpoint ept_c(asio::ip::make_address_v4("127.0.0.1"), 9999); asio::ip::tcp::endpoint ept_c(asio::ip::make_address_v4("0.0.0.0"), 9999);
asio::ip::tcp::endpoint ept_c(asio::ip::tcp::v4(), 9999); // asio::ip::tcp::endpoint ept_c(asio::ip::address_v4::any(), 9999);
std::cout << "ADDR: " << ept_s << "\n";
std::cout << "ADDR: " << ept_c << "\n";
asio::io_context ctx; asio::io_context ctx;
@ -29,21 +32,24 @@ int main()
// srv.asyncConnect(ept_c, s_ctx); // srv.asyncConnect(ept_c, s_ctx);
// auto res = srv.asyncConnect(ept_c, asio::use_awaitable); // auto res = srv.asyncConnect(ept_c, asio::use_awaitable);
srv.asyncAccept(ept_c, [&srv](std::error_code ec) { srv.asyncAccept(
if (!ec) { ept_c,
std::cout << "New connection\n"; [&srv](std::error_code ec) {
if (!ec) {
std::cout << "New connection\n";
srv.asyncReceive( srv.asyncReceive(
[](std::error_code ec, std::string msg) { [](std::error_code ec, std::string msg) {
if (!ec) { if (!ec) {
std::cout << "Received: " << msg << "\n"; std::cout << "Received: [" << msg << "]\n";
} }
}, },
std::chrono::minutes(3)); std::chrono::minutes(3));
} else { } else {
std::cout << "ACCEPT ERR: " << ec.message() << "\n"; std::cout << "ACCEPT ERR: " << ec.message() << "\n";
} }
}); },
std::chrono::minutes(10));
ctx.run(); ctx.run();