diff --git a/libraries/MySensors/Makefile b/libraries/MySensors/Makefile index 20a46ff..03e671d 100644 --- a/libraries/MySensors/Makefile +++ b/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 diff --git a/libraries/MySensors/core/MyGatewayTransportEthernetLinux.cpp b/libraries/MySensors/core/MyGatewayTransportEthernetLinux.cpp index e522828..4eadf49 100644 --- a/libraries/MySensors/core/MyGatewayTransportEthernetLinux.cpp +++ b/libraries/MySensors/core/MyGatewayTransportEthernetLinux.cpp @@ -27,6 +27,26 @@ #include #include +#ifdef MY_GATEWAY_MQTT_CLIENT + #include + + #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 using namespace std;