++

Author: Damir Zainullin

include/ipfixprobe/processPlugin/flowKey.hpp

@@ -3,6 +3,7 @@
 #include <cstdint>
 
 #include "ipAddress.hpp"
+#include "xxhash.h"
 
 namespace ipxp {
 
@@ -29,7 +30,7 @@ struct FlowKey {
     constexpr inline
     std::size_t hash() const noexcept 
     {
-        return 0; //XXH3_64bits(this, sizeof(*this), 0);
+        return XXH3_64bits(this, sizeof(*this));
     }
 };
 

include/ipfixprobe/processPlugin/processPlugin.hpp

@@ -25,7 +25,7 @@ namespace ipxp {
  */
 class FlowRecord;
 class Packet;
-class PacketFeatures;
+class PacketFeatures{};
 
 /**
  * @brief Context passed to plugin methods, containing references to flow and packet.

include/ipfixprobe/processPlugin/processPluginManager.hpp

@@ -5,6 +5,7 @@
 #include "processPlugin.hpp"
 #include "processPluginEntry.hpp"
 #include "flowRecordBuilder.hpp"
+#include "flowKey.hpp"
 #include "../pluginFactory/pluginFactory.hpp"
 #include "../api.hpp"
 
@@ -192,7 +193,7 @@ class IPXP_API ProcessPluginManager {
 
 	std::shared_ptr<FlowRecordBuilder> rebuild()
 	{
-		auto builder = std::make_shared<FlowRecordBuilder>(m_processPlugins, FlowKey::DEFAULT_LAYOUT);
+		auto builder = std::make_shared<FlowRecordBuilder>(m_processPlugins, FlowKey::getLayout());
 		return builder;
 	}
 

src/plugins/storage/cache/src/cache.cpp

@@ -171,8 +171,8 @@ void NHTFlowCache::export_flow(size_t index)
 		m_flow_table[index]->directionalData[Direction::Forward].packets + m_flow_table[index]->directionalData[Direction::Reverse].packets);
 	m_flows_in_cache--;
 
-	ipx_ring_push(m_export_queue, m_flow_table[index].get());
 	std::swap(m_flow_table[index], m_flow_table[m_cache_size + m_qidx]);
+	ipx_ring_push(m_export_queue, &m_flow_table[m_cache_size + m_qidx]);
 	m_flow_table[index]->erase();
 	m_qidx = (m_qidx + 1) % m_qsize;
 }
@@ -338,9 +338,16 @@ int NHTFlowCache::put_pkt(Packet& pkt)
 		return 0;
 	}
 
+	auto features = PacketFeatures{};
+	auto context = FlowContext {
+			.flowRecord = *flow,
+			.packet = pkt,
+			.features = features
+		};
 	if (flow->isEmpty()) {
 		m_flows_in_cache++;
 		flow->createFrom(pkt, hashval);
+		m_manager.processFlowRecord(context);
 		ret = 0; //plugins_post_create(flow->m_flow, pkt);
 
 		//if (ret & FLOW_FLUSH) {