TCP - Add plugin class

Author: BonnyAD9

src/plugins/input/tcp/CMakeLists.txt

@@ -11,6 +11,7 @@ add_library(tcp-input MODULE
     src/Lz4Decoder.cpp
     src/DecoderFactory.cpp
     src/Acceptor.cpp
+    src/Plugin.cpp
 )
 
 if (CMAKE_HOST_SYSTEM_NAME STREQUAL "FreeBSD" OR CMAKE_HOST_SYSTEM_NAME STREQUAL "OpenBSD")

src/plugins/input/tcp/src/Plugin.cpp

@@ -0,0 +1,71 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief Tcp input plugin for ipfixcol2 (source file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "Plugin.hpp"
+
+#include <array>     // array
+#include <stdexcept> // exception
+#include <cstddef>   // size_t
+
+#include <ipfixcol2.h> // ipx_ctx_t, IPX_CTX_ERROR, IPX_CTX_INFO, IPX_CTX_WARNING, ipx_session
+
+#include "Config.hpp"         // Config
+#include "DecoderFactory.hpp" // DecoderFactory
+#include "Connection.hpp"     // Connection
+
+namespace tcp_in {
+
+Plugin::Plugin(ipx_ctx_t *ctx, Config &config) :
+    m_ctx(ctx),
+    m_clients(ctx),
+    m_acceptor(m_clients, DecoderFactory(), ctx)
+{
+    m_acceptor.bind_addresses(config);
+    m_acceptor.start();
+}
+
+void Plugin::get() {
+    constexpr int MAX_CONNECTIONS = 16;
+    std::array<Connection *, MAX_CONNECTIONS> connections{};
+
+    auto count = m_clients.wait_for_connections(connections.begin(), connections.size());
+
+    for (size_t i = 0; i < count; ++i) {
+        try {
+            if (!connections[i]->receive(m_ctx)) {
+                // EOF reached
+                auto session = connections[i]->get_session();
+                IPX_CTX_INFO(m_ctx, "Closing %s", session->ident);
+                m_clients.close_connection(session);
+            }
+        } catch (std::exception &ex) {
+            IPX_CTX_ERROR(m_ctx, "%s", ex.what());
+            auto session = connections[i]->get_session();
+            IPX_CTX_INFO(m_ctx, "Closing %s", session->ident);
+            m_clients.close_connection(session);
+        }
+    }
+}
+
+void Plugin::close_session(const ipx_session *session) noexcept {
+    m_clients.close_connection(session);
+}
+
+Plugin::~Plugin() {
+    try {
+        m_acceptor.stop();
+        m_clients.close_all_connections();
+    } catch (std::exception &ex) {
+        IPX_CTX_WARNING(m_ctx, "%s", ex.what());
+    }
+}
+
+} // namespace tcp_in
+

src/plugins/input/tcp/src/Plugin.hpp

@@ -0,0 +1,60 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief Tcp input plugin for ipfixcol2 (header file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include <ipfixcol2.h> // ipx_ctx_t, ipx_session
+
+#include "Config.hpp"        // Config
+#include "ClientManager.hpp" // ClientManager
+#include "Acceptor.hpp"      // Acceptor
+
+namespace tcp_in {
+
+/** TCP input plugin for ipfixcol2. */
+class Plugin {
+public:
+    /**
+     * @brief Creates new TCP plugin instance.
+     * @param[in] ctx
+     * @param[in] config configuration of the plugin
+     * @throws when fails to init acceptor
+     * @throws when fails to bind to addresses from config
+     */
+    Plugin(ipx_ctx_t *ctx, Config &config);
+
+    // force that Plugin stays in its original memory (so that reference to `m_clients` in acceptor
+    // stays valid)
+    Plugin(const Plugin &) = delete;
+    Plugin(Plugin &&) = delete;
+
+    /**
+     * @brief Wait for the next tcp message and process all received messages.
+     * @throws when fails to wait for connections
+     * @throws when fails to receive
+     */
+    void get();
+
+    /**
+     * @brief Close the given session.
+     * @param session Session to close.
+     */
+    void close_session(const ipx_session *session) noexcept;
+
+    ~Plugin();
+private:
+    ipx_ctx_t *m_ctx;
+    ClientManager m_clients;
+    /** Acceptor thread. */
+    Acceptor m_acceptor;
+};
+
+} // namespace tcp_in
+