[thirdparty/shairport-sync.git] / mqtt.c

#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "config.h"

#include "common.h"
#include "player.h"
#include "rtsp.h"

#include "rtp.h"

#include "dacp.h"
#include "metadata_hub.h"
#include "mqtt.h"
#include <mosquitto.h>

// this holds the mosquitto client
struct mosquitto *global_mosq = NULL;
char *topic = NULL;
int connected = 0;

// mosquitto logging
void _cb_log(__attribute__((unused)) struct mosquitto *mosq, __attribute__((unused)) void *userdata,
             int level, const char *str) {
  switch (level) {
  case MOSQ_LOG_DEBUG:
    debug(3, str);
    break;
  case MOSQ_LOG_INFO:
    debug(3, str);
    break;
  case MOSQ_LOG_NOTICE:
    debug(3, str);
    break;
  case MOSQ_LOG_WARNING:
    inform(str);
    break;
  case MOSQ_LOG_ERR: {
    die("MQTT: Error: %s\n", str);
  }
  }
}

// mosquitto message handler
void on_message(__attribute__((unused)) struct mosquitto *mosq,
                __attribute__((unused)) void *userdata, const struct mosquitto_message *msg) {

  // null-terminate the payload
  char payload[msg->payloadlen + 1];
  memcpy(payload, msg->payload, msg->payloadlen);
  payload[msg->payloadlen] = 0;

  debug(2, "[MQTT]: received Message on topic %s: %s\n", msg->topic, payload);

  // All recognized commands
  char *commands[] = {"command",    "beginff",       "beginrew",   "mutetoggle", "nextitem",
                      "previtem",   "pause",         "playpause",  "play",       "stop",
                      "playresume", "shuffle_songs", "volumedown", "volumeup",   NULL};

  int it = 0;

  // send command if it's a valid one
  while (commands[it] != NULL) {
    if ((size_t)msg->payloadlen >= strlen(commands[it]) &&
        strncmp(msg->payload, commands[it], strlen(commands[it])) == 0) {
      debug(2, "[MQTT]: DACP Command: %s\n", commands[it]);
      send_simple_dacp_command(commands[it]);
      break;
    }
    it++;
  }
}

void on_disconnect(__attribute__((unused)) struct mosquitto *mosq,
                   __attribute__((unused)) void *userdata, __attribute__((unused)) int rc) {
  connected = 0;
  debug(2, "[MQTT]: disconnected");
}

void on_connect(struct mosquitto *mosq, __attribute__((unused)) void *userdata,
                __attribute__((unused)) int rc) {
  connected = 1;
  debug(2, "[MQTT]: connected");

  // subscribe if requested
  if (config.mqtt_enable_remote) {
    char remotetopic[strlen(config.mqtt_topic) + 8];
    snprintf(remotetopic, strlen(config.mqtt_topic) + 8, "%s/remote", config.mqtt_topic);
    mosquitto_subscribe(mosq, NULL, remotetopic, 0);
  }
}

// helper function to publish under a topic and automatically append the main topic
void mqtt_publish(char *topic, char *data_in, uint32_t length_in) {
  char *data = data_in;
  uint32_t length = length_in;

  if ((length == 0) && (config.mqtt_empty_payload_substitute != NULL)) {
    length = strlen(config.mqtt_empty_payload_substitute);
    data = config.mqtt_empty_payload_substitute;
  }

  char fulltopic[strlen(config.mqtt_topic) + strlen(topic) + 3];
  snprintf(fulltopic, strlen(config.mqtt_topic) + strlen(topic) + 2, "%s/%s", config.mqtt_topic,
           topic);
  debug(2, "[MQTT]: publishing under %s", fulltopic);

  int rc;
  if ((rc = mosquitto_publish(global_mosq, NULL, fulltopic, length, data, 0, 0)) !=
      MOSQ_ERR_SUCCESS) {
    switch (rc) {
    case MOSQ_ERR_NO_CONN:
      debug(1, "[MQTT]: Publish failed: not connected to broker");
      break;
    default:
      debug(1, "[MQTT]: Publish failed: unknown error");
      break;
    }
  }
}

