mqtt.c

   1 /*
   2  *      Interaction with MQTT
   3  *
   4  *      (c) 2022-2023 Martin Mares <mj@ucw.cz>
   5  */
   6
   7 #undef LOCAL_DEBUG
   8
   9 #include <ucw/lib.h>
  10 #include <ucw/io.h>
  11 #include <ucw/mainloop.h>
  12
  13 #include <stdio.h>
  14 #include <string.h>
  15
  16 #include <mosquitto.h>
  17
  18 #include "ursaryd.h"
  19
  20 static struct mosquitto *mosq;
  21
  22 int mqtt_pipes[2];
  23 static struct main_file mqtt_rx_pipe;
  24
  25 struct mqtt_pipe_msg {
  26   char topic[100];
  27   char val[100];
  28 };
  29
  30 void mqtt_publish(const char *topic, const char *fmt, ...)
  31 {
  32   va_list args;
  33   va_start(args, fmt);
  34   char m[256];
  35   int l = vsnprintf(m, sizeof(m), fmt, args);
  36   DBG("MQTT > %s %s", topic, m);
  37   if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
  38     msg(L_ERROR, "Mosquitto: publish failed");
  39   va_end(args);
  40 }
  41
  42 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
  43 {
  44   if (!status)
  45     {
  46       msg(L_DEBUG, "MQTT: Connection established, subscribing");
  47       if (mosquitto_subscribe(mosq, NULL, "burrow/lights/catarium/#", 1) != MOSQ_ERR_SUCCESS ||
  48           mosquitto_subscribe(mosq, NULL, "burrow/control/catarium-ir", 1) != MOSQ_ERR_SUCCESS)
  49         die("Mosquitto: subscribe failed");
  50
  51       mqtt_publish("status/ursary", "ok");
  52     }
  53 }
  54
  55 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level UNUSED, const char *message UNUSED)
  56 {
  57   // msg(L_INFO, "MQTT(%d): %s", level, message);
  58 }
  59
  60 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
  61 {
  62   struct mqtt_pipe_msg pm;
  63
  64   if (snprintf(pm.topic, sizeof(pm.topic), m->topic) >= sizeof(pm.topic))
  65     {
  66       msg(L_ERROR, "MQTT: Topic %s too long", m->topic);
  67       return;
  68     }
  69
  70   if (m->payloadlen >= sizeof(pm.val) - 1)
  71     {
  72       msg(L_ERROR, "MQTT: Invalid value for topic %s", m->topic);
  73       return;
  74     }
  75   memcpy(pm.val, m->payload, m->payloadlen);
  76   pm.val[m->payloadlen] = 0;
  77
  78   careful_write(mqtt_pipes[1], &pm, sizeof(pm));
  79 }
  80
  81 static int mqtt_pipe_read_callback(struct main_file *fi)
  82 {
  83   struct mqtt_pipe_msg pm;
  84   if (careful_read(fi->fd, &pm, sizeof(pm)) <= 0)
  85     die("MQTT pipe read error: %m");
  86
  87   DBG("MQTT < %s %s", pm.topic, pm.val);
  88   notify_mqtt(pm.topic, pm.val);
  89
  90   return HOOK_IDLE;
  91 }
  92
  93 void mqtt_init(void)
  94 {
  95   mosquitto_lib_init();
  96   mosq = mosquitto_new("ursary", 1, NULL);
  97   if (!mosq)
  98     die("Mosquitto: initialization failed");
  99
 100   if (pipe(mqtt_pipes) < 0)
 101     die("Cannot create pipes: %m");
 102
 103   mqtt_rx_pipe.fd = mqtt_pipes[0];
 104   mqtt_rx_pipe.read_handler = mqtt_pipe_read_callback;
 105   file_add(&mqtt_rx_pipe);
 106
 107   mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
 108   mosquitto_log_callback_set(mosq, mqtt_log_callback);
 109   mosquitto_message_callback_set(mosq, mqtt_msg_callback);
 110
 111   if (mosquitto_will_set(mosq, "status/ursary", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
 112     die("Mosquitto: unable to set will");
 113
 114   if (mosquitto_tls_set(mosq, "/etc/burrow-mqtt/ca.crt", NULL, "/etc/burrow-mqtt/client.crt", "/etc/burrow-mqtt/client.key", NULL) != MOSQ_ERR_SUCCESS)
 115     die("Mosquitto: unable to set TLS parameters");
 116
 117   if (mosquitto_connect_async(mosq, "burrow-mqtt", 8883, 60) != MOSQ_ERR_SUCCESS)
 118     die("Mosquitto: connect failed");
 119
 120   if (mosquitto_loop_start(mosq))
 121     die("Mosquitto: cannot start service thread");
 122 }