Standarize the project directory for monorepo-like developer experience

Move the clang formatter to the root of the three so all nested projects could use it

Provide README for all other projects

Refactor the code in rtt agregator

Signed-off-by: Filip Wandzio <contact@philw.dev>

6 files changed, 286 insertions, 110 deletions

diff --git a/analysis/rtt/README.txt b/analysis/rtt/README.txt
new file mode 100644
index 0000000..4a45e6d
--- /dev/null
+++ b/analysis/rtt/README.txt
@@ -0,0 +1,40 @@
+Project: MQTT RTT and Throughput Logger
+
+Description:
+This project implements an MQTT client that connects to a broker, subscribes to a topic, and logs message round-trip time (RTT) and throughput statistics to a CSV file and standard output. The logger measures the time difference between message sent timestamps and received timestamps to calculate RTT and maintains a sliding window of message times to estimate throughput in messages per second.
+
+Files:
+
+mqtt_client.h / mqtt_client.c: MQTT client wrapper using the Mosquitto library.
+
+logger.h / logger.c: Logger module for handling incoming MQTT messages and logging RTT/throughput.
+
+main.c: Entry point initializing the MQTT client and logger, then running the MQTT client loop.
+
+Dependencies:
+
+Mosquitto library (libmosquitto)
+
+Standard C libraries (stdio, stdlib, string, sys/time, time)
+
+Build Instructions:
+
+Look at the Dockerfile and Makefile to find out.
+Usage:
+
+Configure MQTT broker address, port, and topic in main.c.
+
+Run the compiled executable.
+
+The program connects to the broker, subscribes to the specified topic, and logs RTT and throughput stats to a CSV file.
+
+Output is also printed to the console.
+
+Notes:
+
+The logger expects the payload of received messages to contain a timestamp string representing the sent time in milliseconds.
+
+The program uses blocking call mosquitto_loop_forever to run the MQTT network loop.
+
+Contact:
+For questions contact Filip Wandzio at contact@philw.dev 
diff --git a/analysis/rtt/src/logger.c b/analysis/rtt/src/logger.c
index ed5f6e5..e1dd2f3 100644
--- a/analysis/rtt/src/logger.c
+++ b/analysis/rtt/src/logger.c
@@ -1,4 +1,3 @@
-
 #include "logger.h"
 #include <stdio.h>
 #include <stdlib.h>
@@ -9,78 +8,120 @@
 #define WINDOW_SEC 1
 #define MAX_TIMES 1000
 
+#define ISO_TIME_BUF_SIZE 32
+
+#define MICROSECONDS_IN_SECOND 1000000
+#define MICROSECONDS_IN_MILLISECOND 1000
+
+#define CSV_HEADER "timestamp,sent_ms,received_ms,rtt_ms,throughput_msg_per_s\n"
+#define CSV_LINE_FORMAT "%s,%lld,%lld,%lld,%d\n"
+
 static FILE *csv_file = NULL;
 static double msg_times[MAX_TIMES];
 static int msg_count = 0;
 