// handler for incoming metadata
void mqtt_process_metadata(uint32_t type, uint32_t code, char *data, uint32_t length) {
  if (global_mosq == NULL || connected != 1) {
    debug(3, "[MQTT]: Client not connected, skipping metadata handling");
    return;
  }
  if (config.mqtt_publish_raw) {
    uint32_t val;
    char topic[] = "____/____";

    val = htonl(type);
    memcpy(topic, &val, 4);
    val = htonl(code);
    memcpy(topic + 5, &val, 4);
    mqtt_publish(topic, data, length);
  }
  if (config.mqtt_publish_parsed) {
    if (type == 'core') {
      int32_t r;
      uint64_t trackid;
      char trackidstring[32];

      switch (code) {
      case 'asar':
        mqtt_publish("artist", data, length);
        break;
      case 'asal':
        mqtt_publish("album", data, length);
        break;
      case 'asfm':
        mqtt_publish("format", data, length);
        break;
      case 'asgn':
        mqtt_publish("genre", data, length);
        break;
      case 'minm':
        mqtt_publish("title", data, length);
        break;
      case 'mper':
        trackid = *(uint64_t *)(data);
        r = snprintf(trackidstring, sizeof(trackidstring), "%" PRIX64 "", trackid);
        mqtt_publish("track_id", trackidstring, r);
      }
    } else if (type == 'ssnc') {
      switch (code) {
      case 'abeg':
        mqtt_publish("active_start", data, length);
        break;
      case 'acre':
        mqtt_publish("active_remote_id", data, length);
        break;
      case 'aend':
        mqtt_publish("active_end", data, length);
        break;
      case 'asal':
        mqtt_publish("songalbum", data, length);
        break;
      case 'asdk':
        mqtt_publish("songdatakind", data,
                     length); // 0 seem to be a timed item, 1 an untimed stream
        break;
      case 'clip':
        mqtt_publish("client_ip", data, length);
        break;
      case 'cdid':
        mqtt_publish("client_device_id", data, length);
        break;
      case 'cmac':
        mqtt_publish("client_mac_address", data, length);
        break;
      case 'cmod':
        mqtt_publish("client_model", data, length);
        break;
      case 'daid':
        mqtt_publish("dacp_id", data, length);
        break;
      case 'phbt':
        mqtt_publish("frame_position_and_time", data, length);
        break;
      case 'phb0':
        mqtt_publish("first_frame_position_and_time", data, length);
        break;
      case 'ofmt':
        mqtt_publish("output_format", data, length);
        break;
      case 'ofps':
        mqtt_publish("output_frame_rate", data, length);
        break;
      case 'pbeg':
        mqtt_publish("play_start", data, length);
        break;
      case 'pend':
        mqtt_publish("play_end", data, length);
        break;
      case 'pfls':
        mqtt_publish("play_flush", data, length);
        break;
      case 'PICT':
        if (config.mqtt_publish_cover) {
          mqtt_publish("cover", data, length);
        }
        break;
      case 'prsm':
        mqtt_publish("play_resume", data, length);
        break;
      case 'pvol':
        mqtt_publish("volume", data, length);
        break;
      case 'snam':
        mqtt_publish("client_name", data, length);
        break;
      case 'styp':
        mqtt_publish("stream_type", data, length);
        break;
      case 'svip':
        mqtt_publish("server_ip", data, length);
        break;
      case 'svna':
        mqtt_publish("service_name", data, length);
        break;
      }
    }
  }

  return;
}

int initialise_mqtt() {
  debug(1, "Initialising MQTT");
  if (config.mqtt_hostname == NULL) {
    debug(1, "[MQTT]: Not initialized, as the hostname is not set");
    return 0;
  }
  int keepalive = 60;
  mosquitto_lib_init();
  if (!(global_mosq = mosquitto_new(config.service_name, true, NULL))) {
    die("[MQTT]: FATAL: Could not create mosquitto object! %d\n", global_mosq);
  }

  if (config.mqtt_cafile != NULL || config.mqtt_capath != NULL || config.mqtt_certfile != NULL ||
      config.mqtt_keyfile != NULL) {
    if (mosquitto_tls_set(global_mosq, config.mqtt_cafile, config.mqtt_capath, config.mqtt_certfile,
                          config.mqtt_keyfile, NULL) != MOSQ_ERR_SUCCESS) {
      die("[MQTT]: TLS Setup failed");
    }
  }

  if (config.mqtt_username != NULL || config.mqtt_password != NULL) {
    if (mosquitto_username_pw_set(global_mosq, config.mqtt_username, config.mqtt_password) !=
        MOSQ_ERR_SUCCESS) {
      die("[MQTT]: Username/Password set failed");
    }
  }
  mosquitto_log_callback_set(global_mosq, _cb_log);

  if (config.mqtt_enable_remote) {
    mosquitto_message_callback_set(global_mosq, on_message);
  }

  mosquitto_disconnect_callback_set(global_mosq, on_disconnect);
  mosquitto_connect_callback_set(global_mosq, on_connect);
  if (mosquitto_connect(global_mosq, config.mqtt_hostname, config.mqtt_port, keepalive)) {
    inform("[MQTT]: Could not establish a mqtt connection");
  }
  if (mosquitto_loop_start(global_mosq) != MOSQ_ERR_SUCCESS) {
    inform("[MQTT]: Could start MQTT Main loop");
  }

  return 0;
}
Commit	Line	Data
a9c8de50	1	#include <inttypes.h>
02694948 TZ	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"
a9c8de50	15	#include "metadata_hub.h"
02694948	16	#include "mqtt.h"
c2e3fa5a	17	#include <mosquitto.h>
02694948	18
c2e3fa5a	19	// this holds the mosquitto client
c445510d	20	struct mosquitto *global_mosq = NULL;
02694948	21	char *topic = NULL;
3d1086ea	22	int connected = 0;
02694948	23
c2e3fa5a MB	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:
9ae649fa	29	debug(3, str);
c2e3fa5a MB	30	break;
c2e3fa5a MB	31	case MOSQ_LOG_INFO:
9ae649fa	32	debug(3, str);
c2e3fa5a MB	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	}
02694948 TZ	43	}
	44	}
	45
