jsonrpc::transport::SocketTransport Class Reference

#include <socket_transport.hpp>

Inheritance diagram for jsonrpc::transport::SocketTransport:

Collaboration diagram for jsonrpc::transport::SocketTransport:

Public Member Functions
	SocketTransport (asio::any_io_executor executor, std::string address, uint16_t port, bool is_server, std::shared_ptr< spdlog::logger > logger=nullptr)
	~SocketTransport () override
	SocketTransport (const SocketTransport &)=delete
auto	operator= (const SocketTransport &) -> SocketTransport &=delete
	SocketTransport (SocketTransport &&)=delete
auto	operator= (SocketTransport &&) -> SocketTransport &=delete
auto	Start () -> asio::awaitable< std::expected< void, error::RpcError > > override
auto	Close () -> asio::awaitable< std::expected< void, error::RpcError > > override
auto	CloseNow () -> void override
auto	SendMessage (std::string message) -> asio::awaitable< std::expected< void, error::RpcError > > override
auto	ReceiveMessage () -> asio::awaitable< std::expected< std::string, error::RpcError > > override
Public Member Functions inherited from jsonrpc::transport::Transport
	Transport (asio::any_io_executor executor, std::shared_ptr< spdlog::logger > logger=nullptr)
	Transport (const Transport &)=delete
	Transport (Transport &&)=delete
auto	operator= (const Transport &) -> Transport &=delete
auto	operator= (Transport &&) -> Transport &=delete
virtual	~Transport ()=default
auto	GetExecutor () const -> asio::any_io_executor
auto	GetStrand () -> asio::strand< asio::any_io_executor > &

Additional Inherited Members
Protected Member Functions inherited from jsonrpc::transport::Transport
auto	Logger () -> std::shared_ptr< spdlog::logger >

Detailed Description

Definition at line 14 of file socket_transport.hpp.

Constructor & Destructor Documentation

SocketTransport() [1/3]

jsonrpc::transport::SocketTransport::SocketTransport	(	asio::any_io_executor	executor,
		std::string	address,
		uint16_t	port,
		bool	is_server,
		std::shared_ptr< spdlog::logger >	logger = nullptr )

Definition at line 11 of file socket_transport.cpp.

    : Transport(std::move(executor), logger),
      socket_(GetExecutor()),
      address_(std::move(address)),
      port_(port),
      is_server_(is_server),
      read_buffer_() {
}

~SocketTransport()

jsonrpc::transport::SocketTransport::~SocketTransport ( )

override

Definition at line 22 of file socket_transport.cpp.

                                  {
  if (!is_closed_) {
    Logger()->debug("SocketTransport destructor triggering CloseNow()");
    try {
      CloseNow();
    } catch (const std::exception &e) {
      Logger()->error("SocketTransport destructor error: {}", e.what());
    }
  }
}

References CloseNow(), and jsonrpc::transport::Transport::Logger().

Here is the call graph for this function:

SocketTransport() [2/3]

jsonrpc::transport::SocketTransport::SocketTransport ( const SocketTransport & )

delete

SocketTransport() [3/3]

jsonrpc::transport::SocketTransport::SocketTransport ( SocketTransport && )

delete

Member Function Documentation

Close()

auto jsonrpc::transport::SocketTransport::Close ( ) -> asio::awaitable<std::expected<void, error::RpcError>>

overridevirtual

Implements jsonrpc::transport::Transport.

Definition at line 81 of file socket_transport.cpp.

                                                       {
  Logger()->debug("SocketTransport closing");
  co_await asio::post(GetStrand(), asio::use_awaitable);
 
  if (is_closed_) {
    Logger()->debug("SocketTransport already closed");
    co_return std::expected<void, error::RpcError>{};
  }
 
  is_closed_ = true;
  is_connected_ = false;
 
  // Clear the message queue
  send_queue_.clear();
 
  Logger()->debug("SocketTransport closing");
 
  // Cancel and close the socket safely
  std::error_code ec;
  if (socket_.is_open()) {
    socket_.cancel(ec);
    if (ec) {
      Logger()->warn(
          "SocketTransport error canceling socket: {}", ec.message());
    }
    socket_.close(ec);
    if (ec) {
      Logger()->warn("SocketTransport error closing socket: {}", ec.message());
    }
  }
 
  // Clean up acceptor if this is a server
  if (is_server_ && acceptor_) {
    acceptor_->cancel(ec);
    if (ec) {
      Logger()->warn(
          "SocketTransport error canceling acceptor: {}", ec.message());
    }
    acceptor_->close(ec);
    if (ec) {
      Logger()->warn(
          "SocketTransport error closing acceptor: {}", ec.message());
    }
  }
 
  Logger()->debug("SocketTransport closed");
  co_return Ok();
}

CloseNow()

void jsonrpc::transport::SocketTransport::CloseNow ( ) -> void

overridevirtual

Implements jsonrpc::transport::Transport.

Definition at line 131 of file socket_transport.cpp.

                               {
  is_closed_ = true;
  is_connected_ = false;
 
  // Clear the message queue
  send_queue_.clear();
 
  auto try_close_socket = [&]() {
    if (!socket_.is_open()) {
      return;
    }
    Logger()->debug("SocketTransport closing socket synchronously");
 
    std::error_code ec;
    socket_.cancel();
    socket_.close(ec);
    if (ec) {
      Logger()->warn("SocketTransport error closing socket: {}", ec.message());
    }
  };
 
  auto try_close_acceptor = [&]() {
    if (!is_server_ || !acceptor_ || !acceptor_->is_open()) {
      return;
    }
    Logger()->debug("SocketTransport closing acceptor synchronously");
 
    std::error_code ec;
    acceptor_->cancel();
    acceptor_->close(ec);
    if (ec) {
      Logger()->warn(
          "SocketTransport error closing acceptor: {}", ec.message());
    }
  };
 
  try {
    try_close_socket();
    try_close_acceptor();
  } catch (const std::exception &e) {
    Logger()->error("SocketTransport error during CloseNow(): {}", e.what());
  }
}

