test_rpcTransport.cpp

Go to the documentation of this file.

 
 
#include <gtest/gtest.h>
 
#include <array>
#include <asio.hpp>
#include <chrono>
#include <cstdint>
#include <memory>
#include <string>
#include <system_error>
#include <thread>
#include <vector>
 
#include "profinet/AsioIoContextRunner.h"
#include "profinet/exceptions.h"
#include "profinet/rpcTransport.h"
 
using profinet::Bytes;
using profinet::rpc::RpcTransport;
using namespace std::chrono_literals;
 
namespace
{
 
std::uint16_t FindFreeUdpPort()
{
  asio::io_context ioContext;
  asio::ip::udp::socket socket(ioContext);
 
  socket.open(asio::ip::udp::v4());
  socket.bind(asio::ip::udp::endpoint(asio::ip::udp::v4(), 0));
 
  return socket.local_endpoint().port();
}
 
std::unique_ptr<asio::ip::udp::socket> CreateBoundUdpSocket(asio::io_context& ioContext, std::uint16_t port = 0)
{
  auto socket = std::make_unique<asio::ip::udp::socket>(ioContext);
 
  socket->open(asio::ip::udp::v4());
  socket->bind(asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), port));
 
  return socket;
}
 
void SendUdpDatagram(std::uint16_t port, const Bytes& data)
{
  asio::io_context ioContext;
  asio::ip::udp::socket socket(ioContext);
 
  socket.open(asio::ip::udp::v4());
 
  const asio::ip::udp::endpoint endpoint(asio::ip::address_v4::loopback(), port);
 
  asio::error_code ec;
  socket.send_to(asio::buffer(data), endpoint, 0, ec);
 
  ASSERT_FALSE(ec) << "UDP send failed: " << ec.message();
}
 
Bytes ReceiveUdpDatagram(asio::ip::udp::socket& socket, asio::ip::udp::endpoint& remoteEndpoint)
{
  std::array<std::uint8_t, profinet::RECEIVE_BUFFER_LENGTH> buffer{};
  asio::error_code ec;
 
  const std::size_t receivedBytes = socket.receive_from(asio::buffer(buffer), remoteEndpoint, 0, ec);
 
  if (ec)
  {
    throw std::system_error(ec, "Failed to receive test UDP datagram");
  }
 
  return Bytes(buffer.begin(), buffer.begin() + receivedBytes);
}
 
class UdpEchoServer
{
 public:
  UdpEchoServer()
      : socket(ioContext)
  {
    socket.open(asio::ip::udp::v4());
    socket.bind(asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), 0));
    port = socket.local_endpoint().port();
  }
 
  ~UdpEchoServer()
  {
    Stop();
  }
 
  UdpEchoServer(const UdpEchoServer&) = delete;
  UdpEchoServer& operator=(const UdpEchoServer&) = delete;
 
  [[nodiscard]] std::uint16_t Port() const
  {
    return port;
  }
 
  void Start()
  {
    Receive();
 
    thread = std::thread(
        [this]
    {
      ioContext.run();
    });
  }
 
  void Stop()
  {
    asio::error_code ec;
 
    socket.cancel(ec);
    socket.close(ec);
 
    ioContext.stop();
 
    if (thread.joinable())
    {
      thread.join();
    }
  }
 
 private:
  void Receive()
  {
    socket.async_receive_from(asio::buffer(receiveBuffer),
                              remoteEndpoint,
                              [&](const asio::error_code& ec, std::size_t bytesReceived)
    {
      if (ec)
      {
        return;
      }
 
      socket.async_send_to(asio::buffer(receiveBuffer.data(), bytesReceived),
                           remoteEndpoint,
                           [&](const asio::error_code& sendEc, std::size_t)
      {
        if (sendEc)
        {
          return;
        }
 
        Receive();
      });
    });
  }
 
  asio::io_context ioContext;
  asio::ip::udp::socket socket;
  asio::ip::udp::endpoint remoteEndpoint;
  std::array<std::uint8_t, profinet::RECEIVE_BUFFER_LENGTH> receiveBuffer{};
  std::uint16_t port{0};
  std::thread thread;
};
 
} // namespace
 
