Improve performance of IpfixDecoder.

Copying the functionality of DecodeBuffer directly to IpfixDecoder
greatly improves performance of the IpfixDecoder.

Author: BonnyAD9

src/plugins/input/tcp/src/IpfixDecoder.cpp

@@ -10,13 +10,65 @@
 
 #include "IpfixDecoder.hpp"
 
+#include <stdexcept>
+
 #include "DecodeBuffer.hpp" // DecodeBuffer
 #include "Reader.hpp"
 
 namespace tcp_in {
 
 void IpfixDecoder::progress() {
-    m_decoded.read_from(m_reader);
+    // Directly calling `m_decoded.read_from` is somehow much slower than copying the code directly
+    // here.
+
+    while (!m_decoded.enough_data()) {
+        if (!read_header()) {
+            break;
+        }
+
+        if (!read_body()) {
+            break;
+        }
+    }
+}
+
+bool IpfixDecoder::read_header() {
+    if (m_decoded_size != 0) {
+        return true;
+    }
+
+    if (!read_until_n(sizeof(fds_ipfix_msg_hdr))) {
+        return false;
+    }
+
+
+    // The header has been successfully read. Load the size of the message.
+    auto hdr = reinterpret_cast<const fds_ipfix_msg_hdr *>(m_part_decoded.data());
+    m_decoded_size = ntohs(hdr->length);
+
+    if (m_decoded_size < sizeof(fds_ipfix_msg_hdr)) {
+        throw std::runtime_error("Invalid IPFIX message header size.");
+    }
+
+    return true;
+}
+
+bool IpfixDecoder::read_body() {
+    // Read the body
+    if (!read_until_n(m_decoded_size)) {
+        // There is not enough data to read the whole body.
+        return false;
+    }
+
+    m_decoded.add(std::move(m_part_decoded));
+    m_part_decoded = ByteVector();
+    m_decoded_size = 0;
+    return true;
+}
+
+bool IpfixDecoder::read_until_n(size_t n) {
+    m_reader.read_until_n(n, m_part_decoded, m_decoded);
+    return m_part_decoded.size() == n;
 }
 
 } // namespace tcp_in

src/plugins/input/tcp/src/IpfixDecoder.hpp

@@ -40,6 +40,13 @@ class IpfixDecoder : public Decoder {
     }
 
 private:
+    bool read_header();
+    bool read_body();
+    bool read_until_n(size_t n);
+
+    ByteVector m_part_decoded;
+    size_t m_decoded_size = 0;
+
     TcpReader m_reader;
     DecodeBuffer m_decoded;
 };