3 files changed, 370 insertions, 0 deletions
diff --git a/source/lib/src/control/commander.cpp b/source/lib/src/control/commander.cpp
new file mode 100644
index 0000000..4490bb7
--- /dev/null
+++ b/source/lib/src/control/commander.cpp
@@ -0,0 +1,80 @@
+#include "wanda/control/commander.hpp"
+
+#include "wanda/proto/message.hpp"
+#include "wanda/proto/version.hpp"
+#include "wanda/std_ext/optional.hpp"
+#include "wanda/system/logging.hpp"
+
+#include <boost/asio/io_context.hpp>
+#include <spdlog/fmt/ostr.h>
+
+namespace wanda::control
+{
+  commander::commander(boost::asio::io_context & service, std::filesystem::path socket, listener & listener)
+      : m_service{service}
+      , m_endpoint{socket.string()}
+      , m_socket{service}
+      , m_listener{listener}
+  {
+  }
+
+  void commander::start()
+  {
+    m_socket.async_connect(m_endpoint, [&](auto const & error) {
+      if (error)
+      {
+        system::get_logger()->error("error while connecting to control interface: '{}'", error.message());
+      }
+      else
+      {
+        system::get_logger()->info("establishing connection to wanda deamon");
+        m_connection = wanda::control::make_connection(std::move(m_socket));
+        m_connection->add(this);
+        m_connection->start();
+        m_connection->send({proto::message_source_controller, proto::message_command_hello, proto::version});
+      }
+    });
+  }
+
+  void commander::stop()
+  {
+    system::get_logger()->info("closing control connection");
+    m_connection->close();
+  }
+
+  void commander::send(proto::command command)
+  {
+    using namespace wanda::std_ext;
+
+    if (!m_connection || m_connection->current_state() != connection::state::established)
+    {
+      system::get_logger()->error("tried to send command without an established connection");
+      m_listener.on_error(*this, "tried to send command without an established connection");
+      return;
+    }
+
+    with(command.message(), [&](auto const & message) { m_connection->send(message); }) ||
+        [&] { system::get_logger()->error("unknown command"); };
+  }
+
+  void commander::on_error(connection::pointer connection, std::error_code error)
+  {
+    system::get_logger()->error("control interface communication error: '{}'", error.message());
+  }
+
+  void commander::on_received(connection::pointer connection, proto::message message)
+  {
+    if (auto state = connection->current_state(); message.command == "HELLO" && state == connection::state::fresh)
+    {
+      system::get_logger()->info("connection to wanda deamon successfully established");
+      connection->update(connection::state::established);
+      m_listener.on_connected(*this);
+    }
+    else
+    {
+      system::get_logger()->error("unexpected message: '{}'", message);
+      m_listener.on_error(*this, "unexpected message '" + static_cast<std::string>(message) + '\'');
+    }
+  }
+
+}  // namespace wanda::control
+\ No newline at end of file
diff --git a/source/lib/src/control/connection.cpp b/source/lib/src/control/connection.cpp
new file mode 100644
index 0000000..97f41dd
--- /dev/null
+++ b/source/lib/src/control/connection.cpp
@@ -0,0 +1,130 @@
+#include "wanda/control/connection.hpp"
+
+#include "wanda/proto/message.hpp"
+
+#include <boost/asio/completion_condition.hpp>
+#include <boost/asio/read_until.hpp>
+#include <boost/asio/write.hpp>
+#include <boost/system/detail/error_code.hpp>
+#include <boost/system/error_code.hpp>
+
+#include <limits>
+
+namespace wanda::control
+{
+  connection::pointer make_connection(connection::protocol::socket && socket)
+  {
+    return std::make_shared<connection>(connection::key{}, std::move(socket));
+  }
+
+  connection::connection(connection::key key, connection::protocol::socket socket)
+      : keyed{key}
+      , m_socket{std::move(socket)}
+  {
+  }
+
+  bool connection::add(listener * listener)
+  {
+    auto [_, inserted] = m_listeners.insert(listener);
+    return inserted;
+  }
+
+  bool connection::remove(listener * listener)
+  {
+    return m_listeners.erase(listener);
+  }
+
+  void connection::start()
+  {
+    if (m_state == state::unknown)
+    {
+      m_state = state::fresh;
+      perform_read();
+    }
+  }
+
+  void connection::send(proto::message message)
+  {
+    m_output << message << '\n';
+    boost::asio::async_write(m_socket, m_out, boost::asio::transfer_exactly(message.size() + 1), [that = shared_from_this(), this](auto const & error, auto const length) {
+      if (error)
+      {
+        // TODO: Handle error
+      }
+      else
+      {
+        m_out.consume(length);
+      }
+    });
+  }
+
+  void connection::close()
+  {
+    auto error = boost::system::error_code{};
+
+    if (m_socket.cancel(error), error)
+    {
+      for (auto & listener : m_listeners)
+      {
+        listener->on_error(shared_from_this(), error);
+      }
+    }
+
+    if (m_socket.close(error), error)
+    {
+      for (auto & listener : m_listeners)
+      {
+        listener->on_error(shared_from_this(), error);
+      }
+    }
+
+    for (auto & listener : m_listeners)
+    {
+      listener->on_close(shared_from_this());
+    }
+    m_listeners.clear();
+  }
+
+  void connection::update(state state)
+  {
+    m_state = state;
+  }
+
+  connection::state connection::current_state() const
+  {
+    return m_state;
+  }
+
+  void connection::perform_read()
+  {
+    boost::asio::async_read_until(m_socket, m_in, '\n', [that = shared_from_this(), this](auto const & error, auto const length) {
+      if (error)
+      {
+        for (auto & listener : m_listeners)
+        {
+          listener->on_error(shared_from_this(), error);
+        }
+        close();
+      }
+      else
+      {
+        auto msg = proto::message{};
+        m_input >> msg;
+        if (!m_input)
+        {
+          m_input.ignore(std::numeric_limits<std::streamsize>::max());
+          m_input.clear();
+        }
+        else
+        {
+          for (auto & listener : m_listeners)
+          {
+            listener->on_received(shared_from_this(), msg);
+          }
+        }
+        perform_read();
+      }
+    });
+  }
+
+}  // namespace wanda::control
+\ No newline at end of file
diff --git a/source/lib/src/control/interface.cpp b/source/lib/src/control/interface.cpp
new file mode 100644
index 0000000..3ebc55a
--- /dev/null
+++ b/source/lib/src/control/interface.cpp
@@ -0,0 +1,160 @@
+#include "wanda/control/interface.hpp"
+
+#include "wanda/proto/version.hpp"
+#include "wanda/std_ext/optional.hpp"
+#include "wanda/system/logging.hpp"
+
+#include <spdlog/fmt/ostr.h>
+#include <boost/system/error_code.hpp>
+#include <boost/asio/error.hpp>
+#include <unistd.h>
+
+#include <algorithm>
+#include <iterator>
+#include <string>
+#include <system_error>
+#include <utility>
+
+namespace wanda::control
+{
+  // 'socket_deleter' implementation
+
+  socket_deleter::~socket_deleter()
+  {
+    if (std::filesystem::exists(path))
+    {
+      std::filesystem::remove(path);
+    }
+  }
+
+  // 'interface' implementation
+
+  interface::interface(interface::key key, boost::asio::io_context & service, interface::protocol::endpoint endpoint, listener & listener)
+      : keyed{key}
+      , m_service{service}
+      , m_endpoint{std::move(endpoint)}
+      , m_socket{m_service}
+      , m_acceptor{m_service}
+      , m_listener{listener}
+  {
+  }
+
+  std::error_code interface::start()
+  {
+    if (auto error = boost::system::error_code{}; m_acceptor.open(m_endpoint.protocol(), error), error)
+    {
+      return error;
+    }
+
+    if (auto error = boost::system::error_code{}; m_acceptor.bind(m_endpoint, error), error)
+    {
+      return error;
+    }
+
+    if (auto error = boost::system::error_code{}; m_acceptor.listen(128, error), error)
+    {
+      return error;
+    }
+    else
+    {
+      perform_accept();
+      return error;
+    }
+  }
+
+  std::error_code interface::shutdown()
+  {
+    for (auto & connection : m_connections)
+    {
+      connection->close();
+    }
+
+    auto error = boost::system::error_code{};
+    return m_acceptor.close(error), error;
+  }
+
+  void interface::perform_accept()
+  {
+    m_acceptor.async_accept(m_socket, [that = shared_from_this(), this](auto const & error) {
+      if (error && error != boost::asio::error::operation_aborted)
+      {
+        system::get_logger()->error("failed to accept connection because '{}'", error.message());
+      }
+      else
+      {
+        system::get_logger()->info("new incoming controller connection");
+        auto [connection, inserted] = m_connections.insert(make_connection(std::move(m_socket)));
+        if (inserted)
+        {
+          (*connection)->add(this);
+          (*connection)->start();
+        }
+        perform_accept();
+      }
+    });
+  }
+
+  void interface::on_close(connection::pointer connection)
+  {
+    if (static_cast<char>(connection->current_state()) >= static_cast<char>(connection::state::established))
+    {
+      system::get_logger()->info("controller connection closed");
+    }
+    else
+    {
+      system::get_logger()->info("controller connection aborted before it could be established");
+    }
+    m_connections.erase(connection);
+  }
+
+  void interface::on_received(connection::pointer connection, proto::message message)
+  {
+    using namespace wanda::std_ext;
+
+    if (m_connections.find(connection) == m_connections.cend())
+    {
+      system::get_logger()->error("received message from an unknown connection");
+      return;
+    }
+
+    if (message.source != proto::message_source_controller)
+    {
+      system::get_logger()->error("received a deamon message");
+      return;
+    }
+
+    if (auto state = connection->current_state(); message.command == proto::message_command_hello && state == connection::state::fresh)
+    {
+      system::get_logger()->info("controller connection established");
+      if (message.argument.has_value())
+      {
+        system::get_logger()->info("remote controller version '{}'", *message.argument);
+      }
+      connection->send({proto::message_source_daemon, proto::message_command_hello, proto::version});
+      connection->update(connection::state::established);
+    }
+    else
+    {
+      with(make_command(message), [&](auto const & command) {
+        m_listener.on_received(*this, command);
+      }) ||
+          [&] { system::get_logger()->warn("ignoring unknown message '{}'", message); };
+    }
+  }
+
+  interface::pointer make_interface(boost::asio::io_context & service, std::filesystem::path socket, interface::listener & listener)
+  {
+    if (std::filesystem::exists(socket))
+    {
+      system::get_logger()->error("socket '{}' exists", socket.native());
+      return {};
+    }
+
+    interface::protocol::endpoint endpoint
+    {
+      socket.string()
+    };
+    return std::make_shared<interface>(interface::key{}, service, std::move(endpoint), listener);
+  }
+
+}  // namespace wanda::control
+\ No newline at end of file