martino/MARTe-Integrated-Components
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
MARTe-Integrated-Components/Test/GTest/TCPSServerGTest.cpp
T
Martino FerrariandClaude Opus 4.6 50d85363a3 feat(tcp): TCPStreamer DataSources, in-process TCPStreamService, stream multiplexing and TCPStreamJoin 
Add TCPStreamer/TCPStreamerClient (TCP transport behind the UDPSServerI/UDPSClientI
seam), an optional single in-process TCPStreamService serving many streams over
worker threads, multiplexed SUBSCRIBE (comma list, stream index in fragmentIdx),
and TCPStreamJoin, which merges several streams into one sample-and-hold stream
paced by a master. StreamHub takes a comma list of streams over one connection.
Includes GTests, E2E scenarios s58-s73 and docs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-11-06 13:41:27 +01:00

456 lines
14 KiB
C++
Raw Permalink Blame History

/**
* @file TCPSServerGTest.cpp
* @brief GTest coverage for TCPSServer (UDPS framing over TCP, producer side).
*
* The server is driven the way UDPStreamer drives it (single thread:
* ServiceClients / SendConfig / SendData) and observed through raw loopback
* sockets, so these tests pin down the wire behaviour.
*
* @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and
* the Development of Fusion Energy ('Fusion for Energy').
* Licensed under the EUPL, Version 1.1 or - as soon they will be approved
* by the European Commission - subsequent versions of the EUPL (the "Licence")
* You may not use this work except in compliance with the Licence.
* You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl
*/
#define DLL_API
#include "gtest/gtest.h"
#include <pthread.h>
#include <unistd.h>
#include "ConfigurationDatabase.h"
#include "Sleep.h"
#include "TCPSServer.h"
#include "TCPTestUtil.h"
using namespace MARTe;
using namespace tcptest;
namespace {
/** Initialise @p server on an ephemeral port with the given limits. */
bool InitServer(TCPSServer &server, uint32 maxPayload = 60000u, uint32 maxClients = 16u,
uint32 depth = 8u, uint32 stallS = 5u, uint32 sndBuf = 0u) {
ConfigurationDatabase cdb;
(void)cdb.Write("Port", static_cast<uint32>(0u));
(void)cdb.Write("MaxPayloadSize", maxPayload);
(void)cdb.Write("MaxClients", maxClients);
(void)cdb.Write("ClientQueueDepth", depth);
(void)cdb.Write("ClientStallTimeout", stallS);
(void)cdb.Write("SendBufferSize", sndBuf);
return server.Initialise(cdb);
}
/** Pump ServiceClients() until @p n clients are attached. */
bool WaitClients(TCPSServer &server, uint32 n, uint32 timeoutMs) {
const uint64 deadline = NowMs() + timeoutMs;
while (NowMs() < deadline) {
server.ServiceClients();
if (server.GetClientCount() == n) {
return true;
}
Sleep::MSec(5u);
}
return false;
}
/** Pump ServiceClients() for @p ms. */
void Pump(TCPSServer &server, uint32 ms) {
const uint64 deadline = NowMs() + ms;
while (NowMs() < deadline) {
server.ServiceClients();
Sleep::MSec(5u);
}
}
struct FastReader {
int fd;
std::vector<uint32> counters;
std::vector<uint8> configPayload;
bool sawConfigFirst;
bool badFrame;
uint32 stopAt;
};
void *FastReaderMain(void *arg) {
FastReader *r = static_cast<FastReader *>(arg);
r->sawConfigFirst = false;
r->badFrame = false;
bool first = true;
Frame f;
for (;;) {
if (!ReadFrame(r->fd, f, 3000)) {
break;
}
if (first) {
r->sawConfigFirst = (f.type == UDPS_TYPE_CONFIG);
first = false;
}
if (f.type == UDPS_TYPE_CONFIG) {
r->configPayload = f.payload;
} else if (f.type == UDPS_TYPE_DATA) {
r->counters.push_back(f.counter);
if (f.payload.size() != 60000u) {
r->badFrame = true;
}
if (f.counter >= r->stopAt) {
break;
}
} else {
r->badFrame = true;
}
}
return NULL;
}
} // namespace
TEST(TCPSServerGTest, TestInitialise_Defaults) {
TCPSServer server;
ConfigurationDatabase cdb;
ASSERT_TRUE(server.Initialise(cdb));
EXPECT_EQ(TCPSServer::TCPS_DEFAULT_MAX_PAYLOAD, server.GetMaxPayloadSize());
EXPECT_EQ(static_cast<uint16>(TCPSServer::TCPS_DEFAULT_PORT), server.GetPort());
EXPECT_FALSE(server.IsMulticast());
EXPECT_FALSE(server.HasClients());
}
TEST(TCPSServerGTest, TestInitialise_RejectsBadValues) {
{
TCPSServer s;
ConfigurationDatabase cdb;
(void)cdb.Write("MaxPayloadSize", static_cast<uint32>(0u));
EXPECT_FALSE(s.Initialise(cdb));
}
{
TCPSServer s;
ConfigurationDatabase cdb;
(void)cdb.Write("MaxPayloadSize", static_cast<uint32>(2000000u));
EXPECT_FALSE(s.Initialise(cdb));
}
{
TCPSServer s;
ConfigurationDatabase cdb;
(void)cdb.Write("ClientQueueDepth", static_cast<uint32>(0u));
EXPECT_FALSE(s.Initialise(cdb));
}
{
TCPSServer s;
ConfigurationDatabase cdb;
(void)cdb.Write("MaxClients", static_cast<uint32>(0u));
EXPECT_FALSE(s.Initialise(cdb));
}
{
TCPSServer s;
ConfigurationDatabase cdb;
(void)cdb.Write("Port", static_cast<uint32>(70000u));
EXPECT_FALSE(s.Initialise(cdb));
}
}
TEST(TCPSServerGTest, TestStartStop_Idempotent) {
TCPSServer server;
ASSERT_TRUE(InitServer(server));
ASSERT_TRUE(server.Start());
EXPECT_TRUE(server.Start());
EXPECT_NE(0u, server.GetPort());
EXPECT_TRUE(server.Stop());
EXPECT_TRUE(server.Stop());
ASSERT_TRUE(server.Start());
EXPECT_NE(0u, server.GetPort());
EXPECT_TRUE(server.Stop());
}
TEST(TCPSServerGTest, TestConfigBeforeData_LateJoin) {
TCPSServer server;
ASSERT_TRUE(InitServer(server));
ASSERT_TRUE(server.Start());
std::vector<uint8> cfg = Pattern(40u, 3u);
ASSERT_TRUE(server.SendConfig(&cfg[0], 40u));
std::vector<uint8> d = Pattern(100u, 9u);
EXPECT_TRUE(server.SendData(1u, &d[0], 100u)); // nobody listening: harmless
int fd = Connect(server.GetPort());
ASSERT_GE(fd, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
EXPECT_TRUE(server.HasClients());
ASSERT_TRUE(server.SendData(2u, &d[0], 100u));
Frame f;
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(UDPS_TYPE_CONFIG, f.type);
EXPECT_EQ(1u, f.totalFragments);
EXPECT_EQ(cfg, f.payload);
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(UDPS_TYPE_DATA, f.type);
EXPECT_EQ(2u, f.counter);
EXPECT_EQ(d, f.payload);
close(fd);
server.Stop();
}
TEST(TCPSServerGTest, TestMultipleClients_InOrder) {
TCPSServer server;
ASSERT_TRUE(InitServer(server, 60000u, 16u, 64u));
ASSERT_TRUE(server.Start());
std::vector<uint8> cfg = Pattern(24u, 1u);
ASSERT_TRUE(server.SendConfig(&cfg[0], 24u));
int fds[3];
for (int i = 0; i < 3; i++) {
fds[i] = Connect(server.GetPort());
ASSERT_GE(fds[i], 0);
}
ASSERT_TRUE(WaitClients(server, 3u, 2000u));
for (uint32 c = 1u; c <= 50u; c++) {
std::vector<uint8> d = Pattern(200u, static_cast<uint8>(c));
ASSERT_TRUE(server.SendData(c, &d[0], 200u));
}
for (int i = 0; i < 3; i++) {
Frame f;
ASSERT_TRUE(ReadFrame(fds[i], f, 2000));
EXPECT_EQ(UDPS_TYPE_CONFIG, f.type);
for (uint32 c = 1u; c <= 50u; c++) {
ASSERT_TRUE(ReadFrame(fds[i], f, 2000)) << "client " << i << " frame " << c;
EXPECT_EQ(UDPS_TYPE_DATA, f.type);
ASSERT_EQ(c, f.counter);
EXPECT_EQ(Pattern(200u, static_cast<uint8>(c)), f.payload);
}
close(fds[i]);
}
EXPECT_EQ(0u, server.GetClientDrops());
server.Stop();
}
TEST(TCPSServerGTest, TestReconfigDeliveredToConnectedClient) {
TCPSServer server;
ASSERT_TRUE(InitServer(server));
ASSERT_TRUE(server.Start());
int fd = Connect(server.GetPort());
ASSERT_GE(fd, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
std::vector<uint8> cfg1 = Pattern(30u, 1u);
std::vector<uint8> cfg2 = Pattern(50u, 2u);
std::vector<uint8> d = Pattern(64u, 5u);
ASSERT_TRUE(server.SendConfig(&cfg1[0], 30u));
ASSERT_TRUE(server.SendData(1u, &d[0], 64u));
Frame f;
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(cfg1, f.payload);
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(1u, f.counter);
ASSERT_TRUE(server.SendConfig(&cfg2[0], 50u));
ASSERT_TRUE(server.SendData(2u, &d[0], 64u));
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(UDPS_TYPE_CONFIG, f.type);
EXPECT_EQ(cfg2, f.payload);
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(UDPS_TYPE_DATA, f.type);
EXPECT_EQ(2u, f.counter);
// A client joining now gets the NEW config.
int fd2 = Connect(server.GetPort());
ASSERT_GE(fd2, 0);
ASSERT_TRUE(WaitClients(server, 2u, 2000u));
ASSERT_TRUE(ReadFrame(fd2, f, 2000));
EXPECT_EQ(UDPS_TYPE_CONFIG, f.type);
EXPECT_EQ(cfg2, f.payload);
close(fd);
close(fd2);
server.Stop();
}
TEST(TCPSServerGTest, TestMaxClientsRefused) {
TCPSServer server;
ASSERT_TRUE(InitServer(server, 60000u, 2u));
ASSERT_TRUE(server.Start());
int fds[3];
for (int i = 0; i < 3; i++) {
fds[i] = Connect(server.GetPort());
ASSERT_GE(fds[i], 0);
}
Pump(server, 300u);
EXPECT_EQ(2u, server.GetClientCount());
int closedCount = 0;
for (int i = 0; i < 3; i++) {
if (PeerClosed(fds[i], 200)) {
closedCount++;
}
close(fds[i]);
}
EXPECT_EQ(1, closedCount);
server.Stop();
}
TEST(TCPSServerGTest, TestPeerCloseEvicts) {
TCPSServer server;
ASSERT_TRUE(InitServer(server));
ASSERT_TRUE(server.Start());
int fd = Connect(server.GetPort());
ASSERT_GE(fd, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
close(fd);
EXPECT_TRUE(WaitClients(server, 0u, 2000u));
EXPECT_FALSE(server.HasClients());
// The slot is reusable.
fd = Connect(server.GetPort());
ASSERT_GE(fd, 0);
EXPECT_TRUE(WaitClients(server, 1u, 2000u));
close(fd);
server.Stop();
}
TEST(TCPSServerGTest, TestOversizeFrameRefused) {
TCPSServer server;
ASSERT_TRUE(InitServer(server));
ASSERT_TRUE(server.Start());
int fd = Connect(server.GetPort());
ASSERT_GE(fd, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
std::vector<uint8> big(TCPSServer::TCPS_MAX_FRAME_PAYLOAD + 1u, 0u);
EXPECT_FALSE(server.SendData(1u, &big[0], static_cast<uint32>(big.size())));
EXPECT_FALSE(server.SendConfig(&big[0], static_cast<uint32>(big.size())));
EXPECT_TRUE(server.HasClients());
// A frame between MaxPayloadSize and the hard limit is still carried.
std::vector<uint8> mid(100000u, 7u);
EXPECT_TRUE(server.SendData(2u, &mid[0], static_cast<uint32>(mid.size())));
Frame f;
ASSERT_TRUE(ReadFrame(fd, f, 2000));
EXPECT_EQ(2u, f.counter);
EXPECT_EQ(mid.size(), f.payload.size());
close(fd);
server.Stop();
}
TEST(TCPSServerGTest, TestSlowClientDrops_FastClientComplete) {
TCPSServer server;
ASSERT_TRUE(InitServer(server, 60000u, 16u, 32u, 0u, 65536u));
ASSERT_TRUE(server.Start());
std::vector<uint8> cfg = Pattern(24u, 1u);
ASSERT_TRUE(server.SendConfig(&cfg[0], 24u));
int slow = Connect(server.GetPort(), 4096);
int fast = Connect(server.GetPort());
ASSERT_GE(slow, 0);
ASSERT_GE(fast, 0);
ASSERT_TRUE(WaitClients(server, 2u, 2000u));
FastReader reader;
reader.fd = fast;
reader.stopAt = 200u;
pthread_t th;
ASSERT_EQ(0, pthread_create(&th, NULL, FastReaderMain, &reader));
for (uint32 c = 1u; c <= 200u; c++) {
std::vector<uint8> d = Pattern(60000u, static_cast<uint8>(c));
ASSERT_TRUE(server.SendData(c, &d[0], 60000u));
server.ServiceClients();
Sleep::MSec(2u);
}
// Drain the slow client: the stream must still be well-formed, begin with
// CONFIG and have strictly increasing counters (gaps = dropped frames).
uint32 slowFrames = 0u;
uint32 lastCounter = 0u;
bool slowFirstConfig = false;
bool slowInOrder = true;
uint32 idle = 0u;
while (idle < 20u) {
server.ServiceClients();
Frame f;
if (ReadFrame(slow, f, 10)) {
idle = 0u;
if (slowFrames == 0u) {
slowFirstConfig = (f.type == UDPS_TYPE_CONFIG);
}
if (f.type == UDPS_TYPE_DATA) {
if (f.counter <= lastCounter) {
slowInOrder = false;
}
lastCounter = f.counter;
if (f.payload.size() != 60000u) {
slowInOrder = false;
}
}
slowFrames++;
} else {
idle++;
}
}
pthread_join(th, NULL);
ASSERT_EQ(200u, reader.counters.size());
for (uint32 i = 0u; i < 200u; i++) {
ASSERT_EQ(i + 1u, reader.counters[i]);
}
EXPECT_TRUE(reader.sawConfigFirst);
EXPECT_FALSE(reader.badFrame);
EXPECT_GT(server.GetClientDrops(), 100u);
EXPECT_TRUE(slowFirstConfig);
EXPECT_TRUE(slowInOrder);
EXPECT_GT(slowFrames, 1u);
EXPECT_LT(slowFrames, 201u); // some frames were dropped for it
EXPECT_EQ(2u, server.GetClientCount());
close(slow);
close(fast);
server.Stop();
}
TEST(TCPSServerGTest, TestStallDisconnect) {
TCPSServer server;
ASSERT_TRUE(InitServer(server, 60000u, 16u, 4u, 1u, 8192u));
ASSERT_TRUE(server.Start());
int slow = Connect(server.GetPort(), 4096);
ASSERT_GE(slow, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
std::vector<uint8> d = Pattern(60000u, 1u);
for (uint32 c = 1u; c <= 6u; c++) {
ASSERT_TRUE(server.SendData(c, &d[0], 60000u));
}
EXPECT_TRUE(server.HasClients()); // not yet: the stall clock is still running
const uint64 t0 = NowMs();
EXPECT_TRUE(WaitClients(server, 0u, 4000u));
EXPECT_GE(NowMs() - t0, 800u); // evicted by the timeout, not at once
EXPECT_EQ(1u, server.GetStallEvictions());
close(slow);
server.Stop();
}
TEST(TCPSServerGTest, TestStallDisabled_KeepsClient) {
TCPSServer server;
ASSERT_TRUE(InitServer(server, 60000u, 16u, 4u, 0u, 8192u));
ASSERT_TRUE(server.Start());
int slow = Connect(server.GetPort(), 4096);
ASSERT_GE(slow, 0);
ASSERT_TRUE(WaitClients(server, 1u, 2000u));
std::vector<uint8> d = Pattern(60000u, 1u);
for (uint32 c = 1u; c <= 6u; c++) {
ASSERT_TRUE(server.SendData(c, &d[0], 60000u));
}
Pump(server, 1500u);
EXPECT_EQ(1u, server.GetClientCount());
EXPECT_EQ(0u, server.GetStallEvictions());
close(slow);
server.Stop();
}
Reference in new issue View Git Blame Copy Permalink