standard_processor.hpp

Go to the documentation of this file.

#pragma once
 
#include <boost/asio/awaitable.hpp>
#include <boost/asio/cancellation_signal.hpp>
#include <boost/asio/experimental/channel_error.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/system/system_error.hpp>
#include <memory>
#include <spdlog/spdlog.h>
#include <utility>
 
namespace radix_relay::core {
 
template<typename Handler> class standard_processor
{
public:
  using in_queue_t = typename Handler::in_queue_t;
  using out_queues_t = typename Handler::out_queues_t;
 
  template<typename... HandlerArgs>
  standard_processor(const std::shared_ptr<boost::asio::io_context> &io_context,
    const std::shared_ptr<in_queue_t> &in_queue,
    const out_queues_t &out_queues,
    HandlerArgs &&...handler_args)
    : io_context_(io_context), in_queue_(in_queue),
      handler_(std::make_shared<Handler>(std::forward<HandlerArgs>(handler_args)..., out_queues))
  {}
 
  auto run_once(std::shared_ptr<boost::asio::cancellation_slot> cancel_slot = nullptr) -> boost::asio::awaitable<void>
  {
    auto evt = co_await in_queue_->pop(cancel_slot);
    handler_->handle(evt);
    co_return;
  }
 
  auto run(std::shared_ptr<boost::asio::cancellation_slot> cancel_slot = nullptr) -> boost::asio::awaitable<void>
  {
    try {
      while (true) { co_await run_once(cancel_slot); }
    } catch (const boost::system::system_error &e) {
      if (e.code() == boost::asio::error::operation_aborted
          or e.code() == boost::asio::experimental::error::channel_cancelled
          or e.code() == boost::asio::experimental::error::channel_closed) {
        spdlog::debug("[standard_processor] Cancelled, exiting run loop");
        co_return;
      } else {
        spdlog::error("[standard_processor] Unexpected error in run loop: {}", e.what());
        throw;
      }
    }
  }
 
private:
  std::shared_ptr<boost::asio::io_context> io_context_;
  std::shared_ptr<in_queue_t> in_queue_;
  std::shared_ptr<Handler> handler_;
};
 
}// namespace radix_relay::core