diff --git a/libraries/MySensors/MySensors.h b/libraries/MySensors/MySensors.h index 8f7b53a..189990e 100644 --- a/libraries/MySensors/MySensors.h +++ b/libraries/MySensors/MySensors.h @@ -200,7 +200,7 @@ #error You must define a unique MY_MQTT_CLIENT_ID for this MQTT client #endif - #include "drivers/pubsubclient/src/PubSubClient.cpp" + #include "drivers/PubSubClient/PubSubClient.cpp" #include "core/MyGatewayTransport.cpp" #include "core/MyGatewayTransportMQTTClient.cpp" #elif defined(MY_GATEWAY_FEATURE) diff --git a/libraries/MySensors/core/MyGatewayTransportEthernet.cpp b/libraries/MySensors/core/MyGatewayTransportEthernet.cpp index 84a4767..2db3b5a 100644 --- a/libraries/MySensors/core/MyGatewayTransportEthernet.cpp +++ b/libraries/MySensors/core/MyGatewayTransportEthernet.cpp @@ -1,4 +1,4 @@ -/** +/* * The MySensors Arduino library handles the wireless radio link and protocol * between your home built sensors/actuators and HA controller of choice. * The sensors forms a self healing radio network with optional repeaters. Each @@ -47,8 +47,8 @@ typedef struct #define EthernetUDP WiFiUDP #if defined(MY_IP_ADDRESS) - IPAddress gateway(MY_IP_GATEWAY_ADDRESS); - IPAddress subnet(MY_IP_SUBNET_ADDRESS); + IPAddress _gatewayIp(MY_IP_GATEWAY_ADDRESS); + IPAddress _subnetIp(MY_IP_SUBNET_ADDRESS); #endif static bool clientsConnected[MY_GATEWAY_MAX_CLIENTS]; #endif @@ -108,7 +108,7 @@ bool gatewayTransportInit() { #endif (void)WiFi.begin(MY_ESP8266_SSID, MY_ESP8266_PASSWORD); #ifdef MY_IP_ADDRESS - WiFi.config(_ethernetGatewayIP, gateway, subnet); + WiFi.config(_ethernetGatewayIP, _gatewayIp, _subnetIp); #endif while (WiFi.status() != WL_CONNECTED) { @@ -123,16 +123,17 @@ bool gatewayTransportInit() { #else #ifdef MY_IP_ADDRESS Ethernet.begin(_ethernetGatewayMAC, _ethernetGatewayIP); - MY_SERIALDEVICE.print(F("IP: ")); - MY_SERIALDEVICE.println(Ethernet.localIP()); #else // Get IP address from DHCP - Ethernet.begin(_ethernetGatewayMAC); - MY_SERIALDEVICE.print(F("IP: ")); - MY_SERIALDEVICE.println(Ethernet.localIP()); - #endif /* IP_ADDRESS_DHCP */ + if (!Ethernet.begin(_ethernetGatewayMAC)) { + MY_SERIALDEVICE.print("DHCP FAILURE..."); + _w5100_spi_en(false); + return false; + } + #endif + MY_SERIALDEVICE.print(F("IP: ")); + MY_SERIALDEVICE.println(Ethernet.localIP()); // give the Ethernet interface a second to initialize - // TODO: use HW delay delay(1000); #endif @@ -277,9 +278,9 @@ bool gatewayTransportAvailable() return protocolParse(_ethernetMsg, inputString[0].string); #else _ethernetServer.read(inputString.string, MY_GATEWAY_MAX_RECEIVE_LENGTH); - _w5100_spi_en(false); inputString.string[packet_size] = 0; debug(PSTR("UDP packet received: %s\n"), inputString.string); + _w5100_spi_en(false); return protocolParse(_ethernetMsg, inputString.string); #endif } @@ -299,9 +300,7 @@ bool gatewayTransportAvailable() clients[i] = _ethernetServer.available(); inputString[i].idx = 0; debug(PSTR("Client %d connected\n"), i); - _w5100_spi_en(false); gatewayTransportSend(buildGw(_msg, I_GATEWAY_READY).set("Gateway startup complete.")); - _w5100_spi_en(true); if (presentation) presentation(); } @@ -335,6 +334,7 @@ bool gatewayTransportAvailable() debug(PSTR("Eth: connect\n")); _w5100_spi_en(false); gatewayTransportSend(buildGw(_msg, I_GATEWAY_READY).set("Gateway startup complete.")); + _w5100_spi_en(true); if (presentation) presentation(); } diff --git a/libraries/MySensors/core/MyGatewayTransportMQTTClient.cpp b/libraries/MySensors/core/MyGatewayTransportMQTTClient.cpp index 2ab0308..f6c36e7 100644 --- a/libraries/MySensors/core/MyGatewayTransportMQTTClient.cpp +++ b/libraries/MySensors/core/MyGatewayTransportMQTTClient.cpp @@ -1,235 +1,49 @@ -/** - * The MySensors Arduino library handles the wireless radio link and protocol - * between your home built sensors/actuators and HA controller of choice. - * The sensors forms a self healing radio network with optional repeaters. Each - * repeater and gateway builds a routing tables in EEPROM which keeps track of the - * network topology allowing messages to be routed to nodes. - * - * Created by Henrik Ekblad - * Copyright (C) 2013-2015 Sensnology AB - * Full contributor list: https://github.com/mysensors/Arduino/graphs/contributors - * - * Documentation: http://www.mysensors.org - * Support Forum: http://forum.mysensors.org - * - * This program is free software; you can redistribute it and/or - * modify it under the terms of the GNU General Public License - * version 2 as published by the Free Software Foundation. - */ +/* +* The MySensors Arduino library handles the wireless radio link and protocol +* between your home built sensors/actuators and HA controller of choice. +* The sensors forms a self healing radio network with optional repeaters. Each +* repeater and gateway builds a routing tables in EEPROM which keeps track of the +* network topology allowing messages to be routed to nodes. +* +* Created by Henrik Ekblad +* Copyright (C) 2013-2015 Sensnology AB +* Full contributor list: https://github.com/mysensors/Arduino/graphs/contributors +* +* Documentation: http://www.mysensors.org +* Support Forum: http://forum.mysensors.org +* +* This program is free software; you can redistribute it and/or +* modify it under the terms of the GNU General Public License +* version 2 as published by the Free Software Foundation. +*/ -#include "MyConfig.h" -#include "MyProtocol.h" -#include "MyGatewayTransport.h" -#include "MyMessage.h" - // Topic structure: MY_MQTT_PUBLISH_TOPIC_PREFIX/NODE-ID/SENSOR-ID/CMD-TYPE/ACK-FLAG/SUB-TYPE -uint8_t protocolH2i(char c); - #if defined MY_CONTROLLER_IP_ADDRESS - IPAddress _brokerIp(MY_CONTROLLER_IP_ADDRESS); + IPAddress _brokerIp(MY_CONTROLLER_IP_ADDRESS); #endif #if defined(MY_GATEWAY_ESP8266) #define EthernetClient WiFiClient #if defined(MY_IP_ADDRESS) - IPAddress _gatewayIp(MY_IP_GATEWAY_ADDRESS); - IPAddress _subnetIp(MY_IP_SUBNET_ADDRESS); - #endif + IPAddress _gatewayIp(MY_IP_GATEWAY_ADDRESS); + IPAddress _subnetIp(MY_IP_SUBNET_ADDRESS); + #endif #else - byte _clientMAC[] = { MY_MAC_ADDRESS }; + byte _MQTT_clientMAC[] = { MY_MAC_ADDRESS }; #endif #if defined(MY_IP_ADDRESS) - IPAddress _clientIp(MY_IP_ADDRESS); + IPAddress _MQTT_clientIp(MY_IP_ADDRESS); #endif -EthernetClient _ethClient; -PubSubClient _client(_ethClient); -bool _connecting = true; -bool _available = false; -char _convBuffer[MAX_PAYLOAD*2+1]; -char _fmtBuffer[MY_GATEWAY_MAX_SEND_LENGTH]; -MyMessage _mqttMsg; - - - -bool gatewayTransportSend(MyMessage &message) { - if (!_client.connected()) - return false; - setIndication(INDICATION_GW_TX); - - snprintf_P(_fmtBuffer, MY_GATEWAY_MAX_SEND_LENGTH, PSTR(MY_MQTT_PUBLISH_TOPIC_PREFIX "/%d/%d/%d/%d/%d"), message.sender, message.sensor, mGetCommand(message), mGetAck(message), message.type); - debug(PSTR("Sending message on topic: %s\n"), _fmtBuffer); - return _client.publish(_fmtBuffer, message.getString(_convBuffer)); -} - - - -void incomingMQTT(char* topic, byte* payload, - unsigned int length) -{ - debug(PSTR("Message arrived on topic: %s\n"), topic); - char *str, *p; - uint8_t i = 0; - uint8_t bvalue[MAX_PAYLOAD]; - uint8_t blen = 0; - uint8_t command = 0; - for (str = strtok_r(topic, "/", &p); str && i <= 5; - str = strtok_r(NULL, "/", &p)) { - switch (i) { - case 0: { - // Topic prefix - if (strcmp(str, MY_MQTT_SUBSCRIBE_TOPIC_PREFIX) != 0) { - // Message not for us or malformed! - return; - } - break; - } - case 1: { - // Node id - _mqttMsg.destination = atoi(str); - break; - } - case 2: { - // Sensor id - _mqttMsg.sensor = atoi(str); - break; - } - case 3: { - // Command type - command = atoi(str); - mSetCommand(_mqttMsg, command); - break; - } - case 4: { - // Ack flag - mSetRequestAck(_mqttMsg, atoi(str)?1:0); - break; - } - case 5: { - // Sub type - _mqttMsg.type = atoi(str); - // Add payload - if (command == C_STREAM) { - blen = 0; - uint8_t val; - while (*payload) { - val = protocolH2i(*payload++) << 4; - val += protocolH2i(*payload++); - bvalue[blen] = val; - blen++; - } - _mqttMsg.set(bvalue, blen); - } else { - char* ca; - ca = (char *) payload; - ca += length; - *ca = '\0'; - _mqttMsg.set((const char*) payload); - } - _available = true; - } - } - i++; - } -} - - -bool reconnectMQTT() { - debug(PSTR("Attempting MQTT connection...\n")); - // Attempt to connect - if (_client.connect(MY_MQTT_CLIENT_ID - #if defined(MY_MQTT_USER) && defined(MY_MQTT_PASSWORD) - , MY_MQTT_USER, MY_MQTT_PASSWORD - #endif - )) { - debug(PSTR("MQTT connected\n")); - // Once connected, publish an announcement... - //_client.publish("outTopic","hello world"); - // ... and resubscribe - _client.subscribe(MY_MQTT_SUBSCRIBE_TOPIC_PREFIX "/+/+/+/+/+"); - return true; - } - return false; -} - -bool gatewayTransportInit() { - _connecting = true; - #if defined(MY_CONTROLLER_IP_ADDRESS) - _client.setServer(_brokerIp, MY_PORT); - #else - _client.setServer(MY_CONTROLLER_URL_ADDRESS, MY_PORT); - #endif - - _client.setCallback(incomingMQTT); - - #if defined(MY_GATEWAY_ESP8266) - // Turn off access point - WiFi.mode (WIFI_STA); - #if defined(MY_ESP8266_HOSTNAME) - WiFi.hostname(MY_ESP8266_HOSTNAME); - #endif - (void)WiFi.begin(MY_ESP8266_SSID, MY_ESP8266_PASSWORD); - #ifdef MY_IP_ADDRESS - WiFi.config(_clientIp, _gatewayIp, _subnetIp); - #endif - while (WiFi.status() != WL_CONNECTED) - { - delay(500); - MY_SERIALDEVICE.print("."); - yield(); - } - MY_SERIALDEVICE.print("IP: "); - MY_SERIALDEVICE.println(WiFi.localIP()); - #else - #ifdef MY_IP_ADDRESS - Ethernet.begin(_clientMAC, _clientIp); - #else - // Get IP address from DHCP - if (!Ethernet.begin(_clientMAC)) - { - MY_SERIALDEVICE.print("DHCP FAILURE..."); - _connecting = false; - return false; - } - MY_SERIALDEVICE.print("IP: "); - MY_SERIALDEVICE.println(Ethernet.localIP()); - #endif - - // give the Ethernet interface a second to initialize - // TODO: use HW delay - wait(1000); - #endif - _connecting = false; - return true; -} - - -bool gatewayTransportAvailable() { - if (_connecting) - return false; - - //keep lease on dhcp address - //Ethernet.maintain(); - if (!_client.connected()) { - //reinitialise client - if (gatewayTransportInit()) - reconnectMQTT(); - return false; - } - _client.loop(); - return _available; -} - -MyMessage & gatewayTransportReceive() { - // Return the last parsed message - _available = false; - return _mqttMsg; -} - +static EthernetClient _MQTT_ethClient; +static PubSubClient _MQTT_client(_MQTT_ethClient); +static bool _MQTT_connecting = true; +static bool _MQTT_available = false; +static MyMessage _MQTT_msg; uint8_t protocolH2i(char c) { uint8_t i = 0; @@ -241,3 +55,185 @@ uint8_t protocolH2i(char c) { i += c - 'A' + 10; return i; } + + +bool gatewayTransportSend(MyMessage &message) { + if (!_MQTT_client.connected()) + return false; + setIndication(INDICATION_GW_TX); + char _fmtBuffer[MY_GATEWAY_MAX_SEND_LENGTH]; + char _convBuffer[MAX_PAYLOAD * 2 + 1]; + snprintf_P(_fmtBuffer, MY_GATEWAY_MAX_SEND_LENGTH, PSTR(MY_MQTT_PUBLISH_TOPIC_PREFIX "/%d/%d/%d/%d/%d"), message.sender, message.sensor, mGetCommand(message), mGetAck(message), message.type); + debug(PSTR("Sending message on topic: %s\n"), _fmtBuffer); + return _MQTT_client.publish(_fmtBuffer, message.getString(_convBuffer)); +} + +void incomingMQTT(char* topic, byte* payload, unsigned int length) { + debug(PSTR("Message arrived on topic: %s\n"), topic); + char *str, *p; + uint8_t i = 0; + uint8_t bvalue[MAX_PAYLOAD]; + uint8_t blen = 0; + uint8_t command = 0; + for (str = strtok_r(topic, "/", &p); str && i <= 5; + str = strtok_r(NULL, "/", &p)) { + switch (i) { + case 0: { + // Topic prefix + if (strcmp(str, MY_MQTT_SUBSCRIBE_TOPIC_PREFIX) != 0) { + // Message not for us or malformed! + return; + } + break; + } + case 1: { + // Node id + _MQTT_msg.destination = atoi(str); + break; + } + case 2: { + // Sensor id + _MQTT_msg.sensor = atoi(str); + break; + } + case 3: { + // Command type + command = atoi(str); + mSetCommand(_MQTT_msg, command); + break; + } + case 4: { + // Ack flag + mSetRequestAck(_MQTT_msg, atoi(str) ? 1 : 0); + break; + } + case 5: { + // Sub type + _MQTT_msg.type = atoi(str); + // Add payload + if (command == C_STREAM) { + blen = 0; + uint8_t val; + while (*payload) { + val = protocolH2i(*payload++) << 4; + val += protocolH2i(*payload++); + bvalue[blen] = val; + blen++; + } + _MQTT_msg.set(bvalue, blen); + } + else { + char* ca; + ca = (char *)payload; + ca += length; + *ca = '\0'; + _MQTT_msg.set((const char*)payload); + } + _MQTT_available = true; + } + } + i++; + } +} + + +bool reconnectMQTT() { + debug(PSTR("Attempting MQTT connection...\n")); + // Attempt to connect + if (_MQTT_client.connect(MY_MQTT_CLIENT_ID +#if defined(MY_MQTT_USER) && defined(MY_MQTT_PASSWORD) + , MY_MQTT_USER, MY_MQTT_PASSWORD +#endif + )) { + debug(PSTR("MQTT connected\n")); + // Once connected, publish an announcement... + //_MQTT_client.publish("outTopic","hello world"); + // ... and resubscribe + _MQTT_client.subscribe(MY_MQTT_SUBSCRIBE_TOPIC_PREFIX "/+/+/+/+/+"); + return true; + } + return false; +} + +bool gatewayTransportConnect() { + #if defined(MY_GATEWAY_ESP8266) + while (WiFi.status() != WL_CONNECTED) { + delay(500); // delay calls yield + MY_SERIALDEVICE.print("."); + } + MY_SERIALDEVICE.print("IP: "); + MY_SERIALDEVICE.println(WiFi.localIP()); + #else + #ifdef MY_IP_ADDRESS + Ethernet.begin(_MQTT_clientMAC, _MQTT_clientIp); + #else + // Get IP address from DHCP + if (!Ethernet.begin(_MQTT_clientMAC)) { + MY_SERIALDEVICE.print("DHCP FAILURE..."); + _MQTT_connecting = false; + return false; + } + #endif + MY_SERIALDEVICE.print("IP: "); + MY_SERIALDEVICE.println(Ethernet.localIP()); + // give the Ethernet interface a second to initialize + delay(1000); + #endif + return true; +} + +bool gatewayTransportInit() { + _MQTT_connecting = true; + #if defined(MY_CONTROLLER_IP_ADDRESS) + _MQTT_client.setServer(_brokerIp, MY_PORT); + #else + _MQTT_client.setServer(MY_CONTROLLER_URL_ADDRESS, MY_PORT); + #endif + + _MQTT_client.setCallback(incomingMQTT); + + #if defined(MY_GATEWAY_ESP8266) + // Turn off access point + WiFi.mode(WIFI_STA); + #if defined(MY_ESP8266_HOSTNAME) + WiFi.hostname(MY_ESP8266_HOSTNAME); + #endif + (void)WiFi.begin(MY_ESP8266_SSID, MY_ESP8266_PASSWORD); + #ifdef MY_IP_ADDRESS + WiFi.config(_MQTT_clientIp, _gatewayIp, _subnetIp); + #endif + #endif + + gatewayTransportConnect(); + + _MQTT_connecting = false; + return true; +} + + + +bool gatewayTransportAvailable() { + + if (_MQTT_connecting) + return false; + + //keep lease on dhcp address + //Ethernet.maintain(); + if (!_MQTT_client.connected()) { + //reinitialise client + if (gatewayTransportConnect()) + reconnectMQTT(); + return false; + } + _MQTT_client.loop(); + return _MQTT_available; +} + +MyMessage & gatewayTransportReceive() { + // Return the last parsed message + _MQTT_available = false; + return _MQTT_msg; +} + + + diff --git a/libraries/MySensors/drivers/pubsubclient/src/PubSubClient.cpp b/libraries/MySensors/drivers/PubSubClient/PubSubClient.cpp similarity index 92% rename from libraries/MySensors/drivers/pubsubclient/src/PubSubClient.cpp rename to libraries/MySensors/drivers/PubSubClient/PubSubClient.cpp index 29f70fc..e66c9e0 100644 --- a/libraries/MySensors/drivers/pubsubclient/src/PubSubClient.cpp +++ b/libraries/MySensors/drivers/PubSubClient/PubSubClient.cpp @@ -122,7 +122,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass } else { result = _client->connect(this->ip, this->port); } - if (result) { + if (result == 1) { nextMsgId = 1; // Leave room in the buffer for header and variable length field uint16_t length = 5; @@ -177,7 +177,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass while (!_client->available()) { unsigned long t = millis(); - if (t-lastInActivity > MQTT_KEEPALIVE*1000UL) { + if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) { _state = MQTT_CONNECTION_TIMEOUT; _client->stop(); return false; @@ -205,14 +205,33 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass return true; } -uint8_t PubSubClient::readByte() { - while(!_client->available()) {} - return _client->read(); +// reads a byte into result +boolean PubSubClient::readByte(uint8_t * result) { + uint32_t previousMillis = millis(); + while(!_client->available()) { + uint32_t currentMillis = millis(); + if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){ + return false; + } + } + *result = _client->read(); + return true; +} + +// reads a byte into result[*index] and increments index +boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){ + uint16_t current_index = *index; + uint8_t * write_address = &(result[current_index]); + if(readByte(write_address)){ + *index = current_index + 1; + return true; + } + return false; } uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { uint16_t len = 0; - buffer[len++] = readByte(); + if(!readByte(buffer, &len)) return 0; bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; uint32_t multiplier = 1; uint16_t length = 0; @@ -221,7 +240,7 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { uint8_t start = 0; do { - digit = readByte(); + if(!readByte(&digit)) return 0; buffer[len++] = digit; length += (digit & 127) * multiplier; multiplier *= 128; @@ -230,8 +249,8 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { if (isPublish) { // Read in topic length to calculate bytes to skip over for Stream writing - buffer[len++] = readByte(); - buffer[len++] = readByte(); + if(!readByte(buffer, &len)) return 0; + if(!readByte(buffer, &len)) return 0; skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; start = 2; if (buffer[0]&MQTTQOS1) { @@ -241,7 +260,7 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { } for (uint16_t i = start;istream) { if (isPublish && len-*lengthLength-2>skip) { this->stream->write(digit); @@ -409,7 +428,7 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) { uint8_t llen = 0; uint8_t digit; uint8_t pos = 0; - uint8_t rc; + uint16_t rc; uint16_t len = length; do { digit = len % 128; @@ -428,7 +447,7 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) { #ifdef MQTT_MAX_TRANSFER_SIZE uint8_t* writeBuf = buf+(4-llen); - uint8_t bytesRemaining = length+1+llen; + uint16_t bytesRemaining = length+1+llen; //Match the length type uint8_t bytesToWrite; boolean result = true; while((bytesRemaining > 0) && result) { @@ -451,7 +470,8 @@ boolean PubSubClient::subscribe(const char* topic) { } boolean PubSubClient::subscribe(const char* topic, uint8_t qos) { - if (qos > 1) { + // original: if (qos < 0 || qos > 1) { (qos is uint8_t, hence qos < 0 impossible, tekka) + if (qos > 1) { return false; } if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { @@ -551,7 +571,7 @@ PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port) { return *this; } -PubSubClient& PubSubClient::setCallback(void(*callback)(char*,uint8_t*,unsigned int)){ +PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE) { this->callback = callback; return *this; } diff --git a/libraries/MySensors/drivers/pubsubclient/src/PubSubClient.h b/libraries/MySensors/drivers/PubSubClient/PubSubClient.h similarity index 91% rename from libraries/MySensors/drivers/pubsubclient/src/PubSubClient.h rename to libraries/MySensors/drivers/PubSubClient/PubSubClient.h index 76d76cc..90ad8ae 100644 --- a/libraries/MySensors/drivers/pubsubclient/src/PubSubClient.h +++ b/libraries/MySensors/drivers/PubSubClient/PubSubClient.h @@ -17,13 +17,24 @@ // MQTT_VERSION : Pick the version //#define MQTT_VERSION MQTT_VERSION_3_1 +#ifndef MQTT_VERSION #define MQTT_VERSION MQTT_VERSION_3_1_1 +#endif // MQTT_MAX_PACKET_SIZE : Maximum packet size +#ifndef MQTT_MAX_PACKET_SIZE #define MQTT_MAX_PACKET_SIZE 128 +#endif // MQTT_KEEPALIVE : keepAlive interval in Seconds +#ifndef MQTT_KEEPALIVE #define MQTT_KEEPALIVE 15 +#endif + +// MQTT_SOCKET_TIMEOUT: socket timeout interval in Seconds +#ifndef MQTT_SOCKET_TIMEOUT +#define MQTT_SOCKET_TIMEOUT 15 +#endif // MQTT_MAX_TRANSFER_SIZE : limit how much data is passed to the network client // in each write call. Needed for the Arduino Wifi Shield. Leave undefined to @@ -62,7 +73,12 @@ #define MQTTQOS1 (1 << 1) #define MQTTQOS2 (2 << 1) -#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*,uint8_t*,unsigned int) +#ifdef ESP8266 +#include +#define MQTT_CALLBACK_SIGNATURE std::function callback +#else +#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int) +#endif /** PubSubClient class */ class PubSubClient { @@ -75,7 +91,8 @@ private: bool pingOutstanding; MQTT_CALLBACK_SIGNATURE; uint16_t readPacket(uint8_t*); - uint8_t readByte(); + boolean readByte(uint8_t * result); + boolean readByte(uint8_t * result, uint16_t * index); boolean write(uint8_t header, uint8_t* buf, uint16_t length); uint16_t writeString(const char* string, uint8_t* buf, uint16_t pos); IPAddress ip; diff --git a/libraries/MySensors/drivers/pubsubclient/CHANGES.txt b/libraries/MySensors/drivers/pubsubclient/CHANGES.txt index 5202dda..8c8bef6 100644 --- a/libraries/MySensors/drivers/pubsubclient/CHANGES.txt +++ b/libraries/MySensors/drivers/pubsubclient/CHANGES.txt @@ -1,3 +1,8 @@ +2.4 + * Add MQTT_SOCKET_TIMEOUT to prevent it blocking indefinitely + whilst waiting for inbound data + * Fixed return code when publishing >256 bytes + 2.3 * Add publish(topic,payload,retained) function diff --git a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_auth/mqtt_auth.ino b/libraries/MySensors/drivers/pubsubclient/examples/mqtt_auth/mqtt_auth.ino deleted file mode 100644 index e9f7b18..0000000 --- a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_auth/mqtt_auth.ino +++ /dev/null @@ -1,43 +0,0 @@ -/* - Basic MQTT example with Authentication - - - connects to an MQTT server, providing username - and password - - publishes "hello world" to the topic "outTopic" - - subscribes to the topic "inTopic" -*/ - -#include -#include -#include - -// Update these with values suitable for your network. -byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -IPAddress ip(172, 16, 0, 100); -IPAddress server(172, 16, 0, 2); - -void callback(char* topic, byte* payload, unsigned int length) { - // handle message arrived -} - -EthernetClient ethClient; -PubSubClient client(server, 1883, callback, ethClient); - -void setup() -{ - Ethernet.begin(mac, ip); - // Note - the default maximum packet size is 128 bytes. If the - // combined length of clientId, username and password exceed this, - // you will need to increase the value of MQTT_MAX_PACKET_SIZE in - // PubSubClient.h - - if (client.connect("arduinoClient", "testuser", "testpass")) { - client.publish("outTopic","hello world"); - client.subscribe("inTopic"); - } -} - -void loop() -{ - client.loop(); -} diff --git a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_basic/mqtt_basic.ino b/libraries/MySensors/drivers/pubsubclient/examples/mqtt_basic/mqtt_basic.ino deleted file mode 100644 index f545ade..0000000 --- a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_basic/mqtt_basic.ino +++ /dev/null @@ -1,77 +0,0 @@ -/* - Basic MQTT example - - This sketch demonstrates the basic capabilities of the library. - It connects to an MQTT server then: - - publishes "hello world" to the topic "outTopic" - - subscribes to the topic "inTopic", printing out any messages - it receives. NB - it assumes the received payloads are strings not binary - - It will reconnect to the server if the connection is lost using a blocking - reconnect function. See the 'mqtt_reconnect_nonblocking' example for how to - achieve the same result without blocking the main loop. - -*/ - -#include -#include -#include - -// Update these with values suitable for your network. -byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -IPAddress ip(172, 16, 0, 100); -IPAddress server(172, 16, 0, 2); - -void callback(char* topic, byte* payload, unsigned int length) { - Serial.print("Message arrived ["); - Serial.print(topic); - Serial.print("] "); - for (int i=0;i -#include -#include - -// Update these with values suitable for your network. -byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -IPAddress ip(172, 16, 0, 100); -IPAddress server(172, 16, 0, 2); - -// Callback function header -void callback(char* topic, byte* payload, unsigned int length); - -EthernetClient ethClient; -PubSubClient client(server, 1883, callback, ethClient); - -// Callback function -void callback(char* topic, byte* payload, unsigned int length) { - // In order to republish this payload, a copy must be made - // as the orignal payload buffer will be overwritten whilst - // constructing the PUBLISH packet. - - // Allocate the correct amount of memory for the payload copy - byte* p = (byte*)malloc(length); - // Copy the payload to the new buffer - memcpy(p,payload,length); - client.publish("outTopic", p, length); - // Free the memory - free(p); -} - -void setup() -{ - - Ethernet.begin(mac, ip); - if (client.connect("arduinoClient")) { - client.publish("outTopic","hello world"); - client.subscribe("inTopic"); - } -} - -void loop() -{ - client.loop(); -} diff --git a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_stream/mqtt_stream.ino b/libraries/MySensors/drivers/pubsubclient/examples/mqtt_stream/mqtt_stream.ino deleted file mode 100644 index 67c2287..0000000 --- a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_stream/mqtt_stream.ino +++ /dev/null @@ -1,57 +0,0 @@ -/* - Example of using a Stream object to store the message payload - - Uses SRAM library: https://github.com/ennui2342/arduino-sram - but could use any Stream based class such as SD - - - connects to an MQTT server - - publishes "hello world" to the topic "outTopic" - - subscribes to the topic "inTopic" -*/ - -#include -#include -#include -#include - -// Update these with values suitable for your network. -byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -IPAddress ip(172, 16, 0, 100); -IPAddress server(172, 16, 0, 2); - -SRAM sram(4, SRAM_1024); - -void callback(char* topic, byte* payload, unsigned int length) { - sram.seek(1); - - // do something with the message - for(uint8_t i=0; i256 bytes + +2.3 + * Add publish(topic,payload,retained) function + +2.2 + * Change code layout to match Arduino Library reqs + +2.1 + * Add MAX_TRANSFER_SIZE def to chunk messages if needed + * Reject topic/payloads that exceed MQTT_MAX_PACKET_SIZE + +2.0 + * Add (and default to) MQTT 3.1.1 support + * Fix PROGMEM handling for Intel Galileo/ESP8266 + * Add overloaded constructors for convenience + * Add chainable setters for server/callback/client/stream + * Add state function to return connack return code + 1.9 * Do not split MQTT packets over multiple calls to _client->write() * API change: All constructors now require an instance of Client - to be passed in. + to be passed in. * Fixed example to match 1.8 api changes - dpslwk * Added username/password support - WilHall * Added publish_P - publishes messages from PROGMEM - jobytaffey @@ -10,7 +32,7 @@ * KeepAlive interval is configurable in PubSubClient.h * Maximum packet size is configurable in PubSubClient.h * API change: Return boolean rather than int from various functions - * API change: Length parameter in message callback changed + * API change: Length parameter in message callback changed from int to unsigned int * Various internal tidy-ups around types 1.7 @@ -44,4 +66,3 @@ * The maximum message size, including header, is 128 bytes * The keepalive interval is set to 30 seconds * No support for Will messages - diff --git a/libraries/PubSubClient/LICENSE.txt b/libraries/PubSubClient/LICENSE.txt index f512cba..217df35 100755 --- a/libraries/PubSubClient/LICENSE.txt +++ b/libraries/PubSubClient/LICENSE.txt @@ -1,4 +1,4 @@ -Copyright (c) 2008-2012 Nicholas O'Leary +Copyright (c) 2008-2015 Nicholas O'Leary Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the diff --git a/libraries/PubSubClient/PubSubClient.cpp b/libraries/PubSubClient/PubSubClient.cpp deleted file mode 100755 index dbb538a..0000000 --- a/libraries/PubSubClient/PubSubClient.cpp +++ /dev/null @@ -1,427 +0,0 @@ -/* - PubSubClient.cpp - A simple client for MQTT. - Nicholas O'Leary - http://knolleary.net -*/ - -#include "PubSubClient.h" -#include - -PubSubClient::PubSubClient() { - this->_client = NULL; - this->stream = NULL; -} - -PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client) { - this->_client = &client; - this->callback = callback; - this->ip = ip; - this->port = port; - this->domain = NULL; - this->stream = NULL; -} - -PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client) { - this->_client = &client; - this->callback = callback; - this->domain = domain; - this->port = port; - this->stream = NULL; -} - -PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client, Stream& stream) { - this->_client = &client; - this->callback = callback; - this->ip = ip; - this->port = port; - this->domain = NULL; - this->stream = &stream; -} - -PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client, Stream& stream) { - this->_client = &client; - this->callback = callback; - this->domain = domain; - this->port = port; - this->stream = &stream; -} - -boolean PubSubClient::connect(char *id) { - return connect(id,NULL,NULL,0,0,0,0); -} - -boolean PubSubClient::connect(char *id, char *user, char *pass) { - return connect(id,user,pass,0,0,0,0); -} - -boolean PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) -{ - return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage); -} - -boolean PubSubClient::connect(char *id, char *user, char *pass, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) { - if (!connected()) { - int result = 0; - - if (domain != NULL) { - result = _client->connect(this->domain, this->port); - } else { - result = _client->connect(this->ip, this->port); - } - - if (result) { - nextMsgId = 1; - uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p',MQTTPROTOCOLVERSION}; - // Leave room in the buffer for header and variable length field - uint16_t length = 5; - unsigned int j; - for (j = 0;j<9;j++) { - buffer[length++] = d[j]; - } - - uint8_t v; - if (willTopic) { - v = 0x06|(willQos<<3)|(willRetain<<5); - } else { - v = 0x02; - } - - if(user != NULL) { - v = v|0x80; - - if(pass != NULL) { - v = v|(0x80>>1); - } - } - - buffer[length++] = v; - - buffer[length++] = ((MQTT_KEEPALIVE) >> 8); - buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF); - length = writeString(id,buffer,length); - if (willTopic) { - length = writeString(willTopic,buffer,length); - length = writeString(willMessage,buffer,length); - } - - if(user != NULL) { - length = writeString(user,buffer,length); - if(pass != NULL) { - length = writeString(pass,buffer,length); - } - } - - write(MQTTCONNECT,buffer,length-5); - - lastInActivity = lastOutActivity = millis(); - - while (!_client->available()) { - unsigned long t = millis(); - if (t-lastInActivity > MQTT_KEEPALIVE*1000UL) { - _client->stop(); - return false; - } - } - uint8_t llen; - uint16_t len = readPacket(&llen); - - if (len == 4 && buffer[3] == 0) { - lastInActivity = millis(); - pingOutstanding = false; - return true; - } - } - _client->stop(); - } - return false; -} - -uint8_t PubSubClient::readByte() { - while(!_client->available()) {} - return _client->read(); -} - -uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { - uint16_t len = 0; - buffer[len++] = readByte(); - bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; - uint32_t multiplier = 1; - uint16_t length = 0; - uint8_t digit = 0; - uint16_t skip = 0; - uint8_t start = 0; - - do { - digit = readByte(); - buffer[len++] = digit; - length += (digit & 127) * multiplier; - multiplier *= 128; - } while ((digit & 128) != 0); - *lengthLength = len-1; - - if (isPublish) { - // Read in topic length to calculate bytes to skip over for Stream writing - buffer[len++] = readByte(); - buffer[len++] = readByte(); - skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; - start = 2; - if (buffer[0]&MQTTQOS1) { - // skip message id - skip += 2; - } - } - - for (uint16_t i = start;istream) { - if (isPublish && len-*lengthLength-2>skip) { - this->stream->write(digit); - } - } - if (len < MQTT_MAX_PACKET_SIZE) { - buffer[len] = digit; - } - len++; - } - - if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { - len = 0; // This will cause the packet to be ignored. - } - - return len; -} - -boolean PubSubClient::loop() { - if (connected()) { - unsigned long t = millis(); - if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) { - if (pingOutstanding) { - _client->stop(); - return false; - } else { - buffer[0] = MQTTPINGREQ; - buffer[1] = 0; - _client->write(buffer,2); - lastOutActivity = t; - lastInActivity = t; - pingOutstanding = true; - } - } - if (_client->available()) { - uint8_t llen; - uint16_t len = readPacket(&llen); - uint16_t msgId = 0; - uint8_t *payload; - if (len > 0) { - lastInActivity = t; - uint8_t type = buffer[0]&0xF0; - if (type == MQTTPUBLISH) { - if (callback) { - uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; - char topic[tl+1]; - for (uint16_t i=0;i0 - if ((buffer[0]&0x06) == MQTTQOS1) { - msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1]; - payload = buffer+llen+3+tl+2; - callback(topic,payload,len-llen-3-tl-2); - - buffer[0] = MQTTPUBACK; - buffer[1] = 2; - buffer[2] = (msgId >> 8); - buffer[3] = (msgId & 0xFF); - _client->write(buffer,4); - lastOutActivity = t; - - } else { - payload = buffer+llen+3+tl; - callback(topic,payload,len-llen-3-tl); - } - } - } else if (type == MQTTPINGREQ) { - buffer[0] = MQTTPINGRESP; - buffer[1] = 0; - _client->write(buffer,2); - } else if (type == MQTTPINGRESP) { - pingOutstanding = false; - } - } - } - return true; - } - return false; -} - -boolean PubSubClient::publish(char* topic, char* payload) { - return publish(topic,(uint8_t*)payload,strlen(payload),false); -} - -boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plength) { - return publish(topic, payload, plength, false); -} - -boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plength, boolean retained) { - if (connected()) { - // Leave room in the buffer for header and variable length field - uint16_t length = 5; - length = writeString(topic,buffer,length); - uint16_t i; - for (i=0;i 0) { - digit |= 0x80; - } - buffer[pos++] = digit; - llen++; - } while(len>0); - - pos = writeString(topic,buffer,pos); - - rc += _client->write(buffer,pos); - - for (i=0;iwrite((char)pgm_read_byte_near(payload + i)); - } - - lastOutActivity = millis(); - - return rc == tlen + 4 + plength; -} - -boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) { - uint8_t lenBuf[4]; - uint8_t llen = 0; - uint8_t digit; - uint8_t pos = 0; - uint8_t rc; - uint8_t len = length; - do { - digit = len % 128; - len = len / 128; - if (len > 0) { - digit |= 0x80; - } - lenBuf[pos++] = digit; - llen++; - } while(len>0); - - buf[4-llen] = header; - for (int i=0;iwrite(buf+(4-llen),length+1+llen); - - lastOutActivity = millis(); - return (rc == 1+llen+length); -} - -boolean PubSubClient::subscribe(char* topic) { - return subscribe(topic, 0); -} - -boolean PubSubClient::subscribe(char* topic, uint8_t qos) { - if (qos > 1) - return false; - - if (connected()) { - // Leave room in the buffer for header and variable length field - uint16_t length = 5; - nextMsgId++; - if (nextMsgId == 0) { - nextMsgId = 1; - } - buffer[length++] = (nextMsgId >> 8); - buffer[length++] = (nextMsgId & 0xFF); - length = writeString(topic, buffer,length); - buffer[length++] = qos; - return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5); - } - return false; -} - -boolean PubSubClient::unsubscribe(char* topic) { - if (connected()) { - uint16_t length = 5; - nextMsgId++; - if (nextMsgId == 0) { - nextMsgId = 1; - } - buffer[length++] = (nextMsgId >> 8); - buffer[length++] = (nextMsgId & 0xFF); - length = writeString(topic, buffer,length); - return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5); - } - return false; -} - -void PubSubClient::disconnect() { - buffer[0] = MQTTDISCONNECT; - buffer[1] = 0; - _client->write(buffer,2); - _client->stop(); - lastInActivity = lastOutActivity = millis(); -} - -uint16_t PubSubClient::writeString(char* string, uint8_t* buf, uint16_t pos) { - char* idp = string; - uint16_t i = 0; - pos += 2; - while (*idp) { - buf[pos++] = *idp++; - i++; - } - buf[pos-i-2] = (i >> 8); - buf[pos-i-1] = (i & 0xFF); - return pos; -} - - -boolean PubSubClient::connected() { - boolean rc; - if (_client == NULL ) { - rc = false; - } else { - rc = (int)_client->connected(); - if (!rc) _client->stop(); - } - return rc; -} - diff --git a/libraries/PubSubClient/PubSubClient.h b/libraries/PubSubClient/PubSubClient.h deleted file mode 100755 index f2faff8..0000000 --- a/libraries/PubSubClient/PubSubClient.h +++ /dev/null @@ -1,81 +0,0 @@ -/* - PubSubClient.h - A simple client for MQTT. - Nicholas O'Leary - http://knolleary.net -*/ - -#ifndef PubSubClient_h -#define PubSubClient_h - -#include -#include "Client.h" -#include "Stream.h" - -// MQTT_MAX_PACKET_SIZE : Maximum packet size -#define MQTT_MAX_PACKET_SIZE 128 - -// MQTT_KEEPALIVE : keepAlive interval in Seconds -#define MQTT_KEEPALIVE 15 - -#define MQTTPROTOCOLVERSION 3 -#define MQTTCONNECT 1 << 4 // Client request to connect to Server -#define MQTTCONNACK 2 << 4 // Connect Acknowledgment -#define MQTTPUBLISH 3 << 4 // Publish message -#define MQTTPUBACK 4 << 4 // Publish Acknowledgment -#define MQTTPUBREC 5 << 4 // Publish Received (assured delivery part 1) -#define MQTTPUBREL 6 << 4 // Publish Release (assured delivery part 2) -#define MQTTPUBCOMP 7 << 4 // Publish Complete (assured delivery part 3) -#define MQTTSUBSCRIBE 8 << 4 // Client Subscribe request -#define MQTTSUBACK 9 << 4 // Subscribe Acknowledgment -#define MQTTUNSUBSCRIBE 10 << 4 // Client Unsubscribe request -#define MQTTUNSUBACK 11 << 4 // Unsubscribe Acknowledgment -#define MQTTPINGREQ 12 << 4 // PING Request -#define MQTTPINGRESP 13 << 4 // PING Response -#define MQTTDISCONNECT 14 << 4 // Client is Disconnecting -#define MQTTReserved 15 << 4 // Reserved - -#define MQTTQOS0 (0 << 1) -#define MQTTQOS1 (1 << 1) -#define MQTTQOS2 (2 << 1) - -class PubSubClient { -private: - Client* _client; - uint8_t buffer[MQTT_MAX_PACKET_SIZE]; - uint16_t nextMsgId; - unsigned long lastOutActivity; - unsigned long lastInActivity; - bool pingOutstanding; - void (*callback)(char*,uint8_t*,unsigned int); - uint16_t readPacket(uint8_t*); - uint8_t readByte(); - boolean write(uint8_t header, uint8_t* buf, uint16_t length); - uint16_t writeString(char* string, uint8_t* buf, uint16_t pos); - uint8_t *ip; - char* domain; - uint16_t port; - Stream* stream; -public: - PubSubClient(); - PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client); - PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client, Stream&); - PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client); - PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client, Stream&); - boolean connect(char *); - boolean connect(char *, char *, char *); - boolean connect(char *, char *, uint8_t, uint8_t, char *); - boolean connect(char *, char *, char *, char *, uint8_t, uint8_t, char*); - void disconnect(); - boolean publish(char *, char *); - boolean publish(char *, uint8_t *, unsigned int); - boolean publish(char *, uint8_t *, unsigned int, boolean); - boolean publish_P(char *, uint8_t *, unsigned int, boolean); - boolean subscribe(char *); - boolean subscribe(char *, uint8_t qos); - boolean unsubscribe(char *); - boolean loop(); - boolean connected(); -}; - - -#endif diff --git a/libraries/PubSubClient/README.md b/libraries/PubSubClient/README.md new file mode 100644 index 0000000..8317691 --- /dev/null +++ b/libraries/PubSubClient/README.md @@ -0,0 +1,47 @@ +# Arduino Client for MQTT + +This library provides a client for doing simple publish/subscribe messaging with +a server that supports MQTT. + +## Examples + +The library comes with a number of example sketches. See File > Examples > PubSubClient +within the Arduino application. + +Full API documentation is available here: http://pubsubclient.knolleary.net + +## Limitations + + - It can only publish QoS 0 messages. It can subscribe at QoS 0 or QoS 1. + - The maximum message size, including header, is **128 bytes** by default. This + is configurable via `MQTT_MAX_PACKET_SIZE` in `PubSubClient.h`. + - The keepalive interval is set to 15 seconds by default. This is configurable + via `MQTT_KEEPALIVE` in `PubSubClient.h`. + - The client uses MQTT 3.1.1 by default. It can be changed to use MQTT 3.1 by + changing value of `MQTT_VERSION` in `PubSubClient.h`. + + +## Compatible Hardware + +The library uses the Arduino Ethernet Client api for interacting with the +underlying network hardware. This means it Just Works with a growing number of +boards and shields, including: + + - Arduino Ethernet + - Arduino Ethernet Shield + - Arduino YUN – use the included `YunClient` in place of `EthernetClient`, and + be sure to do a `Bridge.begin()` first + - Arduino WiFi Shield - if you want to send packets > 90 bytes with this shield, + enable the `MQTT_MAX_TRANSFER_SIZE` define in `PubSubClient.h`. + - Sparkfun WiFly Shield – [library](https://github.com/dpslwk/WiFly) + - TI CC3000 WiFi - [library](https://github.com/sparkfun/SFE_CC3000_Library) + - Intel Galileo/Edison + - ESP8266 + +The library cannot currently be used with hardware based on the ENC28J60 chip – +such as the Nanode or the Nuelectronics Ethernet Shield. For those, there is an +[alternative library](https://github.com/njh/NanodeMQTT) available. + +## License + +This code is released under the MIT License. diff --git a/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino b/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino index c8a519a..e9f7b18 100755 --- a/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino +++ b/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino @@ -1,6 +1,6 @@ /* Basic MQTT example with Authentication - + - connects to an MQTT server, providing username and password - publishes "hello world" to the topic "outTopic" @@ -13,8 +13,8 @@ // Update these with values suitable for your network. byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -byte server[] = { 172, 16, 0, 2 }; -byte ip[] = { 172, 16, 0, 100 }; +IPAddress ip(172, 16, 0, 100); +IPAddress server(172, 16, 0, 2); void callback(char* topic, byte* payload, unsigned int length) { // handle message arrived @@ -26,6 +26,11 @@ PubSubClient client(server, 1883, callback, ethClient); void setup() { Ethernet.begin(mac, ip); + // Note - the default maximum packet size is 128 bytes. If the + // combined length of clientId, username and password exceed this, + // you will need to increase the value of MQTT_MAX_PACKET_SIZE in + // PubSubClient.h + if (client.connect("arduinoClient", "testuser", "testpass")) { client.publish("outTopic","hello world"); client.subscribe("inTopic"); @@ -36,4 +41,3 @@ void loop() { client.loop(); } - diff --git a/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino b/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino index 72276cd..f545ade 100755 --- a/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino +++ b/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino @@ -1,9 +1,16 @@ /* - Basic MQTT example - - - connects to an MQTT server + Basic MQTT example + + This sketch demonstrates the basic capabilities of the library. + It connects to an MQTT server then: - publishes "hello world" to the topic "outTopic" - - subscribes to the topic "inTopic" + - subscribes to the topic "inTopic", printing out any messages + it receives. NB - it assumes the received payloads are strings not binary + + It will reconnect to the server if the connection is lost using a blocking + reconnect function. See the 'mqtt_reconnect_nonblocking' example for how to + achieve the same result without blocking the main loop. + */ #include @@ -12,27 +19,59 @@ // Update these with values suitable for your network. byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -byte server[] = { 172, 16, 0, 2 }; -byte ip[] = { 172, 16, 0, 100 }; +IPAddress ip(172, 16, 0, 100); +IPAddress server(172, 16, 0, 2); void callback(char* topic, byte* payload, unsigned int length) { - // handle message arrived + Serial.print("Message arrived ["); + Serial.print(topic); + Serial.print("] "); + for (int i=0;i @@ -20,8 +20,8 @@ // Update these with values suitable for your network. byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -byte server[] = { 172, 16, 0, 2 }; -byte ip[] = { 172, 16, 0, 100 }; +IPAddress ip(172, 16, 0, 100); +IPAddress server(172, 16, 0, 2); // Callback function header void callback(char* topic, byte* payload, unsigned int length); @@ -34,7 +34,7 @@ void callback(char* topic, byte* payload, unsigned int length) { // In order to republish this payload, a copy must be made // as the orignal payload buffer will be overwritten whilst // constructing the PUBLISH packet. - + // Allocate the correct amount of memory for the payload copy byte* p = (byte*)malloc(length); // Copy the payload to the new buffer @@ -46,7 +46,7 @@ void callback(char* topic, byte* payload, unsigned int length) { void setup() { - + Ethernet.begin(mac, ip); if (client.connect("arduinoClient")) { client.publish("outTopic","hello world"); @@ -58,4 +58,3 @@ void loop() { client.loop(); } - diff --git a/libraries/MySensors/drivers/pubsubclient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino b/libraries/PubSubClient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino rename to libraries/PubSubClient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino diff --git a/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino b/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino index a416ee2..67c2287 100755 --- a/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino +++ b/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino @@ -16,8 +16,8 @@ // Update these with values suitable for your network. byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; -byte server[] = { 172, 16, 0, 2 }; -byte ip[] = { 172, 16, 0, 100 }; +IPAddress ip(172, 16, 0, 100); +IPAddress server(172, 16, 0, 2); SRAM sram(4, SRAM_1024); @@ -29,7 +29,7 @@ void callback(char* topic, byte* payload, unsigned int length) { Serial.write(sram.read()); } Serial.println(); - + // Reset position for the next message to be stored sram.seek(1); } @@ -47,7 +47,7 @@ void setup() sram.begin(); sram.seek(1); - + Serial.begin(9600); } @@ -55,4 +55,3 @@ void loop() { client.loop(); } - diff --git a/libraries/PubSubClient/keywords.txt b/libraries/PubSubClient/keywords.txt index d91582e..b979588 100755 --- a/libraries/PubSubClient/keywords.txt +++ b/libraries/PubSubClient/keywords.txt @@ -1,5 +1,5 @@ ####################################### -# Syntax Coloring Map For Ultrasound +# Syntax Coloring Map For PubSubClient ####################################### ####################################### @@ -15,11 +15,16 @@ PubSubClient KEYWORD1 connect KEYWORD2 disconnect KEYWORD2 publish KEYWORD2 +publish_P KEYWORD2 subscribe KEYWORD2 +unsubscribe KEYWORD2 loop KEYWORD2 connected KEYWORD2 +setServer KEYWORD2 +setCallback KEYWORD2 +setClient KEYWORD2 +setStream KEYWORD2 ####################################### # Constants (LITERAL1) ####################################### - diff --git a/libraries/MySensors/drivers/pubsubclient/library.properties b/libraries/PubSubClient/library.properties similarity index 98% rename from libraries/MySensors/drivers/pubsubclient/library.properties rename to libraries/PubSubClient/library.properties index 1e555f1..3ceeda8 100644 --- a/libraries/MySensors/drivers/pubsubclient/library.properties +++ b/libraries/PubSubClient/library.properties @@ -1,5 +1,5 @@ name=PubSubClient -version=2.3 +version=2.6 author=Nick O'Leary maintainer=Nick O'Leary sentence=A client library for MQTT messaging. diff --git a/libraries/PubSubClient/src/PubSubClient.cpp b/libraries/PubSubClient/src/PubSubClient.cpp new file mode 100644 index 0000000..e66c9e0 --- /dev/null +++ b/libraries/PubSubClient/src/PubSubClient.cpp @@ -0,0 +1,591 @@ +/* + PubSubClient.cpp - A simple client for MQTT. + Nick O'Leary + http://knolleary.net +*/ + +#include "PubSubClient.h" +#include "Arduino.h" + +PubSubClient::PubSubClient() { + this->_state = MQTT_DISCONNECTED; + this->_client = NULL; + this->stream = NULL; + setCallback(NULL); +} + +PubSubClient::PubSubClient(Client& client) { + this->_state = MQTT_DISCONNECTED; + setClient(client); + this->stream = NULL; +} + +PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(addr, port); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(addr,port); + setClient(client); + setStream(stream); +} +PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(addr, port); + setCallback(callback); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(addr,port); + setCallback(callback); + setClient(client); + setStream(stream); +} + +PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(ip, port); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(ip,port); + setClient(client); + setStream(stream); +} +PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(ip, port); + setCallback(callback); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(ip,port); + setCallback(callback); + setClient(client); + setStream(stream); +} + +PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(domain,port); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(domain,port); + setClient(client); + setStream(stream); +} +PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { + this->_state = MQTT_DISCONNECTED; + setServer(domain,port); + setCallback(callback); + setClient(client); + this->stream = NULL; +} +PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { + this->_state = MQTT_DISCONNECTED; + setServer(domain,port); + setCallback(callback); + setClient(client); + setStream(stream); +} + +boolean PubSubClient::connect(const char *id) { + return connect(id,NULL,NULL,0,0,0,0); +} + +boolean PubSubClient::connect(const char *id, const char *user, const char *pass) { + return connect(id,user,pass,0,0,0,0); +} + +boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { + return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage); +} + +boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { + if (!connected()) { + int result = 0; + + if (domain != NULL) { + result = _client->connect(this->domain, this->port); + } else { + result = _client->connect(this->ip, this->port); + } + if (result == 1) { + nextMsgId = 1; + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + unsigned int j; + +#if MQTT_VERSION == MQTT_VERSION_3_1 + uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p', MQTT_VERSION}; +#define MQTT_HEADER_VERSION_LENGTH 9 +#elif MQTT_VERSION == MQTT_VERSION_3_1_1 + uint8_t d[7] = {0x00,0x04,'M','Q','T','T',MQTT_VERSION}; +#define MQTT_HEADER_VERSION_LENGTH 7 +#endif + for (j = 0;j>1); + } + } + + buffer[length++] = v; + + buffer[length++] = ((MQTT_KEEPALIVE) >> 8); + buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF); + length = writeString(id,buffer,length); + if (willTopic) { + length = writeString(willTopic,buffer,length); + length = writeString(willMessage,buffer,length); + } + + if(user != NULL) { + length = writeString(user,buffer,length); + if(pass != NULL) { + length = writeString(pass,buffer,length); + } + } + + write(MQTTCONNECT,buffer,length-5); + + lastInActivity = lastOutActivity = millis(); + + while (!_client->available()) { + unsigned long t = millis(); + if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) { + _state = MQTT_CONNECTION_TIMEOUT; + _client->stop(); + return false; + } + } + uint8_t llen; + uint16_t len = readPacket(&llen); + + if (len == 4) { + if (buffer[3] == 0) { + lastInActivity = millis(); + pingOutstanding = false; + _state = MQTT_CONNECTED; + return true; + } else { + _state = buffer[3]; + } + } + _client->stop(); + } else { + _state = MQTT_CONNECT_FAILED; + } + return false; + } + return true; +} + +// reads a byte into result +boolean PubSubClient::readByte(uint8_t * result) { + uint32_t previousMillis = millis(); + while(!_client->available()) { + uint32_t currentMillis = millis(); + if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){ + return false; + } + } + *result = _client->read(); + return true; +} + +// reads a byte into result[*index] and increments index +boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){ + uint16_t current_index = *index; + uint8_t * write_address = &(result[current_index]); + if(readByte(write_address)){ + *index = current_index + 1; + return true; + } + return false; +} + +uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { + uint16_t len = 0; + if(!readByte(buffer, &len)) return 0; + bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; + uint32_t multiplier = 1; + uint16_t length = 0; + uint8_t digit = 0; + uint16_t skip = 0; + uint8_t start = 0; + + do { + if(!readByte(&digit)) return 0; + buffer[len++] = digit; + length += (digit & 127) * multiplier; + multiplier *= 128; + } while ((digit & 128) != 0); + *lengthLength = len-1; + + if (isPublish) { + // Read in topic length to calculate bytes to skip over for Stream writing + if(!readByte(buffer, &len)) return 0; + if(!readByte(buffer, &len)) return 0; + skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; + start = 2; + if (buffer[0]&MQTTQOS1) { + // skip message id + skip += 2; + } + } + + for (uint16_t i = start;istream) { + if (isPublish && len-*lengthLength-2>skip) { + this->stream->write(digit); + } + } + if (len < MQTT_MAX_PACKET_SIZE) { + buffer[len] = digit; + } + len++; + } + + if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { + len = 0; // This will cause the packet to be ignored. + } + + return len; +} + +boolean PubSubClient::loop() { + if (connected()) { + unsigned long t = millis(); + if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) { + if (pingOutstanding) { + this->_state = MQTT_CONNECTION_TIMEOUT; + _client->stop(); + return false; + } else { + buffer[0] = MQTTPINGREQ; + buffer[1] = 0; + _client->write(buffer,2); + lastOutActivity = t; + lastInActivity = t; + pingOutstanding = true; + } + } + if (_client->available()) { + uint8_t llen; + uint16_t len = readPacket(&llen); + uint16_t msgId = 0; + uint8_t *payload; + if (len > 0) { + lastInActivity = t; + uint8_t type = buffer[0]&0xF0; + if (type == MQTTPUBLISH) { + if (callback) { + uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; + char topic[tl+1]; + for (uint16_t i=0;i0 + if ((buffer[0]&0x06) == MQTTQOS1) { + msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1]; + payload = buffer+llen+3+tl+2; + callback(topic,payload,len-llen-3-tl-2); + + buffer[0] = MQTTPUBACK; + buffer[1] = 2; + buffer[2] = (msgId >> 8); + buffer[3] = (msgId & 0xFF); + _client->write(buffer,4); + lastOutActivity = t; + + } else { + payload = buffer+llen+3+tl; + callback(topic,payload,len-llen-3-tl); + } + } + } else if (type == MQTTPINGREQ) { + buffer[0] = MQTTPINGRESP; + buffer[1] = 0; + _client->write(buffer,2); + } else if (type == MQTTPINGRESP) { + pingOutstanding = false; + } + } + } + return true; + } + return false; +} + +boolean PubSubClient::publish(const char* topic, const char* payload) { + return publish(topic,(const uint8_t*)payload,strlen(payload),false); +} + +boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained) { + return publish(topic,(const uint8_t*)payload,strlen(payload),retained); +} + +boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength) { + return publish(topic, payload, plength, false); +} + +boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) { + if (connected()) { + if (MQTT_MAX_PACKET_SIZE < 5 + 2+strlen(topic) + plength) { + // Too long + return false; + } + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + length = writeString(topic,buffer,length); + uint16_t i; + for (i=0;i 0) { + digit |= 0x80; + } + buffer[pos++] = digit; + llen++; + } while(len>0); + + pos = writeString(topic,buffer,pos); + + rc += _client->write(buffer,pos); + + for (i=0;iwrite((char)pgm_read_byte_near(payload + i)); + } + + lastOutActivity = millis(); + + return rc == tlen + 4 + plength; +} + +boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) { + uint8_t lenBuf[4]; + uint8_t llen = 0; + uint8_t digit; + uint8_t pos = 0; + uint16_t rc; + uint16_t len = length; + do { + digit = len % 128; + len = len / 128; + if (len > 0) { + digit |= 0x80; + } + lenBuf[pos++] = digit; + llen++; + } while(len>0); + + buf[4-llen] = header; + for (int i=0;i 0) && result) { + bytesToWrite = (bytesRemaining > MQTT_MAX_TRANSFER_SIZE)?MQTT_MAX_TRANSFER_SIZE:bytesRemaining; + rc = _client->write(writeBuf,bytesToWrite); + result = (rc == bytesToWrite); + bytesRemaining -= rc; + writeBuf += rc; + } + return result; +#else + rc = _client->write(buf+(4-llen),length+1+llen); + lastOutActivity = millis(); + return (rc == 1+llen+length); +#endif +} + +boolean PubSubClient::subscribe(const char* topic) { + return subscribe(topic, 0); +} + +boolean PubSubClient::subscribe(const char* topic, uint8_t qos) { + // original: if (qos < 0 || qos > 1) { (qos is uint8_t, hence qos < 0 impossible, tekka) + if (qos > 1) { + return false; + } + if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { + // Too long + return false; + } + if (connected()) { + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + nextMsgId++; + if (nextMsgId == 0) { + nextMsgId = 1; + } + buffer[length++] = (nextMsgId >> 8); + buffer[length++] = (nextMsgId & 0xFF); + length = writeString((char*)topic, buffer,length); + buffer[length++] = qos; + return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5); + } + return false; +} + +boolean PubSubClient::unsubscribe(const char* topic) { + if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { + // Too long + return false; + } + if (connected()) { + uint16_t length = 5; + nextMsgId++; + if (nextMsgId == 0) { + nextMsgId = 1; + } + buffer[length++] = (nextMsgId >> 8); + buffer[length++] = (nextMsgId & 0xFF); + length = writeString(topic, buffer,length); + return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5); + } + return false; +} + +void PubSubClient::disconnect() { + buffer[0] = MQTTDISCONNECT; + buffer[1] = 0; + _client->write(buffer,2); + _state = MQTT_DISCONNECTED; + _client->stop(); + lastInActivity = lastOutActivity = millis(); +} + +uint16_t PubSubClient::writeString(const char* string, uint8_t* buf, uint16_t pos) { + const char* idp = string; + uint16_t i = 0; + pos += 2; + while (*idp) { + buf[pos++] = *idp++; + i++; + } + buf[pos-i-2] = (i >> 8); + buf[pos-i-1] = (i & 0xFF); + return pos; +} + + +boolean PubSubClient::connected() { + boolean rc; + if (_client == NULL ) { + rc = false; + } else { + rc = (int)_client->connected(); + if (!rc) { + if (this->_state == MQTT_CONNECTED) { + this->_state = MQTT_CONNECTION_LOST; + _client->flush(); + _client->stop(); + } + } + } + return rc; +} + +PubSubClient& PubSubClient::setServer(uint8_t * ip, uint16_t port) { + IPAddress addr(ip[0],ip[1],ip[2],ip[3]); + return setServer(addr,port); +} + +PubSubClient& PubSubClient::setServer(IPAddress ip, uint16_t port) { + this->ip = ip; + this->port = port; + this->domain = NULL; + return *this; +} + +PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port) { + this->domain = domain; + this->port = port; + return *this; +} + +PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE) { + this->callback = callback; + return *this; +} + +PubSubClient& PubSubClient::setClient(Client& client){ + this->_client = &client; + return *this; +} + +PubSubClient& PubSubClient::setStream(Stream& stream){ + this->stream = &stream; + return *this; +} + +int PubSubClient::state() { + return this->_state; +} diff --git a/libraries/PubSubClient/src/PubSubClient.h b/libraries/PubSubClient/src/PubSubClient.h new file mode 100644 index 0000000..90ad8ae --- /dev/null +++ b/libraries/PubSubClient/src/PubSubClient.h @@ -0,0 +1,145 @@ +/* + PubSubClient.h - A simple client for MQTT. + Nick O'Leary + http://knolleary.net +*/ + +#ifndef PubSubClient_h +#define PubSubClient_h + +#include +#include "IPAddress.h" +#include "Client.h" +#include "Stream.h" + +#define MQTT_VERSION_3_1 3 +#define MQTT_VERSION_3_1_1 4 + +// MQTT_VERSION : Pick the version +//#define MQTT_VERSION MQTT_VERSION_3_1 +#ifndef MQTT_VERSION +#define MQTT_VERSION MQTT_VERSION_3_1_1 +#endif + +// MQTT_MAX_PACKET_SIZE : Maximum packet size +#ifndef MQTT_MAX_PACKET_SIZE +#define MQTT_MAX_PACKET_SIZE 128 +#endif + +// MQTT_KEEPALIVE : keepAlive interval in Seconds +#ifndef MQTT_KEEPALIVE +#define MQTT_KEEPALIVE 15 +#endif + +// MQTT_SOCKET_TIMEOUT: socket timeout interval in Seconds +#ifndef MQTT_SOCKET_TIMEOUT +#define MQTT_SOCKET_TIMEOUT 15 +#endif + +// MQTT_MAX_TRANSFER_SIZE : limit how much data is passed to the network client +// in each write call. Needed for the Arduino Wifi Shield. Leave undefined to +// pass the entire MQTT packet in each write call. +//#define MQTT_MAX_TRANSFER_SIZE 80 + +// Possible values for client.state() +#define MQTT_CONNECTION_TIMEOUT -4 +#define MQTT_CONNECTION_LOST -3 +#define MQTT_CONNECT_FAILED -2 +#define MQTT_DISCONNECTED -1 +#define MQTT_CONNECTED 0 +#define MQTT_CONNECT_BAD_PROTOCOL 1 +#define MQTT_CONNECT_BAD_CLIENT_ID 2 +#define MQTT_CONNECT_UNAVAILABLE 3 +#define MQTT_CONNECT_BAD_CREDENTIALS 4 +#define MQTT_CONNECT_UNAUTHORIZED 5 + +#define MQTTCONNECT 1 << 4 // Client request to connect to Server +#define MQTTCONNACK 2 << 4 // Connect Acknowledgment +#define MQTTPUBLISH 3 << 4 // Publish message +#define MQTTPUBACK 4 << 4 // Publish Acknowledgment +#define MQTTPUBREC 5 << 4 // Publish Received (assured delivery part 1) +#define MQTTPUBREL 6 << 4 // Publish Release (assured delivery part 2) +#define MQTTPUBCOMP 7 << 4 // Publish Complete (assured delivery part 3) +#define MQTTSUBSCRIBE 8 << 4 // Client Subscribe request +#define MQTTSUBACK 9 << 4 // Subscribe Acknowledgment +#define MQTTUNSUBSCRIBE 10 << 4 // Client Unsubscribe request +#define MQTTUNSUBACK 11 << 4 // Unsubscribe Acknowledgment +#define MQTTPINGREQ 12 << 4 // PING Request +#define MQTTPINGRESP 13 << 4 // PING Response +#define MQTTDISCONNECT 14 << 4 // Client is Disconnecting +#define MQTTReserved 15 << 4 // Reserved + +#define MQTTQOS0 (0 << 1) +#define MQTTQOS1 (1 << 1) +#define MQTTQOS2 (2 << 1) + +#ifdef ESP8266 +#include +#define MQTT_CALLBACK_SIGNATURE std::function callback +#else +#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int) +#endif + +/** PubSubClient class */ +class PubSubClient { +private: + Client* _client; + uint8_t buffer[MQTT_MAX_PACKET_SIZE]; + uint16_t nextMsgId; + unsigned long lastOutActivity; + unsigned long lastInActivity; + bool pingOutstanding; + MQTT_CALLBACK_SIGNATURE; + uint16_t readPacket(uint8_t*); + boolean readByte(uint8_t * result); + boolean readByte(uint8_t * result, uint16_t * index); + boolean write(uint8_t header, uint8_t* buf, uint16_t length); + uint16_t writeString(const char* string, uint8_t* buf, uint16_t pos); + IPAddress ip; + const char* domain; + uint16_t port; + Stream* stream; + int _state; +public: + PubSubClient(); //!< PubSubClient + PubSubClient(Client& client); //!< PubSubClient + PubSubClient(IPAddress, uint16_t, Client& client); //!< PubSubClient + PubSubClient(IPAddress, uint16_t, Client& client, Stream&); //!< PubSubClient + PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); //!< PubSubClient + PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); //!< PubSubClient + PubSubClient(uint8_t *, uint16_t, Client& client); //!< PubSubClient + PubSubClient(uint8_t *, uint16_t, Client& client, Stream&); //!< PubSubClient + PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); //!< PubSubClient + PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); //!< PubSubClient + PubSubClient(const char*, uint16_t, Client& client); //!< PubSubClient + PubSubClient(const char*, uint16_t, Client& client, Stream&); //!< PubSubClient + PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); //!< PubSubClient + PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); //!< PubSubClient + + PubSubClient& setServer(IPAddress ip, uint16_t port); //!< setServer + PubSubClient& setServer(uint8_t * ip, uint16_t port); //!< setServer + PubSubClient& setServer(const char * domain, uint16_t port); //!< setServer + PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE); //!< setCallback + PubSubClient& setClient(Client& client); //!< setClient + PubSubClient& setStream(Stream& stream); //!< setStream + + boolean connect(const char* id); //!< connect + boolean connect(const char* id, const char* user, const char* pass); //!< connect + boolean connect(const char* id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); //!< connect + boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); //!< connect + void disconnect(); //!< disconnect + boolean publish(const char* topic, const char* payload); //!< publish + boolean publish(const char* topic, const char* payload, boolean retained); //!< publish + boolean publish(const char* topic, const uint8_t * payload, unsigned int plength); //!< publish + boolean publish(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); //!< publish + boolean publish_P(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); //!< publish_P + boolean subscribe(const char* topic); //!< subscribe + boolean subscribe(const char* topic, uint8_t qos); //!< subscribe + boolean unsubscribe(const char* topic); //!< unsubscribe + boolean loop(); //!< loop + boolean connected(); //!< connected + int state(); //!< state +}; + + +#endif diff --git a/libraries/MySensors/drivers/pubsubclient/tests/.gitignore b/libraries/PubSubClient/tests/.gitignore similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/.gitignore rename to libraries/PubSubClient/tests/.gitignore diff --git a/libraries/MySensors/drivers/pubsubclient/tests/Makefile b/libraries/PubSubClient/tests/Makefile similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/Makefile rename to libraries/PubSubClient/tests/Makefile diff --git a/libraries/MySensors/drivers/pubsubclient/tests/README.md b/libraries/PubSubClient/tests/README.md similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/README.md rename to libraries/PubSubClient/tests/README.md diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/connect_spec.cpp b/libraries/PubSubClient/tests/src/connect_spec.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/connect_spec.cpp rename to libraries/PubSubClient/tests/src/connect_spec.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/keepalive_spec.cpp b/libraries/PubSubClient/tests/src/keepalive_spec.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/keepalive_spec.cpp rename to libraries/PubSubClient/tests/src/keepalive_spec.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Arduino.h b/libraries/PubSubClient/tests/src/lib/Arduino.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Arduino.h rename to libraries/PubSubClient/tests/src/lib/Arduino.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.cpp b/libraries/PubSubClient/tests/src/lib/BDDTest.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.cpp rename to libraries/PubSubClient/tests/src/lib/BDDTest.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.h b/libraries/PubSubClient/tests/src/lib/BDDTest.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.h rename to libraries/PubSubClient/tests/src/lib/BDDTest.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.cpp b/libraries/PubSubClient/tests/src/lib/Buffer.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.cpp rename to libraries/PubSubClient/tests/src/lib/Buffer.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.h b/libraries/PubSubClient/tests/src/lib/Buffer.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.h rename to libraries/PubSubClient/tests/src/lib/Buffer.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Client.h b/libraries/PubSubClient/tests/src/lib/Client.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Client.h rename to libraries/PubSubClient/tests/src/lib/Client.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.cpp b/libraries/PubSubClient/tests/src/lib/IPAddress.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.cpp rename to libraries/PubSubClient/tests/src/lib/IPAddress.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.h b/libraries/PubSubClient/tests/src/lib/IPAddress.h similarity index 78% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.h rename to libraries/PubSubClient/tests/src/lib/IPAddress.h index d5d69c6..3871074 100644 --- a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.h +++ b/libraries/PubSubClient/tests/src/lib/IPAddress.h @@ -48,24 +48,24 @@ public: // Overloaded cast operator to allow IPAddress objects to be used where a pointer // to a four-byte uint8_t array is expected - /** @brief operator overload */ - operator uint32_t() { return *((uint32_t*)_address); }; - /** @brief operator overload */ - bool operator==(const IPAddress& addr) { return (*((uint32_t*)_address)) == (*((uint32_t*)addr._address)); }; - /** @brief operator overload */ - bool operator==(const uint8_t* addr); + /** @brief operator overload */ + operator uint32_t() { return *((uint32_t*)_address); }; + /** @brief operator overload */ + bool operator==(const IPAddress& addr) { return (*((uint32_t*)_address)) == (*((uint32_t*)addr._address)); }; + /** @brief operator overload */ + bool operator==(const uint8_t* addr); // Overloaded index operator to allow getting and setting individual octets of the address - /** @brief operator overload */ - uint8_t operator[](int index) const { return _address[index]; }; - /** @brief operator overload */ - uint8_t& operator[](int index) { return _address[index]; }; + /** @brief operator overload */ + uint8_t operator[](int index) const { return _address[index]; }; + /** @brief operator overload */ + uint8_t& operator[](int index) { return _address[index]; }; // Overloaded copy operators to allow initialisation of IPAddress objects from other types - /** @brief operator overload */ - IPAddress& operator=(const uint8_t *address); - /** @brief operator overload */ - IPAddress& operator=(uint32_t address); + /** @brief operator overload */ + IPAddress& operator=(const uint8_t *address); + /** @brief operator overload */ + IPAddress& operator=(uint32_t address); friend class EthernetClass; diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.cpp b/libraries/PubSubClient/tests/src/lib/ShimClient.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.cpp rename to libraries/PubSubClient/tests/src/lib/ShimClient.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.h b/libraries/PubSubClient/tests/src/lib/ShimClient.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.h rename to libraries/PubSubClient/tests/src/lib/ShimClient.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.cpp b/libraries/PubSubClient/tests/src/lib/Stream.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.cpp rename to libraries/PubSubClient/tests/src/lib/Stream.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.h b/libraries/PubSubClient/tests/src/lib/Stream.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.h rename to libraries/PubSubClient/tests/src/lib/Stream.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/lib/trace.h b/libraries/PubSubClient/tests/src/lib/trace.h similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/lib/trace.h rename to libraries/PubSubClient/tests/src/lib/trace.h diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/publish_spec.cpp b/libraries/PubSubClient/tests/src/publish_spec.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/publish_spec.cpp rename to libraries/PubSubClient/tests/src/publish_spec.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/receive_spec.cpp b/libraries/PubSubClient/tests/src/receive_spec.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/receive_spec.cpp rename to libraries/PubSubClient/tests/src/receive_spec.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/src/subscribe_spec.cpp b/libraries/PubSubClient/tests/src/subscribe_spec.cpp similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/src/subscribe_spec.cpp rename to libraries/PubSubClient/tests/src/subscribe_spec.cpp diff --git a/libraries/MySensors/drivers/pubsubclient/tests/testcases/__init__.py b/libraries/PubSubClient/tests/testcases/__init__.py similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/testcases/__init__.py rename to libraries/PubSubClient/tests/testcases/__init__.py diff --git a/libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_basic.py b/libraries/PubSubClient/tests/testcases/mqtt_basic.py similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_basic.py rename to libraries/PubSubClient/tests/testcases/mqtt_basic.py diff --git a/libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_publish_in_callback.py b/libraries/PubSubClient/tests/testcases/mqtt_publish_in_callback.py similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_publish_in_callback.py rename to libraries/PubSubClient/tests/testcases/mqtt_publish_in_callback.py diff --git a/libraries/MySensors/drivers/pubsubclient/tests/testcases/settings.py b/libraries/PubSubClient/tests/testcases/settings.py similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/testcases/settings.py rename to libraries/PubSubClient/tests/testcases/settings.py diff --git a/libraries/MySensors/drivers/pubsubclient/tests/testsuite.py b/libraries/PubSubClient/tests/testsuite.py similarity index 100% rename from libraries/MySensors/drivers/pubsubclient/tests/testsuite.py rename to libraries/PubSubClient/tests/testsuite.py