Initial implementation of MQTT

Initial implementation of MQTT alongside ethernet gateway.

libraries/MySensors/Makefile

@@ -33,7 +33,9 @@ ifeq ($(PIREV),$(filter $(PIREV),a01041 a21041 0010))
 	CXXFLAGS+=-D__PI_BPLUS
 endif 
 
+COMPILER_FLAGS=examples_RPi/compiler_flags
 GATEWAY=examples_RPi/PiGateway
+GATEWAYMQTT=examples_RPi/PiGatewayMQTT
 GATEWAY_SOURCES=examples_RPi/PiGateway.cpp $(wildcard ./utility/*.cpp)
 GATEWAY_OBJECTS=$(patsubst %.cpp,%.o,$(GATEWAY_SOURCES))
 DEPS+=$(patsubst %.cpp,%.d,$(GATEWAY_SOURCES))
@@ -41,24 +43,37 @@ DEPS+=$(patsubst %.cpp,%.d,$(GATEWAY_SOURCES))
 RF24H = /usr/local/include/RF24
 CINCLUDE=-I. -I./core -I$(RF24H)
 
-.PHONY: all clean install uninstall
+.PHONY: all gateway gatewaymqtt clean install uninstall force
 
 all: $(GATEWAY)
 
+mqtt: $(GATEWAYMQTT)
+
+$(COMPILER_FLAGS): force
+	echo '$(CXXFLAGS)' | cmp -s - $@ || echo '$(CXXFLAGS)' > $@
+
+$(GATEWAY_OBJECTS): $(COMPILER_FLAGS)
+
 # Basic Gateway Build
 $(GATEWAY): LDFLAGS += -pthread
 $(GATEWAY): $(GATEWAY_OBJECTS)
 	$(CXX) $(LDFLAGS) -o $@ $(GATEWAY_OBJECTS)
+	
+# MQTT Gateway Build
+$(GATEWAYMQTT): CXXFLAGS += -DMY_GATEWAY_MQTT_CLIENT
+$(GATEWAYMQTT): LDFLAGS += -pthread -lmosquitto
+$(GATEWAYMQTT): $(GATEWAY_OBJECTS)
+	$(CXX) $(LDFLAGS) -o $@ $(GATEWAY_OBJECTS)
 
 # Include all .d files
 -include $(DEPS)
-
+	
 %.o: %.cpp
 	$(CXX) $(CXXFLAGS) $(CINCLUDE) -MMD -c -o $@ $<
 
 # The Cleaner
 clean:
-	rm -rf build $(GATEWAY_OBJECTS) $(GATEWAY) $(DEPS)
+	rm -rf build $(GATEWAY_OBJECTS) $(GATEWAY) $(GATEWAYMQTT) $(DEPS) $(COMPILER_FLAGS)
 
 install: all install-gatewaybasic install-gatewayserial install-gatewayethernet install-initscripts
 

libraries/MySensors/core/MyGatewayTransportEthernetLinux.cpp

@@ -27,6 +27,26 @@
 #include <signal.h>
 #include <list>
 
+#ifdef MY_GATEWAY_MQTT_CLIENT
+	#include <mosquitto.h>
+	
+	#ifndef MQTT_IP
+		#define MQTT_IP "127.0.0.1"
+	#endif
+	#ifndef MQTT_PORT
+		#define MQTT_PORT 1883
+	#endif
+	#ifndef MQTT_KEEPALIVE
+		#define MQTT_KEEPALIVE 60
+	#endif
+	#ifndef MY_MQTT_PUBLISH_TOPIC_PREFIX
+		#define MY_MQTT_PUBLISH_TOPIC_PREFIX "mygateway1-out"
+	#endif
+	#ifndef MY_MQTT_SUBSCRIBE_TOPIC_PREFIX
+		#define MY_MQTT_SUBSCRIBE_TOPIC_PREFIX "mygateway1-in"
+	#endif
+#endif
+
 //TODO
 #ifdef MY_USE_UDP
 	#error UDP not supported for this type of gateway
@@ -49,10 +69,19 @@ void *get_in_addr(struct sockaddr *sa);
 void *waiting_controllers(void* thread_arg);
 void *connected_controller(void* thread_arg);
 
+#ifdef MY_GATEWAY_MQTT_CLIENT
+void *mqtt_thread(void *);
+void parsemqtt_and_send(char *topic, const char *payload);
+struct mosquitto *mosq;
+#endif
+
 bool gatewayTransportInit() {
 	long int sockfd;
 	pthread_t thread_id;
 	pthread_attr_t attr;
+#ifdef MY_GATEWAY_MQTT_CLIENT
+	pthread_attr_t mqtt_attr;
+#endif
 	char *ip = NULL;
 
 	// ignore SIGPIPE generated by send() to a disconnected client
@@ -84,6 +113,14 @@ bool gatewayTransportInit() {
 	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 	pthread_create(&thread_id, &attr, &waiting_controllers, (void *)sockfd);
 	pthread_attr_destroy(&attr);
+	
+#ifdef MY_GATEWAY_MQTT_CLIENT
+	/* mqtt thread */
+	pthread_attr_init(&mqtt_attr);
+	pthread_attr_setdetachstate(&mqtt_attr, PTHREAD_CREATE_DETACHED);
+	pthread_create(&thread_id, &mqtt_attr, &mqtt_thread, NULL);
+	pthread_attr_destroy(&mqtt_attr);
+#endif
 
 	return true;
 }