class RpcTransportTests : public ::testing::Test
{
 protected:
  void SetUp() override
  {
    ioContext = std::make_unique<asio::io_context>();
    runner = std::make_unique<profinet::AsioIoContextRunner>(*ioContext);
 
    // workGuard = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(asio::make_work_guard(*ioContext));
 
    // worker = std::jthread([&]
    // {
    //   ioContext->run();
    // });
  }
 
  void TearDown() override
  {
    // workGuard.reset();
    // ioContext->stop();
 
    // if (worker.joinable())
    //{
    //   worker.join();
    // }
    runner.reset();
    ioContext.reset();
  }
 
  std::unique_ptr<asio::io_context> ioContext;
 
  // std::unique_ptr<asio::executor_work_guard<asio::io_context::executor_type>> workGuard;
  // std::jthread worker;
 
  std::unique_ptr<profinet::AsioIoContextRunner> runner;
};
 
TEST_F(RpcTransportTests, ConstructorCreatesClosedTransport)
{
  RpcTransport transport(*ioContext);
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, OpenCreatesUdpSocket)
{
  RpcTransport transport(*ioContext);
 
  ASSERT_FALSE(transport.IsOpen());
 
  transport.Open("127.0.0.1", 34964);
 
  EXPECT_TRUE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, OpenListenerBindsRequestedPort)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", port, true);
 
  EXPECT_TRUE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, OpenRejectsInvalidIpv4Address)
{
  RpcTransport transport(*ioContext);
 
  EXPECT_THROW(transport.Open("not-an-ip-address", 34964), std::system_error);
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, OpenListenerRejectsOccupiedPort)
{
  asio::io_context occupiedIoContext;
  asio::ip::udp::socket occupiedSocket(occupiedIoContext);
 
  occupiedSocket.open(asio::ip::udp::v4());
  occupiedSocket.bind(asio::ip::udp::endpoint(asio::ip::address_v4::any(), 0));
 
  const std::uint16_t port = occupiedSocket.local_endpoint().port();
 
  RpcTransport transport(*ioContext);
 
  EXPECT_THROW(transport.Open("127.0.0.1", port, true), std::system_error);
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, OpenUpdatesEndpointWhenTransportIsAlreadyOpen)
{
  auto receiver = CreateBoundUdpSocket(*ioContext);
  const std::uint16_t receiverPort = receiver->local_endpoint().port();
 
  const std::uint16_t unusedPort = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", unusedPort);
 
  ASSERT_TRUE(transport.IsOpen());
 
  transport.Open("127.0.0.1", receiverPort);
 
  const Bytes request{0x10, 0x20, 0x30, 0x40};
 
  ASSERT_NO_THROW(transport.Send(request));
 
  asio::ip::udp::endpoint senderEndpoint;
  const Bytes received = ReceiveUdpDatagram(*receiver, senderEndpoint);
 
  EXPECT_EQ(received, request);
}
 
TEST_F(RpcTransportTests, ReceiveReturnsDatagramAndSenderEndpoint)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port, true);
 
  const Bytes expected{0x01, 0x02, 0x03, 0x04, 0xAA, 0xBB};
 
  SendUdpDatagram(port, expected);
 
  asio::ip::udp::endpoint remoteEndpoint;
  const Bytes received = transport.Receive(1000ms, remoteEndpoint);
 
  EXPECT_EQ(received, expected);
  EXPECT_EQ(remoteEndpoint.address(), asio::ip::address_v4::loopback());
  EXPECT_NE(remoteEndpoint.port(), 0);
}
 
TEST_F(RpcTransportTests, ReceiveFailsWhenTransportIsClosed)
{
  RpcTransport transport(*ioContext);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  EXPECT_THROW(transport.Receive(100ms, remoteEndpoint), std::system_error);
}
 
TEST_F(RpcTransportTests, ReceiveRejectsZeroTimeout)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port, true);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  EXPECT_THROW(transport.Receive(0ms, remoteEndpoint), std::system_error);
}
 
TEST_F(RpcTransportTests, ReceiveRejectsNegativeTimeout)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port, true);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  EXPECT_THROW(transport.Receive(-1ms, remoteEndpoint), std::system_error);
}
 
