Forwarder: handle periodic message

Adds support for newly introduced "periodic message".

Without this change, unfinished data transfers might never finish if no
input data is coming.

Author: sedmicha

src/plugins/output/forwarder/src/Forwarder.cpp

@@ -105,6 +105,14 @@ void Forwarder::handle_ipfix_message(ipx_msg_ipfix_t *msg)
     }
 }
 
+void
+Forwarder::handle_periodic_message(ipx_msg_periodic_t *msg)
+{
+    for (auto &host : m_hosts) {
+        host->advance_transfers();
+    }
+}
+
 void
 Forwarder::forward_to_all(ipx_msg_ipfix_t *msg)
 {

src/plugins/output/forwarder/src/Forwarder.h

@@ -80,6 +80,13 @@ class Forwarder {
     void
     handle_ipfix_message(ipx_msg_ipfix_t *msg);
 
+    /**
+     * \brief Handle a periodic message
+     * \param msg  The periodic message
+     */
+    void
+    handle_periodic_message(ipx_msg_periodic_t *msg);
+
     /**
      * \brief The destructor - finalize the forwarder
      */
@@ -101,4 +108,5 @@ class Forwarder {
 
     void
     forward_round_robin(ipx_msg_ipfix_t *msg);
-};
+
+};

src/plugins/output/forwarder/src/Host.cpp

@@ -142,6 +142,17 @@ Host::forward_message(ipx_msg_ipfix_t *msg)
     return true;
 }
 
+void
+Host::advance_transfers()
+{
+    for (auto &p : m_session_to_connection) {
+        Connection &connection = *p.second.get();
+        if (connection.check_connected()) {
+            connection.advance_transfers();
+        }
+    }
+}
+
 Host::~Host()
 {
     for (auto &p : m_session_to_connection) {

src/plugins/output/forwarder/src/Host.h

@@ -100,6 +100,12 @@ class Host {
     bool
     forward_message(ipx_msg_ipfix_t *msg);
 
+    /**
+     * \brief Advance the unfinished transfers
+     */
+    void
+    advance_transfers();
+
 private:
     const std::string &m_ident;
 
@@ -116,4 +122,4 @@ class Host {
     Connector &m_connector;
 
     std::unordered_map<const ipx_session *, std::unique_ptr<Connection>> m_session_to_connection;
-};
+};

src/plugins/output/forwarder/src/main.cpp

@@ -88,7 +88,7 @@ ipx_plugin_init(ipx_ctx_t *ctx, const char *xml_config)
         return IPX_ERR_DENIED;
     }
 
-    ipx_msg_mask_t mask = IPX_MSG_IPFIX | IPX_MSG_SESSION;
+    ipx_msg_mask_t mask = IPX_MSG_IPFIX | IPX_MSG_SESSION | IPX_MSG_PERIODIC;
     ipx_ctx_subscribe(ctx, &mask, NULL);
     ipx_ctx_private_set(ctx, forwarder.release());
 
@@ -121,6 +121,10 @@ ipx_plugin_process(ipx_ctx_t *ctx, void *priv, ipx_msg_t *msg)
             forwarder.handle_session_message(ipx_msg_base2session(msg));
             break;
 
+        case IPX_MSG_PERIODIC:
+            forwarder.handle_periodic_message(ipx_msg_base2periodic(msg));
+            break;
+
         default: assert(0);
         }