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-13 22:24:12 +03:00
parent f215ea7a6d
commit 9818b5f2b8
1 changed files with 31 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
net/adc_netservice_asio.h
View File

@ -126,7 +126,7 @@ public:
if (ctx.use_future) { if (ctx.use_future) {
return _socket.async_connect( return _socket.async_connect(
endpoint, asio::use_future([&ctx, timer = std::move(timer)](std::error_code ec) { timer->cancel(); })); endpoint, asio::use_future([&ctx, timer = std::move(timer)](std::error_code) { timer->cancel(); }));
} else { } else {
return _socket.async_connect(endpoint, [&ctx, timer = std::move(timer)](std::error_code ec) { return _socket.async_connect(endpoint, [&ctx, timer = std::move(timer)](std::error_code ec) {
timer->cancel(); timer->cancel();
@ -149,6 +149,9 @@ public:
try { try {
acceptor = typename TRANSPORT_PROTOT::acceptor(_ioContext, endpoint); acceptor = typename TRANSPORT_PROTOT::acceptor(_ioContext, endpoint);
} catch (std::system_error err) { } catch (std::system_error err) {
if (ctx.use_future) { // emulation of asio::use_future behaivior?!
throw;
}
ctx.accept_comp_token(err.code()); ctx.accept_comp_token(err.code());
return; return;
} }
@ -157,7 +160,7 @@ public:
if (ctx.use_future) { if (ctx.use_future) {
return _socket.async_accept( return _socket.async_accept(
endpoint, asio::use_future([&ctx, timer = std::move(timer)](std::error_code ec) { timer->cancel(); })); endpoint, asio::use_future([&ctx, timer = std::move(timer)](std::error_code) { timer->cancel(); }));
} else { } else {
return _socket.async_accept(endpoint, [&ctx, timer = std::move(timer)](std::error_code ec) { return _socket.async_accept(endpoint, [&ctx, timer = std::move(timer)](std::error_code ec) {
timer->cancel(); timer->cancel();
@ -182,15 +185,34 @@ public:
}, },
[&ctx, s_res, timer = std::move(timer), this](std::error_code ec, size_t) { [&ctx, s_res, timer = std::move(timer), this](std::error_code ec, size_t) {
timer->cancel(); timer->cancel();
if (ec) { R msg;
return;
if (!ec) {
std::string_view net_pack{std::get<0>(*s_res), std::get<1>(*s_res)};
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
auto begin_it = (const char*)traits::asio_streambuff_iter_t::begin(_streamBuffer.data());
auto end_it = (const char*)traits::asio_streambuff_iter_t::end(_streamBuffer.data());
// static_cast<std::ranges::iterator_t<std::string_view>>(_streamBuffer.data().data());
// auto end_it = begin_it + _streamBuffer.data().size();
*s_res = this->search(std::span(begin_it, end_it));
if (std::get<2>(*s_res)) {
net_pack = std::string_view{std::get<0>(*s_res), std::get<1>(*s_res)};
std::ranges::copy(this->fromProto(net_pack), std::back_inserter(msg));
_streamBuffer.consume(net_pack.size());
// TODO: insert to queue
} else {
break;
}
}
} }
R msg;
std::string_view net_pack{std::get<0>(*s_res), std::get<1>(*s_res)};
std::ranges::copy(this->fromProto(net_pack), std::back_inserter(msg));
_streamBuffer.consume(net_pack.size());
ctx.accept_comp_token(ec, std::move(msg)); ctx.accept_comp_token(ec, std::move(msg));
}); });