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