TCP - partially implement DecoderFactory

src/plugins/input/tcp/CMakeLists.txt

@@ -7,6 +7,7 @@ add_library(tcp-input MODULE
     src/DecodeBuffer.cpp
     src/Connection.cpp
     src/ClientManager.cpp
+    src/DecoderFactory.cpp
 )
 
 install(

src/plugins/input/tcp/src/ClientManager.cpp

@@ -13,6 +13,7 @@
 #include <stdexcept> // runtime_error, exception
 #include <string>    // string
 #include <memory>    // unique_ptr, make_unique
+#include <cerrno>    // errno, EINTR
 
 #include <sys/epoll.h> // epoll_create, epoll_event, EPOLLIN, EPOLL_CTL_ADD
 #include <fcntl.h>     // fcntl, F_GETFL, F_SETFL, O_NONBLOCK

src/plugins/input/tcp/src/DecodeBuffer.cpp

@@ -68,7 +68,7 @@ void DecodeBuffer::read_from(
 }
 
 const uint8_t *DecodeBuffer::read_header(const uint8_t *data, size_t size) {
-    constexpr size_t HDR_SIZE = sizeof(struct fds_ipfix_msg_hdr);
+    constexpr size_t HDR_SIZE = sizeof(fds_ipfix_msg_hdr);
 
     auto filled = m_part_decoded.size();
     auto remaining = 0;
@@ -80,7 +80,7 @@ const uint8_t *DecodeBuffer::read_header(const uint8_t *data, size_t size) {
         }
     }
 
-    auto hdr = reinterpret_cast<const struct fds_ipfix_msg_hdr *>(m_part_decoded.get());
+    auto hdr = reinterpret_cast<const fds_ipfix_msg_hdr *>(m_part_decoded.get());
     size_t length = hdr->length;
 
     m_part_decoded.reserve(length);

src/plugins/input/tcp/src/DecoderFactory.cpp

@@ -0,0 +1,90 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief Factory for creating decoders (source file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "DecoderFactory.hpp"
+
+#include <memory>    // unique_ptr
+#include <array>     // array
+#include <cstdint>   // uint8_t, uint16_t, uint32_t
+#include <stdexcept> // runtime_error
+#include <cerrno>    // errno
+#include <string>    // string
+
+#include <sys/socket.h> // recv, MSG_PEEK, MSG_WAITALL
+#include <netinet/in.h> // ntohl, ntohs
+
+#include <ipfixcol2.h> // ipx_strerror
+
+#include "Decoder.hpp" // Decoder
+
+namespace tcp_in {
+
+using namespace std;
+
+constexpr uint16_t IPFIX_MAGIC = 10;
+constexpr uint32_t LZ4_MAGIC = 0;
+
+DecoderFactory::DecoderFactory() {};
+
+unique_ptr<Decoder> DecoderFactory::detect_decoder(int fd) {
+    // number of bytes neaded to detect the decoder
+    constexpr size_t MAX_MAGIC_LEN = 4;
+
+    array<uint8_t, MAX_MAGIC_LEN> buf{};
+
+    auto res = recv(fd, buf.begin(), buf.size(), MSG_PEEK | MSG_WAITALL);
+    if (res == -1) {
+        const char *err_msg;
+        ipx_strerror(errno, err_msg);
+        throw runtime_error("Failed to receive start of first message: " + string(err_msg));
+    }
+
+    constexpr const char *not_enough_data_err =
+        "Failed to read enough bytes to recognize the decoder";
+
+    // check decoders in order from shortest magic number to longest
+
+    if (res < 2) {
+        throw runtime_error(not_enough_data_err);
+    }
+
+    // IPFIX decoder
+    auto magic_u16 = ntohs(*reinterpret_cast<uint16_t *>(buf.begin()));
+    if (magic_u16 == IPFIX_MAGIC) {
+        return create_ipfix_decoder(fd);
+    }
+
+    if (res < 4) {
+        throw runtime_error(not_enough_data_err);
+    }
+
+    // LZ4 decoder
+    auto magic_u32 = ntohl(*reinterpret_cast<uint16_t *>(buf.begin()));
+    if (magic_u32 == LZ4_MAGIC) {
+        return create_lz4_decoder(fd);
+    }
+
+    throw runtime_error("Failed to recognize the decoder.");
+}
+
+unique_ptr<Decoder> DecoderFactory::create_ipfix_decoder(int fd) {
+    (void)fd;
+    // TODO
+    throw runtime_error("Not implemented");
+}
+
+unique_ptr<Decoder> DecoderFactory::create_lz4_decoder(int fd) {
+    (void)fd;
+    // TODO
+    throw runtime_error("Not implemented");
+}
+
+} // namespace tcp_in
+

src/plugins/input/tcp/src/DecoderFactory.hpp

@@ -29,6 +29,10 @@ class DecoderFactory {
      * @return Instance of the correct decoder, nullptr no decoder matches the data.
      */
     std::unique_ptr<Decoder> detect_decoder(int fd);
+
+private:
+    std::unique_ptr<Decoder> create_ipfix_decoder(int fd);
+    std::unique_ptr<Decoder> create_lz4_decoder(int fd);
 };
 
 } // namespace tcp_in