1 files changed, 38 insertions, 4 deletions
diff --git a/ttwhy/io.cppm b/ttwhy/io.cppm
index be87d36..09f16c8 100644
--- a/ttwhy/io.cppm
+++ b/ttwhy/io.cppm
@@ -1,7 +1,9 @@
 module;
 
 #include <asio.hpp>
+#include <asio/experimental/awaitable_operators.hpp>
 
+#include <chrono>
 #include <format>
 #include <span>
 #include <vector>
@@ -61,6 +63,12 @@ namespace ttwhy::io
   export template<typename InputStream, typename AppRouter>
   auto read_events(InputStream & stream, AppRouter & router) -> asio::awaitable<void>
   {
+    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<input_event>{};
     queue.reserve(16);
 
@@ -69,8 +77,36 @@ namespace ttwhy::io
 
     while (true)
     {
-      auto [error, bytes_read] =
-          co_await stream.async_read_some(asio::buffer(raw_buffer), asio::as_tuple(asio::use_awaitable));
+      auto error = asio::error_code{};
+      auto bytes_read = std::size_t{};
+
+      if (scanner.is_pending())
+      {
+        timer.expires_after(500ms);
+
+        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)
       {
@@ -82,14 +118,12 @@ namespace ttwhy::io
       }
 
       auto const byte_span = std::span<char const>{raw_buffer.data(), bytes_read};
-
       scanner.process(byte_span);
 
       for (auto const & event : queue)
       {
         co_await router.process(event);
       }
-
       queue.clear();
     }
   }