mqtt.c

   1 /*
   2  *      Interaction with MQTT
   3  *
   4  *      (c) 2022 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         die("Mosquitto: subscribe failed");
  49
  50       mqtt_publish("status/ursary", "ok");
  51     }
  52 }
  53
  54 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level UNUSED, const char *message UNUSED)
  55 {
  56   // msg(L_INFO, "MQTT(%d): %s", level, message);
  57 }
  58
  59 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
  60 {
  61   struct mqtt_pipe_msg pm;
  62
  63   if (snprintf(pm.topic, sizeof(pm.topic), m->topic) >= sizeof(pm.topic))
  64     {
  65       msg(L_ERROR, "MQTT: Topic %s too long", m->topic);
  66       return;
  67     }
  68
  69   if (m->payloadlen >= sizeof(pm.val) - 1)
  70     {
  71       msg(L_ERROR, "MQTT: Invalid value for topic %s", m->topic);
  72       return;
  73     }
  74   memcpy(pm.val, m->payload, m->payloadlen);
  75   pm.val[m->payloadlen] = 0;
  76
  77   careful_write(mqtt_pipes[1], &pm, sizeof(pm));
  78 }
  79
  80 static int mqtt_pipe_read_callback(struct main_file *fi)
  81 {
  82   struct mqtt_pipe_msg pm;
  83   if (careful_read(fi->fd, &pm, sizeof(pm)) <= 0)
  84     die("MQTT pipe read error: %m");
  85
  86   DBG("MQTT < %s %s", pm.topic, pm.val);
  87   notify_mqtt(pm.topic, pm.val);
  88
  89   return HOOK_IDLE;
  90 }
  91
  92 void mqtt_init(void)
  93 {
  94   mosquitto_lib_init();
  95   mosq = mosquitto_new("ursary", 1, NULL);
  96   if (!mosq)
  97     die("Mosquitto: initialization failed");
  98
  99   if (pipe(mqtt_pipes) < 0)
 100     die("Cannot create pipes: %m");
 101
 102   mqtt_rx_pipe.fd = mqtt_pipes[0];
 103   mqtt_rx_pipe.read_handler = mqtt_pipe_read_callback;
 104   file_add(&mqtt_rx_pipe);
 105
 106   mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
 107   mosquitto_log_callback_set(mosq, mqtt_log_callback);
 108   mosquitto_message_callback_set(mosq, mqtt_msg_callback);
 109
 110   if (mosquitto_will_set(mosq, "status/ursary", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
 111     die("Mosquitto: unable to set will");
 112
 113   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)
 114     die("Mosquitto: unable to set TLS parameters");
 115
 116   if (mosquitto_connect_async(mosq, "burrow-mqtt", 8883, 60) != MOSQ_ERR_SUCCESS)
 117     die("Mosquitto: connect failed");
 118
 119   if (mosquitto_loop_start(mosq))
 120     die("Mosquitto: cannot start service thread");
 121 }