c2e3fa5a MB	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
dfd90bac	55	debug(2, "[MQTT]: received Message on topic %s: %s\n", msg->topic, payload);
c2e3fa5a MB	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
cb38e90e	65	while (commands[it] != NULL) {
c2e3fa5a MB	66	if ((size_t)msg->payloadlen >= strlen(commands[it]) &&
c2e3fa5a MB	67	strncmp(msg->payload, commands[it], strlen(commands[it])) == 0) {
dfd90bac	68	debug(2, "[MQTT]: DACP Command: %s\n", commands[it]);
02694948 TZ	69	send_simple_dacp_command(commands[it]);
	70	break;
	71	}
cb38e90e	72	it++;
02694948 TZ	73	}
	74	}
	75
c2e3fa5a MB	76	void on_disconnect(__attribute__((unused)) struct mosquitto *mosq,
c2e3fa5a MB	77	__attribute__((unused)) void *userdata, __attribute__((unused)) int rc) {
3d1086ea	78	connected = 0;
dfd90bac	79	debug(2, "[MQTT]: disconnected");
02694948 TZ	80	}
02694948 TZ	81
c2e3fa5a MB	82	void on_connect(struct mosquitto mosq, __attribute__((unused)) void userdata,
c2e3fa5a MB	83	__attribute__((unused)) int rc) {
3d1086ea	84	connected = 1;
dfd90bac	85	debug(2, "[MQTT]: connected");
c2e3fa5a MB	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);
02694948 TZ	92	}
	93	}
	94
c2e3fa5a	95	// helper function to publish under a topic and automatically append the main topic
3d720d65 MB	96	void mqtt_publish(char topic, char data_in, uint32_t length_in) {
	97	char *data = data_in;
	98	uint32_t length = length_in;
a382ac66	99
3d720d65 MB	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	}
a382ac66	104
c2e3fa5a MB	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);
dfd90bac	108	debug(2, "[MQTT]: publishing under %s", fulltopic);
a382ac66	109
3d1086ea	110	int rc;
c2e3fa5a MB	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;
3d1086ea TZ	120	}
3d1086ea TZ	121	}
02694948 TZ	122	}
02694948 TZ	123
c2e3fa5a MB	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) {
3d1086ea TZ	127	debug(3, "[MQTT]: Client not connected, skipping metadata handling");
	128	return;
	129	}
