From dd4ad8bde6aede233e211c25b5d974625798a42b Mon Sep 17 00:00:00 2001 From: vmasdani Date: Wed, 16 Jan 2019 16:28:20 +0700 Subject: [PATCH] First upload, ver 0.9.0 --- README.md | 3 + .../AntaresMQTTPublish/AntaresMQTTPublish.ino | 20 + examples/mqtt_auth/mqtt_auth.ino | 43 ++ examples/mqtt_basic/mqtt_basic.ino | 77 +++ examples/mqtt_esp8266/mqtt_esp8266.ino | 132 ++++ .../mqtt_large_message/mqtt_large_message.ino | 179 +++++ .../mqtt_publish_in_callback.ino | 60 ++ .../mqtt_reconnect_nonblocking.ino | 67 ++ examples/mqtt_stream/mqtt_stream.ino | 57 ++ keywords.txt | 30 + library.properties | 9 + src/AntaresESP8266MQTT.cpp | 87 +++ src/AntaresESP8266MQTT.h | 30 + src/PubSubClient.cpp | 653 ++++++++++++++++++ src/PubSubClient.h | 173 +++++ 15 files changed, 1620 insertions(+) create mode 100644 README.md create mode 100644 examples/AntaresMQTTPublish/AntaresMQTTPublish.ino create mode 100644 examples/mqtt_auth/mqtt_auth.ino create mode 100644 examples/mqtt_basic/mqtt_basic.ino create mode 100644 examples/mqtt_esp8266/mqtt_esp8266.ino create mode 100644 examples/mqtt_large_message/mqtt_large_message.ino create mode 100644 examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino create mode 100644 examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino create mode 100644 examples/mqtt_stream/mqtt_stream.ino create mode 100644 keywords.txt create mode 100644 library.properties create mode 100644 src/AntaresESP8266MQTT.cpp create mode 100644 src/AntaresESP8266MQTT.h create mode 100644 src/PubSubClient.cpp create mode 100644 src/PubSubClient.h diff --git a/README.md b/README.md new file mode 100644 index 0000000..3e77f3d --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# Antares ESP8266 MQTT +**On progress** +A Library to simplify the process of MQTT publication and subscription to Antares IoT Platform. diff --git a/examples/AntaresMQTTPublish/AntaresMQTTPublish.ino b/examples/AntaresMQTTPublish/AntaresMQTTPublish.ino new file mode 100644 index 0000000..16e1fd0 --- /dev/null +++ b/examples/AntaresMQTTPublish/AntaresMQTTPublish.ino @@ -0,0 +1,20 @@ +#include + +#define ACCESSKEY "your-access-key" +#define WIFISSID "your-wifi-ssid" +#define PASSWORD "your-wifi-password" + +#define projectName = "your-project-name"; +#define deviceName = "your-device-name"; + +AntaresESP8266MQTT antares(ACCESSKEY); + +void setup() { + Serial.begin(115200); + antares.setDebug(true); + antares.wifiConnection(WIFISSID, PASSWORD); + antares.setMqttServer(); +} +void loop() { + antares.checkMqttConnection(); +} diff --git a/examples/mqtt_auth/mqtt_auth.ino b/examples/mqtt_auth/mqtt_auth.ino new file mode 100644 index 0000000..e9f7b18 --- /dev/null +++ b/examples/mqtt_auth/mqtt_auth.ino @@ -0,0 +1,43 @@ +/* + 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/examples/mqtt_basic/mqtt_basic.ino b/examples/mqtt_basic/mqtt_basic.ino new file mode 100644 index 0000000..f545ade --- /dev/null +++ b/examples/mqtt_basic/mqtt_basic.ino @@ -0,0 +1,77 @@ +/* + 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 Preferences -> Additional Boards Manager URLs": + http://arduino.esp8266.com/stable/package_esp8266com_index.json + - Open the "Tools -> Board -> Board Manager" and click install for the ESP8266" + - Select your ESP8266 in "Tools -> Board" + +*/ + +#include +#include + +// Update these with values suitable for your network. + +const char* ssid = "........"; +const char* password = "........"; +const char* mqtt_server = "broker.mqtt-dashboard.com"; + +WiFiClient espClient; +PubSubClient client(espClient); +long lastMsg = 0; +char msg[50]; +int value = 0; + +void setup_wifi() { + + delay(10); + // We start by connecting to a WiFi network + Serial.println(); + Serial.print("Connecting to "); + Serial.println(ssid); + + WiFi.begin(ssid, password); + + while (WiFi.status() != WL_CONNECTED) { + delay(500); + Serial.print("."); + } + + randomSeed(micros()); + + Serial.println(""); + Serial.println("WiFi connected"); + Serial.println("IP address: "); + Serial.println(WiFi.localIP()); +} + +void callback(char* topic, byte* payload, unsigned int length) { + Serial.print("Message arrived ["); + Serial.print(topic); + Serial.print("] "); + for (int i = 0; i < length; i++) { + Serial.print((char)payload[i]); + } + Serial.println(); + + // Switch on the LED if an 1 was received as first character + if ((char)payload[0] == '1') { + digitalWrite(BUILTIN_LED, LOW); // Turn the LED on (Note that LOW is the voltage level + // but actually the LED is on; this is because + // it is active low on the ESP-01) + } else { + digitalWrite(BUILTIN_LED, HIGH); // Turn the LED off by making the voltage HIGH + } + +} + +void reconnect() { + // Loop until we're reconnected + while (!client.connected()) { + Serial.print("Attempting MQTT connection..."); + // Create a random client ID + String clientId = "ESP8266Client-"; + clientId += String(random(0xffff), HEX); + // Attempt to connect + if (client.connect(clientId.c_str())) { + Serial.println("connected"); + // Once connected, publish an announcement... + client.publish("outTopic", "hello world"); + // ... and resubscribe + client.subscribe("inTopic"); + } else { + Serial.print("failed, rc="); + Serial.print(client.state()); + Serial.println(" try again in 5 seconds"); + // Wait 5 seconds before retrying + delay(5000); + } + } +} + +void setup() { + pinMode(BUILTIN_LED, OUTPUT); // Initialize the BUILTIN_LED pin as an output + Serial.begin(115200); + setup_wifi(); + client.setServer(mqtt_server, 1883); + client.setCallback(callback); +} + +void loop() { + + if (!client.connected()) { + reconnect(); + } + client.loop(); + + long now = millis(); + if (now - lastMsg > 2000) { + lastMsg = now; + ++value; + snprintf (msg, 50, "hello world #%ld", value); + Serial.print("Publish message: "); + Serial.println(msg); + client.publish("outTopic", msg); + } +} diff --git a/examples/mqtt_large_message/mqtt_large_message.ino b/examples/mqtt_large_message/mqtt_large_message.ino new file mode 100644 index 0000000..e048c3e --- /dev/null +++ b/examples/mqtt_large_message/mqtt_large_message.ino @@ -0,0 +1,179 @@ +/* + Long message ESP8266 MQTT example + + This sketch demonstrates sending arbitrarily large messages in combination + with the ESP8266 board/library. + + It connects to an MQTT server then: + - publishes "hello world" to the topic "outTopic" + - subscribes to the topic "greenBottles/#", printing out any messages + it receives. NB - it assumes the received payloads are strings not binary + - If the sub-topic is a number, it publishes a "greenBottles/lyrics" message + with a payload consisting of the lyrics to "10 green bottles", replacing + 10 with the number given in the sub-topic. + + 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. + + To install the ESP8266 board, (using Arduino 1.6.4+): + - Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs": + http://arduino.esp8266.com/stable/package_esp8266com_index.json + - Open the "Tools -> Board -> Board Manager" and click install for the ESP8266" + - Select your ESP8266 in "Tools -> Board" + +*/ + +#include +#include + +// Update these with values suitable for your network. + +const char* ssid = "........"; +const char* password = "........"; +const char* mqtt_server = "broker.mqtt-dashboard.com"; + +WiFiClient espClient; +PubSubClient client(espClient); +long lastMsg = 0; +char msg[50]; +int value = 0; + +void setup_wifi() { + + delay(10); + // We start by connecting to a WiFi network + Serial.println(); + Serial.print("Connecting to "); + Serial.println(ssid); + + WiFi.begin(ssid, password); + + while (WiFi.status() != WL_CONNECTED) { + delay(500); + Serial.print("."); + } + + randomSeed(micros()); + + Serial.println(""); + Serial.println("WiFi connected"); + Serial.println("IP address: "); + Serial.println(WiFi.localIP()); +} + +void callback(char* topic, byte* payload, unsigned int length) { + Serial.print("Message arrived ["); + Serial.print(topic); + Serial.print("] "); + for (int i = 0; i < length; i++) { + Serial.print((char)payload[i]); + } + Serial.println(); + + // Find out how many bottles we should generate lyrics for + String topicStr(topic); + int bottleCount = 0; // assume no bottles unless we correctly parse a value from the topic + if (topicStr.indexOf('/') >= 0) { + // The topic includes a '/', we'll try to read the number of bottles from just after that + topicStr.remove(0, topicStr.indexOf('/')+1); + // Now see if there's a number of bottles after the '/' + bottleCount = topicStr.toInt(); + } + + if (bottleCount > 0) { + // Work out how big our resulting message will be + int msgLen = 0; + for (int i = bottleCount; i > 0; i--) { + String numBottles(i); + msgLen += 2*numBottles.length(); + if (i == 1) { + msgLen += 2*String(" green bottle, standing on the wall\n").length(); + } else { + msgLen += 2*String(" green bottles, standing on the wall\n").length(); + } + msgLen += String("And if one green bottle should accidentally fall\nThere'll be ").length(); + switch (i) { + case 1: + msgLen += String("no green bottles, standing on the wall\n\n").length(); + break; + case 2: + msgLen += String("1 green bottle, standing on the wall\n\n").length(); + break; + default: + numBottles = i-1; + msgLen += numBottles.length(); + msgLen += String(" green bottles, standing on the wall\n\n").length(); + break; + }; + } + + // Now we can start to publish the message + client.beginPublish("greenBottles/lyrics", msgLen, false); + for (int i = bottleCount; i > 0; i--) { + for (int j = 0; j < 2; j++) { + client.print(i); + if (i == 1) { + client.print(" green bottle, standing on the wall\n"); + } else { + client.print(" green bottles, standing on the wall\n"); + } + } + client.print("And if one green bottle should accidentally fall\nThere'll be "); + switch (i) { + case 1: + client.print("no green bottles, standing on the wall\n\n"); + break; + case 2: + client.print("1 green bottle, standing on the wall\n\n"); + break; + default: + client.print(i-1); + client.print(" green bottles, standing on the wall\n\n"); + break; + }; + } + // Now we're done! + client.endPublish(); + } +} + +void reconnect() { + // Loop until we're reconnected + while (!client.connected()) { + Serial.print("Attempting MQTT connection..."); + // Create a random client ID + String clientId = "ESP8266Client-"; + clientId += String(random(0xffff), HEX); + // Attempt to connect + if (client.connect(clientId.c_str())) { + Serial.println("connected"); + // Once connected, publish an announcement... + client.publish("outTopic", "hello world"); + // ... and resubscribe + client.subscribe("greenBottles/#"); + } else { + Serial.print("failed, rc="); + Serial.print(client.state()); + Serial.println(" try again in 5 seconds"); + // Wait 5 seconds before retrying + delay(5000); + } + } +} + +void setup() { + pinMode(BUILTIN_LED, OUTPUT); // Initialize the BUILTIN_LED pin as an output + Serial.begin(115200); + setup_wifi(); + client.setServer(mqtt_server, 1883); + client.setCallback(callback); +} + +void loop() { + + if (!client.connected()) { + reconnect(); + } + client.loop(); +} diff --git a/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino b/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino new file mode 100644 index 0000000..42afb2a --- /dev/null +++ b/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino @@ -0,0 +1,60 @@ +/* + Publishing in the callback + + - connects to an MQTT server + - subscribes to the topic "inTopic" + - when a message is received, republishes it to "outTopic" + + This example shows how to publish messages within the + callback function. The callback function header needs to + be declared before the PubSubClient constructor and the + actual callback defined afterwards. + This ensures the client reference in the callback function + is valid. + +*/ + +#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); + +// 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/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino b/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino new file mode 100644 index 0000000..080b739 --- /dev/null +++ b/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino @@ -0,0 +1,67 @@ +/* + Reconnecting MQTT example - non-blocking + + This sketch demonstrates how to keep the client connected + using a non-blocking reconnect function. If the client loses + its connection, it attempts to reconnect every 5 seconds + without blocking the main loop. + +*/ + +#include +#include +#include + +// Update these with values suitable for your hardware/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(ethClient); + +long lastReconnectAttempt = 0; + +boolean reconnect() { + if (client.connect("arduinoClient")) { + // Once connected, publish an announcement... + client.publish("outTopic","hello world"); + // ... and resubscribe + client.subscribe("inTopic"); + } + return client.connected(); +} + +void setup() +{ + client.setServer(server, 1883); + client.setCallback(callback); + + Ethernet.begin(mac, ip); + delay(1500); + lastReconnectAttempt = 0; +} + + +void loop() +{ + if (!client.connected()) { + long now = millis(); + if (now - lastReconnectAttempt > 5000) { + lastReconnectAttempt = now; + // Attempt to reconnect + if (reconnect()) { + lastReconnectAttempt = 0; + } + } + } else { + // Client connected + + client.loop(); + } + +} diff --git a/examples/mqtt_stream/mqtt_stream.ino b/examples/mqtt_stream/mqtt_stream.ino new file mode 100644 index 0000000..67c2287 --- /dev/null +++ b/examples/mqtt_stream/mqtt_stream.ino @@ -0,0 +1,57 @@ +/* + 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; i +sentence=A library to simplify the process of fetching/deploying data to Antares IoT platform via MQTT +paragraph=A library to simplify the process of fetching/deploying data to Antares IoT platform via MQTT +category=Communication +url= +architectures=esp8266 diff --git a/src/AntaresESP8266MQTT.cpp b/src/AntaresESP8266MQTT.cpp new file mode 100644 index 0000000..4266acd --- /dev/null +++ b/src/AntaresESP8266MQTT.cpp @@ -0,0 +1,87 @@ +#include "AntaresESP8266MQTT.h" + +WiFiClient espClient; +PubSubClient client(espClient); + +AntaresESP8266MQTT::AntaresESP8266MQTT(String accessKey) { + _accessKey = accessKey; +} + +void AntaresESP8266MQTT::setMqttServer() { + if(WiFi.status() != WL_CONNECTED) { + printDebug("[ANTARES] Unable to connect to MQTT server.\n"); + } + else { + printDebug("[ANTARES] Setting MQTT server \"" + String(_mqttServer) + "\" on port " + String(_mqttPort) + "\n"); + client.setServer(_mqttServer, _mqttPort); + } + +} + +void AntaresESP8266MQTT::checkMqttConnection() { + if(!client.connected()) { + while(!client.connected()) { + printDebug("[ANTARES] Attempting MQTT connection...\n"); + if(client.connect("ESP8266-TESTVALIAN")) { + printDebug("[ANTARES] Connected!"); + client.publish("testvalian", "connect!"); + } + else { + printDebug("[ANTARES] Failed, rc=" + String(client.state()) + ", Will try again in 5 secs.\n"); + delay(5000); + } + } + } + client.loop(); +} + +bool AntaresESP8266MQTT::wifiConnection(String SSID, String wifiPassword) { + char ssidChar[sizeof(SSID)]; + char wifiPasswordChar[sizeof(wifiPassword)]; + + SSID.toCharArray(ssidChar, sizeof(SSID)); + wifiPassword.toCharArray(wifiPasswordChar, sizeof(wifiPassword)); + + int count = 0; + _wifiSSID = ssidChar; + _wifiPass = wifiPasswordChar; + + WiFi.begin(_wifiSSID, _wifiPass); + printDebug("[ANTARES] Trying to connect to " + SSID + "...\n"); + + for (count=0;count<20;count++) + { + delay(500); + printDebug("."); + } + + if(WiFi.status() != WL_CONNECTED) { + printDebug("[ANTARES] Could not connect to " + SSID + ".\n"); + return false; + } + else { + WiFi.setAutoReconnect(true); + printDebug("\n[ANTARES] WiFi Connected!\n"); + printDebug("[ANTARES] IP Address: " + ipToString(WiFi.localIP()) + "\n"); + return true; + } +} + + + +void AntaresESP8266MQTT::printDebug(String text) { + if(_debug) { + Serial.print(text); + } +} + +bool AntaresESP8266MQTT::setDebug(bool trueFalse) { + _debug = trueFalse; +} + +String AntaresESP8266MQTT::ipToString(IPAddress ip) { + String s=""; + for (int i=0; i<4; i++) + s += i ? "." + String(ip[i]) : String(ip[i]); + return s; +} diff --git a/src/AntaresESP8266MQTT.h b/src/AntaresESP8266MQTT.h new file mode 100644 index 0000000..6df356d --- /dev/null +++ b/src/AntaresESP8266MQTT.h @@ -0,0 +1,30 @@ +#ifndef ANTARESESP8266MQTT_H +#define ANTARESESP8266MQTT_H + +#include +#include +#include + +class AntaresESP8266MQTT { + +private: + const char* _mqttServer = "platform.antares.id"; + const int _mqttPort = 1883; + const int _secureMqttPort = 8883; + bool _debug; + char* _wifiSSID; + char* _wifiPass; + String _accessKey; + +public: + AntaresESP8266MQTT(String accessKey); + bool wifiConnection(String SSID, String wifiPassword); + bool setDebug(bool trueFalse); + void printDebug(String text); + String ipToString(IPAddress ip); + + void setMqttServer(); + void checkMqttConnection(); +}; + +#endif diff --git a/src/PubSubClient.cpp b/src/PubSubClient.cpp new file mode 100644 index 0000000..0fa420d --- /dev/null +++ b/src/PubSubClient.cpp @@ -0,0 +1,653 @@ +/* + 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,1); +} + +boolean PubSubClient::connect(const char *id, const char *user, const char *pass) { + return connect(id,user,pass,0,0,0,0,1); +} + +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,1); +} + +boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { + return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1); +} + +boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession) { + 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 = MQTT_MAX_HEADER_SIZE; + 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); + + CHECK_STRING_LENGTH(length,id) + length = writeString(id,buffer,length); + if (willTopic) { + CHECK_STRING_LENGTH(length,willTopic) + length = writeString(willTopic,buffer,length); + CHECK_STRING_LENGTH(length,willMessage) + length = writeString(willMessage,buffer,length); + } + + if(user != NULL) { + CHECK_STRING_LENGTH(length,user) + length = writeString(user,buffer,length); + if(pass != NULL) { + CHECK_STRING_LENGTH(length,pass) + length = writeString(pass,buffer,length); + } + } + + write(MQTTCONNECT,buffer,length-MQTT_MAX_HEADER_SIZE); + + 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()) { + yield(); + 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 (len == 5) { + // Invalid remaining length encoding - kill the connection + _state = MQTT_DISCONNECTED; + _client->stop(); + return 0; + } + 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]; /* topic length in bytes */ + memmove(buffer+llen+2,buffer+llen+3,tl); /* move topic inside buffer 1 byte to front */ + buffer[llen+2+tl] = 0; /* end the topic as a 'C' string with \x00 */ + char *topic = (char*) buffer+llen+2; + // msgId only present for QOS>0 + 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; + } + } else if (!connected()) { + // readPacket has closed the connection + return 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 < MQTT_MAX_HEADER_SIZE + 2+strlen(topic) + plength) { + // Too long + return false; + } + // Leave room in the buffer for header and variable length field + uint16_t length = MQTT_MAX_HEADER_SIZE; + 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::beginPublish(const char* topic, unsigned int plength, boolean retained) { + if (connected()) { + // Send the header and variable length field + uint16_t length = MQTT_MAX_HEADER_SIZE; + length = writeString(topic,buffer,length); + uint16_t i; + uint8_t header = MQTTPUBLISH; + if (retained) { + header |= 1; + } + size_t hlen = buildHeader(header, buffer, plength+length-MQTT_MAX_HEADER_SIZE); + uint16_t rc = _client->write(buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen)); + lastOutActivity = millis(); + return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen))); + } + return false; +} + +int PubSubClient::endPublish() { + return 1; +} + +size_t PubSubClient::write(uint8_t data) { + lastOutActivity = millis(); + return _client->write(data); +} + +size_t PubSubClient::write(const uint8_t *buffer, size_t size) { + lastOutActivity = millis(); + return _client->write(buffer,size); +} + +size_t PubSubClient::buildHeader(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 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+(MQTT_MAX_HEADER_SIZE-hlen),length+hlen); + lastOutActivity = millis(); + return (rc == hlen+length); +#endif +} + +boolean PubSubClient::subscribe(const char* topic) { + return subscribe(topic, 0); +} + +boolean PubSubClient::subscribe(const char* topic, uint8_t qos) { + 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 = MQTT_MAX_HEADER_SIZE; + 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-MQTT_MAX_HEADER_SIZE); + } + 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 = MQTT_MAX_HEADER_SIZE; + 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-MQTT_MAX_HEADER_SIZE); + } + return false; +} + +void PubSubClient::disconnect() { + buffer[0] = MQTTDISCONNECT; + buffer[1] = 0; + _client->write(buffer,2); + _state = MQTT_DISCONNECTED; + _client->flush(); + _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/src/PubSubClient.h b/src/PubSubClient.h new file mode 100644 index 0000000..2fd6f1d --- /dev/null +++ b/src/PubSubClient.h @@ -0,0 +1,173 @@ +/* + 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) + +// Maximum size of fixed header and variable length size header +#define MQTT_MAX_HEADER_SIZE 5 + +#if defined(ESP8266) || defined(ESP32) +#include +#define MQTT_CALLBACK_SIGNATURE std::function callback +#else +#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int) +#endif + +#define CHECK_STRING_LENGTH(l,s) if (l+2+strlen(s) > MQTT_MAX_PACKET_SIZE) {_client->stop();return false;} + +class PubSubClient : public Print { +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); + // Build up the header ready to send + // Returns the size of the header + // Note: the header is built at the end of the first MQTT_MAX_HEADER_SIZE bytes, so will start + // (MQTT_MAX_HEADER_SIZE - ) bytes into the buffer + size_t buildHeader(uint8_t header, uint8_t* buf, uint16_t length); + IPAddress ip; + const char* domain; + uint16_t port; + Stream* stream; + int _state; +public: + PubSubClient(); + PubSubClient(Client& client); + PubSubClient(IPAddress, uint16_t, Client& client); + PubSubClient(IPAddress, uint16_t, Client& client, Stream&); + PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); + PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); + PubSubClient(uint8_t *, uint16_t, Client& client); + PubSubClient(uint8_t *, uint16_t, Client& client, Stream&); + PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); + PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); + PubSubClient(const char*, uint16_t, Client& client); + PubSubClient(const char*, uint16_t, Client& client, Stream&); + PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); + PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); + + PubSubClient& setServer(IPAddress ip, uint16_t port); + PubSubClient& setServer(uint8_t * ip, uint16_t port); + PubSubClient& setServer(const char * domain, uint16_t port); + PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE); + PubSubClient& setClient(Client& client); + PubSubClient& setStream(Stream& stream); + + boolean connect(const char* id); + boolean connect(const char* id, const char* user, const char* pass); + boolean connect(const char* id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); + boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); + boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession); + void disconnect(); + boolean publish(const char* topic, const char* payload); + boolean publish(const char* topic, const char* payload, boolean retained); + boolean publish(const char* topic, const uint8_t * payload, unsigned int plength); + boolean publish(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); + boolean publish_P(const char* topic, const char* payload, boolean retained); + boolean publish_P(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); + // Start to publish a message. + // This API: + // beginPublish(...) + // one or more calls to write(...) + // endPublish() + // Allows for arbitrarily large payloads to be sent without them having to be copied into + // a new buffer and held in memory at one time + // Returns 1 if the message was started successfully, 0 if there was an error + boolean beginPublish(const char* topic, unsigned int plength, boolean retained); + // Finish off this publish message (started with beginPublish) + // Returns 1 if the packet was sent successfully, 0 if there was an error + int endPublish(); + // Write a single byte of payload (only to be used with beginPublish/endPublish) + virtual size_t write(uint8_t); + // Write size bytes from buffer into the payload (only to be used with beginPublish/endPublish) + // Returns the number of bytes written + virtual size_t write(const uint8_t *buffer, size_t size); + boolean subscribe(const char* topic); + boolean subscribe(const char* topic, uint8_t qos); + boolean unsubscribe(const char* topic); + boolean loop(); + boolean connected(); + int state(); +}; + + +#endif