TEST_F(RpcTransportTests, ReceiveThrowsOnTimeout)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port, true);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  EXPECT_THROW(transport.Receive(20ms, remoteEndpoint), profinet::RPCTimeoutError);
}
 
TEST_F(RpcTransportTests, ReceiveThrowsSystemErrorWhenSocketIsCancelled)
{
  const std::uint16_t port = FindFreeUdpPort();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port, true);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  asio::steady_timer closeTimer(*ioContext);
  closeTimer.expires_after(1ms);
  closeTimer.async_wait([&transport](const asio::error_code& ec)
  {
    if (!ec)
    {
      transport.Close();
    }
  });
 
  EXPECT_THROW(transport.Receive(500ms, remoteEndpoint), std::system_error);
}
 
TEST_F(RpcTransportTests, SendToTransmitsDatagramToSuppliedEndpoint)
{
  auto receiver = CreateBoundUdpSocket(*ioContext);
  const std::uint16_t receiverPort = receiver->local_endpoint().port();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  const Bytes request{0x10, 0x20, 0x30, 0x40};
  const asio::ip::udp::endpoint target(asio::ip::address_v4::loopback(), receiverPort);
 
  ASSERT_NO_THROW(transport.SendTo(request, target));
 
  asio::ip::udp::endpoint senderEndpoint;
  const Bytes received = ReceiveUdpDatagram(*receiver, senderEndpoint);
 
  EXPECT_EQ(received, request);
  EXPECT_EQ(senderEndpoint.address(), asio::ip::address_v4::loopback());
  EXPECT_NE(senderEndpoint.port(), 0);
}
 
TEST_F(RpcTransportTests, SendToFailsWhenTransportIsClosed)
{
  RpcTransport transport(*ioContext);
 
  const Bytes request{0x01, 0x02, 0x03};
  const asio::ip::udp::endpoint target(asio::ip::address_v4::loopback(), 34964);
 
  EXPECT_THROW(transport.SendTo(request, target), std::system_error);
}
 
TEST_F(RpcTransportTests, SendToReportsUnderlyingSocketError)
{
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  const Bytes request{0x01, 0x02, 0x03};
  const asio::ip::udp::endpoint target(asio::ip::address_v6::loopback(), 34964);
 
  EXPECT_THROW(transport.SendTo(request, target), std::system_error);
}
 
TEST_F(RpcTransportTests, SendReceiveReturnsEchoResponse)
{
  UdpEchoServer server;
  server.Start();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", server.Port());
 
  const Bytes request{0x01, 0x02, 0x03, 0x04, 0xAA, 0xBB};
 
  const Bytes response = transport.SendReceive(request, 1000ms);
 
  EXPECT_EQ(response, request);
 
  server.Stop();
}
TEST_F(RpcTransportTests, SendReceiveUsesExternallyDrivenIoContext)
{
  UdpEchoServer server;
  server.Start();
 
  asio::io_context ioContext;
  profinet::AsioIoContextRunner ioContextRunner(ioContext);
 
  RpcTransport transport(ioContext);
  transport.Open("127.0.0.1", server.Port());
 
  const Bytes request{
      0x10,
      0x20,
      0x30,
      0x40,
      0xAA,
      0xBB};
 
  const Bytes response = transport.SendReceive(request, 1000ms);
 
  EXPECT_EQ(response, request);
 
  server.Stop();
}
 
TEST_F(RpcTransportTests, SendReceiveThrowsOnTimeout)
{
  auto silentServer = CreateBoundUdpSocket(*ioContext);
  const std::uint16_t port = silentServer->local_endpoint().port();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port);
 
  const Bytes request{0x01, 0x02, 0x03};
 
  EXPECT_THROW(transport.SendReceive(request, 20ms), profinet::RPCTimeoutError);
}
 
TEST_F(RpcTransportTests, SendReceiveFailsWhenTransportIsClosed)
{
  RpcTransport transport(*ioContext);
 
  const Bytes request{0x01, 0x02, 0x03};
 
  EXPECT_THROW(transport.SendReceive(request, 100ms), std::system_error);
}
 
