Implement MQTT connect

2 files changed, 68 insertions, 10 deletions

diff --git a/backends/mqtt.c b/backends/mqtt.c
index d9cf2d8..8c4a9fd 100644
--- a/backends/mqtt.c
+++ b/backends/mqtt.c
@@ -2,10 +2,13 @@
 #define DEBUG
 
 #include <string.h>
+#include <time.h>
 
 #include "libmmbackend.h"
 #include "mqtt.h"
 
+static uint64_t last_maintenance = 0;
+
 //TODO
 // * Periodic connection retries
 
@@ -75,6 +78,14 @@ static int mqtt_parse_hostspec(instance* inst, char* hostspec){
 	return 0;
 }
 
+static int mqtt_generate_instanceid(instance* inst){
+	mqtt_instance_data* data = (mqtt_instance_data*) inst->impl;
+	char clientid[23] = "";
+
+	snprintf(clientid, sizeof(clientid), "MIDIMonster-%d-%s", (uint32_t) time(NULL), inst->name);
+	return mmbackend_strdup(&(data->client_id), clientid);
+}
+
 static size_t mqtt_varint_decode(uint8_t* buffer, uint32_t* result){
 	size_t value = 0, offset = 0;
 	do {
@@ -98,6 +109,24 @@ static size_t mqtt_varint_encode(size_t value, size_t maxlen, uint8_t* buffer){
 	return offset;
 }
 
+static size_t mqtt_push_binary(uint8_t* buffer, size_t buffer_length, uint8_t* content, size_t length){
+	if(buffer_length < length + 2 || length > 65535){
+		LOG("Failed to push length-prefixed data blob, buffer size exceeded");
+		return 0;
+	}
+
+	buffer[0] = (length >> 8) & 0xFF;
+	buffer[1] = length & 0xFF;
+
+	memcpy(buffer + 2, content, length);
+	return length + 2;
+}
+
+static size_t mqtt_push_utf8(uint8_t* buffer, size_t buffer_length, char* content){
+	//FIXME might want to validate the string for valid UTF-8
+	return mqtt_push_binary(buffer, buffer_length, (uint8_t*) content, strlen(content));
+}
+
 static void mqtt_disconnect(instance* inst){
 	mqtt_instance_data* data = (mqtt_instance_data*) inst->impl;
 
@@ -133,7 +162,7 @@ static int mqtt_configure(char* option, char* value){
 }
 
 static int mqtt_reconnect(instance* inst){
-	uint8_t variable_header[MQTT_BUFFER_LENGTH] = {0x00, 0x04, 'M', 'Q', 'T', 'T', 0x05, 0x00 /*flags*/, (MQTT_KEEPALIVE >> 8) & 0xFF, MQTT_KEEPALIVE & 0xFF};
+	uint8_t variable_header[MQTT_BUFFER_LENGTH] = {0x00, 0x04, 'M', 'Q', 'T', 'T', MQTT_VERSION, 0x00 /*flags*/, (MQTT_KEEPALIVE >> 8) & 0xFF, MQTT_KEEPALIVE & 0xFF};
 	uint8_t payload[MQTT_BUFFER_LENGTH];
 	size_t vh_offset = 10, payload_offset = 0;
 	mqtt_instance_data* data = (mqtt_instance_data*) inst->impl;
@@ -162,14 +191,15 @@ static int mqtt_reconnect(instance* inst){
 		return 1;
 	}
 
-	//prepare CONNECT message
+	//prepare CONNECT message flags
 	variable_header[7] = 0x02 /*clean start*/ | (data->user ? 0x80 : 0x00) | (data->user ? 0x40 : 0x00);
+
 	//TODO set session expiry interval option
 	//TODO re-use previos session on reconnect
-	
+
 	//push number of option bytes (as a varint, no less) before actually pushing the option data.
 	//obviously someone thought saving 3 whole bytes in exchange for not being able to sequentially creating the package was smart..
-	variable_header[vh_offset++] = 7;
+	variable_header[vh_offset++] = 8;
 	//push maximum packet size option
 	variable_header[vh_offset++] = 0x27;
 	variable_header[vh_offset++] = (MQTT_BUFFER_LENGTH >> 24) & 0xFF;
@@ -181,13 +211,14 @@ static int mqtt_reconnect(instance* inst){
 	variable_header[vh_offset++] = 0xFF;
 	variable_header[vh_offset++] = 0xFF;
 
-	//push client_id as utf8
-	//payload_offset += mqtt_push_utf8();
+	//prepare CONNECT payload
+	//push client id
+	payload_offset += mqtt_push_utf8(payload + payload_offset, sizeof(payload) - payload_offset, data->client_id);
 	if(data->user){
-		//push user name as utf8
+		payload_offset += mqtt_push_utf8(payload + payload_offset, sizeof(payload) - payload_offset, data->user);
 	}
 	if(data->password){
-		//push password as binary
+		payload_offset += mqtt_push_utf8(payload + payload_offset, sizeof(payload) - payload_offset, data->password);
 	}
 
 	mqtt_transmit(inst, MSG_CONNECT, vh_offset, variable_header, payload_offset, payload);
@@ -218,6 +249,15 @@ static int mqtt_configure_instance(instance* inst, char* option, char* value){
 		}
 		return 0;
 	}
+	else if(!strcmp(option, "clientid")){
+		if(strlen(value)){
+			mmbackend_strdup(&(data->client_id), value);
+			return 0;
+		}
+		else{
+			return mqtt_generate_instanceid(inst);
+		}
+	}
 
 	LOGPF("Unknown instance configuration option %s on instance %s", option, inst->name);
 	return 1;
@@ -233,6 +273,10 @@ static int mqtt_instance(instance* inst){
 
 	data->fd = -1;
 	inst->impl = data;
+
+	if(mqtt_generate_instanceid(inst)){
+		return 1;
+	}
 	return 0;
 }
 
@@ -247,8 +291,17 @@ static int mqtt_set(instance* inst, size_t num, channel** c, channel_value* v){
 }
 
 static int mqtt_handle(size_t num, managed_fd* fds){
-	LOG("Handling");
-	//TODO
+	size_t n = 0;
+
+	//for(n = 0; n < num; n++){
+	//}
+
+	//keepalive/reconnect processing
+	if(last_maintenance && mm_timestamp() - last_maintenance >= MQTT_KEEPALIVE * 1000){
+		//TODO run reconnects
+		last_maintenance = mm_timestamp();
+	}
+
 	return 0;
 }
 
@@ -268,6 +321,8 @@ static int mqtt_start(size_t n, instance** inst){
 		}
 	}
 
+	//initialize maintenance timer
+	last_maintenance = mm_timestamp();
 	return 0;
 }
 
diff --git a/backends/mqtt.h b/backends/mqtt.h
index 1c8b47d..df63319 100644
--- a/backends/mqtt.h
+++ b/backends/mqtt.h
@@ -14,6 +14,7 @@ static int mqtt_shutdown(size_t n, instance** inst);
 #define MQTT_TLS_PORT "8883"
 #define MQTT_BUFFER_LENGTH 8192
 #define MQTT_KEEPALIVE 10 
+#define MQTT_VERSION 0x05
 
 enum {
 	MSG_RESERVED = 0x00,
@@ -47,6 +48,8 @@ typedef struct /*_mqtt_instance_data*/ {
 	char** channel;
 
 	int fd;
+	uint8_t receive_buffer[MQTT_BUFFER_LENGTH];
+	size_t receive_offset;
 } mqtt_instance_data;
 
 //per-channel