c2e3fa5a	130	if (config.mqtt_publish_raw) {
02694948 TZ	131	uint32_t val;
02694948 TZ	132	char topic[] = "____/____";
c2e3fa5a MB	133
	134	val = htonl(type);
	135	memcpy(topic, &val, 4);
	136	val = htonl(code);
	137	memcpy(topic + 5, &val, 4);
02694948 TZ	138	mqtt_publish(topic, data, length);
02694948 TZ	139	}
c2e3fa5a MB	140	if (config.mqtt_publish_parsed) {
c2e3fa5a MB	141	if (type == 'core') {
a9c8de50 CC	142	int32_t r;
	143	uint64_t trackid;
	144	char trackidstring[32];
	145
02694948	146	switch (code) {
c2e3fa5a MB	147	case 'asar':
	148	mqtt_publish("artist", data, length);
	149	break;
	150	case 'asal':
	151	mqtt_publish("album", data, length);
	152	break;
a9c8de50 CC	153	case 'asfm':
a9c8de50 CC	154	mqtt_publish("format", data, length);
c2e3fa5a MB	155	break;
	156	case 'asgn':
	157	mqtt_publish("genre", data, length);
	158	break;
a9c8de50 CC	159	case 'minm':
a9c8de50 CC	160	mqtt_publish("title", data, length);
c2e3fa5a	161	break;
a9c8de50 CC	162	case 'mper':
	163	trackid = (uint64_t )(data);
	164	r = snprintf(trackidstring, sizeof(trackidstring), "%" PRIX64 "", trackid);
	165	mqtt_publish("track_id", trackidstring, r);
02694948	166	}
c2e3fa5a	167	} else if (type == 'ssnc') {
02694948	168	switch (code) {
a9c8de50 CC	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;
c2e3fa5a MB	178	case 'asal':
	179	mqtt_publish("songalbum", data, length);
	180	break;
d67909b8	181	case 'asdk':
e8b8c6ed MB	182	mqtt_publish("songdatakind", data,
e8b8c6ed MB	183	length); // 0 seem to be a timed item, 1 an untimed stream
d67909b8	184	break;
e8b8c6ed	185	case 'clip':
c2e3fa5a MB	186	mqtt_publish("client_ip", data, length);
c2e3fa5a MB	187	break;
82c7f4ef CC	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;
a9c8de50 CC	197	case 'daid':
a9c8de50 CC	198	mqtt_publish("dacp_id", data, length);
f3b2d070	199	break;
8e5393ec MB	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;
d67909b8 MB	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;
c2e3fa5a MB	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;
c2e3fa5a	221	case 'PICT':
60b9347a	222	if (config.mqtt_publish_cover) {
c2e3fa5a MB	223	mqtt_publish("cover", data, length);
	224	}
	225	break;
a9c8de50 CC	226	case 'prsm':
	227	mqtt_publish("play_resume", data, length);
	228	break;
	229	case 'pvol':
	230	mqtt_publish("volume", data, length);
	231	break;
82c7f4ef CC	232	case 'snam':
	233	mqtt_publish("client_name", data, length);
	234	break;
d67909b8 MB	235	case 'styp':
	236	mqtt_publish("stream_type", data, length);
	237	break;
a9c8de50 CC	238	case 'svip':
	239	mqtt_publish("server_ip", data, length);
	240	break;
3e4b67e9 MB	241	case 'svna':
	242	mqtt_publish("service_name", data, length);
	243	break;
02694948 TZ	244	}
	245	}
	246	}
	247
	248	return;
	249	}
	250
02694948 TZ	251	int initialise_mqtt() {
02694948 TZ	252	debug(1, "Initialising MQTT");
c2e3fa5a	253	if (config.mqtt_hostname == NULL) {
02694948 TZ	254	debug(1, "[MQTT]: Not initialized, as the hostname is not set");
	255	return 0;
	256	}
	257	int keepalive = 60;
	258	mosquitto_lib_init();
c2e3fa5a	259	if (!(global_mosq = mosquitto_new(config.service_name, true, NULL))) {
c445510d	260	die("[MQTT]: FATAL: Could not create mosquitto object! %d\n", global_mosq);
02694948 TZ	261	}
02694948 TZ	262
c2e3fa5a MB	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) {
02694948 TZ	267	die("[MQTT]: TLS Setup failed");
	268	}
	269	}
	270
c2e3fa5a MB	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) {
02694948 TZ	274	die("[MQTT]: Username/Password set failed");
	275	}
	276	}
c445510d	277	mosquitto_log_callback_set(global_mosq, _cb_log);
c2e3fa5a MB	278
c2e3fa5a MB	279	if (config.mqtt_enable_remote) {
c445510d	280	mosquitto_message_callback_set(global_mosq, on_message);
02694948	281	}
c2e3fa5a	282
c445510d TZ	283	mosquitto_disconnect_callback_set(global_mosq, on_disconnect);
c445510d TZ	284	mosquitto_connect_callback_set(global_mosq, on_connect);
c2e3fa5a	285	if (mosquitto_connect(global_mosq, config.mqtt_hostname, config.mqtt_port, keepalive)) {
02694948 TZ	286	inform("[MQTT]: Could not establish a mqtt connection");
02694948 TZ	287	}
c2e3fa5a	288	if (mosquitto_loop_start(global_mosq) != MOSQ_ERR_SUCCESS) {
02694948 TZ	289	inform("[MQTT]: Could start MQTT Main loop");
	290	}
	291
	292	return 0;
	293	}