From d05b9343f69a0841043963088e0003368e205bd9 Mon Sep 17 00:00:00 2001 From: Felix Morgner Date: Tue, 22 Sep 2026 22:47:33 +0200 Subject: lib: restructure into vertical slices --- ttwhy/terminal/readers.cppm | 94 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) create mode 100644 ttwhy/terminal/readers.cppm (limited to 'ttwhy/terminal/readers.cppm') diff --git a/ttwhy/terminal/readers.cppm b/ttwhy/terminal/readers.cppm new file mode 100644 index 0000000..ffa5687 --- /dev/null +++ b/ttwhy/terminal/readers.cppm @@ -0,0 +1,94 @@ +module; + +#include +#include + +#include +#include +#include +#include + +export module ttwhy.terminal:readers; + +import :events; +import :policies; +import :scanner; +import ttwhy.core; + +namespace ttwhy::terminal +{ + + export template AppRouter> + auto read_events(InputStream & stream, AppRouter & router) -> asio::awaitable + { + using namespace asio::experimental::awaitable_operators; + using namespace std::chrono_literals; + + auto executor = co_await asio::this_coro::executor; + auto timer = asio::steady_timer{executor}; + + auto queue = std::vector{}; + queue.reserve(16); + + auto sink = [&queue](auto const & event) { + queue.push_back(event); + }; + + auto scanner = terminal::scanner{sink}; + + auto raw_buffer = std::array{}; + + while (true) + { + auto error = asio::error_code{}; + auto bytes_read = 0uz; + + if (scanner.is_pending()) + { + timer.expires_after(50ms); + + auto result = co_await (stream.async_read_some(asio::buffer(raw_buffer), asio::as_tuple(asio::use_awaitable)) || + timer.async_wait(asio::as_tuple(asio::use_awaitable))); + + if (result.index() == 0) + { + std::tie(error, bytes_read) = std::get<0>(result); + } + else + { + scanner.timeout(); + for (auto const & event : queue) + { + co_await router.process(event); + } + queue.clear(); + continue; + } + } + else + { + std::tie(error, bytes_read) = + co_await stream.async_read_some(asio::buffer(raw_buffer), asio::as_tuple(asio::use_awaitable)); + } + + if (error) + { + if (error == asio::error::interrupted) + { + continue; + } + co_return; + } + + auto const byte_span = std::span{raw_buffer.data(), bytes_read}; + scanner.process(byte_span); + + for (auto const & event : queue) + { + co_await router.process(event); + } + queue.clear(); + } + } + +} // namespace ttwhy::terminal -- cgit v1.2.3