--- /dev/null
+/*#############################################################################
+# #
+# telemetryd - The IPFire Telemetry Collection Service #
+# Copyright (C) 2026 IPFire Development Team #
+# #
+# This program is free software: you can redistribute it and/or modify #
+# it under the terms of the GNU General Public License as published by #
+# the Free Software Foundation, either version 3 of the License, or #
+# (at your option) any later version. #
+# #
+# This program is distributed in the hope that it will be useful, #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
+# GNU General Public License for more details. #
+# #
+# You should have received a copy of the GNU General Public License #
+# along with this program. If not, see <http://www.gnu.org/licenses/>. #
+# #
+#############################################################################*/
+
+#include <stdlib.h>
+
+#include <curl/curl.h>
+
+#include <systemd/sd-event.h>
+
+#include "ctx.h"
+#include "client.h"
+
+struct td_client {
+ td_ctx* ctx;
+ int nrefs;
+
+ // Event Loop
+ sd_event* loop;
+
+ // cURL Multi Handle
+ CURLM* curl;
+
+ // Timer
+ sd_event_source* timer;
+
+ // How many are still running?
+ int still_running;
+};
+
+static int td_client_action(td_client* self) {
+ void* p = NULL;
+
+ CURLMsg* msg = NULL;
+ int msgs_left = 0;
+
+ for (;;) {
+ // Read the next message
+ msg = curl_multi_info_read(self->curl, &msgs_left);
+ if (!msg)
+ break;
+
+ switch (msg->msg) {
+ case CURLMSG_DONE:
+ // Update reference to transfer
+ curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &p);
+
+#if 0
+ // Call the done callback
+ r = td_request_done(p, self->loop, msg->data.result);
+ if (r < 0) {
+ switch (-r) {
+ case EAGAIN:
+ // Launch it again
+ // XXX TODO
+
+ // Move to the next message
+ continue;
+
+ // Handle signal that the handle should not be dequeued, yet
+ case EINPROGRESS:
+ continue;
+
+ // Raise any errors
+ default:
+ return r;
+ }
+ }
+#endif
+ break;
+
+ default:
+ ERROR(self->ctx, "Received unhandled cURL message %u\n", msg->msg);
+ break;
+ }
+ }
+
+ return 0;
+}
+
+static int td_client_timer_callback(CURLM* multi, long timeout_ms, void* data) {
+ td_client* self = data;
+ int r;
+
+ // A negative value indicates that we should remove the timer
+ if (timeout_ms < 0) {
+ if (self->timer) {
+ r = sd_event_source_set_enabled(self->timer, SD_EVENT_OFF);
+ if (r < 0)
+ ERROR(self->ctx, "Could not disarm the timer: %s\n", strerror(-r));
+ }
+
+ return 0;
+ }
+
+ // Set the timer
+ r = sd_event_source_set_time_relative(self->timer, timeout_ms * 1000);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not set timer: %s\n", strerror(-r));
+
+ return r;
+ }
+
+ // Have the timer fire once
+ r = sd_event_source_set_enabled(self->timer, SD_EVENT_ONESHOT);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not enable the timer: %s\n", strerror(-r));
+
+ return r;
+ }
+
+ DEBUG(self->ctx, "cURL set a timer for %ldms\n", timeout_ms);
+
+ return 0;
+}
+
+static int td_client_timer_fired(sd_event_source* s, uint64_t usec, void* data) {
+ td_client* self = data;
+ int r;
+
+ DEBUG(self->ctx, "cURL timer fired\n");
+
+ r = curl_multi_socket_action(self->curl, CURL_SOCKET_TIMEOUT, 0, &self->still_running);
+ if (r)
+ return r;
+
+ // Check for any messages
+ return td_client_action(self);
+}
+
+static int td_client_setup_timer(td_client* self) {
+ int r;
+
+ // Create a new timer
+ r = sd_event_add_time_relative(self->loop, &self->timer, CLOCK_MONOTONIC, 0, 0,
+ td_client_timer_fired, self);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not set timer: %s\n", strerror(-r));
+
+ return r;
+ }
+
+ return 0;
+}
+
+static int td_client_socket(sd_event_source* s, int fd, uint32_t events, void* data) {
+ td_client* self = data;
+ int r;
+
+ int action = 0;
+
+ if (events & EPOLLIN)
+ action |= CURL_CSELECT_IN;
+
+ if (events & EPOLLOUT)
+ action |= CURL_CSELECT_OUT;
+
+ //DEBUG(self->ctx, "cURL has activity on socket %d\n", fd);
+
+ // Inform cURL about some socket activity
+ r = curl_multi_socket_action(self->curl, fd, action, &self->still_running);
+ if (r)
+ return r;
+
+ // Check for any messages
+ r = td_client_action(self);
+ if (r)
+ return r;
+
+ // Disarm the timer
+ if (self->still_running <= 0) {
+ if (self->timer) {
+ r = sd_event_source_set_enabled(self->timer, SD_EVENT_OFF);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not disarm the timer: %s\n", strerror(-r));
+
+ return r;
+ }
+ }
+ }
+
+ return 0;
+}
+
+static int td_client_socket_callback(CURL* e, curl_socket_t fd, int what, void* data, void* p) {
+ td_client* self = data;
+ sd_event_source* s = p;
+ uint32_t events = 0;
+ int r;
+
+ // Remove the socket?
+ if (what == CURL_POLL_REMOVE) {
+ // Disable the event
+ r = sd_event_source_set_enabled(s, SD_EVENT_OFF);
+ if (r < 0)
+ ERROR(self->ctx, "Could not disable fd %d: %s\n", fd, strerror(-r));
+
+ DEBUG(self->ctx, "cURL deregistered socket %d\n", fd);
+
+ return r;
+ }
+
+ // Do we need to read from this socket?
+ if (what & CURL_POLL_IN)
+ events |= EPOLLIN;
+
+ // Do we need to write to this socket?
+ if (what & CURL_POLL_OUT)
+ events |= EPOLLOUT;
+
+ // Change events?
+ if (s) {
+ r = sd_event_source_set_io_events(s, events);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not change events for socket %d: %s\n",
+ fd, strerror(-r));
+
+ return r;
+ }
+
+ DEBUG(self->ctx, "cURL changed socket %d\n", fd);
+
+ return 0;
+ }
+
+ // Add the socket to the event loop
+ r = sd_event_add_io(self->loop, &s, fd, events, td_client_socket, self);
+ if (r < 0) {
+ ERROR(self->ctx, "Could not register socket %d: %s\n", fd, strerror(-r));
+
+ goto ERROR;
+ }
+
+ // Store the event source
+ curl_multi_assign(self->curl, fd, sd_event_source_ref(s));
+
+ DEBUG(self->ctx, "cURL registered socket %d\n", fd);
+
+ERROR:
+ if (s)
+ sd_event_source_unref(s);
+
+ return r;
+}
+
+static int td_client_setup_curl(td_client* self) {
+ int r;
+
+ // Initialize cURL
+ r = curl_global_init(CURL_GLOBAL_DEFAULT);
+ if (r) {
+ ERROR(self->ctx, "Could not initialize cURL: %d\n", r);
+ return r;
+ }
+
+ // Create a new multi handle
+ self->curl = curl_multi_init();
+ if (!self->curl) {
+ ERROR(self->ctx, "Could not create cURL multi handle\n");
+ return 1;
+ }
+
+ // Register with the event loop
+ r = curl_multi_setopt(self->curl, CURLMOPT_TIMERFUNCTION, td_client_timer_callback);
+ if (r) {
+ ERROR(self->ctx, "Could not register the timer function: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ r = curl_multi_setopt(self->curl, CURLMOPT_TIMERDATA, self);
+ if (r) {
+ ERROR(self->ctx, "Could not register the timer data: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ r = curl_multi_setopt(self->curl, CURLMOPT_SOCKETFUNCTION, td_client_socket_callback);
+ if (r) {
+ ERROR(self->ctx, "Could not register the socket function: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ r = curl_multi_setopt(self->curl, CURLMOPT_SOCKETDATA, self);
+ if (r) {
+ ERROR(self->ctx, "Could not register the socket data: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ // Set the number of open idle connections
+ r = curl_multi_setopt(self->curl, CURLMOPT_MAXCONNECTS, 16L);
+ if (r) {
+ ERROR(self->ctx, "Could not set max parallel transfers: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ // Do not open more than eight connections to the same host
+ r = curl_multi_setopt(self->curl, CURLMOPT_MAX_HOST_CONNECTIONS, 8L);
+ if (r) {
+ ERROR(self->ctx, "Could not set max host connections: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ // Limit total number of open connections
+ r = curl_multi_setopt(self->curl, CURLMOPT_MAX_TOTAL_CONNECTIONS, 64L);
+ if (r) {
+ ERROR(self->ctx, "Could not set max total connections: %s\n",
+ curl_multi_strerror(r));
+
+ return r;
+ }
+
+ return 0;
+}
+
+static void td_client_free(td_client* self) {
+ if (self->timer)
+ sd_event_source_unref(self->timer);
+ if (self->curl)
+ curl_multi_cleanup(self->curl);
+ if (self->loop)
+ sd_event_unref(self->loop);
+ if (self->ctx)
+ td_ctx_unref(self->ctx);
+ free(self);
+}
+
+int td_client_create(td_client** client, td_ctx* ctx, td_daemon* daemon) {
+ td_client* self = NULL;
+ int r;
+
+ // Allocate some memory
+ self = calloc(1, sizeof(*self));
+ if (!self)
+ return -errno;
+
+ // Store a reference to the context
+ self->ctx = td_ctx_ref(ctx);
+
+ // Initialize the reference counter
+ self->nrefs = 1;
+
+ // Fetch the event loop
+ self->loop = td_daemon_loop(daemon);
+
+ // Setup timer
+ r = td_client_setup_timer(self);
+ if (r < 0)
+ goto ERROR;
+
+ // Setup cURL
+ r = td_client_setup_curl(self);
+ if (r < 0)
+ goto ERROR;
+
+ // Return the pointer
+ *client = self;
+
+ return 0;
+
+ERROR:
+ td_client_free(self);
+ return r;
+}
+
+td_client* td_client_ref(td_client* self) {
+ ++self->nrefs;
+ return self;
+}
+
+td_client* td_client_unref(td_client* self) {
+ if (--self->nrefs > 0)
+ return self;
+
+ td_client_free(self);
+ return NULL;
+}
projects / telemetry.git / commitdiff
