Add UnixFileChannel

.vscode/tasks.json

@@ -31,7 +31,15 @@
                 "isDefault": true
             },
             "problemMatcher": ["$gcc"],
-            "detail": "Generated task by Debugger."
+            "detail": "Generated task by Debugger.",
+            "presentation": {
+                "echo": true,
+                "reveal": "always",
+                "focus": false,
+                "panel": "shared",
+                "showReuseMessage": true,
+                "clear": true
+            }
         }
     ],
     "version": "2.0.0"

cpp/modules/l4/CMakeLists.txt

@@ -4,12 +4,11 @@ add_library(module_l4 STATIC)
 
 target_sources(module_l4
     PRIVATE
+        src/Channel.cpp
         src/impl/ChannelUnixTcp.cpp
         src/impl/ClientsRepoImpl.cpp
-    PUBLIC
-        include/Channel.h
-        include/ChannelUnixTcp.h
-        include/data.h
+        src/impl/ChannelUnixPipeImpl.cpp
+        src/impl/ChannelUnixFile.cpp
     )
 
 target_include_directories(module_l4 PUBLIC
@@ -19,6 +18,7 @@ target_include_directories(module_l4 PUBLIC
 
 add_executable(module_l4_tests
     test/test_ChannelUnixTcp.cpp
+    test/test_ChannelUnixPipeImpl.cpp
     )
 target_include_directories(module_l4_tests PRIVATE ./include/)
 target_link_libraries(module_l4_tests PUBLIC Catch2::Catch2WithMain module_l4)

cpp/modules/l4/include/Channel.h

@@ -1,18 +1,48 @@
 #pragma once
 #include <memory>
+#include <expected>
 
 #include "data.h"
 #include "SerializerL4.h"
 
 namespace styxlib
 {
+    enum class PacketHeaderSize : uint8_t
+    {
+        Size1Byte = 1,
+        Size2Bytes = 2,
+        Size4Bytes = 4
+    };
+
+    inline uint8_t to_uint8_t(const PacketHeaderSize &headerSize)
+    {
+        return static_cast<uint8_t>(headerSize);
+    }
+
+    /**
+     * 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.
+     */
+    std::expected<uint8_t, ErrorCode> setPacketSize(
+        const PacketHeaderSize &headerSize,
+        uint8_t *buffer,
+        Size bufferSize,
+        Size packetSize);
+
     class ChannelRx
     {
     protected:
         DeserializerL4Ptr deserializer;
+
     public:
-        ChannelRx(DeserializerL4Ptr deserializer) : deserializer(deserializer) {
-            if (deserializer == nullptr) {
+        ChannelRx(DeserializerL4Ptr deserializer) : deserializer(deserializer)
+        {
+            if (deserializer == nullptr)
+            {
                 throw std::invalid_argument("Deserializer cannot be null");
             }
         }
@@ -23,12 +53,6 @@ namespace styxlib
     {
     public:
         virtual ~ChannelTx() = default;
-        virtual SizeResult sendBuffer(const StyxBuffer buffer, Size size) = 0;
-    };
-
-    class ChannelTxOneToMany {
-    public:
-        virtual ~ChannelTxOneToMany() = default;
         virtual SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) = 0;
     };
 }

cpp/modules/l4/include/data.h

@@ -16,6 +16,7 @@ namespace styxlib
         CantCreateSocketPoll,
         PacketTooLarge,
         UnknownClient,
+        BufferTooSmall,
     };
 
     using StyxString = std::string;
@@ -26,7 +27,9 @@ namespace styxlib
     using Fid = uint32_t;
     using Tag = uint16_t;
     using Type = uint16_t;
-    using ClientId = int;
+    using ClientId = uint16_t;
+
+    constexpr ClientId InvalidClientId = 0;
 
     class SerializerL4;
     class DeserializerL4;

cpp/modules/l4/include/impl/ChannelUnixFile.h

@@ -0,0 +1,61 @@
+#pragma once
+
+#include <vector>
+#include <cstdint>
+
+#include "data.h"
+#include "Channel.h"
+
+namespace styxlib
+{
+    using FileDescriptor = int;
+    constexpr FileDescriptor InvalidFileDescriptor = -1;
+
+    struct ReadBuffer
+    {
+        std::vector<uint8_t> buffer;
+        Size currentSize{0};
+        bool isDirty{false};
+    };
+
+    class ChannelUnixFile : public ChannelTx, public ChannelRx
+    {
+    public:
+        struct Configuration
+        {
+            PacketHeaderSize packetSizeHeader{PacketHeaderSize::Size2Bytes};
+            uint16_t iounit{8192};
+            DeserializerL4Ptr deserializer{nullptr};
+            Configuration(
+                PacketHeaderSize packetSizeHeader,
+                uint16_t iounit,
+                DeserializerL4Ptr deserializer)
+                : packetSizeHeader(packetSizeHeader),
+                  iounit(iounit),
+                  deserializer(deserializer) {}
+        };
+        class FileDescriptorPair
+        {
+        public:
+            FileDescriptor readFd;
+            FileDescriptor writeFd;
+            FileDescriptorPair(FileDescriptor readFd, FileDescriptor writeFd)
+                : readFd(readFd), writeFd(writeFd) {}
+        };
+
+    protected:
+        FileDescriptorPair fds;
+        Configuration config;
+        ReadBuffer readBufferData;
+
+        virtual void closeDescriptors();
+    public:
+        ChannelUnixFile(const Configuration &config);
+        ~ChannelUnixFile() override;
+        SizeResult sendBuffer(
+            ClientId clientId,
+            const StyxBuffer buffer,
+            Size size) override;
+        void readBufferBlocking();
+    };
+} // namespace styxlib

cpp/modules/l4/include/impl/ChannelUnixPipeImpl.h

@@ -0,0 +1,38 @@
+#pragma once
+
+#include <expected>
+#include <future>
+#include <atomic>
+#include <memory>
+#include <thread>
+
+#include "ChannelUnixFile.h"
+
+namespace styxlib
+{
+    class ChannelUnixPipeImpl : public ChannelUnixFile
+    {
+    public:
+        using StartResult = std::expected<ChannelUnixFile::FileDescriptorPair, ErrorCode>;
+    private:
+        std::thread serverThread;
+        std::atomic<bool> running{false};
+        std::atomic<bool> stopRequested{false};
+        std::unique_ptr<std::promise<StartResult>> startPromise;
+        FileDescriptorPair rxFds;
+        FileDescriptorPair txFds;
+        bool isDescriptorOwned{false};
+    private:
+        void workThreadFunction();
+    protected:
+        void closeDescriptors() override;        
+    public:
+        ChannelUnixPipeImpl(const ChannelUnixFile::Configuration &config);
+        std::future<StartResult> start();
+        std::future<ErrorCode> connect(const FileDescriptorPair& fds);
+        bool isStarted() const;
+        bool isConnected() const;
+        std::future<void> stop();
+        FileDescriptorPair getClientFileDescriptors() const;
+    };
+} // namespace styxlib

cpp/modules/l4/include/ChannelUnixTcp.h → cpp/modules/l4/include/impl/ChannelUnixTcp.h

@@ -10,26 +10,11 @@
 #include "Channel.h"
 #include "ClientsRepo.h"
 #include "impl/ProgressObservableMutexImpl.h"
+#include "impl/ChannelUnixFile.h"
 
 namespace styxlib
 {
-    using Socket = int;
-
-    enum class PacketHeaderSize : uint8_t {
-        Size1Byte = 1,
-        Size2Bytes = 2,
-        Size4Bytes = 4
-    };
-
-    inline uint8_t to_uint8_t(const PacketHeaderSize &headerSize) {
-        return static_cast<uint8_t>(headerSize);
-    }
-
-    struct ReadBuffer {
-        std::vector<uint8_t> buffer;
-        Size currentSize{0};
-        bool isDirty{false};
-    };
+    using Socket = FileDescriptor;
 
     class ChannelUnixTcpTx : public ChannelTx {
     protected:
@@ -43,7 +28,7 @@ namespace styxlib
         ChannelUnixTcpTx &operator=(ChannelUnixTcpTx &&) = delete;
         ChannelUnixTcpTx &operator=(const ChannelUnixTcpTx &) = delete;
         virtual ~ChannelUnixTcpTx() = default;
-        SizeResult sendBuffer(const StyxBuffer buffer, Size size) override;
+        SizeResult sendBuffer(ClientId clientId, const StyxBuffer buffer, Size size) override;
     };
 
     class ChannelUnixTcpClient : public ChannelUnixTcpTx, public ChannelRx
@@ -81,7 +66,7 @@ namespace styxlib
         bool isConnected() const;
     };
 
-    class ChannelUnixTcpServer : public ChannelRx, public ChannelTxOneToMany
+    class ChannelUnixTcpServer : public ChannelRx, public ChannelTx
     {
     public:
         class Configuration

cpp/modules/l4/include/impl/ClientsRepoImpl.h

@@ -8,7 +8,7 @@ namespace styxlib
     class ClientsRepoImpl : public styxlib::ClientsRepo
     {
     private:
-        std::atomic_int nextId{0};
+        std::atomic<ClientId> nextId{InvalidClientId + 1};
 
     public:
         styxlib::ClientId getNextClientId() override;

cpp/modules/l4/src/Channel.cpp

@@ -0,0 +1,39 @@
+#include "Channel.h"
+
+namespace styxlib
+{
+    std::expected<uint8_t, ErrorCode> 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<uint8_t>(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;
+        }
+        return static_cast<uint8_t>(headerSize);
+    }
+} // namespace styxlib