diff --git a/.github/workflows/cpp_tests.yml b/.github/workflows/cpp_tests.yml index e7a67249..f15a3d2d 100644 --- a/.github/workflows/cpp_tests.yml +++ b/.github/workflows/cpp_tests.yml @@ -31,22 +31,39 @@ jobs: - uses: actions/checkout@v4 - - name: Configure CMake + - name: Configure CMake for CPP23 env: CC: gcc-14 - CXX: g++-14 + CXX: g++-14 working-directory: ${{github.workspace}}/ # Configure CMake in a 'build' subdirectory. `CMAKE_BUILD_TYPE` is only required if you are using a single-configuration generator such as make. # See https://cmake.org/cmake/help/latest/variable/CMAKE_BUILD_TYPE.html?highlight=cmake_build_type - run: cmake -B ${{github.workspace}}/build -S ${{github.workspace}}/cpp -DCMAKE_BUILD_TYPE=${{env.BUILD_TYPE}} + run: cmake -B ${{github.workspace}}/build23 -S ${{github.workspace}}/cpp -DCMAKE_BUILD_TYPE=${{env.BUILD_TYPE}} -DCPP_VERSION=23 - - name: Build + - name: Build CPP23 working-directory: ${{github.workspace}}/ # Build your program with the given configuration - run: cmake --build ${{github.workspace}}/build --config ${{env.BUILD_TYPE}} + run: cmake --build ${{github.workspace}}/build23 --config ${{env.BUILD_TYPE}} - name: Run CTest - working-directory: ${{github.workspace}}/build/ + working-directory: ${{github.workspace}}/build23/ + run: ctest --rerun-failed --output-on-failure + + - name: Configure CMake for CPP17 + env: + CC: gcc-14 + CXX: g++-14 + working-directory: ${{github.workspace}}/ + # Configure CMake in a 'build' subdirectory. `CMAKE_BUILD_TYPE` is only required if you are using a single-configuration generator such as make. + # See https://cmake.org/cmake/help/latest/variable/CMAKE_BUILD_TYPE.html?highlight=cmake_build_type + run: cmake -B ${{github.workspace}}/build17 -S ${{github.workspace}}/cpp -DCMAKE_BUILD_TYPE=${{env.BUILD_TYPE}} -DCPP_VERSION=17 + + - name: Build CPP17 + working-directory: ${{github.workspace}}/ + run: cmake --build ${{github.workspace}}/build17 --config ${{env.BUILD_TYPE}} + + - name: Run CTest + working-directory: ${{github.workspace}}/build17/ run: ctest --rerun-failed --output-on-failure convert_to_draft: diff --git a/.gitignore b/.gitignore index d1bb1e43..9ddf30d5 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ -build +build17 +build23 dist bin diff --git a/.vscode/tasks.json b/.vscode/tasks.json index ad298b74..eadf19fe 100644 --- a/.vscode/tasks.json +++ b/.vscode/tasks.json @@ -1,15 +1,16 @@ { "tasks": [ { - "label": "CMakePrepare", + "label": "CMakePrepare CPP23", "type": "shell", "command": "cmake", "args": [ "-B", - "${workspaceFolder}/build", + "${workspaceFolder}/build23", "-S", "${workspaceFolder}/cpp", - "-DCMAKE_BUILD_TYPE=Debug" + "-DCMAKE_BUILD_TYPE=Debug", + "-DCPP_VERSION=23" ], "group": { "kind": "build", @@ -19,12 +20,54 @@ "detail": "Generated task by Debugger." }, { - "label": "CMakeBuild", + "label": "CMakeBuild CPP23", "type": "shell", "command": "cmake", "args": [ "--build", - "${workspaceFolder}/build" + "${workspaceFolder}/build23" + ], + "group": { + "kind": "build", + "isDefault": true + }, + "problemMatcher": ["$gcc"], + "detail": "Generated task by Debugger.", + "presentation": { + "echo": true, + "reveal": "always", + "focus": false, + "panel": "shared", + "showReuseMessage": true, + "clear": true + } + }, + { + "label": "CMakePrepare CPP17", + "type": "shell", + "command": "cmake", + "args": [ + "-B", + "${workspaceFolder}/build17", + "-S", + "${workspaceFolder}/cpp", + "-DCMAKE_BUILD_TYPE=Debug", + "-DCPP_VERSION=17" + ], + "group": { + "kind": "build", + "isDefault": true + }, + "problemMatcher": ["$gcc"], + "detail": "Generated task by Debugger." + }, + { + "label": "CMakeBuild CPP17", + "type": "shell", + "command": "cmake", + "args": [ + "--build", + "${workspaceFolder}/build17" ], "group": { "kind": "build", diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 9418e81b..44238253 100644 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -7,7 +7,12 @@ include(CTest) # option(BUILD_MODULE_L5 "Build the L5 module" ON) -set(CMAKE_CXX_STANDARD 23) +set(CPP_VERSION "23" CACHE STRING "C++ standard version to use (17 or 23)") +set_property(CACHE CPP_VERSION PROPERTY STRINGS "17" "23") +if(NOT CPP_VERSION STREQUAL "17" AND NOT CPP_VERSION STREQUAL "23") + message(FATAL_ERROR "CPP_VERSION must be 17 or 23 (got '${CPP_VERSION}')") +endif() +set(CMAKE_CXX_STANDARD ${CPP_VERSION}) set(CMAKE_CXX_STANDARD_REQUIRED True) set(CMAKE_C_STANDARD 99) set(CMAKE_C_STANDARD_REQUIRED True) diff --git a/cpp/modules/l4/CMakeLists.txt b/cpp/modules/l4/CMakeLists.txt index 6c8c55c9..414578eb 100644 --- a/cpp/modules/l4/CMakeLists.txt +++ b/cpp/modules/l4/CMakeLists.txt @@ -1,13 +1,14 @@ CMAKE_MINIMUM_REQUIRED(VERSION 3.22) add_library(module_l4 STATIC) +add_executable(module_l4_tests + test/test_dummy.cpp) target_sources(module_l4 PRIVATE - src/Channel.cpp - src/impl/ChannelUnixTcp.cpp + src/ChannelRx.cpp + src/ChannelTx.cpp src/impl/ClientsRepoImpl.cpp - src/impl/ChannelUnixPipeImpl.cpp src/impl/ChannelUnixFile.cpp ) @@ -16,10 +17,6 @@ target_include_directories(module_l4 PUBLIC $ ) -add_executable(module_l4_tests - test/test_ChannelUnixTcp.cpp - test/test_ChannelUnixPipeImpl.cpp - ) if (USE_STD_MEMORY) target_compile_definitions(module_l4 PRIVATE USE_STD_MEMORY) @@ -27,20 +24,38 @@ if (USE_STD_MEMORY) endif() if(CMAKE_CXX_STANDARD GREATER_EQUAL 23) - target_sources(module_l4 PRIVATE src/cxx_23/ChannelTx.cpp) + message("Build for C++23") + target_sources(module_l4 PRIVATE + src/impl/ChannelUnixTcp.cpp + src/impl/ChannelUnixUdp.cpp + src/impl/ChannelUnixPipeImpl.cpp + src/impl/ChannelUnixSocket.cpp + ) + + + target_sources(module_l4_tests PRIVATE + test/test_ChannelUnixTcp.cpp + test/test_ChannelUnixPipeImpl.cpp + ) + + # This helps CMake provide better error messages if the compiler doesn't fully + # support the C++23 standard. + target_compile_features(module_l4_tests PRIVATE cxx_std_23) + elseif(CMAKE_CXX_STANDARD GREATER_EQUAL 17) - target_sources(module_l4 PRIVATE src/cxx_17/ChannelTx.cpp) + + message("Build for C++17") + # target_sources(module_l4 PRIVATE src/cxx_17/ChannelTx.cpp) + else() + message(FATAL_ERROR "module_l4 requires C++17 or higher. Please set CMAKE_CXX_STANDARD to 17 or above.") + endif() target_include_directories(module_l4_tests PRIVATE ./include/) target_link_libraries(module_l4_tests PUBLIC Catch2::Catch2WithMain module_l4) -# This helps CMake provide better error messages if the compiler doesn't fully -# support the C++23 standard. -target_compile_features(module_l4_tests PRIVATE cxx_std_23) - catch_discover_tests(module_l4_tests PROPERTIES TIMEOUT 10) diff --git a/cpp/modules/l4/include/Channel.h b/cpp/modules/l4/include/Channel.h deleted file mode 100644 index 4a4ac993..00000000 --- a/cpp/modules/l4/include/Channel.h +++ /dev/null @@ -1,18 +0,0 @@ -#pragma once - -#include "data.h" -#include "SerializerL4.h" - -namespace styxlib -{ - class ChannelRx - { - protected: - DeserializerL4Ptr deserializer; - - public: - ChannelRx(); - ErrorCode setDeserializer(DeserializerL4Ptr deserializer); - virtual ~ChannelRx() = default; - }; -} \ No newline at end of file diff --git a/cpp/modules/l4/include/ChannelRx.h b/cpp/modules/l4/include/ChannelRx.h new file mode 100644 index 00000000..65d70c0a --- /dev/null +++ b/cpp/modules/l4/include/ChannelRx.h @@ -0,0 +1,31 @@ +#pragma once + +#include "data.h" +#include "SerializerL4.h" + +namespace styxlib +{ + class ChannelRx + { + protected: + DeserializerL4Ptr deserializer; + + public: + ChannelRx(); + ErrorCode setDeserializer(DeserializerL4Ptr deserializer); + virtual ~ChannelRx() = default; + }; + + /** + * Reads the first X bytes of buffer and interprets them as a packet size. + * @param headerSize The size of the packet header (1, 2, or 4 bytes). + * @param buffer The buffer to read from. + * @param bufferSize The number of valid bytes available in the buffer. + * @return The decoded packet payload size, or an ErrorCode if the buffer + * is too small to hold the header. + */ + SizeResult getPacketSize( + const PacketHeaderSize &headerSize, + const uint8_t *buffer, + Size bufferSize); +} \ No newline at end of file diff --git a/cpp/modules/l4/include/ChannelTx.h b/cpp/modules/l4/include/ChannelTx.h index ba611aae..dcb9ff5a 100644 --- a/cpp/modules/l4/include/ChannelTx.h +++ b/cpp/modules/l4/include/ChannelTx.h @@ -10,13 +10,19 @@ namespace styxlib virtual ~ChannelTx() = default; virtual SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) = 0; }; + + /** + * Sets the packet size in the provided buffer according to the specified header size. + * @param headerSize The size of the packet header (1, 2, or 4 bytes). + * @param buffer The buffer where the packet size will be set. + * @param bufferSize The size of the buffer. + * @param packetSize The size of the packet to set. + * @return The number of bytes used for the header, or an ErrorCode if an error occurs. + */ + SizeResult setPacketSize( + const PacketHeaderSize &headerSize, + uint8_t *buffer, + Size bufferSize, + Size packetSize); } -#if __cplusplus >= 202302L - #include "cxx_23/ChannelTx.h" -#elif __cplusplus >= 201703L - #include "cxx_17/ChannelTx.h" -#else - // Handle older standards (C++14, C++11, etc.) - #error "This library requires at least C++17." -#endif \ No newline at end of file diff --git a/cpp/modules/l4/include/SerializerL4.h b/cpp/modules/l4/include/SerializerL4.h index 3fc23388..7fd17004 100644 --- a/cpp/modules/l4/include/SerializerL4.h +++ b/cpp/modules/l4/include/SerializerL4.h @@ -1,6 +1,5 @@ #pragma once #include "data.h" -#include "Channel.h" namespace styxlib { diff --git a/cpp/modules/l4/include/cxx_17/ChannelTx.h b/cpp/modules/l4/include/cxx_17/ChannelTx.h deleted file mode 100644 index 8f60517a..00000000 --- a/cpp/modules/l4/include/cxx_17/ChannelTx.h +++ /dev/null @@ -1,18 +0,0 @@ -#pragma once - -namespace styxlib -{ - /** - * Sets the packet size in the provided buffer according to the specified header size. - * @param headerSize The size of the packet header (1, 2, or 4 bytes). - * @param buffer The buffer where the packet size will be set. - * @param bufferSize The size of the buffer. - * @param packetSize The size of the packet to set. - * @return The number of bytes used for the header, or an ErrorCode if an error occurs. - */ - SizeResult setPacketSize( - const PacketHeaderSize &headerSize, - uint8_t *buffer, - Size bufferSize, - Size packetSize); -} \ No newline at end of file diff --git a/cpp/modules/l4/include/cxx_17/data.h b/cpp/modules/l4/include/cxx_17/data.h index b01a74cb..8c84c08f 100644 --- a/cpp/modules/l4/include/cxx_17/data.h +++ b/cpp/modules/l4/include/cxx_17/data.h @@ -15,5 +15,12 @@ namespace styxlib bool has_value() const { return _error == ErrorCode::Success; } }; + class Unexpected: public ExpectedSizeResult + { + public: + Unexpected(ErrorCode error) : ExpectedSizeResult(error) {} + }; + using SizeResult = ExpectedSizeResult; + } \ No newline at end of file diff --git a/cpp/modules/l4/include/cxx_23/ChannelTx.h b/cpp/modules/l4/include/cxx_23/ChannelTx.h deleted file mode 100644 index 8f60517a..00000000 --- a/cpp/modules/l4/include/cxx_23/ChannelTx.h +++ /dev/null @@ -1,18 +0,0 @@ -#pragma once - -namespace styxlib -{ - /** - * Sets the packet size in the provided buffer according to the specified header size. - * @param headerSize The size of the packet header (1, 2, or 4 bytes). - * @param buffer The buffer where the packet size will be set. - * @param bufferSize The size of the buffer. - * @param packetSize The size of the packet to set. - * @return The number of bytes used for the header, or an ErrorCode if an error occurs. - */ - SizeResult setPacketSize( - const PacketHeaderSize &headerSize, - uint8_t *buffer, - Size bufferSize, - Size packetSize); -} \ No newline at end of file diff --git a/cpp/modules/l4/include/cxx_23/data.h b/cpp/modules/l4/include/cxx_23/data.h index e660b330..5cd1b8f1 100644 --- a/cpp/modules/l4/include/cxx_23/data.h +++ b/cpp/modules/l4/include/cxx_23/data.h @@ -4,4 +4,5 @@ namespace styxlib { using SizeResult = std::expected; + using Unexpected = std::unexpected; } \ No newline at end of file diff --git a/cpp/modules/l4/include/data.h b/cpp/modules/l4/include/data.h index 8e2a3910..6edb1322 100644 --- a/cpp/modules/l4/include/data.h +++ b/cpp/modules/l4/include/data.h @@ -19,7 +19,8 @@ namespace styxlib UnknownClient, BufferTooSmall, InvalidHeaderSize, - NullptrArgument + NullptrArgument, + SendFailed }; enum class PacketHeaderSize : uint8_t diff --git a/cpp/modules/l4/include/impl/ChannelUnixFile.h b/cpp/modules/l4/include/impl/ChannelUnixFile.h index f4f55229..afcd5697 100644 --- a/cpp/modules/l4/include/impl/ChannelUnixFile.h +++ b/cpp/modules/l4/include/impl/ChannelUnixFile.h @@ -4,7 +4,7 @@ #include #include "data.h" -#include "Channel.h" +#include "ChannelRx.h" #include "ChannelTx.h" namespace styxlib diff --git a/cpp/modules/l4/include/impl/ChannelUnixSocket.h b/cpp/modules/l4/include/impl/ChannelUnixSocket.h new file mode 100644 index 00000000..0c357ad4 --- /dev/null +++ b/cpp/modules/l4/include/impl/ChannelUnixSocket.h @@ -0,0 +1,224 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include "ChannelRx.h" +#include "ChannelTx.h" +#include "ClientsRepo.h" +#include "impl/ChannelUnixFile.h" +#include "impl/ProgressObservableMutexImpl.h" + +namespace styxlib +{ + using Socket = FileDescriptor; + + /** + * Common base class for socket-based channel transmitters (TCP, UDP). + * Holds the socket descriptor and packet header size, and provides + * a single-call sendBuffer() that prepends the length header and sends + * header + payload atomically (required for UDP datagrams). + */ + class ChannelUnixSocketTx : public ChannelTx { + protected: + std::optional socket = std::nullopt; + PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; + public: + ChannelUnixSocketTx( + PacketHeaderSize packetSizeHeader, + std::optional socket + ); + ChannelUnixSocketTx(const ChannelUnixSocketTx &) = delete; + ChannelUnixSocketTx(ChannelUnixSocketTx &&) = delete; + ChannelUnixSocketTx &operator=(ChannelUnixSocketTx &&) = delete; + ChannelUnixSocketTx &operator=(const ChannelUnixSocketTx &) = delete; + virtual ~ChannelUnixSocketTx() = default; + + /** + * Sends the buffer with a length header prepended. + * Header and payload are combined into one ::send() call so that + * both TCP and UDP (datagram) transports work correctly. + * Returns the number of payload bytes sent without the header + * (i.e. in success case it should return exactly same value as the size parameter), + * or an error code on failure. + */ + SizeResult sendBuffer( + ClientId clientId, + const StyxBuffer buffer, + Size size) override; + }; + + /** + * Common base for socket-based client channels (TCP, UDP). + * Handles the shared Configuration, disconnect(), isConnected(), + * and the destructor. Subclasses only need to implement connect(), + * which creates the protocol-specific socket and calls ::connect(). + */ + class ChannelUnixSocketClient : public ChannelUnixSocketTx, public ChannelRx + { + public: + struct Configuration + { + std::string address; + uint16_t port; + PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; + uint16_t iounit{8192}; + DeserializerL4Ptr deserializer{nullptr}; + Configuration( + const std::string &address, + uint16_t port, + PacketHeaderSize packetSizeHeader, + uint16_t iounit, + DeserializerL4Ptr deserializer) + : address(address), + port(port), + packetSizeHeader(packetSizeHeader), + iounit(iounit), + deserializer(deserializer) {} + }; + + protected: + const Configuration configuration; + + public: + explicit ChannelUnixSocketClient(const Configuration &config); + ChannelUnixSocketClient(ChannelUnixSocketClient &&) = delete; + ChannelUnixSocketClient &operator=(ChannelUnixSocketClient &&) = delete; + ~ChannelUnixSocketClient() override; + + virtual std::future connect() = 0; + std::future disconnect(); + bool isConnected() const; + }; + + /** + * Common base for socket-based server channels (TCP, UDP). + * + * Handles the shared lifecycle (start / stop / isStarted), the poll loop, + * client bookkeeping, processBuffers, and cleanupClosedSockets. + * + * Subclasses must implement createServerSocket() to set up the protocol- + * specific listen/bind socket. Subclasses may also override + * acceptClients(), readDataFromSocket(), handlePollEvents(), + * cleanupClosedSockets(), and sendBuffer() to customise per-protocol + * behaviour (e.g. UDP does not call accept() and uses recvfrom/sendto). + */ + class ChannelUnixSocketServer : public ChannelRx, public ChannelTx + { + public: + class Configuration + { + public: + const uint16_t port; + const std::shared_ptr clientsRepo{nullptr}; + const PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; + const uint16_t iounit{8192}; + const DeserializerL4Ptr deserializer{nullptr}; + const uint16_t maxClients{16}; + + Configuration( + uint16_t port, + std::shared_ptr clientsRepo, + PacketHeaderSize packetSizeHeader, + uint16_t iounit, + DeserializerL4Ptr deserializer, + uint8_t maxClients) + : port(port), + clientsRepo(clientsRepo), + packetSizeHeader(packetSizeHeader), + iounit(iounit), + deserializer(deserializer), + maxClients(maxClients) + { + } + }; + + struct ClientInfo + { + ClientId id{0}; + std::string address{""}; + uint16_t port{0}; + }; + + protected: + class ClientFullInfo : public ClientInfo, public ReadBuffer + { + }; + + const Configuration configuration; + std::thread serverThread; + Socket serverSocketFd{InvalidFileDescriptor}; + + // Sockets / pseudo-sockets -> per-client state + std::map socketToClientInfoMapFull; + std::vector socketsToClose; + + // Per-client transmit channels (used by connection-oriented protocols) + std::map> clientIdToChannelClient; + + ProgressObservableMutexImpl> clientsObserver; + std::atomic running{false}; + std::atomic stopRequested{false}; + std::unique_ptr> startPromise; + std::vector pollFds; + + // ── Extension points ───────────────────────────────────────────────── + /** + * Create, configure and bind (and optionally listen on) the server + * socket. Must return a valid fd on success or set startPromise with + * an error code and return InvalidFileDescriptor on failure. + */ + virtual Socket createServerSocket() = 0; + + /** Called when the server-socket poll fd fires with POLLIN. + * Default: calls accept() in a loop until EAGAIN. + * Return true to indicate a new client was accepted (loop continues). */ + virtual bool acceptClients(Socket serverSocket); + + /** Called when a client-socket poll fd fires with POLLIN. + * Default: reads via recv() into the client's ReadBuffer. */ + virtual void readDataFromSocket(Socket clientFd); + + /** Dispatch POLLIN / POLLERR / POLLHUP events for all ready fds. + * Default implementation: server fd => acceptClients loop, + * client fd => readDataFromSocket, + * POLLERR|POLLHUP => mark socket for closure. */ + virtual void handlePollEvents(Socket serverSocket, size_t numEvents); + + /** Remove sockets that were queued in socketsToClose. + * Default: calls ::close(), removes from pollFds and maps. */ + virtual void cleanupClosedSockets(); + + /** Process dirty ReadBuffers held in socketToClientInfoMapFull and + * forward complete packets to the deserializer. Not virtual – the + * framing logic is identical for all transports. */ + void processBuffers(); + + void workThreadFunction(); + + /** Helper to rebuild and publish the clients observer vector. */ + void publishClients(); + + /** Called at the end of stop() so subclasses can clean up protocol- + * specific state (e.g. address maps in the UDP server). */ + virtual void onStop() {} + + public: + explicit ChannelUnixSocketServer(const Configuration &config); + ~ChannelUnixSocketServer() override; + + SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) override; + std::future start(); + std::future stop(); + bool isStarted() const; + ProgressObserver> &getClientsObserver() + { + return clientsObserver; + } + }; +} // namespace styxlib diff --git a/cpp/modules/l4/include/impl/ChannelUnixTcp.h b/cpp/modules/l4/include/impl/ChannelUnixTcp.h index 0a892500..17279839 100644 --- a/cpp/modules/l4/include/impl/ChannelUnixTcp.h +++ b/cpp/modules/l4/include/impl/ChannelUnixTcp.h @@ -1,143 +1,66 @@ #pragma once -#include -#include -#include -#include #include -#include -#include "Channel.h" +#include "ChannelRx.h" #include "ChannelTx.h" #include "ClientsRepo.h" #include "impl/ProgressObservableMutexImpl.h" -#include "impl/ChannelUnixFile.h" +#include "impl/ChannelUnixSocket.h" namespace styxlib { - using Socket = FileDescriptor; - - class ChannelUnixTcpTx : public ChannelTx { - protected: - std::optional socket = std::nullopt; - PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; + /** + * TCP-specific transmitter. Delegates all send logic to ChannelUnixSocketTx. + */ + class ChannelUnixTcpTx : public ChannelUnixSocketTx { public: - ChannelUnixTcpTx(PacketHeaderSize packetSizeHeader, std::optional socket) : - packetSizeHeader(packetSizeHeader), socket(socket) {} + ChannelUnixTcpTx( + PacketHeaderSize packetSizeHeader, + std::optional socket) + : ChannelUnixSocketTx(packetSizeHeader, socket) {} ChannelUnixTcpTx(const ChannelUnixTcpTx &) = delete; ChannelUnixTcpTx(ChannelUnixTcpTx &&) = delete; ChannelUnixTcpTx &operator=(ChannelUnixTcpTx &&) = delete; ChannelUnixTcpTx &operator=(const ChannelUnixTcpTx &) = delete; virtual ~ChannelUnixTcpTx() = default; - SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) override; }; - class ChannelUnixTcpClient : public ChannelUnixTcpTx, public ChannelRx + class ChannelUnixTcpClient : public ChannelUnixSocketClient { public: - struct Configuration - { - std::string address; - uint16_t port; - PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; - uint16_t iounit{8192}; - DeserializerL4Ptr deserializer{nullptr}; - Configuration( - const std::string &address, - uint16_t port, - PacketHeaderSize packetSizeHeader, - uint16_t iounit, - DeserializerL4Ptr deserializer) - : address(address), - port(port), - packetSizeHeader(packetSizeHeader), - iounit(iounit), - deserializer(deserializer) {} - }; + // Re-export the base Configuration so existing call sites remain unchanged. + using Configuration = ChannelUnixSocketClient::Configuration; - private: - const Configuration configuration; - public: - ChannelUnixTcpClient(const Configuration &config); + explicit ChannelUnixTcpClient(const Configuration &config); ChannelUnixTcpClient(ChannelUnixTcpClient &&) = delete; ChannelUnixTcpClient &operator=(ChannelUnixTcpClient &&) = delete; - ~ChannelUnixTcpClient() override; - std::future connect(); - std::future disconnect(); - bool isConnected() const; + ~ChannelUnixTcpClient() override = default; + std::future connect() override; }; - class ChannelUnixTcpServer : public ChannelRx, public ChannelTx + /** + * TCP server channel. + * Inherits the common server lifecycle, poll loop, buffer processing and + * client bookkeeping from ChannelUnixSocketServer. Only the TCP-specific + * socket setup (SOCK_STREAM / listen) and per-client accept/recv logic + * is implemented here. + */ + class ChannelUnixTcpServer : public ChannelUnixSocketServer { public: - class Configuration - { - public: - const uint16_t port; - const std::shared_ptr clientsRepo{nullptr}; - const PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes}; - const uint16_t iounit{8192}; - const DeserializerL4Ptr deserializer{nullptr}; - const uint16_t maxClients{16}; + // Re-export the shared Configuration and ClientInfo types. + using Configuration = ChannelUnixSocketServer::Configuration; + using ClientInfo = ChannelUnixSocketServer::ClientInfo; - Configuration( - uint16_t port, - std::shared_ptr clientsRepo, - PacketHeaderSize packetSizeHeader, - uint16_t iounit, - DeserializerL4Ptr deserializer, - uint8_t maxClients) - : port(port), - clientsRepo(clientsRepo), - packetSizeHeader(packetSizeHeader), - iounit(iounit), - deserializer(deserializer), - maxClients(maxClients) - { - } - }; - struct ClientInfo - { - ClientId id{0}; - std::string address{""}; - uint16_t port{0}; - }; + explicit ChannelUnixTcpServer(const Configuration &config); + ChannelUnixTcpServer(ChannelUnixTcpServer &&) = delete; + ChannelUnixTcpServer &operator=(ChannelUnixTcpServer &&) = delete; + ~ChannelUnixTcpServer() override = default; protected: - class ClientFullInfo : public ClientInfo, public ReadBuffer - { - }; - const Configuration configuration; - std::thread serverThread; - // Sockets - std::map socketToClientInfoMapFull; - std::vector socketsToClose; - // Clients - std::map> clientIdToChannelClient; - - ProgressObservableMutexImpl> clientsObserver; - std::atomic running{false}; - std::atomic stopRequested{false}; - std::unique_ptr> startPromise; - std::vector pollFds; - - void workThreadFunction(); - virtual bool acceptClients(Socket serverSocket); - virtual void readDataFromSocket(Socket clientFd); - void handlePollEvents(Socket serverSocket, size_t numEvents); // Handle poll events - void cleanupClosedSockets(); // Clean up sockets marked for closure - void processBuffers(); // Check dirty buffers and send data to deserializer - - public: - ChannelUnixTcpServer(const Configuration &config); - ~ChannelUnixTcpServer() override; - SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) override; - std::future start(); - std::future stop(); - ProgressObserver> &getClientsObserver() - { - return clientsObserver; - } - bool isStarted() const; + Socket createServerSocket() override; + bool acceptClients(Socket serverSocket) override; + void readDataFromSocket(Socket clientFd) override; }; } \ No newline at end of file diff --git a/cpp/modules/l4/include/impl/ChannelUnixUdp.h b/cpp/modules/l4/include/impl/ChannelUnixUdp.h new file mode 100644 index 00000000..91cc61e5 --- /dev/null +++ b/cpp/modules/l4/include/impl/ChannelUnixUdp.h @@ -0,0 +1,98 @@ +#pragma once + +#include +#include +#include +#include + +#include "impl/ChannelUnixSocket.h" + +namespace styxlib +{ + /** + * Maximum UDP payload size that fits in a single Ethernet frame without + * fragmentation: Ethernet MTU (1500) - IPv4 header (20) - UDP header (8). + * iounit must not exceed this value. + */ + constexpr uint16_t UdpMaxPayloadBytes = 1472; + + /** + * UDP transmitter channel. + * + * Uses a "connected" UDP socket (created with SOCK_DGRAM and bound to a + * remote address via ::connect()) so that the inherited sendBuffer() can + * use a plain ::send() call, sending header + payload as a single datagram. + */ + class ChannelUnixUdpTx : public ChannelUnixSocketTx { + public: + ChannelUnixUdpTx( + PacketHeaderSize packetSizeHeader, + std::optional socket) + : ChannelUnixSocketTx(packetSizeHeader, socket) {} + ChannelUnixUdpTx(const ChannelUnixUdpTx &) = delete; + ChannelUnixUdpTx(ChannelUnixUdpTx &&) = delete; + ChannelUnixUdpTx &operator=(ChannelUnixUdpTx &&) = delete; + ChannelUnixUdpTx &operator=(const ChannelUnixUdpTx &) = delete; + virtual ~ChannelUnixUdpTx() = default; + }; + + /** + * UDP client channel (Tx + Rx). + * + * connect() creates a SOCK_DGRAM socket and calls ::connect() to fix the + * remote address, enabling plain ::send() / ::recv() calls without + * specifying the peer on every call. + */ + class ChannelUnixUdpClient : public ChannelUnixSocketClient + { + public: + using Configuration = ChannelUnixSocketClient::Configuration; + + explicit ChannelUnixUdpClient(const Configuration &config); + ChannelUnixUdpClient(ChannelUnixUdpClient &&) = delete; + ChannelUnixUdpClient &operator=(ChannelUnixUdpClient &&) = delete; + ~ChannelUnixUdpClient() override = default; + std::future connect() override; + }; + + /** + * UDP server channel. + * + * Binds a single SOCK_DGRAM socket to the configured port. Each unique + * (source-IP, source-port) pair that sends a datagram is treated as a + * distinct "client" and assigned a ClientId on first contact. There is + * no connection-level acknowledgement – reliability is left to higher + * protocol layers. + * + * Inherits lifecycle management, the poll loop, buffer framing + * (processBuffers) and the clients observer from ChannelUnixSocketServer. + * It overrides only the UDP-specific socket setup, datagram reading and + * send-back logic. + */ + class ChannelUnixUdpServer : public ChannelUnixSocketServer + { + public: + using Configuration = ChannelUnixSocketServer::Configuration; + using ClientInfo = ChannelUnixSocketServer::ClientInfo; + + explicit ChannelUnixUdpServer(const Configuration &config); + ChannelUnixUdpServer(ChannelUnixUdpServer &&) = delete; + ChannelUnixUdpServer &operator=(ChannelUnixUdpServer &&) = delete; + ~ChannelUnixUdpServer() override = default; + + SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) override; + + protected: + // address string "ip:port" -> assigned ClientId + std::map addressToClientId; + // ClientId -> remote socket address (for sendto) + std::map clientIdToSockAddr; + + Socket createServerSocket() override; + void handlePollEvents(Socket serverSocket, size_t numEvents) override; + void readDataFromSocket(Socket serverFd) override; + void cleanupClosedSockets() override; + void onStop() override; + }; +} // namespace styxlib + diff --git a/cpp/modules/l4/src/Channel.cpp b/cpp/modules/l4/src/Channel.cpp deleted file mode 100644 index 51ef5722..00000000 --- a/cpp/modules/l4/src/Channel.cpp +++ /dev/null @@ -1,19 +0,0 @@ -#include "Channel.h" - -namespace styxlib -{ - ChannelRx::ChannelRx() : deserializer(nullptr) - { - } - - ErrorCode ChannelRx::setDeserializer(DeserializerL4Ptr deserializer) - { - if (deserializer == nullptr) - { - return ErrorCode::NullptrArgument; - } - this->deserializer = deserializer; - return ErrorCode::Success; - } - -} // namespace styxlib \ No newline at end of file diff --git a/cpp/modules/l4/src/ChannelRx.cpp b/cpp/modules/l4/src/ChannelRx.cpp new file mode 100644 index 00000000..3ccfb1ef --- /dev/null +++ b/cpp/modules/l4/src/ChannelRx.cpp @@ -0,0 +1,46 @@ +#include "ChannelRx.h" + +namespace styxlib +{ + ChannelRx::ChannelRx() : deserializer(nullptr) + { + } + + ErrorCode ChannelRx::setDeserializer(DeserializerL4Ptr deserializer) + { + if (deserializer == nullptr) + { + return ErrorCode::NullptrArgument; + } + this->deserializer = deserializer; + return ErrorCode::Success; + } + + SizeResult getPacketSize( + const PacketHeaderSize &headerSize, + const uint8_t *buffer, + Size bufferSize) + { + const uint8_t needed = static_cast(headerSize); + if (bufferSize < needed) + { + return Unexpected(ErrorCode::BufferTooSmall); + } + switch (headerSize) + { + case PacketHeaderSize::Size1Byte: + return static_cast(buffer[0]); + case PacketHeaderSize::Size2Bytes: + return static_cast( + (static_cast(buffer[0]) << 8) | + static_cast(buffer[1])); + case PacketHeaderSize::Size4Bytes: + return static_cast( + (static_cast(buffer[0]) << 24) | + (static_cast(buffer[1]) << 16) | + (static_cast(buffer[2]) << 8) | + static_cast(buffer[3])); + } + return Unexpected(ErrorCode::InvalidHeaderSize); + } +} // namespace styxlib \ No newline at end of file diff --git a/cpp/modules/l4/src/cxx_17/ChannelTx.cpp b/cpp/modules/l4/src/ChannelTx.cpp similarity index 78% rename from cpp/modules/l4/src/cxx_17/ChannelTx.cpp rename to cpp/modules/l4/src/ChannelTx.cpp index 8724b223..e06309b8 100644 --- a/cpp/modules/l4/src/cxx_17/ChannelTx.cpp +++ b/cpp/modules/l4/src/ChannelTx.cpp @@ -7,26 +7,26 @@ namespace styxlib Size bufferSize, Size packetSize) { if (bufferSize < 4) { - return SizeResult(ErrorCode::BufferTooSmall); + return Unexpected(ErrorCode::BufferTooSmall); } switch (headerSize) { case PacketHeaderSize::Size1Byte: if (packetSize > 0xFF) { - return SizeResult(ErrorCode::PacketTooLarge); + return Unexpected(ErrorCode::PacketTooLarge); } buffer[0] = static_cast(packetSize); break; case PacketHeaderSize::Size2Bytes: if (packetSize > 0xFFFF) { - return SizeResult(ErrorCode::PacketTooLarge); + return Unexpected(ErrorCode::PacketTooLarge); } buffer[1] = packetSize & 0xFF; buffer[0] = (packetSize >> 8) & 0xFF; break; case PacketHeaderSize::Size4Bytes: if (packetSize > 0xFFFFFFFF) { - return SizeResult(ErrorCode::PacketTooLarge); + return Unexpected(ErrorCode::PacketTooLarge); } buffer[3] = packetSize & 0xFF; buffer[2] = (packetSize >> 8) & 0xFF; @@ -34,7 +34,7 @@ namespace styxlib buffer[0] = (packetSize >> 24) & 0xFF; break; default: - return SizeResult(ErrorCode::InvalidHeaderSize); + return Unexpected(ErrorCode::InvalidHeaderSize); } return SizeResult(static_cast(headerSize)); } diff --git a/cpp/modules/l4/src/cxx_23/ChannelTx.cpp b/cpp/modules/l4/src/cxx_23/ChannelTx.cpp deleted file mode 100644 index e6a928a1..00000000 --- a/cpp/modules/l4/src/cxx_23/ChannelTx.cpp +++ /dev/null @@ -1,41 +0,0 @@ -#include "ChannelTx.h" - -namespace styxlib -{ - SizeResult setPacketSize(const PacketHeaderSize &headerSize, - uint8_t* buffer, - Size bufferSize, - Size packetSize) { - if (bufferSize < 4) { - return std::unexpected(ErrorCode::BufferTooSmall); - } - switch (headerSize) - { - case PacketHeaderSize::Size1Byte: - if (packetSize > 0xFF) { - return std::unexpected(ErrorCode::PacketTooLarge); - } - buffer[0] = static_cast(packetSize); - break; - case PacketHeaderSize::Size2Bytes: - if (packetSize > 0xFFFF) { - return std::unexpected(ErrorCode::PacketTooLarge); - } - buffer[1] = packetSize & 0xFF; - buffer[0] = (packetSize >> 8) & 0xFF; - break; - case PacketHeaderSize::Size4Bytes: - if (packetSize > 0xFFFFFFFF) { - return std::unexpected(ErrorCode::PacketTooLarge); - } - buffer[3] = packetSize & 0xFF; - buffer[2] = (packetSize >> 8) & 0xFF; - buffer[1] = (packetSize >> 16) & 0xFF; - buffer[0] = (packetSize >> 24) & 0xFF; - break; - default: - return std::unexpected(ErrorCode::InvalidHeaderSize); - } - return static_cast(headerSize); - } -} // namespace styxlib \ No newline at end of file diff --git a/cpp/modules/l4/src/impl/ChannelUnixFile.cpp b/cpp/modules/l4/src/impl/ChannelUnixFile.cpp index 2b542053..c38bb364 100644 --- a/cpp/modules/l4/src/impl/ChannelUnixFile.cpp +++ b/cpp/modules/l4/src/impl/ChannelUnixFile.cpp @@ -33,28 +33,28 @@ namespace styxlib { if (fds.writeFd == InvalidFileDescriptor) { - return std::unexpected(ErrorCode::NotConnected); + return Unexpected(ErrorCode::NotConnected); } // Send the buffer over the file descriptor uint8_t packetSizeBuffer[4] = {0}; - std::expected headerSize = setPacketSize( + SizeResult headerSize = setPacketSize( config.packetSizeHeader, packetSizeBuffer, sizeof(packetSizeBuffer), size); if (!headerSize.has_value()) { - return std::unexpected(headerSize.error()); + return Unexpected(headerSize.error()); } ssize_t result = ::write(fds.writeFd, packetSizeBuffer, headerSize.value()); if (result < 0) { std::cerr << "Error writing packet size to fd " << fds.writeFd << ": " << strerror(errno) << std::endl; - return std::unexpected(ErrorCode::NotConnected); + return Unexpected(ErrorCode::NotConnected); } result = ::write(fds.writeFd, buffer, size); if (result < 0) { - return std::unexpected(ErrorCode::NotConnected); + return Unexpected(ErrorCode::NotConnected); } return static_cast(result); } diff --git a/cpp/modules/l4/src/impl/ChannelUnixPipeImpl.cpp b/cpp/modules/l4/src/impl/ChannelUnixPipeImpl.cpp index c074c77b..133e04a5 100644 --- a/cpp/modules/l4/src/impl/ChannelUnixPipeImpl.cpp +++ b/cpp/modules/l4/src/impl/ChannelUnixPipeImpl.cpp @@ -43,7 +43,7 @@ namespace styxlib } else { - startPromise->set_value(std::unexpected(ErrorCode::AlreadyStarted)); + startPromise->set_value(Unexpected(ErrorCode::AlreadyStarted)); } return startPromise->get_future(); } @@ -82,7 +82,7 @@ namespace styxlib if (pipe(pipeFds) == -1) { running.store(false); - startPromise->set_value(std::unexpected(ErrorCode::CantCreateSocket)); + startPromise->set_value(Unexpected(ErrorCode::CantCreateSocket)); return; } rxFds.readFd = pipeFds[0]; @@ -91,7 +91,7 @@ namespace styxlib { closeDescriptors(); running.store(false); - startPromise->set_value(std::unexpected(ErrorCode::CantCreateSocket)); + startPromise->set_value(Unexpected(ErrorCode::CantCreateSocket)); return; } txFds.readFd = pipeFds[0]; diff --git a/cpp/modules/l4/src/impl/ChannelUnixSocket.cpp b/cpp/modules/l4/src/impl/ChannelUnixSocket.cpp new file mode 100644 index 00000000..1aa762e2 --- /dev/null +++ b/cpp/modules/l4/src/impl/ChannelUnixSocket.cpp @@ -0,0 +1,345 @@ +#include "impl/ChannelUnixSocket.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace styxlib +{ + ChannelUnixSocketTx::ChannelUnixSocketTx( + PacketHeaderSize packetSizeHeader, + std::optional socket) + : packetSizeHeader(packetSizeHeader), + socket(socket) + { + } + + SizeResult ChannelUnixSocketTx::sendBuffer( + ClientId clientId, + const StyxBuffer buffer, + Size size) + { + if (!socket.has_value()) + { + return Unexpected(ErrorCode::NotConnected); + } + + uint8_t packetSizeBuffer[4] = {0}; + SizeResult headerSize = setPacketSize( + packetSizeHeader, + packetSizeBuffer, + sizeof(packetSizeBuffer), + size); + if (!headerSize.has_value()) + { + return Unexpected(headerSize.error()); + } + + // Combine header and payload into a single buffer so the send is + // atomic. This is essential for UDP (each ::send() maps to one + // datagram) and harmless for TCP. + // TODO allocate IOUnit buffer at construction and reuse it here to avoid this copy. + std::vector combined(headerSize.value() + size); + std::memcpy(combined.data(), packetSizeBuffer, headerSize.value()); + std::memcpy(combined.data() + headerSize.value(), buffer, size); + ssize_t bytesSent = static_cast(::send(socket.value(), combined.data(), combined.size(), 0)); + return bytesSent - headerSize.value(); + } + + ChannelUnixSocketClient::ChannelUnixSocketClient(const Configuration &config) + : ChannelUnixSocketTx(config.packetSizeHeader, std::nullopt), + ChannelRx(), + configuration(config) + { + if (setDeserializer(config.deserializer) != ErrorCode::Success) { + throw std::invalid_argument("Deserializer cannot be null"); + } + } + + ChannelUnixSocketClient::~ChannelUnixSocketClient() + { + disconnect().get(); + } + + std::future ChannelUnixSocketClient::disconnect() + { + return std::async( + std::launch::async, + [this]() + { + if (socket.has_value()) + { + ::close(socket.value()); + socket = std::nullopt; + } + }); + } + + bool ChannelUnixSocketClient::isConnected() const + { + return socket.has_value(); + } + + // ── ChannelUnixSocketServer ─────────────────────────────────────────────── + + ChannelUnixSocketServer::ChannelUnixSocketServer(const Configuration &config) + : ChannelRx(), configuration(config) + { + if (setDeserializer(config.deserializer) != ErrorCode::Success) { + throw std::invalid_argument("Deserializer cannot be null"); + } + if (configuration.clientsRepo == nullptr) + { + throw std::invalid_argument("ClientsRepo must be provided in configuration"); + } + } + + ChannelUnixSocketServer::~ChannelUnixSocketServer() + { + stop().get(); + } + + SizeResult ChannelUnixSocketServer::sendBuffer( + ClientId clientId, + const StyxBuffer buffer, + Size size) + { + if (!isStarted()) + { + return Unexpected(ErrorCode::NotConnected); + } + auto it = clientIdToChannelClient.find(clientId); + if (it == clientIdToChannelClient.end()) + { + return Unexpected(ErrorCode::UnknownClient); + } + return it->second->sendBuffer(InvalidClientId, buffer, size); + } + + std::future ChannelUnixSocketServer::start() + { + startPromise = std::make_unique>(); + if (!isStarted()) + { + stopRequested.store(false); + serverThread = std::thread([this]() + { this->workThreadFunction(); }); + } + else + { + startPromise->set_value(ErrorCode::AlreadyStarted); + } + return startPromise->get_future(); + } + + std::future ChannelUnixSocketServer::stop() + { + return std::async( + std::launch::async, + [this]() + { + stopRequested.store(true); + if (this->serverThread.joinable()) + { + this->serverThread.join(); + } + this->startPromise = nullptr; + clientIdToChannelClient.clear(); + socketToClientInfoMapFull.clear(); + clientsObserver.setData(std::vector{}, true); + onStop(); + }); + } + + bool ChannelUnixSocketServer::isStarted() const + { + return running.load(); + } + + void ChannelUnixSocketServer::workThreadFunction() + { + running.store(true); + + Socket serverSocket = createServerSocket(); + if (serverSocket == InvalidFileDescriptor) + { + running.store(false); + return; + } + serverSocketFd = serverSocket; + + pollFds.push_back({serverSocket, POLLIN, 0}); + startPromise->set_value(ErrorCode::Success); + + while (!stopRequested.load()) + { + int numEvents = poll(pollFds.data(), pollFds.size(), 100); + if (stopRequested.load()) + break; + if (numEvents < 0) + { + std::cerr << "Poll error: " << strerror(errno) << std::endl; + break; + } + else if (numEvents == 0) + { + continue; + } + else + { + handlePollEvents(serverSocket, numEvents); + if (stopRequested.load()) + break; + processBuffers(); + cleanupClosedSockets(); + } + } + + ::close(serverSocket); + serverSocketFd = InvalidFileDescriptor; + pollFds.clear(); + running.store(false); + stopRequested.store(false); + } + + bool ChannelUnixSocketServer::acceptClients(Socket /*serverSocket*/) + { + // Default: connection-oriented protocols should override this. + return false; + } + + void ChannelUnixSocketServer::readDataFromSocket(Socket clientFd) + { + auto it = socketToClientInfoMapFull.find(clientFd); + if (it == socketToClientInfoMapFull.end()) + { + socketsToClose.push_back(clientFd); + return; + } + ClientFullInfo &readBuffer = it->second; + Size leftForRead = readBuffer.buffer.size() - readBuffer.currentSize; + ssize_t bytesRead = recv(clientFd, readBuffer.buffer.data() + readBuffer.currentSize, leftForRead, 0); + if (bytesRead > 0) + { + readBuffer.currentSize += bytesRead; + readBuffer.isDirty = true; + } + else + { + std::cerr << "Error reading from client socket " << clientFd << ": " << strerror(errno) << std::endl; + socketsToClose.push_back(clientFd); + } + } + + void ChannelUnixSocketServer::handlePollEvents(Socket serverSocket, size_t /*numEvents*/) + { + for (const auto &pollItem : pollFds) + { + if (pollItem.revents & POLLIN) + { + if (pollItem.fd == serverSocket) + { + while (acceptClients(serverSocket)) + { + // accept all pending connections + } + } + else + { + readDataFromSocket(pollItem.fd); + } + } + if (pollItem.revents & (POLLERR | POLLHUP)) + { + socketsToClose.push_back(pollItem.fd); + std::cerr << "Error or hang-up on socket " << pollItem.fd << std::endl; + } + } + } + + void ChannelUnixSocketServer::cleanupClosedSockets() + { + for (Socket fd : socketsToClose) + { + ::close(fd); + pollFds.erase(std::remove_if(pollFds.begin(), pollFds.end(), + [fd](const pollfd &p) { return p.fd == fd; }), + pollFds.end()); + auto it = std::find_if( + socketToClientInfoMapFull.begin(), + socketToClientInfoMapFull.end(), + [fd](const auto &pair) { return pair.first == fd; }); + if (it != socketToClientInfoMapFull.end()) + { + clientIdToChannelClient.erase(it->second.id); + } + socketToClientInfoMapFull.erase(fd); + } + if (!socketsToClose.empty()) + { + publishClients(); + } + socketsToClose.clear(); + } + + void ChannelUnixSocketServer::processBuffers() + { + const uint8_t headerBytes = to_uint8_t(configuration.packetSizeHeader); + for (auto &pair : socketToClientInfoMapFull) + { + ClientFullInfo &readBuffer = pair.second; + while (readBuffer.isDirty && readBuffer.currentSize > headerBytes) + { + auto packetSizeResult = getPacketSize( + configuration.packetSizeHeader, + readBuffer.buffer.data(), + readBuffer.currentSize); + if (!packetSizeResult.has_value()) + { + // Not enough data even for the header – wait for more. + break; + } + const Size packetSize = packetSizeResult.value(); + const Size packetSizeWithHeader = packetSize + headerBytes; + if (readBuffer.currentSize >= packetSizeWithHeader) + { + deserializer->handleBuffer( + pair.second.id, + readBuffer.buffer.data() + headerBytes, + packetSize); + Size remainingSize = readBuffer.currentSize - packetSizeWithHeader; + if (remainingSize > 0) + { + std::memmove(readBuffer.buffer.data(), + readBuffer.buffer.data() + packetSizeWithHeader, + remainingSize); + } + readBuffer.currentSize = remainingSize; + } + else + { + // Full payload not yet received – wait for more data. + break; + } + } + readBuffer.isDirty = false; + } + } + + void ChannelUnixSocketServer::publishClients() + { + clientsObserver.setData( + socketToClientInfoMapFull | std::views::transform([](const auto &pair) { + return ClientInfo{ + .id = pair.second.id, + .address = pair.second.address, + .port = pair.second.port}; + }) | std::ranges::to(), + false); + } +} // namespace styxlib diff --git a/cpp/modules/l4/src/impl/ChannelUnixTcp.cpp b/cpp/modules/l4/src/impl/ChannelUnixTcp.cpp index e47b4001..08852b94 100644 --- a/cpp/modules/l4/src/impl/ChannelUnixTcp.cpp +++ b/cpp/modules/l4/src/impl/ChannelUnixTcp.cpp @@ -11,44 +11,8 @@ namespace styxlib { ChannelUnixTcpClient::ChannelUnixTcpClient(const Configuration &config) - : ChannelRx(), - ChannelUnixTcpTx(config.packetSizeHeader, std::nullopt), - configuration(config) + : ChannelUnixSocketClient(config) { - if (setDeserializer(config.deserializer) != ErrorCode::Success) { - throw std::invalid_argument("Deserializer cannot be null"); - } - } - - ChannelUnixTcpClient::~ChannelUnixTcpClient() - { - disconnect().get(); - } - - SizeResult ChannelUnixTcpTx::sendBuffer( - ClientId clientId, - const StyxBuffer buffer, - Size size) - { - if (socket.has_value()) - { - // Send the buffer over the TCP socket - uint8_t packetSizeBuffer[4] = {0}; - std::expected headerSize = setPacketSize( - packetSizeHeader, - packetSizeBuffer, - sizeof(packetSizeBuffer), - size); - if (!headerSize.has_value()) { - return std::unexpected(headerSize.error()); - } - ::send(socket.value(), packetSizeBuffer, headerSize.value(), 0); - return ::send(socket.value(), buffer, size, 0); - } - else - { - return std::unexpected(ErrorCode::NotConnected); - } } std::future ChannelUnixTcpClient::connect() @@ -87,117 +51,28 @@ namespace styxlib }); } - std::future ChannelUnixTcpClient::disconnect() - { - return std::async( - std::launch::async, - [this]() - { - if (socket.has_value()) - { - ::close(socket.value()); - socket = std::nullopt; - } - }); - } - - bool ChannelUnixTcpClient::isConnected() const - { - return socket.has_value(); - } + // ── ChannelUnixTcpServer ────────────────────────────────────────────────── ChannelUnixTcpServer::ChannelUnixTcpServer(const Configuration &config) - : ChannelRx(), configuration(config) + : ChannelUnixSocketServer(config) { - if (setDeserializer(config.deserializer) != ErrorCode::Success) { - throw std::invalid_argument("Deserializer cannot be null"); - } - if (configuration.clientsRepo == nullptr) - { - throw std::invalid_argument("ClientsRepo must be provided in configuration"); - } - } - - ChannelUnixTcpServer::~ChannelUnixTcpServer() - { - stop().get(); } - SizeResult ChannelUnixTcpServer::sendBuffer( - ClientId clientId, - const StyxBuffer buffer, - Size size) + Socket ChannelUnixTcpServer::createServerSocket() { - if (!isStarted()) - { - return std::unexpected(ErrorCode::NotConnected); - } - - auto it = clientIdToChannelClient.find(clientId); - if (it == clientIdToChannelClient.end()) - { - return std::unexpected(ErrorCode::UnknownClient); - } - return it->second->sendBuffer(InvalidClientId, buffer, size); - } - - std::future ChannelUnixTcpServer::start() - { - startPromise = std::make_unique>(); - if (!isStarted()) - { - stopRequested.store(false); - serverThread = std::thread([this]() - { this->workThreadFunction(); }); - } - else - { - startPromise->set_value(ErrorCode::AlreadyStarted); - } - return startPromise->get_future(); - } - - std::future ChannelUnixTcpServer::stop() - { - return std::async( - std::launch::async, - [this]() - { - stopRequested.store(true); - if (this->serverThread.joinable()) - { - this->serverThread.join(); - } - this->startPromise = nullptr; - // clean structures - clientIdToChannelClient.clear(); - socketToClientInfoMapFull.clear(); - clientsObserver.setData(std::vector{}, true); - }); - } - - bool ChannelUnixTcpServer::isStarted() const - { - return running.load(); - } - - void ChannelUnixTcpServer::workThreadFunction() - { - running.store(true); - // Create the server socket int serverSocket = ::socket(AF_INET, SOCK_STREAM, 0); if (serverSocket < 0) { startPromise->set_value(ErrorCode::CantCreateSocket); - return; + return InvalidFileDescriptor; } int opt = 1; if (setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) { - close(serverSocket); + ::close(serverSocket); startPromise->set_value(ErrorCode::CantCreateSocket); - return; + return InvalidFileDescriptor; } sockaddr_in serverAddress; @@ -206,60 +81,36 @@ namespace styxlib serverAddress.sin_port = htons(configuration.port); if (::bind(serverSocket, - reinterpret_cast(&serverAddress), - sizeof(serverAddress)) < 0) + reinterpret_cast(&serverAddress), + sizeof(serverAddress)) < 0) { - close(serverSocket); + ::close(serverSocket); startPromise->set_value(ErrorCode::CantBindSocket); - return; + return InvalidFileDescriptor; } if (::listen(serverSocket, configuration.maxClients) < 0) { - close(serverSocket); + ::close(serverSocket); startPromise->set_value(ErrorCode::CantListenSocket); - return; + return InvalidFileDescriptor; } - // Set server socket to non-blocking mode + + // Non-blocking to allow the poll loop to time out gracefully int flags = fcntl(serverSocket, F_GETFL, 0); fcntl(serverSocket, F_SETFL, flags | O_NONBLOCK); - pollFds.push_back({serverSocket, POLLIN, 0}); - - startPromise->set_value(ErrorCode::Success); - - while (!stopRequested.load()) - { - int num_events = poll(pollFds.data(), pollFds.size(), 100); // 100 ms timeout - if (stopRequested.load()) - break; - if (num_events < 0) - { - std::cerr << "Poll error: " << strerror(errno) << std::endl; - break; - } else if (num_events == 0) { - // Timeout, no events - continue; - } else { - handlePollEvents(serverSocket, num_events); - if (stopRequested.load()) - break; - processBuffers(); - cleanupClosedSockets(); - } - } - ::close(serverSocket); - running.store(false); - stopRequested.store(false); + return serverSocket; } bool ChannelUnixTcpServer::acceptClients(Socket serverSocket) { - int clientSocket = accept(serverSocket, nullptr, nullptr); + int clientSocket = ::accept(serverSocket, nullptr, nullptr); if (clientSocket < 0) { return false; } + ClientFullInfo clientInfo; clientInfo.id = configuration.clientsRepo->getNextClientId(); clientInfo.buffer = std::vector(configuration.iounit); @@ -277,136 +128,37 @@ namespace styxlib } socketToClientInfoMapFull.insert({clientSocket, clientInfo}); - // Create a ChannelTx for the client + auto client = std::make_shared( - configuration.packetSizeHeader, + configuration.packetSizeHeader, clientSocket); - // Store the client with its socket as the ID clientIdToChannelClient[clientInfo.id] = client; - clientsObserver.setData( - socketToClientInfoMapFull | std::views::transform([](const auto &pair) { - return ClientInfo { - .id = pair.second.id, - .address = pair.second.address, - .port = pair.second.port - }; - }) - | std::ranges::to(), false); + + publishClients(); pollFds.push_back({clientSocket, POLLIN | POLLPRI | POLLERR | POLLHUP, 0}); return true; } - void ChannelUnixTcpServer::handlePollEvents(Socket serverSocket, size_t numEvents) { - for (const auto &pollItem: pollFds) { - if (pollItem.revents & POLLIN) { - if (pollItem.fd == serverSocket) { - // New incoming connections - while (acceptClients(serverSocket)) { - // Accept all pending connections - } - } else { - readDataFromSocket(pollItem.fd); - } - } - if (pollItem.revents & (POLLERR | POLLHUP)) { - socketsToClose.push_back(pollItem.fd); - std::cerr << "Error or hang-up on socket " << pollItem.fd << std::endl; - } - } - } - - void ChannelUnixTcpServer::readDataFromSocket(Socket clientFd) { + void ChannelUnixTcpServer::readDataFromSocket(Socket clientFd) + { auto it = socketToClientInfoMapFull.find(clientFd); - if (it == socketToClientInfoMapFull.end()) { + if (it == socketToClientInfoMapFull.end()) + { socketsToClose.push_back(clientFd); return; } ClientFullInfo &readBuffer = it->second; Size leftForRead = readBuffer.buffer.size() - readBuffer.currentSize; - ssize_t bytesRead = recv(clientFd, readBuffer.buffer.data() + readBuffer.currentSize, leftForRead, 0); - if (bytesRead > 0) { + ssize_t bytesRead = ::recv(clientFd, readBuffer.buffer.data() + readBuffer.currentSize, leftForRead, 0); + if (bytesRead > 0) + { readBuffer.currentSize += bytesRead; readBuffer.isDirty = true; - } else { - std::cerr << "Error reading from client socket " << clientFd << ": " << strerror(errno) << std::endl; - socketsToClose.push_back(clientFd); - } - } - - void ChannelUnixTcpServer::cleanupClosedSockets() { - for (Socket fd : socketsToClose) { - close(fd); - pollFds.erase(std::remove_if(pollFds.begin(), pollFds.end(), - [fd](const pollfd &p) { return p.fd == fd; }), - pollFds.end()); - auto it = std::find_if( - socketToClientInfoMapFull.begin(), - socketToClientInfoMapFull.end(), - [fd](const auto &pair) { return pair.first == fd; }); - - if (it != socketToClientInfoMapFull.end()) { - clientIdToChannelClient.erase(it->second.id); - } - socketToClientInfoMapFull.erase(fd); - } - - if (!socketsToClose.empty()) { - clientsObserver.setData(socketToClientInfoMapFull - | std::views::transform([](const auto &pair) { - return ClientInfo { - .id = pair.second.id, - .address = pair.second.address, - .port = pair.second.port - }; - }) - | std::ranges::to(), false); } - socketsToClose.clear(); - } - - void ChannelUnixTcpServer::processBuffers() { - auto headerSize = to_uint8_t(configuration.packetSizeHeader); - for (auto &pair : socketToClientInfoMapFull) { - ClientFullInfo &readBuffer = pair.second; - while (readBuffer.isDirty && readBuffer.currentSize > headerSize) { - uint32_t packetSize = 0; - switch (configuration.packetSizeHeader) - { - case PacketHeaderSize::Size1Byte: - packetSize = readBuffer.buffer[0]; - break; - case PacketHeaderSize::Size2Bytes: - packetSize = readBuffer.buffer[1] | (readBuffer.buffer[0] << 8); - break; - case PacketHeaderSize::Size4Bytes: - packetSize = readBuffer.buffer[3] | - (readBuffer.buffer[2] << 8) | - (readBuffer.buffer[1] << 16) | - (readBuffer.buffer[0] << 24); - break; - } - uint32_t packetSizeWithHeader = packetSize + to_uint8_t(configuration.packetSizeHeader); - if (readBuffer.currentSize >= packetSizeWithHeader) { - // Process the buffer with the deserializer - auto it = socketToClientInfoMapFull.find(pair.first); - if (it != socketToClientInfoMapFull.end()) { - ClientId clientId = it->second.id; - deserializer->handleBuffer(clientId, readBuffer.buffer.data() + to_uint8_t(configuration.packetSizeHeader), packetSize); - } - // move buffer - Size remainingSize = readBuffer.currentSize - packetSizeWithHeader; - if (remainingSize > 0) { - std::memmove(readBuffer.buffer.data(), - readBuffer.buffer.data() + packetSizeWithHeader, - remainingSize); - } - readBuffer.currentSize = remainingSize; - } else { - // Not enough data to process complete packet; waiting for more data - break; - } - } - readBuffer.isDirty = false; + else + { + std::cerr << "Error reading from TCP client socket " << clientFd << ": " << strerror(errno) << std::endl; + socketsToClose.push_back(clientFd); } } -} // namespace styxlib \ No newline at end of file +} // namespace styxlib diff --git a/cpp/modules/l4/src/impl/ChannelUnixUdp.cpp b/cpp/modules/l4/src/impl/ChannelUnixUdp.cpp new file mode 100644 index 00000000..5ca6eac2 --- /dev/null +++ b/cpp/modules/l4/src/impl/ChannelUnixUdp.cpp @@ -0,0 +1,280 @@ +#include "impl/ChannelUnixUdp.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "ChannelRx.h" + +namespace styxlib +{ + ChannelUnixUdpClient::ChannelUnixUdpClient(const Configuration &config) + : ChannelUnixSocketClient(config) + { + } + + std::future ChannelUnixUdpClient::connect() + { + return std::async( + std::launch::async, + [this] + { + if (isConnected()) + { + return ErrorCode::AlreadyStarted; + } + + // Create a UDP socket. + int sock = ::socket(AF_INET, SOCK_DGRAM, 0); + if (sock < 0) + { + return ErrorCode::CantCreateSocket; + } + + // "Connect" the UDP socket to fix the remote address so that + // subsequent ::send() calls work without specifying the peer. + sockaddr_in serverAddress{}; + serverAddress.sin_family = AF_INET; + serverAddress.sin_port = htons(configuration.port); + inet_pton(AF_INET, configuration.address.c_str(), &serverAddress.sin_addr); + + int result = ::connect(sock, + reinterpret_cast(&serverAddress), + sizeof(serverAddress)); + if (result == 0) + { + this->socket = sock; + return ErrorCode::Success; + } + else + { + ::close(sock); + return ErrorCode::NotConnected; + } + }); + } + + // ── ChannelUnixUdpServer ────────────────────────────────────────────────── + + ChannelUnixUdpServer::ChannelUnixUdpServer(const Configuration &config) + : ChannelUnixSocketServer(config) + { + if (configuration.iounit > UdpMaxPayloadBytes) + { + throw std::invalid_argument( + "iounit (" + std::to_string(configuration.iounit) + + ") exceeds the maximum UDP payload size for a single Ethernet " + "frame without fragmentation (" + + std::to_string(UdpMaxPayloadBytes) + " bytes)"); + } + } + + Socket ChannelUnixUdpServer::createServerSocket() + { + int sock = ::socket(AF_INET, SOCK_DGRAM, 0); + if (sock < 0) + { + startPromise->set_value(ErrorCode::CantCreateSocket); + return InvalidFileDescriptor; + } + + int opt = 1; + if (::setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) + { + ::close(sock); + startPromise->set_value(ErrorCode::CantCreateSocket); + return InvalidFileDescriptor; + } + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = INADDR_ANY; + addr.sin_port = htons(configuration.port); + + if (::bind(sock, reinterpret_cast(&addr), sizeof(addr)) < 0) + { + ::close(sock); + startPromise->set_value(ErrorCode::CantBindSocket); + return InvalidFileDescriptor; + } + + // Non-blocking so the poll loop can time out gracefully + int flags = fcntl(sock, F_GETFL, 0); + if (flags == -1) + { + ::close(sock); + startPromise->set_value(ErrorCode::CantCreateSocket); + return InvalidFileDescriptor; + } + if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1) + { + ::close(sock); + startPromise->set_value(ErrorCode::CantCreateSocket); + return InvalidFileDescriptor; + } + + return sock; + } + + void ChannelUnixUdpServer::handlePollEvents(Socket serverSocket, size_t /*numEvents*/) + { + // Each UDP datagram is one complete Styx packet: length-header + payload. + // There is nothing to accumulate – deliver each datagram to the + // deserializer immediately after stripping the framing header. + const uint8_t headerBytes = to_uint8_t(configuration.packetSizeHeader); + std::vector datagram(configuration.iounit); + + for (const auto &pollItem : pollFds) + { + if (pollItem.revents & POLLIN) + { + // Drain all pending datagrams from the single server socket. + while (true) + { + sockaddr_in peerAddr{}; + socklen_t addrLen = sizeof(peerAddr); + + ssize_t bytesRead = ::recvfrom( + serverSocket, + datagram.data(), datagram.size(), + MSG_DONTWAIT, + reinterpret_cast(&peerAddr), &addrLen); + + if (bytesRead <= 0) + break; // EAGAIN / EWOULDBLOCK – no more datagrams right now + + if (static_cast(bytesRead) <= headerBytes) + { + std::cerr << "UDP: datagram too short to contain a header – discarded" << std::endl; + continue; + } + const Size payloadSize = static_cast(bytesRead) - headerBytes; + const auto packetSizeResult = getPacketSize( + configuration.packetSizeHeader, + datagram.data(), + bytesRead); + if (!packetSizeResult.has_value()) + { + std::cerr << "UDP: failed to parse packet size from header – discarded" << + std::endl; + continue; + } + if (packetSizeResult.value() != payloadSize) + { + std::cerr << "UDP: payload size mismatch (header says " << + packetSizeResult.value() << " bytes, but recvfrom got " << + payloadSize << " bytes) – discarded" << + std::endl; + continue; + } + + + + // ── Identify / register the client ────────────────────── + char ipStr[INET_ADDRSTRLEN] = {0}; + ::inet_ntop(AF_INET, &peerAddr.sin_addr, ipStr, sizeof(ipStr)); + uint16_t peerPort = ntohs(peerAddr.sin_port); + std::string addrKey = std::string(ipStr) + ":" + std::to_string(peerPort); + + ClientId clientId; + auto addrIt = addressToClientId.find(addrKey); + if (addrIt == addressToClientId.end()) + { + clientId = configuration.clientsRepo->getNextClientId(); + addressToClientId[addrKey] = clientId; + clientIdToSockAddr[clientId] = peerAddr; + + // Register in the base map (buffer unused) so that + // publishClients() iteration works unchanged. + ClientFullInfo info; + info.id = clientId; + info.address = ipStr; + info.port = peerPort; + // No ReadBuffer needed – UDP datagrams go directly to + // the deserializer. isDirty stays false forever so + // processBuffers() is a no-op for UDP entries. + socketToClientInfoMapFull.insert({static_cast(clientId), std::move(info)}); + publishClients(); + } + else + { + clientId = addrIt->second; + } + + // ── Deliver payload directly to the deserializer ───────── + // Skip the framing header; the remaining bytes are the + // complete Styx message payload. + deserializer->handleBuffer( + clientId, + datagram.data() + headerBytes, + payloadSize); + } + } + if (pollItem.revents & (POLLERR | POLLHUP)) + { + std::cerr << "Error on UDP server socket " << pollItem.fd << std::endl; + stopRequested.store(true); + } + } + } + + void ChannelUnixUdpServer::readDataFromSocket(Socket /*serverFd*/) + { + // UDP server reads all datagrams inside handlePollEvents via recvfrom. + // This override exists only to satisfy the base-class virtual – it is + // never called in the UDP server code path. + } + + void ChannelUnixUdpServer::cleanupClosedSockets() + { + // UDP has no per-client sockets to close; pseudo-socket keys in + // socketToClientInfoMapFull are just ClientIds, not real fds. + socketsToClose.clear(); + } + + SizeResult ChannelUnixUdpServer::sendBuffer( + ClientId clientId, + const StyxBuffer buffer, + Size size) + { + if (!isStarted()) + return Unexpected(ErrorCode::NotConnected); + + auto addrIt = clientIdToSockAddr.find(clientId); + if (addrIt == clientIdToSockAddr.end()) + return Unexpected(ErrorCode::UnknownClient); + + const uint8_t headerBytes = to_uint8_t(configuration.packetSizeHeader); + std::vector datagram(headerBytes + size); + auto headerResult = setPacketSize( + configuration.packetSizeHeader, + datagram.data(), datagram.size(), size); + if (!headerResult) + return Unexpected(headerResult.error()); + + std::memcpy(datagram.data() + headerBytes, buffer, size); + + ssize_t sent = ::sendto( + serverSocketFd, + datagram.data(), datagram.size(), 0, + reinterpret_cast(&addrIt->second), + sizeof(sockaddr_in)); + + if (sent < 0) + return Unexpected(ErrorCode::SendFailed); + + // Return the number of payload bytes sent (excluding the header) + return static_cast(sent) - headerBytes; + } + + void ChannelUnixUdpServer::onStop() + { + addressToClientId.clear(); + clientIdToSockAddr.clear(); + } +} // namespace styxlib + diff --git a/cpp/modules/l4/test/test_ChannelUnixTcp.cpp b/cpp/modules/l4/test/test_ChannelUnixTcp.cpp index 83a89451..186ad927 100644 --- a/cpp/modules/l4/test/test_ChannelUnixTcp.cpp +++ b/cpp/modules/l4/test/test_ChannelUnixTcp.cpp @@ -98,13 +98,13 @@ class TestSuite } }; -TEST_CASE_METHOD(TestSuite, "Server starts and stops", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestSuite, "ChannelUnixTcpServer:starts and stops", "[ChannelUnixTcpServer]") { waitStartServer(); waitStopServer(); } -TEST_CASE_METHOD(TestSuite, "test_ServerAcceptsConnections", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestSuite, "ChannelUnixTcpServer:accepts connections", "[ChannelUnixTcpServer]") { waitStartServer(); auto future = server->getClientsObserver().waitAsync(); @@ -128,7 +128,7 @@ TEST_CASE_METHOD(TestSuite, "test_ServerAcceptsConnections", "[ChannelUnixTcpSer waitStopServer(); } -TEST_CASE_METHOD(TestSuite, "test_ServerReceiveMessages", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestSuite, "ChannelUnixTcpServer:receive messages", "[ChannelUnixTcpServer]") { waitStartServer(); connectClient(); @@ -152,7 +152,7 @@ TEST_CASE_METHOD(TestSuite, "test_ServerReceiveMessages", "[ChannelUnixTcpServer server->stop().get(); } -TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents accept connections", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestChannelUnixTcpServer, "ChannelUnixTcpServer:handlePollEvents accept connections", "[ChannelUnixTcpServer]") { pollFds.clear(); pollFds.push_back({ .fd = 1, .events = POLLIN, .revents = POLLIN }); // Simulate server socket ready to accept @@ -160,7 +160,7 @@ TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents accept connections" REQUIRE(acceptCalled == 1); } -TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents read data from socket", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestChannelUnixTcpServer, "ChannelUnixTcpServer:handlePollEvents read data from socket", "[ChannelUnixTcpServer]") { pollFds.clear(); pollFds.push_back({ .fd = 1, .events = POLLIN, .revents = 0 }); // Simulate server socket ready to accept @@ -171,7 +171,7 @@ TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents read data from sock REQUIRE(readDataFromSocketCalled == 1); } -TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents should cleanup closed sockets", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestChannelUnixTcpServer, "ChannelUnixTcpServer:handlePollEvents should cleanup closed sockets", "[ChannelUnixTcpServer]") { dontCallRealMethods = false; // let the real methods be called pollFds.clear(); @@ -189,7 +189,7 @@ TEST_CASE_METHOD(TestChannelUnixTcpServer, "handlePollEvents should cleanup clos REQUIRE(readDataFromSocketCalled == 1); } -TEST_CASE_METHOD(TestChannelUnixTcpServer, "test_processBuffers", "[ChannelUnixTcpServer]") +TEST_CASE_METHOD(TestChannelUnixTcpServer, "ChannelUnixTcpServer:processBuffers", "[ChannelUnixTcpServer]") { dontCallRealMethods = false; // let the real methods be called diff --git a/cpp/modules/l4/test/test_dummy.cpp b/cpp/modules/l4/test/test_dummy.cpp new file mode 100644 index 00000000..444c89e5 --- /dev/null +++ b/cpp/modules/l4/test/test_dummy.cpp @@ -0,0 +1,6 @@ +#include + +TEST_CASE("Dummy test", "[dummy]") +{ + REQUIRE(1 + 1 == 2); +} \ No newline at end of file diff --git a/cpp/modules/l5/include/Client.h b/cpp/modules/l5/include/Client.h index bf741254..d91ffc7f 100644 --- a/cpp/modules/l5/include/Client.h +++ b/cpp/modules/l5/include/Client.h @@ -1,7 +1,7 @@ #pragma once #include "messages/base/StyxMessage.h" -#include "Channel.h" +#include "ChannelRx.h" #include "ChannelTx.h" namespace styxlib