-static long long current_time_ms() {
-  struct timeval tv;
-  gettimeofday(&tv, NULL);
-  return ((long long)tv.tv_sec * 1000) + (tv.tv_usec / 1000);
+/**
+ * @brief Returns the current time in milliseconds since the Epoch.
+ *
+ * @return Current time in milliseconds (int64).
+ */
+static long long current_time_ms(void);
+
+/**
+ * @brief Returns the current time in seconds (with microsecond precision) since
+ * the Epoch.
+ *
+ * @return Current time in seconds (double).
+ */
+static double current_time_sec(void);
+
+/**
+ * @brief Updates the sliding window of message timestamps and returns the count
+ * of messages within WINDOW_SEC.
+ *
+ * @param now Current timestamp in seconds.
+ * @return Number of messages in the last WINDOW_SEC seconds.
+ */
+static int update_throughput_window(double now);
+
+static long long current_time_ms(void)
+{
+        struct timeval tv;
+        gettimeofday(&tv, NULL);
+        return ((long long)tv.tv_sec * 1000) +
+               (tv.tv_usec / MICROSECONDS_IN_MILLISECOND);
 }
 
-static double current_time_sec() {
-  struct timeval tv;
-  gettimeofday(&tv, NULL);
-  return (double)tv.tv_sec + (tv.tv_usec / 1e6);
+static double current_time_sec(void)
+{
+        struct timeval tv;
+        gettimeofday(&tv, NULL);
+        return (double)tv.tv_sec +
+               (tv.tv_usec / (double)MICROSECONDS_IN_SECOND);
 }
 
-static int update_throughput_window(double now) {
-  int i, new_count = 0;
-  for (i = 0; i < msg_count; i++) {
-    if (now - msg_times[i] < WINDOW_SEC) {
-      msg_times[new_count++] = msg_times[i];
-    }
-  }
-  msg_times[new_count++] = now;
-  msg_count = new_count;
-  return msg_count;
+static int update_throughput_window(double current_time)
+{
+        int msg_index, window_count = 0;
+        for (msg_index = 0; msg_index < msg_count; msg_index++) {
+                if (current_time - msg_times[msg_index] < WINDOW_SEC)
+                        msg_times[window_count++] = msg_times[msg_index];
+        }
+        msg_times[window_count++] = current_time;
+        msg_count = window_count;
+        return msg_count;
 }
 
-void logger_init(const char *filename) {
-  csv_file = fopen(filename, "w");
-  if (!csv_file) {
-    perror("Cannot open CSV file");
-    exit(EXIT_FAILURE);
-  }
-  fprintf(csv_file,
-          "timestamp,sent_ms,received_ms,rtt_ms,throughput_msg_per_s\n");
-  fflush(csv_file);
+void logger_init(const char *filename)
+{
+        csv_file = fopen(filename, "w");
+        if (!csv_file) {
+                perror("Cannot open CSV file");
+                exit(EXIT_FAILURE);
+        }
+        fprintf(csv_file, CSV_HEADER);
+        fflush(csv_file);
 }
 
-void logger_cleanup() {
-  if (csv_file) {
-    fclose(csv_file);
-    csv_file = NULL;
-  }
+void logger_cleanup(void)
+{
+        if (csv_file) {
+                fclose(csv_file);
+                csv_file = NULL;
+        }
 }
 
-void logger_handle_message(const void *payload, int payloadlen) {
-  char *payload_str = malloc(payloadlen + 1);
-  if (!payload_str)
-    return;
-  memcpy(payload_str, payload, payloadlen);
-  payload_str[payloadlen] = '\0';
+void logger_handle_message(const void *payload, int payloadlen)
+{
+        if (payloadlen <= 0)
+                return;
+
+        char *payload_str = malloc(payloadlen + 1);
+        if (!payload_str)
+                return;
+
+        memcpy(payload_str, payload, payloadlen);
+        payload_str[payloadlen] = '\0';
 
-  long long sent_ms = atoll(payload_str);
-  free(payload_str);
+        long long sent_ms = atoll(payload_str);
+        free(payload_str);
 
-  long long received_ms = current_time_ms();
-  long long rtt = received_ms - sent_ms;
+        long long received_ms = current_time_ms();
+        long long rtt = received_ms - sent_ms;
 
-  double now_sec = current_time_sec();
-  int throughput = update_throughput_window(now_sec);
+        double now_sec = current_time_sec();
+        int throughput = update_throughput_window(now_sec);
 
-  time_t now = time(NULL);
-  struct tm *tm_info = localtime(&now);
-  char iso_time[32];
-  strftime(iso_time, sizeof(iso_time), "%Y-%m-%dT%H:%M:%S", tm_info);
+        time_t now = time(NULL);
+        struct tm *tm_info = localtime(&now);
+        char iso_time[ISO_TIME_BUF_SIZE];
+        strftime(iso_time, sizeof(iso_time), "%Y-%m-%dT%H:%M:%S", tm_info);
 
-  if (csv_file) {
-    fprintf(csv_file, "%s,%lld,%lld,%lld,%d\n", iso_time, sent_ms, received_ms,
-            rtt, throughput);
-    fflush(csv_file);
-  }
+        if (csv_file) {
+                fprintf(csv_file, CSV_LINE_FORMAT, iso_time, sent_ms,
+                        received_ms, rtt, throughput);
+                fflush(csv_file);
+        }
 
-  printf("RTT: %lld ms | Throughput: %d msg/s\n", rtt, throughput);
+        printf("RTT: %lld ms | Throughput: %d msg/s\n", rtt, throughput);
 }
diff --git a/analysis/rtt/src/logger.h b/analysis/rtt/src/logger.h
index f1602c9..904c327 100644
--- a/analysis/rtt/src/logger.h
+++ b/analysis/rtt/src/logger.h
@@ -1,9 +1,25 @@
-
 #ifndef LOGGER_H
 #define LOGGER_H
 
+/**
+ * @brief Initializes the logger by opening the CSV file and writing the header.
+ *
+ * @param filename Path to the CSV log file.
+ */
 void logger_init(const char *filename);
-void logger_cleanup();
+
+/**
+ * @brief Closes the CSV file and cleans up resources.
+ */
+void logger_cleanup(void);
+
+/**
+ * @brief Processes an incoming message payload, parses the timestamp,
+ * calculates RTT and throughput, and logs this data to the CSV and stdout.
+ *
+ * @param payload Pointer to the message payload.
+ * @param payloadlen Length of the payload in bytes.
+ */
 void logger_handle_message(const void *payload, int payloadlen);
 
-#endif // LOGGER_H
+#endif
diff --git a/analysis/rtt/src/main.c b/analysis/rtt/src/main.c
index 13f7716..163b5f6 100644
--- a/analysis/rtt/src/main.c
+++ b/analysis/rtt/src/main.c
@@ -1,4 +1,3 @@
-
 #include "logger.h"
 #include "mqtt_client.h"
 #include <stdio.h>
@@ -8,21 +7,26 @@
 #define TOPIC "device/echo/in"
 #define CSV_FILE "rtt_throughput_log.csv"
 
-int main() {
-  mqtt_client_t client;
-
-  logger_init(CSV_FILE);
-
-  if (mqtt_client_init(&client, BROKER_ADDRESS, BROKER_PORT, TOPIC) != 0) {
-    fprintf(stderr, "MQTT initialization error\n");
-    return 1;
-  }
-
-  printf("Connection established.Waiting for messages...\n");
-  mqtt_client_loop(&client);
-
-  logger_cleanup();
-  mqtt_client_cleanup(&client);
-
-  return 0;
+/**
+ * @brief Entry point for MQTT client program.
+ *
+ * Initializes the logger and MQTT client, then enters the MQTT loop to receive
+ * messages. Cleans up resources on exit.
+ *
+ * @return int Exit code (0 on success, 1 on failure).
+ */
+int main(void)
+{
+        mqtt_client_t client;
+        logger_init(CSV_FILE);
+
+        if (mqtt_client_init(&client, BROKER_ADDRESS, BROKER_PORT, TOPIC))
+                return fprintf(stderr, "MQTT initialization error\n"), 1;
+
+        printf("Connection established. Waiting for messages...\n");
+        mqtt_client_loop(&client);
+        logger_cleanup();
+        mqtt_client_cleanup(&client);
+
+        return 0;
 }
diff --git a/analysis/rtt/src/mqtt_client.c b/analysis/rtt/src/mqtt_client.c
index 43a06d8..edc394d 100644
--- a/analysis/rtt/src/mqtt_client.c
+++ b/analysis/rtt/src/mqtt_client.c
@@ -3,46 +3,98 @@
 #include <stdio.h>
 #include <stdlib.h>
 
-#define WINDOW_SEC 1
+#define MQTT_KEEPALIVE_SECONDS 60
+#define MQTT_QOS_LEVEL 0
+#define LOOP_TIMEOUT_MS -1
+#define LOOP_MAX_MESSAGES 1
 
+/**
+ * @brief Callback called when a message arrives on a subscribed topic.
+ *
+ * @param mosq      Mosquitto client instance.
+ * @param userdata  User data pointer (unused).
+ * @param msg       Incoming message data.
+ */
 static void on_message(struct mosquitto *mosq, void *userdata,
-                       const struct mosquitto_message *msg);
+                       const struct mosquitto_message *message);
+
+/**
+ * @brief Helper function to handle errors by printing a message,
+ *        cleaning up the MQTT client, and returning failure.
+ *
+ * @param client           MQTT client instance.
+ * @param error_message    Error message to print.
+ * @param mosq_error_code  Mosquitto error code.
+ * @return int             Always returns 1 to indicate failure.
+ */
+static int fail_with_cleanup(mqtt_client_t *client, const char *error_message,
+                             int mosq_error_code);
 
 int mqtt_client_init(mqtt_client_t *client, const char *broker_address,
-                     int port, const char *topic) {
-  mosquitto_lib_init();
-  client->mosq = mosquitto_new(NULL, true, NULL);
-  if (!client->mosq) {
-    fprintf(stderr, "Cannot create MQTT client\n");
-    return 1;
-  }
-
-  mosquitto_message_callback_set(client->mosq, on_message);
-
-  if (mosquitto_connect(client->mosq, broker_address, port, 60) !=
-      MOSQ_ERR_SUCCESS) {
-    fprintf(stderr, "Nie można połączyć się z brokerem MQTT\n");
-    return 1;
-  }
-
-  if (mosquitto_subscribe(client->mosq, NULL, topic, 0) != MOSQ_ERR_SUCCESS) {
-    fprintf(stderr, "Cannot subscribe to topic\n");
-    return 1;
-  }
-
-  return 0;
+                     int broker_port, const char *topic)
+{
+        if (!client || !broker_address || !topic)
+                return fprintf(stderr, "Invalid MQTT init parameters\n"), 1;
+
+        int init_result = mosquitto_lib_init();
+        if (init_result != MOSQ_ERR_SUCCESS) {
+                fprintf(stderr, "Mosquitto library initialization failed: %s\n",
+                        mosquitto_strerror(init_result));
+                return 1;
+        }
+
+        client->mosq = mosquitto_new(NULL, true, NULL);
+        if (!client->mosq)
+                return fprintf(stderr, "Cannot create MQTT client\n"), 1;
+
+        mosquitto_message_callback_set(client->mosq, on_message);
+
+        int connect_result = mosquitto_connect(
+            client->mosq, broker_address, broker_port, MQTT_KEEPALIVE_SECONDS);
+        if (connect_result != MOSQ_ERR_SUCCESS)
+                return fail_with_cleanup(
+                    client, "Cannot connect to MQTT broker", connect_result);
+
+        int subscribe_result =
+            mosquitto_subscribe(client->mosq, NULL, topic, MQTT_QOS_LEVEL);
+        if (subscribe_result != MOSQ_ERR_SUCCESS)
+                return fail_with_cleanup(client, "Cannot subscribe to topic",
+                                         subscribe_result);
+
+        return 0;
 }
 
-void mqtt_client_cleanup(mqtt_client_t *client) {
-  mosquitto_destroy(client->mosq);
-  mosquitto_lib_cleanup();
+void mqtt_client_cleanup(mqtt_client_t *client)
+{
+        if (!client)
+                return;
+
+        if (client->mosq)
+                mosquitto_destroy(client->mosq);
+
+        mosquitto_lib_cleanup();
 }
 
-void mqtt_client_loop(mqtt_client_t *client) {
-  mosquitto_loop_forever(client->mosq, -1, 1);
+void mqtt_client_loop(mqtt_client_t *client)
+{
+        if (client && client->mosq)
+                mosquitto_loop_forever(client->mosq, LOOP_TIMEOUT_MS,
+                                       LOOP_MAX_MESSAGES);
 }
 
 static void on_message(struct mosquitto *mosq, void *userdata,
-                       const struct mosquitto_message *msg) {
-  logger_handle_message(msg->payload, msg->payloadlen);
+                       const struct mosquitto_message *message)
+{
+        if (message && message->payload)
+                logger_handle_message(message->payload, message->payloadlen);
+}
+
+static int fail_with_cleanup(mqtt_client_t *client, const char *error_message,
+                             int mosq_error_code)
+{
+        fprintf(stderr, "%s: %s\n", error_message,
+                mosquitto_strerror(mosq_error_code));
+        if (client && client->mosq)
+                mosquitto_destroy(client->mosq);
+        return 1;
 }
diff --git a/analysis/rtt/src/mqtt_client.h b/analysis/rtt/src/mqtt_client.h
index 781e742..1faa4ad 100644
--- a/analysis/rtt/src/mqtt_client.h
+++ b/analysis/rtt/src/mqtt_client.h
@@ -1,16 +1,39 @@
-
 #ifndef MQTT_CLIENT_H
 #define MQTT_CLIENT_H
 
 #include <mosquitto.h>
 
 typedef struct {
-  struct mosquitto *mosq;
+        struct mosquitto *mosq;
 } mqtt_client_t;
 
+/**
+ * @brief Initialize the MQTT client, connect to the broker, and subscribe to a
+ * topic.
+ *
+ * @param client          Pointer to mqtt_client_t struct.
+ * @param broker_address  MQTT broker IP or hostname.
+ * @param port            Broker port number.
+ * @param topic           Topic to subscribe to.
+ * @return int            0 on success, 1 on failure.
+ */
 int mqtt_client_init(mqtt_client_t *client, const char *broker_address,
                      int port, const char *topic);
+
+/**
+ * @brief Cleanup MQTT client resources and free associated memory.
+ *
+ * @param client Pointer to mqtt_client_t struct.
+ */
 void mqtt_client_cleanup(mqtt_client_t *client);
+
+/**
+ * @brief Run the MQTT client loop to process network events.
+ *
+ * This function blocks indefinitely.
+ *
+ * @param client Pointer to mqtt_client_t struct.
+ */
 void mqtt_client_loop(mqtt_client_t *client);
 
-#endif // MQTT_CLIENT_H
+#endif