@@ -98,6 +135,13 @@ bool gatewayTransportSend(MyMessage &message)
 		if (send(controllers[i], ethernetMsg, strlen(ethernetMsg), 0) == -1)
 			perror("send");
 	}
+	
+#ifdef MY_GATEWAY_MQTT_CLIENT
+	char *mqttMsg = protocolFormatMQTTTopic(MY_MQTT_PUBLISH_TOPIC_PREFIX, message);
+	//connection, message id, topic, bytes, data, QOS, retain
+	mosquitto_publish(mosq, NULL, mqttMsg, strlen(message.getString(_convBuffer)), message.getString(_convBuffer), 0, false);
+#endif
+	
 	return true;
 }
 
@@ -302,3 +346,177 @@ void *connected_controller(void* thread_arg)
 	close(sockfd);
 	return NULL;
 }
+
+#ifdef MY_GATEWAY_MQTT_CLIENT
+void mqtt_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
+{
+	(void)mosq;
+	(void)userdata;
+	
+	if(message->payloadlen){
+		printf("Got a MQTT message %s %s\n", message->topic, (char*)message->payload);
+		
+		std::string strpayload((char *)message->payload);
+		strpayload = strpayload.substr(0,(int)message->payloadlen);
+	
+		parsemqtt_and_send(message->topic, strpayload.c_str());
+	}else{
+		printf("%s (null)\n", message->topic);
+	}
+	fflush(stdout);
+}
+
+void mqtt_connect_callback(struct mosquitto *mosq, void *userdata, int result)
+{
+	(void)mosq;
+	(void)userdata;
+	
+	if(!result){
+		printf("Connected!\n");
+		/* Subscribe to broker information topics on successful connect. */
+		//connection, message id, topic, qos
+		mosquitto_subscribe(mosq, NULL, protocolFormatMQTTSubscribe(MY_MQTT_SUBSCRIBE_TOPIC_PREFIX), 0);
+	}else{
+		fprintf(stderr, "Connect failed\n");
+	}
+}
+
+void mqtt_subscribe_callback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
+{	
+	(void)mosq;
+	(void)userdata;
+
+	int i;
+
+	printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
+	for(i=1; i<qos_count; i++){
+		printf(", %d", granted_qos[i]);
+	}
+	printf("\n");
+}
+
+void mqtt_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
+{
+	(void)mosq;
+	(void)userdata;
+	(void)level;
+	/* Pring all log messages regardless of level. */
+	printf("Log: %s\n", str);
+}
+
+void *mqtt_thread(void *)
+{
+	mosquitto_lib_init();
+	bool clean_session = true; //will clean out subscriptions  on disconnect
+	mosq = mosquitto_new(NULL, clean_session, NULL);
+	if(!mosq){
+		fprintf(stderr, "Error: Out of memory.\n");
+		exit(1);
+	}
+
+	mosquitto_log_callback_set(mosq, mqtt_log_callback);
+	mosquitto_connect_callback_set(mosq, mqtt_connect_callback);
+	mosquitto_message_callback_set(mosq, mqtt_message_callback);
+	mosquitto_subscribe_callback_set(mosq, mqtt_subscribe_callback);
+
+	if(mosquitto_connect(mosq, MQTT_IP, MQTT_PORT, MQTT_KEEPALIVE)){
+		fprintf(stderr, "Unable to connect.\n");
+		exit(1);
+	}
+
+	mosquitto_loop_forever(mosq, -1, 1);
+
+	mosquitto_destroy(mosq);
+	mosquitto_lib_cleanup();
+
+	return NULL;
+}
+
+void parsemqtt_and_send(char *topic, const char *payload)
+{
+	//Buffer for parsemqtt_and_send.  Do not want to re-create it every time we get a message
+	MyMessage msg;
+	char *str, *p;
+
+	char* topic_copy = strdup(topic);
+	
+
+	// TODO: Check if we should send ack or not.
+	int i = 0;
+	for (str = strtok_r(topic_copy,"/",&p) ; str && i<4 ; str = strtok_r(NULL,"/",&p))
+	{
+		// Example: /mySensors/105/4/V_LIGHT
+		// 0: mySensors - Prefix
+		// 1: 105 - node
+		// 2: 4 - sensor
+		// 3: V_LIGHT - type
+		if (i == 0) {
+			//TODO: Add warning when we receive a message from an unexpected broker prefix.
+			//if (strcmp_P(str,broker)!=0) {  //look for MQTT_BROKER_PREFIX
+			//	return;                 //Message not for us or malformatted!
+			//}
+		} else if (i==1) {
+			printf("Destination parse %s\n", str);
+			msg.destination = atoi(str);    //NodeID
+		} else if (i==2) {
+			msg.sensor = atoi(str);         //SensorID
+		} else if (i==3) {
+			unsigned char match=255;         //SensorType
+			
+			//Support for numeric types
+			if ( atoi(str)!=0 || (str[0]=='0' && str[1] =='\0') ) {
+				match=atoi(str);
+			}
+			
+			/*
+			//Check through all the data type string attary to find a match.
+			//? Not sure if there is any need to be checking for other types like sensor and internal
+			//? as those don't seem like they should be outside settable thought MQTT.
+			for (uint8_t j=0; match == 255 && j<sizeof(mysensor_data_str)/sizeof(mysensor_data_str[0]); j++) {
+				if (strcmp(str,mysensor_data_str[j])==0) { //compare sensors
+					match=j;
+				}
+			}
+			*/
+			
+			msg.type = match;
+		} else if (i==4) {
+			mSetAck(msg, atoi(str)); //Ack
+		}
+		
+		i++;
+	}
+	
+	msg.sender = GATEWAY_ADDRESS;
+	msg.last = GATEWAY_ADDRESS;
+	mSetCommand(msg, C_SET);
+	mSetRequestAck(msg, 0);
+	msg.set(payload);
+	
+	if(msg.destination != 0 && msg.sensor != 0 && msg.type != 255)
+	{
+		ethernetMsg_q.push_back(msg);
+			printf("Message not sent!\n");
+
+		// Forward the data to Ethernet
+		// Likely this is a duplicate from a C_SET that we received and published
+		// in msg_callback.  Not much we can do to avoid the duplicate as there is 
+		// no way to tell if we performed the publish that is triggering  this callback.
+		char *ethernetMsg = protocolFormat(msg);
+
+		for (uint8_t i = 0; i < MY_GATEWAY_MAX_CLIENTS; i++) {
+			if (controllers[i] == -1)
+				continue;
+			if (send(controllers[i], ethernetMsg, strlen(ethernetMsg), 0) == -1)
+				perror("send");
+		}
+	}	
+	else
+	{
+		printf("Recieved a bad MQTT message: '%s':'%s'\n destination:%i, sensor:%i, type:%i\n", 
+		topic, payload, msg.destination, msg.sensor, msg.type);
+	}
+		
+	free(topic_copy);
+}
+#endif

libraries/MySensors/examples_RPi/PiGateway.cpp

@@ -39,6 +39,19 @@
 // How many clients should be able to connect to this gateway (default 1)
 #define MY_GATEWAY_MAX_CLIENTS 2
 
+/* MQTT BUILD */
+#ifdef MY_GATEWAY_MQTT_CLIENT
+//Linux MQTT Defines - Uncomment and modify as needed.  Written as defaults.
+//#define MQTT_IP "127.0.0.1"
+//#define MQTT_PORT 1883
+//#define MQTT_KEEPALIVE 60
+// Enable these if your MQTT broker requires usenrame/password
+//#define MY_MQTT_USER "username"
+//#define MY_MQTT_PASSWORD "password"
+//#define MY_MQTT_PUBLISH_TOPIC_PREFIX "mygateway1-out"
+//#define MY_MQTT_SUBSCRIBE_TOPIC_PREFIX "mygateway1-in"
+#endif
+
 #include <MySensor.h>
 
 using namespace std;