From: Michael Tremer <michael.tremer@ipfire.org>
Date: Sat, 27 Sep 2025 15:41:14 +0000 (+0000)
Subject: daemon: Call a function to regularly flush the queue
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f592d470045ee85869460cce371af3b9c9fd45d4;p=telemetry.git

daemon: Call a function to regularly flush the queue

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>
---

diff --git a/src/daemon/daemon.c b/src/daemon/daemon.c
index 741c840..95be5ed 100644
--- a/src/daemon/daemon.c
+++ b/src/daemon/daemon.c
@@ -32,6 +32,8 @@
 #include "module.h"
 #include "modules.h"
 
+#define FLUSH_QUEUE_INTERVAL 60000000 // 60s (for testing)
+
 struct collecty_queue_object {
 	STAILQ_ENTRY(collecty_queue_object) nodes;
 
@@ -60,6 +62,7 @@ struct collecty_daemon {
 		sd_event_source* sigterm;
 		sd_event_source* sigint;
 		sd_event_source* modules_init;
+		sd_event_source* flush;
 	} events;
 
 	// Write Queue
@@ -134,10 +137,41 @@ static void collecty_daemon_free_queue_object(struct collecty_queue_object* o) {
 	free(o);
 }
 
+static int collecty_daemon_flush(sd_event_source* source, uint64_t usec, void* data) {
+	collecty_daemon* self = data;
+	int r;
+
+	DEBUG(self->ctx, "Flushing the queue...\n");
+
+	// TODO
+
+	// Call again soon...
+	r = sd_event_source_set_time(source, usec + FLUSH_QUEUE_INTERVAL);
+	if (r < 0)
+		return r;
+
+	// Enable the timer
+	r = sd_event_source_set_enabled(source, SD_EVENT_ONESHOT);
+	if (r < 0)
+		return r;
+
+	return 0;
+}
+
 static int collecty_daemon_setup_queue(collecty_daemon* self) {
+	int r;
+
 	// Initialize the queue
 	STAILQ_INIT(&self->queue);
 
+	// Create a timer which regularly flushes the queue
+	r = sd_event_add_time_relative(self->loop, &self->events.flush,
+		CLOCK_MONOTONIC, FLUSH_QUEUE_INTERVAL, 0, collecty_daemon_flush, self);
+	if (r < 0) {
+		ERROR(self->ctx, "Failed to set up the flush timer: %s\n", strerror(-r));
+		return r;
+	}
+
 	return 0;
 }
 
@@ -157,6 +191,8 @@ static void collecty_daemon_free(collecty_daemon* self) {
 		collecty_daemon_free_queue_object(o);
 	}
 
+	if (self->events.flush)
+		sd_event_source_unref(self->events.flush);
 	if (self->events.modules_init)
 		sd_event_source_unref(self->events.modules_init);
 	if (self->events.sigterm)