rpcTransport.cpp

Go to the documentation of this file.

 
 
#include "profinet/rpcTransport.h"
 
#include <algorithm>
#include <atomic>
#include <future>
#include <memory>
#include <system_error>
#include <utility>
#include <vector>
 
#include "profinet/exceptions.h"
 
namespace profinet::rpc
{
namespace
{
 
struct ReceiveOperation
{
  explicit ReceiveOperation(asio::io_context& ioContext)
      : timer(ioContext),
        buffer(RECEIVE_BUFFER_LENGTH)
  {
  }
 
  asio::steady_timer timer;
  Bytes buffer;
  asio::ip::udp::endpoint remoteEndpoint;
 
  asio::error_code error;
  std::size_t receivedBytes = 0;
 
  std::atomic_bool timedOut{false};
  std::atomic_bool completed{false};
 
  std::promise<void> completion;
 
  void Complete()
  {
    if (!completed.exchange(true))
    {
      completion.set_value();
    }
  }
};
} // namespace
 
RpcTransport::RpcTransport(asio::io_context& ioContext)
    : ioContext(ioContext),
      socket(ioContext)
{
}
void RpcTransport::Open(const std::string& address, std::uint16_t port, bool isListener)
{
  asio::error_code ec;
  const auto ipAddress = asio::ip::make_address_v4(address, ec);
 
  if (ec)
  {
    throw std::system_error(ec, "Invalid IPv4 address: " + address);
  }
 
  if (!socket.is_open())
  {
    socket.open(asio::ip::udp::v4(), ec);
    if (ec)
    {
      throw std::system_error(ec, "Failed to open UDP socket");
    }
 
    // Allow socket reuse if needed
    socket.set_option(asio::ip::udp::socket::reuse_address(true), ec);
    if (isListener)
    {
      // BIND to the specified local interface and port!
      const asio::ip::udp::endpoint listenEndpoint(ipAddress, port);
      socket.bind(listenEndpoint, ec);
      if (ec)
      {
        socket.close();
        throw std::system_error(ec, "Failed to bind UDP socket to port " + std::to_string(port));
      }
    }
  }
 
  if (!isListener)
  {
    // Target endpoint for outgoing client requests
    endpoint = asio::ip::udp::endpoint(ipAddress, port);
  }
}
/*
void RpcTransport::Open(
    const std::string& address,
    std::uint16_t port,
    bool isListener)
{
  asio::error_code ec;
 
  const auto ipAddress =
      asio::ip::make_address_v4(address, ec);
 
  if (ec)
  {
    throw std::system_error(
        ec,
        "Invalid IPv4 address: " + address);
  }
 
  endpoint =
      asio::ip::udp::endpoint(ipAddress, port);
 
  if (!socket.is_open())
  {
    socket.open(
        asio::ip::udp::v4(),
        ec);
 
    if (ec)
    {
      throw std::system_error(
          ec,
          "Failed to open UDP socket");
    }
  }
}
*/
Bytes RpcTransport::Receive(std::chrono::milliseconds timeout, asio::ip::udp::endpoint& remoteEndpoint)
{
  if (!socket.is_open())
  {
    throw std::system_error(asio::error::not_connected,
                            "UDP socket is not open");
  }
 
  if (timeout <= std::chrono::milliseconds::zero())
  {
    throw std::system_error(asio::error::invalid_argument,
                            "UDP receive timeout must be greater than zero");
  }
 
  auto operation = std::make_shared<ReceiveOperation>(ioContext);
  auto completion = operation->completion.get_future();
 
  asio::post(ioContext, [&, operation, timeout]
  {
    operation->timer.expires_after(timeout);
 
    socket.async_receive_from(asio::buffer(operation->buffer),
                              operation->remoteEndpoint,
                              [operation](const asio::error_code& ec, std::size_t bytesReceived)
    {
      operation->error = ec;
      operation->receivedBytes = bytesReceived;
 
      if (!operation->timedOut.load())
      {
        asio::error_code timerError;
        operation->timer.cancel(timerError);
        operation->Complete();
      }
    });
 
    operation->timer.async_wait([&, operation](const asio::error_code& ec)
    {
      if (ec)
      {
        return;
      }
 
      operation->timedOut = true;
 
      asio::error_code cancelError;
      socket.cancel(cancelError);
 
      operation->Complete();
    });
  });
 
  completion.wait();
 
  if (operation->timedOut)
  {
    throw RPCTimeoutError("UDP receive timed out");
  }
 
  if (operation->error)
  {
    throw std::system_error(operation->error,
                            "Failed to receive UDP datagram");
  }
 
  remoteEndpoint = operation->remoteEndpoint;
 
  operation->buffer.resize(operation->receivedBytes);
 
  return std::move(operation->buffer);
}
 
void RpcTransport::SendTo(
    const Bytes& data,
    const asio::ip::udp::endpoint& targetEndpoint)
{
  if (!socket.is_open())
  {
    throw std::system_error(
        asio::error::not_connected,
        "UDP socket is not open");
  }
 
  asio::error_code ec;
 
  socket.send_to(
      asio::buffer(data),
      targetEndpoint,
      0,
      ec);
 
  if (ec)
  {
    throw std::system_error(
        ec,
        "Failed to send UDP datagram");
  }
}
 
Bytes RpcTransport::SendReceive(const Bytes& data, std::chrono::milliseconds timeout)
{
  if (!socket.is_open())
  {
    throw std::system_error(asio::error::not_connected,
                            "UDP socket is not open");
  }
 
  if (timeout <= std::chrono::milliseconds::zero())
  {
    throw std::system_error(asio::error::invalid_argument,
                            "UDP receive timeout must be greater than zero");
  }
 
  auto operation = std::make_shared<ReceiveOperation>(ioContext);
  auto completion = operation->completion.get_future();
 
  const auto targetEndpoint = endpoint;
 
  asio::post(ioContext, [&, operation, data, timeout, targetEndpoint]
  {
    asio::error_code sendError;
 
    socket.send_to(asio::buffer(data), targetEndpoint, 0, sendError);
 
    if (sendError)
    {
      operation->error = sendError;
      operation->Complete();
      return;
    }
 
    operation->timer.expires_after(timeout);
 
    socket.async_receive_from(asio::buffer(operation->buffer),
                              operation->remoteEndpoint,
                              [operation](const asio::error_code& ec, std::size_t bytesReceived)
    {
      operation->error = ec;
      operation->receivedBytes = bytesReceived;
 
      if (!operation->timedOut.load())
      {
        asio::error_code timerError;
        operation->timer.cancel(timerError);
        operation->Complete();
      }
    });
 
    operation->timer.async_wait([&, operation](const asio::error_code& ec)
    {
      if (ec)
      {
        return;
      }
 
      operation->timedOut = true;
 
      asio::error_code cancelError;
      socket.cancel(cancelError);
 
      operation->Complete();
    });
  });
 
  completion.wait();
 
  if (operation->timedOut)
  {
    throw RPCTimeoutError("UDP receive timed out");
  }
 
  if (operation->error)
  {
    throw std::system_error(operation->error,
                            "Failed to send/receive UDP datagram");
  }
 
  operation->buffer.resize(operation->receivedBytes);
 
  return std::move(operation->buffer);
}
 
void RpcTransport::Send(const Bytes& data)
{
  if (!socket.is_open())
  {
    throw std::system_error(
        asio::error::not_connected,
        "UDP socket is not open");
  }
 
  asio::error_code ec;
 
  socket.send_to(
      asio::buffer(data),
      endpoint,
      0,
      ec);
 
  if (ec)
  {
    throw std::system_error(
        ec,
        "Failed to send UDP datagram");
  }
}
 
void RpcTransport::Close() noexcept
{
  asio::error_code ec;
 
  if (socket.is_open())
  {
    socket.cancel(ec);
    socket.close(ec);
  }
}
 
bool RpcTransport::IsOpen() const noexcept
{
  return socket.is_open();
}
 
} // namespace profinet::rpc