1 files changed, 48 insertions, 0 deletions

diff --git a/analysis/rtt/src/mqtt_client.c b/analysis/rtt/src/mqtt_client.c
new file mode 100644
index 0000000..43a06d8
--- /dev/null
+++ b/analysis/rtt/src/mqtt_client.c
@@ -0,0 +1,48 @@
+#include "mqtt_client.h"
+#include "logger.h"
+#include <stdio.h>
+#include <stdlib.h>
+
+#define WINDOW_SEC 1
+
+static void on_message(struct mosquitto *mosq, void *userdata,
+                       const struct mosquitto_message *msg);
+
+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;
+}
+
+void mqtt_client_cleanup(mqtt_client_t *client) {
+  mosquitto_destroy(client->mosq);
+  mosquitto_lib_cleanup();
+}
+
+void mqtt_client_loop(mqtt_client_t *client) {
+  mosquitto_loop_forever(client->mosq, -1, 1);
+}
+
+static void on_message(struct mosquitto *mosq, void *userdata,
+                       const struct mosquitto_message *msg) {
+  logger_handle_message(msg->payload, msg->payloadlen);
+}