lib/io: refactor I/O module

4 files changed, 291 insertions, 0 deletions

diff --git a/ttwhy/io/mod.cppm b/ttwhy/io/mod.cppm
new file mode 100644
index 0000000..8deff38
--- /dev/null
+++ b/ttwhy/io/mod.cppm
@@ -0,0 +1,5 @@
+export module ttwhy.io;
+
+export import :readers;
+export import :scoped_attributes;
+export import :signal_handling;
diff --git a/ttwhy/io/readers.cppm b/ttwhy/io/readers.cppm
new file mode 100644
index 0000000..d15cc7a
--- /dev/null
+++ b/ttwhy/io/readers.cppm
@@ -0,0 +1,93 @@
+module;
+
+#include <asio.hpp>
+#include <asio/experimental/awaitable_operators.hpp>
+
+#include <array>
+#include <chrono>
+#include <span>
+#include <vector>
+
+export module ttwhy.io:readers;
+
+import ttwhy.routers;
+import ttwhy.scanners;
+
+namespace ttwhy::io
+{
+
+  export template<typename InputStream, router 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<scanners::input_event>{};
+    queue.reserve(16);
+
+    auto sink = [&queue](auto const & event) {
+      queue.push_back(event);
+    };
+
+    using terminal_policy = ttwhy::scanners::associated_terminal_policy_t<AppRouter>;
+    auto scanner = scanners::terminal_scanner<decltype(sink), terminal_policy>{sink};
+
+    auto raw_buffer = std::array<char, 64>{};
+
+    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<char const>{raw_buffer.data(), bytes_read};
+      scanner.process(byte_span);
+
+      for (auto const & event : queue)
+      {
+        co_await router.process(event);
+      }
+      queue.clear();
+    }
+  }
+
+}  // namespace ttwhy::io
diff --git a/ttwhy/io/scoped_attributes.cppm b/ttwhy/io/scoped_attributes.cppm
new file mode 100644
index 0000000..5e42e39
--- /dev/null
+++ b/ttwhy/io/scoped_attributes.cppm
@@ -0,0 +1,135 @@
+module;
+
+#include <termios.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <system_error>
+#include <utility>
+
+export module ttwhy.io:scoped_attributes;
+
+namespace ttwhy::io
+{
+
+  auto static read_attributes(int file_descriptor) -> termios
+  {
+    auto active_attributes = termios{};
+    if (::tcgetattr(file_descriptor, &active_attributes))
+    {
+      throw std::system_error(errno, std::system_category(), "failed to read termios attributes");
+    }
+    return active_attributes;
+  }
+
+  auto static get_line_discipline_flag(int file_descriptor, int flag) -> bool
+  {
+    auto active_attributes = read_attributes(file_descriptor);
+    return active_attributes.c_lflag & flag;
+  }
+
+  auto static set_line_discipline_flag(int file_descriptor, int flag, bool enabled) -> void
+  {
+    auto active_attributes = read_attributes(file_descriptor);
+
+    if ((active_attributes.c_lflag & flag) == enabled)
+    {
+      return;
+    }
+
+    if (enabled)
+    {
+      active_attributes.c_lflag = active_attributes.c_lflag | flag;
+    }
+    else
+    {
+      active_attributes.c_lflag = active_attributes.c_lflag & ~flag;
+    }
+
+    if (::tcsetattr(file_descriptor, TCSANOW, &active_attributes))
+    {
+      throw std::system_error(errno, std::system_category(), "failed to write termios attributes");
+    }
+  }
+
+  //! Store the current TC attributes and restore them on destruction.
+  export struct scoped_attributes
+  {
+    explicit scoped_attributes(int file_descriptor)
+        : m_file_descriptor{-1}
+        , m_attributes{}
+    {
+      if (isatty(file_descriptor))
+      {
+        m_file_descriptor = dup(file_descriptor);
+        m_attributes = read_attributes(m_file_descriptor);
+      }
+    }
+
+    scoped_attributes(scoped_attributes const &) = delete;
+
+    scoped_attributes(scoped_attributes && other)
+        : m_file_descriptor{std::exchange(other.m_file_descriptor, -1)}
+        , m_attributes{std::move(other.m_attributes)}
+    {}
+
+    ~scoped_attributes()
+    {
+      if (m_file_descriptor > 0)
+      {
+        ::tcsetattr(m_file_descriptor, TCSANOW, &m_attributes);
+        ::close(m_file_descriptor);
+      }
+    }
+
+    auto operator=(scoped_attributes const &) -> scoped_attributes & = delete;
+
+    auto operator=(scoped_attributes && other) -> scoped_attributes &
+    {
+      std::ranges::swap(m_file_descriptor, other.m_file_descriptor);
+      std::ranges::swap(m_attributes, other.m_attributes);
+      return *this;
+    }
+
+    //! Check if canonical mode is currently active
+    [[nodiscard]] auto canonical_mode() const noexcept -> bool
+    {
+      return get_line_discipline_flag(m_file_descriptor, ICANON);
+    }
+
+    //! Set whether canonical mode is active or not.
+    auto canonical_mode(bool enabled) & -> scoped_attributes &
+    {
+      set_line_discipline_flag(m_file_descriptor, ICANON, enabled);
+      return *this;
+    }
+
+    auto canonical_mode(bool enabled) && -> scoped_attributes
+    {
+      set_line_discipline_flag(m_file_descriptor, ICANON, enabled);
+      return std::move(*this);
+    }
+
+    [[nodiscard]] auto echo() const noexcept -> bool
+    {
+      return get_line_discipline_flag(m_file_descriptor, ECHO);
+    }
+
+    auto echo(bool enabled) & -> scoped_attributes &
+    {
+      set_line_discipline_flag(m_file_descriptor, ECHO, enabled);
+      return *this;
+    }
+
+    auto echo(bool enabled) && -> scoped_attributes
+    {
+      set_line_discipline_flag(m_file_descriptor, ECHO, enabled);
+      return std::move(*this);
+    }
+
+  private:
+    int m_file_descriptor;
+    termios m_attributes;
+  };
+
+}  // namespace ttwhy::io
diff --git a/ttwhy/io/signal_handling.cppm b/ttwhy/io/signal_handling.cppm
new file mode 100644
index 0000000..1f3fecc
--- /dev/null
+++ b/ttwhy/io/signal_handling.cppm
@@ -0,0 +1,58 @@
+module;
+
+#include <asio.hpp>
+
+#include <csignal>
+#include <format>
+#include <system_error>
+
+export module ttwhy.io:signal_handling;
+
+namespace ttwhy::io
+{
+
+  export template<typename Stream, typename TerminalAttributes>
+  auto handle_signals(Stream & stream, TerminalAttributes & attributes) -> asio::awaitable<std::error_code>
+  {
+    auto executor = co_await asio::this_coro::executor;
+    auto signals = asio::signal_set{executor, SIGINT, SIGTERM};
+    signals.add(SIGTSTP);
+    signals.add(SIGCONT);
+
+    while (true)
+    {
+      auto [error, signal] = co_await signals.async_wait(asio::as_tuple(asio::use_awaitable));
+      if (error)
+      {
+        co_return error;
+      }
+
+      switch (signal)
+      {
+        case SIGINT:
+        case SIGTERM:
+        {
+          auto message = std::format("Received signal {}, exiting ...\n", signal);
+          co_await asio::async_write(stream, asio::buffer(message), asio::use_awaitable);
+          co_return asio::error_code{};
+        }
+        case SIGTSTP:
+        {
+          attributes.canonical_mode(true).echo(true);
+          ::raise(SIGSTOP);
+          break;
+        }
+        case SIGCONT:
+        {
+          attributes.canonical_mode(false).echo(false);
+          break;
+        }
+      }
+
+      if (!error)
+      {
+      }
+    }
+  }
+
+}  // namespace ttwhy::io