TEST_F(RpcTransportTests, SendReceiveRejectsZeroTimeout)
{
  UdpEchoServer server;
  server.Start();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", server.Port());
 
  const Bytes request{0x01, 0x02, 0x03};
 
  EXPECT_THROW(transport.SendReceive(request, 0ms), std::system_error);
 
  server.Stop();
}
 
TEST_F(RpcTransportTests, SendReceiveRejectsNegativeTimeout)
{
  UdpEchoServer server;
  server.Start();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", server.Port());
 
  const Bytes request{0x01, 0x02, 0x03};
 
  EXPECT_THROW(transport.SendReceive(request, -1ms), std::system_error);
 
  server.Stop();
}
 
TEST_F(RpcTransportTests, SendReceiveThrowsSystemErrorWhenSocketIsCancelled)
{
  auto silentServer = CreateBoundUdpSocket(*ioContext);
  const std::uint16_t port = silentServer->local_endpoint().port();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", port);
 
  const Bytes request{0x01, 0x02, 0x03};
 
  asio::steady_timer closeTimer(*ioContext);
  closeTimer.expires_after(1ms);
  closeTimer.async_wait(
      [&transport](const asio::error_code& ec)
  {
    if (!ec)
    {
      transport.Close();
    }
  });
 
  EXPECT_THROW(transport.SendReceive(request, 500ms), std::system_error);
}
 
TEST_F(RpcTransportTests, SendTransmitsDatagramWithoutWaiting)
{
  auto receiver = CreateBoundUdpSocket(*ioContext);
  const std::uint16_t receiverPort = receiver->local_endpoint().port();
 
  RpcTransport transport(*ioContext);
  transport.Open("127.0.0.1", receiverPort);
 
  const Bytes request{0x10, 0x20, 0x30, 0x40};
 
  ASSERT_NO_THROW(transport.Send(request));
 
  asio::ip::udp::endpoint senderEndpoint;
  const Bytes received = ReceiveUdpDatagram(*receiver, senderEndpoint);
 
  EXPECT_EQ(received, request);
}
 
TEST_F(RpcTransportTests, SendFailsWhenTransportIsClosed)
{
  RpcTransport transport(*ioContext);
 
  const Bytes request{0x01, 0x02, 0x03};
 
  EXPECT_THROW(transport.Send(request), std::system_error);
}
 
TEST_F(RpcTransportTests, CloseClosesSocket)
{
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  ASSERT_TRUE(transport.IsOpen());
 
  transport.Close();
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, CloseIsIdempotent)
{
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  EXPECT_NO_THROW(transport.Close());
  EXPECT_NO_THROW(transport.Close());
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, CloseOnClosedTransportDoesNothing)
{
  RpcTransport transport(*ioContext);
 
  EXPECT_FALSE(transport.IsOpen());
 
  EXPECT_NO_THROW(transport.Close());
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, IsOpenReturnsTrueForOpenTransport)
{
  RpcTransport transport(*ioContext);
 
  EXPECT_FALSE(transport.IsOpen());
 
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  EXPECT_TRUE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, IsOpenReturnsFalseForClosedTransport)
{
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", FindFreeUdpPort());
 
  ASSERT_TRUE(transport.IsOpen());
 
  transport.Close();
 
  EXPECT_FALSE(transport.IsOpen());
}
 
TEST_F(RpcTransportTests, ReceiveUsesExternallyDrivenIoContext)
{
  const std::uint16_t port = FindFreeUdpPort();
  RpcTransport transport(*ioContext);
 
  transport.Open("127.0.0.1", port, true);
 
  const Bytes expected{
      0x01,
      0x02,
      0x03,
      0x04,
      0xAA,
      0xBB};
 
  // Send the datagram before Receive() starts. UDP retains the datagram in
  // the kernel receive queue, so the asynchronous receive operation can
  // consume it once it is registered with the socket.
  SendUdpDatagram(port, expected);
 
  asio::ip::udp::endpoint remoteEndpoint;
 
  const Bytes received = transport.Receive(1000ms, remoteEndpoint);
 
  EXPECT_EQ(received, expected);
  EXPECT_EQ(remoteEndpoint.address(), asio::ip::address_v4::loopback());
  EXPECT_NE(remoteEndpoint.port(), 0);
}