Updated to latest fluvio-client-cpp version

Author: KunoVonHagen

.gitignore

@@ -0,0 +1 @@
+.idea

CMakeLists.txt

@@ -5,7 +5,7 @@ include(ZeekPlugin)
 include(FetchContent)
 FetchContent_Declare(
   fluvio_client_cpp
-  URL https://github.com/stefanDeveloper/fluvio-client-cpp/releases/download/v0.0.2/fluvio-client-cpp-linux-x64.tar.gz
+  URL https://github.com/stefanDeveloper/fluvio-client-cpp/releases/download/v0.1.0/fluvio-client-cpp-linux-x64.tar.gz
 )
 FetchContent_MakeAvailable(fluvio_client_cpp)
 find_package(fluvio_client_cpp CONFIG REQUIRED PATHS ${fluvio_client_cpp_SOURCE_DIR})

README.md

@@ -1,6 +1,6 @@
 # Zeek-Fluvio Plugin
 
-A high-performance Zeek log writer plugin that bridges telemetry payloads seamlessly into **Fluvio** streams. Built natively on the [Fluvio C++ Client](https://github.com/infinyon/fluvio-client-cpp), this plugin ensures low-latency data ingestion for security operations and network analysis.
+A high-performance Zeek log writer plugin that bridges telemetry payloads seamlessly into **Fluvio** streams. Built natively on the [Fluvio C++ Client](https://github.com/stefanDeveloper/fluvio-client-cpp), this plugin ensures low-latency data ingestion for security operations and network analysis.
 
 ## 🚀 Features
 
@@ -24,9 +24,9 @@ zkg install zeek-fluvio
 If you prefer to build from source:
 
 ```bash
-git clone https://github.com/infinyon/zeek-fluvio
+git clone https://github.com/stefanDeveloper/zeek-fluvio.git
 cd zeek-fluvio
-./configure --zeek-dist=/path/to/zeek/source
+./configure
 make -C build
 sudo make -C build install
 ```

src/FluvioWriter.cc

@@ -4,30 +4,29 @@
 namespace logging {
 namespace writer {
 
-FluvioWriter::FluvioWriter(zeek::logging::WriterFrontend* frontend) : zeek::logging::WriterBackend(frontend), client(nullptr), producer(nullptr) {
+FluvioWriter::FluvioWriter(zeek::logging::WriterFrontend* frontend) : zeek::logging::WriterBackend(frontend) {
     formatter = new zeek::threading::formatter::JSON(this, zeek::threading::formatter::JSON::TS_EPOCH);
 }
 
 FluvioWriter::~FluvioWriter() {
     delete formatter;
-    if (producer) {
-        fluvio_c_producer_free(producer);
-    }
-    if (client) {
-        fluvio_c_client_free(client);
-    }
 }
 
 bool FluvioWriter::DoInit(const zeek::logging::WriterBackend::WriterInfo& info, int num_fields, const zeek::threading::Field* const* fields) {
     topic_name = info.path;
 
-    if (fluvio_c_connect(&client) != 0) {
-        Error("Failed to connect to Fluvio Cluster");
+    try {
+        this->client = Fluvio::connect();
+    } catch (const std::exception& e) {
+        std::string err = std::string("Failed to connect to Fluvio: ") + e.what();
+        Error(err.c_str());
         return false;
     }
 
-    if (fluvio_c_create_producer(client, topic_name.c_str(), &producer) != 0) {
-        std::string err = "Failed to instantiate producer for topic: " + topic_name;
+    try {
+        this->producer = (*this->client)->topic_producer(topic_name);
+    } catch (const std::exception& e) {
+        std::string err = std::string("Failed to create producer for topic '") + topic_name + "': " + e.what();
         Error(err.c_str());
         return false;
     }
@@ -36,16 +35,19 @@ bool FluvioWriter::DoInit(const zeek::logging::WriterBackend::WriterInfo& info,
 }
 
 bool FluvioWriter::DoWrite(int num_fields, const zeek::threading::Field* const* fields, zeek::threading::Value** vals) {
-    if (!producer) return false;
+    if (!this->producer) return false;
 
     zeek::ODesc desc;
-    formatter->Describe(&desc, num_fields, fields, vals);
+    this->formatter->Describe(&desc, num_fields, fields, vals);
 
     std::string payload = desc.Description();
-    if (fluvio_c_producer_send(producer, nullptr, 0, (const uint8_t*)payload.data(), payload.size(), nullptr) != 0) {
-        Warning("Dropped record due to internal queue/send limits");
-        return false;
-    }
+    auto out = (*this->producer)->send(
+        rust::Slice<const uint8_t>(nullptr, 0),
+        rust::Slice<const uint8_t>(
+            reinterpret_cast<const uint8_t*>(payload.data()),
+            payload.size()
+        )
+    );
 
     return true;
 }
@@ -54,12 +56,13 @@ bool FluvioWriter::DoSetBuf(bool enabled) { return true; }
 bool FluvioWriter::DoRotate(const char* rotated_path, double open, double close, bool terminating) { return true; }
 
 bool FluvioWriter::DoFlush(double network_time) {
-    if (producer) fluvio_c_producer_flush(producer);
+    if (!this->producer) return false;
+    (*this->producer)->flush();
     return true;
 }
 
 bool FluvioWriter::DoFinish(double network_time) {
-    if (producer) fluvio_c_producer_flush(producer);
+    if (this->producer) (*this->producer)->flush();
     return true;
 }
 

src/FluvioWriter.h

@@ -1,4 +1,6 @@
 #pragma once
+#include "rust/cxx.h"
+#include "fluvio-client-cpp/src/lib.rs.h"
 #include <zeek/logging/WriterBackend.h>
 #include <zeek/threading/formatters/JSON.h>
 #include <fluvio.h>
@@ -27,8 +29,8 @@ class FluvioWriter : public zeek::logging::WriterBackend {
 
 private:
     zeek::threading::formatter::JSON* formatter;
-    fluvio_client_t* client;
-    fluvio_topic_producer_t* producer;
+    std::optional<rust::Box<Fluvio>> client;
+    std::optional<rust::Box<TopicProducerPool>> producer;
     std::string topic_name;
 };