mqtt.c

   1 #include <inttypes.h>
   2 #include <stdio.h>
   3 #include <stdlib.h>
   4 #include <string.h>
   5
   6 #include "config.h"
   7
   8 #include "common.h"
   9 #include "player.h"
  10 #include "rtsp.h"
  11
  12 #include "rtp.h"
  13
  14 #include "dacp.h"
  15 #include "metadata_hub.h"
  16 #include "mqtt.h"
  17 #include <mosquitto.h>
  18
  19 // this holds the mosquitto client
  20 struct mosquitto *global_mosq = NULL;
  21 char *topic = NULL;
  22 int connected = 0;
  23
  24 // mosquitto logging
  25 void _cb_log(__attribute__((unused)) struct mosquitto *mosq, __attribute__((unused)) void *userdata,
  26              int level, const char *str) {
  27   switch (level) {
  28   case MOSQ_LOG_DEBUG:
  29     debug(3, str);
  30     break;
  31   case MOSQ_LOG_INFO:
  32     debug(3, str);
  33     break;
  34   case MOSQ_LOG_NOTICE:
  35     debug(3, str);
  36     break;
  37   case MOSQ_LOG_WARNING:
  38     inform(str);
  39     break;
  40   case MOSQ_LOG_ERR: {
  41     die("MQTT: Error: %s\n", str);
  42   }
  43   }
  44 }
  45
  46 // mosquitto message handler
  47 void on_message(__attribute__((unused)) struct mosquitto *mosq,
  48                 __attribute__((unused)) void *userdata, const struct mosquitto_message *msg) {
  49
  50   // null-terminate the payload
  51   char payload[msg->payloadlen + 1];
  52   memcpy(payload, msg->payload, msg->payloadlen);
  53   payload[msg->payloadlen] = 0;
  54
  55   debug(2, "[MQTT]: received Message on topic %s: %s\n", msg->topic, payload);
  56
  57   // All recognized commands
  58   char *commands[] = {"command",    "beginff",       "beginrew",   "mutetoggle", "nextitem",
  59                       "previtem",   "pause",         "playpause",  "play",       "stop",
  60                       "playresume", "shuffle_songs", "volumedown", "volumeup",   NULL};
  61
  62   int it = 0;
  63
  64   // send command if it's a valid one
  65   while (commands[it] != NULL) {
  66     if ((size_t)msg->payloadlen >= strlen(commands[it]) &&
  67         strncmp(msg->payload, commands[it], strlen(commands[it])) == 0) {
  68       debug(2, "[MQTT]: DACP Command: %s\n", commands[it]);
  69       send_simple_dacp_command(commands[it]);
  70       break;
  71     }
  72     it++;
  73   }
  74 }
  75
  76 void on_disconnect(__attribute__((unused)) struct mosquitto *mosq,
  77                    __attribute__((unused)) void *userdata, __attribute__((unused)) int rc) {
  78   connected = 0;
  79   debug(2, "[MQTT]: disconnected");
  80 }
  81
  82 void on_connect(struct mosquitto *mosq, __attribute__((unused)) void *userdata,
  83                 __attribute__((unused)) int rc) {
  84   connected = 1;
  85   debug(2, "[MQTT]: connected");
  86
  87   // subscribe if requested
  88   if (config.mqtt_enable_remote) {
  89     char remotetopic[strlen(config.mqtt_topic) + 8];
  90     snprintf(remotetopic, strlen(config.mqtt_topic) + 8, "%s/remote", config.mqtt_topic);
  91     mosquitto_subscribe(mosq, NULL, remotetopic, 0);
  92   }
  93 }
  94
  95 // helper function to publish under a topic and automatically append the main topic
  96 void mqtt_publish(char *topic, char *data_in, uint32_t length_in) {
  97   char *data = data_in;
  98   uint32_t length = length_in;
  99
 100   if ((length == 0) && (config.mqtt_empty_payload_substitute != NULL)) {
 101     length = strlen(config.mqtt_empty_payload_substitute);
 102     data = config.mqtt_empty_payload_substitute;
 103   }
 104
 105   char fulltopic[strlen(config.mqtt_topic) + strlen(topic) + 3];
 106   snprintf(fulltopic, strlen(config.mqtt_topic) + strlen(topic) + 2, "%s/%s", config.mqtt_topic,
 107            topic);
 108   debug(2, "[MQTT]: publishing under %s", fulltopic);
 109
 110   int rc;
 111   if ((rc = mosquitto_publish(global_mosq, NULL, fulltopic, length, data, 0, 0)) !=
 112       MOSQ_ERR_SUCCESS) {
 113     switch (rc) {
 114     case MOSQ_ERR_NO_CONN:
 115       debug(1, "[MQTT]: Publish failed: not connected to broker");
 116       break;
 117     default:
 118       debug(1, "[MQTT]: Publish failed: unknown error");
 119       break;
 120     }
 121   }
 122 }
 123
 124 // handler for incoming metadata
 125 void mqtt_process_metadata(uint32_t type, uint32_t code, char *data, uint32_t length) {
 126   if (global_mosq == NULL || connected != 1) {
 127     debug(3, "[MQTT]: Client not connected, skipping metadata handling");
 128     return;
 129   }
 130   if (config.mqtt_publish_raw) {
 131     uint32_t val;
 132     char topic[] = "____/____";
 133
 134     val = htonl(type);
 135     memcpy(topic, &val, 4);
 136     val = htonl(code);
 137     memcpy(topic + 5, &val, 4);
 138     mqtt_publish(topic, data, length);
 139   }
 140   if (config.mqtt_publish_parsed) {
 141     if (type == 'core') {
 142       int32_t r;
 143       uint64_t trackid;
 144       char trackidstring[32];
 145
 146       switch (code) {
 147       case 'asar':
 148         mqtt_publish("artist", data, length);
 149         break;
 150       case 'asal':
 151         mqtt_publish("album", data, length);
 152         break;
 153       case 'asfm':
 154         mqtt_publish("format", data, length);
 155         break;
 156       case 'asgn':
 157         mqtt_publish("genre", data, length);
 158         break;
 159       case 'minm':
 160         mqtt_publish("title", data, length);
 161         break;
 162       case 'mper':
 163         trackid = *(uint64_t *)(data);
 164         r = snprintf(trackidstring, sizeof(trackidstring), "%" PRIX64 "", trackid);
 165         mqtt_publish("track_id", trackidstring, r);
 166       }
 167     } else if (type == 'ssnc') {
 168       switch (code) {
 169       case 'abeg':
 170         mqtt_publish("active_start", data, length);
 171         break;
 172       case 'acre':
 173         mqtt_publish("active_remote_id", data, length);
 174         break;
 175       case 'aend':
 176         mqtt_publish("active_end", data, length);
 177         break;
 178       case 'asal':
 179         mqtt_publish("songalbum", data, length);
 180         break;
 181       case 'asdk':
 182         mqtt_publish("songdatakind", data,
 183                      length); // 0 seem to be a timed item, 1 an untimed stream
 184         break;
 185       case 'clip':
 186         mqtt_publish("client_ip", data, length);
 187         break;
 188       case 'cdid':
 189         mqtt_publish("client_device_id", data, length);
 190         break;
 191       case 'cmac':
 192         mqtt_publish("client_mac_address", data, length);
 193         break;
 194       case 'cmod':
 195         mqtt_publish("client_model", data, length);
 196         break;
 197       case 'daid':
 198         mqtt_publish("dacp_id", data, length);
 199         break;
 200       case 'phbt':
 201         mqtt_publish("frame_position_and_time", data, length);
 202         break;
 203       case 'phb0':
 204         mqtt_publish("first_frame_position_and_time", data, length);
 205         break;
 206       case 'ofmt':
 207         mqtt_publish("output_format", data, length);
 208         break;
 209       case 'ofps':
 210         mqtt_publish("output_frame_rate", data, length);
 211         break;
 212       case 'pbeg':
 213         mqtt_publish("play_start", data, length);
 214         break;
 215       case 'pend':
 216         mqtt_publish("play_end", data, length);
 217         break;
 218       case 'pfls':
 219         mqtt_publish("play_flush", data, length);
 220         break;
 221       case 'PICT':
 222         if (config.mqtt_publish_cover) {
 223           mqtt_publish("cover", data, length);
 224         }
 225         break;
 226       case 'prsm':
 227         mqtt_publish("play_resume", data, length);
 228         break;
 229       case 'pvol':
 230         mqtt_publish("volume", data, length);
 231         break;
 232       case 'snam':
 233         mqtt_publish("client_name", data, length);
 234         break;
 235       case 'styp':
 236         mqtt_publish("stream_type", data, length);
 237         break;
 238       case 'svip':
 239         mqtt_publish("server_ip", data, length);
 240         break;
 241       case 'svna':
 242         mqtt_publish("service_name", data, length);
 243         break;
 244       }
 245     }
 246   }
 247
 248   return;
 249 }
 250
 251 int initialise_mqtt() {
 252   debug(1, "Initialising MQTT");
 253   if (config.mqtt_hostname == NULL) {
 254     debug(1, "[MQTT]: Not initialized, as the hostname is not set");
 255     return 0;
 256   }
 257   int keepalive = 60;
 258   mosquitto_lib_init();
 259   if (!(global_mosq = mosquitto_new(config.service_name, true, NULL))) {
 260     die("[MQTT]: FATAL: Could not create mosquitto object! %d\n", global_mosq);
 261   }
 262
 263   if (config.mqtt_cafile != NULL || config.mqtt_capath != NULL || config.mqtt_certfile != NULL ||
 264       config.mqtt_keyfile != NULL) {
 265     if (mosquitto_tls_set(global_mosq, config.mqtt_cafile, config.mqtt_capath, config.mqtt_certfile,
 266                           config.mqtt_keyfile, NULL) != MOSQ_ERR_SUCCESS) {
 267       die("[MQTT]: TLS Setup failed");
 268     }
 269   }
 270
 271   if (config.mqtt_username != NULL || config.mqtt_password != NULL) {
 272     if (mosquitto_username_pw_set(global_mosq, config.mqtt_username, config.mqtt_password) !=
 273         MOSQ_ERR_SUCCESS) {
 274       die("[MQTT]: Username/Password set failed");
 275     }
 276   }
 277   mosquitto_log_callback_set(global_mosq, _cb_log);
 278
 279   if (config.mqtt_enable_remote) {
 280     mosquitto_message_callback_set(global_mosq, on_message);
 281   }
 282
 283   mosquitto_disconnect_callback_set(global_mosq, on_disconnect);
 284   mosquitto_connect_callback_set(global_mosq, on_connect);
 285   if (mosquitto_connect(global_mosq, config.mqtt_hostname, config.mqtt_port, keepalive)) {
 286     inform("[MQTT]: Could not establish a mqtt connection");
 287   }
 288   if (mosquitto_loop_start(global_mosq) != MOSQ_ERR_SUCCESS) {
 289     inform("[MQTT]: Could start MQTT Main loop");
 290   }
 291
 292   return 0;
 293 }