References jsonrpc::transport::Transport::Logger().

Referenced by ~SocketTransport().

Here is the call graph for this function:

Here is the caller graph for this function:

operator=() [1/2]

auto jsonrpc::transport::SocketTransport::operator= ( const SocketTransport & ) -> SocketTransport &=delete

delete

operator=() [2/2]

auto jsonrpc::transport::SocketTransport::operator= ( SocketTransport && ) -> SocketTransport &=delete

delete

ReceiveMessage()

auto jsonrpc::transport::SocketTransport::ReceiveMessage ( ) -> asio::awaitable<std::expected<std::string, error::RpcError>>

overridevirtual

Implements jsonrpc::transport::Transport.

Definition at line 251 of file socket_transport.cpp.

                                                            {
  co_await asio::post(GetStrand(), asio::use_awaitable);
 
  if (is_closed_) {
    Logger()->warn(
        "SocketTransport ReceiveMessage() called after transport was closed");
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError,
        "ReceiveMessage() called after transport was closed");
  }
 
  if (!is_started_) {
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError,
        "Transport not started before receiving message");
  }
 
  if (!socket_.is_open()) {
    Logger()->warn("SocketTransport ReceiveMessage() socket not open");
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError, "Socket not open in ReceiveMessage()");
  }
 
  message_buffer_.clear();
 
  std::error_code ec;
  size_t bytes_read = co_await socket_.async_read_some(
      asio::buffer(read_buffer_),
      asio::redirect_error(asio::use_awaitable, ec));
 
  if (ec) {
    if (ec == asio::error::eof) {
      Logger()->debug(
          "SocketTransport EOF received, connection closed by peer");
      is_connected_ = false;
      co_return RpcError::UnexpectedFromCode(
          RpcErrorCode::kTransportError, "Connection closed by peer");
    } else if (ec == asio::error::operation_aborted) {
      Logger()->debug("SocketTransport Read operation aborted");
      co_return RpcError::UnexpectedFromCode(
          RpcErrorCode::kTransportError, "Receive operation aborted");
    } else {
      Logger()->error(
          "SocketTransport ASIO error in ReceiveMessage(): {}", ec.message());
      co_return RpcError::UnexpectedFromCode(
          RpcErrorCode::kTransportError, "Receive error: " + ec.message());
    }
  }
 
  if (bytes_read == 0) {
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError, "Connection closed by peer (no data)");
  }
 
  message_buffer_.append(read_buffer_.data(), bytes_read);
  Logger()->debug("SocketTransport received {} bytes", bytes_read);
  co_return std::move(message_buffer_);
}

References jsonrpc::error::RpcError::UnexpectedFromCode().

Here is the call graph for this function:

SendMessage()

auto jsonrpc::transport::SocketTransport::SendMessage ( std::string message ) -> asio::awaitable<std::expected<void, error::RpcError>>

overridevirtual

Implements jsonrpc::transport::Transport.

Definition at line 179 of file socket_transport.cpp.

                                                       {
  co_await asio::post(GetStrand(), asio::use_awaitable);
 
  if (is_closed_) {
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError,
        "SendMessage() called on closed transport");
  }
 
  if (!is_started_) {
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError,
        "Transport not started before sending message");
  }
 
  if (!socket_.is_open()) {
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError, "Socket not open in SendMessage()");
  }
 
  Logger()->debug("Queuing {} bytes to send to socket", message.size());
  send_queue_.push_back(std::move(message));
 
  // If there's no active sending task, start one
  if (!sending_.exchange(true)) {
    asio::co_spawn(GetStrand(), SendMessageLoop(), asio::detached);
  }
 
  co_return Ok();
}

References jsonrpc::error::RpcError::UnexpectedFromCode().

Here is the call graph for this function:

Start()

auto jsonrpc::transport::SocketTransport::Start ( ) -> asio::awaitable<std::expected<void, error::RpcError>>

overridevirtual

Implements jsonrpc::transport::Transport.

Definition at line 33 of file socket_transport.cpp.

                                                       {
  Logger()->debug("SocketTransport starting");
  co_await asio::post(GetStrand(), asio::use_awaitable);
 
  if (is_started_) {
    Logger()->debug("SocketTransport already started");
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError, "SocketTransport already started");
  }
 
  if (is_closed_) {
    Logger()->error("SocketTransport cannot start a closed transport");
    co_return RpcError::UnexpectedFromCode(
        RpcErrorCode::kTransportError, "Cannot start a closed transport");
  }
 
  std::expected<void, error::RpcError> result;
 
  if (is_server_) {
    Logger()->debug(
        "SocketTransport starting server at {}:{}", address_, port_);
    result = co_await BindAndListen();
    if (!result) {
      Logger()->error(
          "SocketTransport error starting server: {}",
          result.error().Message());
      co_return result;
    }
  } else {
    Logger()->debug(
        "Connecting SocketTransport client to {}:{}", address_, port_);
    result = co_await Connect();
    if (!result) {
      Logger()->error(
          "SocketTransport error connecting client: {}",
          result.error().Message());
      co_return result;
    }
    Logger()->debug(
        "SocketTransport client connected to {}:{}", address_, port_);
  }
 
  is_started_ = true;
  Logger()->debug("SocketTransport successfully started");
  co_return Ok();
}

References jsonrpc::error::RpcError::UnexpectedFromCode().

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• include/jsonrpc/transport/socket_transport.hpp
• src/transport/socket_transport.cpp