TCP - continue work on the skeleton

src/plugins/input/tcp/src/Connection.hpp

@@ -27,14 +27,28 @@ namespace tcp_in
  */
 class Connection {
 public:
+    /**
+     * @brief Creates new connection with this TCP connection file descriptor.
+     * @param fd file descriptor of the new tcp connection.
+     */
+    Connection(int fd);
+
     /**
      * @brief Reads from the TCP session while there is data or unil a full message is readed.
      *
      * @param ctx Ipx context (used to send the readed message)
      */
     void recieve(ipx_ctx_t *ctx);
 
-    void close(ipx_ctx_t *ctx /* TODO: epoll parameters */);
+    /**
+     * @brief Closes this connection and removes it from the epoll. Allocated data is freed in
+     *        destructor.
+     *
+     * @param ctx
+     * @param epoll_fd Epoll in which this connection is.
+     */
+    void close(ipx_ctx_t *ctx, int epoll_fd);
+
 private:
     Session session;
     /**

src/plugins/input/tcp/src/ConnectionVec.hpp

@@ -0,0 +1,56 @@
+/**
+ * \file
+ * \author Jakub Antonín Štigler <[email protected]>
+ * \brief Mutexed vector of tcp connections (header file)
+ * \date 2024
+ *
+ * Copyright: (C) 2023 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include <vector> // std::vector
+#include <memory> // std::unique_ptr
+#include <mutex>  // std::mutex, std::lock_guard
+
+#include <ipfixcol2.h> // ipx_session
+
+#include "Connection.hpp" // Connection
+
+namespace tcp_in
+{
+
+class ConnectionVec {
+public:
+    // TODO: methods for waiting for connection and getting the connections
+
+    /**
+     * @brief Adds connection to the vector and epoll.
+     * @param fd file descriptor of the new tcp connection.
+     */
+    void add_connection(int fd);
+
+    /**
+     * @brief Removes connection from the vector based on its session. This is safe only for the
+     *        main thread (not the acceptor thread).
+     * @param session session of the connection to remove.
+     */
+    void remove_connection(struct ipx_session *session);
+private:
+    /**
+     * @brief Locks the session for safe adding, removing is safe only for the main thread (not the
+     *        acceptor thread).
+     * @return Lock guard, vec is unlocked when it goes out of scope.
+     */
+    inline std::lock_guard<std::mutex> get_lock() {
+        return std::lock_guard(m_mutex);
+    }
+
+    int m_epoll_fd;
+    std::mutex m_mutex;
+    std::vector<std::unique_ptr<Connection>> m_connections;
+};
+
+} // namespace tcp_in
+

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

@@ -13,11 +13,12 @@
 #include <vector> // std::vector
 #include <thread> // std::thread
 #include <memory> // std::unique_ptr
+#include <mutex>
 
 #include <ipfixcol2.h> // ipx_ctx_t, ipx_session
 
-#include "Config.hpp"  // Config
-#include "Session.hpp" // Session
+#include "Config.hpp"        // Config
+#include "ConnectionVec.hpp" // ConnectionVec
 
 namespace tcp_in
 {
@@ -43,7 +44,7 @@ class Plugin {
     }
 
     /**
-     * @brief Wait for the next tcp message and process it.
+     * @brief Wait for the next tcp message and process all recieved messages.
      */
     void get();
 
@@ -55,9 +56,9 @@ class Plugin {
 
 private:
     ipx_ctx_t *m_ctx;
-    std::vector<std::unique_ptr<Session>> m_sessions;
+    ConnectionVec m_connections;
 
-    // TODO: epoll, mutexes, ...
+    // TODO: class for acceptor
     std::thread m_acceptor;
 };
 

src/plugins/input/tcp/src/Session.hpp

@@ -22,7 +22,11 @@ namespace tcp_in
  */
 class Session {
 public:
-    // TODO: constructor
+    /**
+     * @brief Creates new session associated with the tcp connection.
+     * @param fd file descriptor of the new tcp connection.
+     */
+    Session(int fd);
 
     /**
      * @brief Reads at most `len` bytes from the TCP socket. Doesn't block, if there is not enough data, less is readed.
@@ -39,7 +43,12 @@ class Session {
      */
     void pass_msg(ipx_ctx_t *ctx, ByteVector &msg);
 
-    void close(ipx_ctx_t *ctx /* TODO: epoll parameters */);
+    /**
+     * @brief Closes the connection and session.
+     * @param ctx
+     * @param epoll_fd file descriptor of the epoll in which this connection is.
+     */
+    void close(ipx_ctx_t *ctx, int epoll_fd);
 private:
     struct ipx_session *m_session;
     int m_fd;