MySmartHome/MySensors
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Cleanup MyGatewayXXX and fix MQTT (#497)

Browse Source
This commit is contained in:
Olivier
2016-07-02 11:01:50 +02:00
committed by GitHub
parent 60495eaf5c
commit 77bb94c16f
54 changed files with 1196 additions and 1082 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/MySensors/MySensors.h
+1 -1
View File
@@ -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)
libraries/MySensors/core/MyGatewayTransportEthernet.cpp
+14 -14
View File
@@ -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();
}
libraries/MySensors/core/MyGatewayTransportMQTTClient.cpp
+211 -215
View File
@@ -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 <henrik.ekblad@mysensors.org>
* 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 <henrik.ekblad@mysensors.org>
* 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;
}
libraries/MySensors/drivers/pubsubclient/src/PubSubClient.cpp → libraries/MySensors/drivers/PubSubClient/PubSubClient.cpp
+34 -14
View File
@@ -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;i<length;i++) {
digit = readByte();
if(!readByte(&digit)) return 0;
if (this->stream) {
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;
}
libraries/MySensors/drivers/pubsubclient/src/PubSubClient.h → libraries/MySensors/drivers/PubSubClient/PubSubClient.h
+19 -2
View File
@@ -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 <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> 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;
libraries/MySensors/drivers/pubsubclient/CHANGES.txt
+5
View File
@@ -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
libraries/MySensors/drivers/pubsubclient/examples/mqtt_auth/mqtt_auth.ino
-43
View File
@@ -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 <SPI.h>
#include <Ethernet.h>
#include <PubSubClient.h>
// 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();
}
libraries/MySensors/drivers/pubsubclient/examples/mqtt_basic/mqtt_basic.ino
-77
View File
@@ -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 <SPI.h>
#include <Ethernet.h>
#include <PubSubClient.h>
// 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<length;i++) {
Serial.print((char)payload[i]);
}
Serial.println();
}
EthernetClient ethClient;
PubSubClient client(ethClient);
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect("arduinoClient")) {
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()
{
Serial.begin(57600);
client.setServer(server, 1883);
client.setCallback(callback);
Ethernet.begin(mac, ip);
// Allow the hardware to sort itself out
delay(1500);
}
void loop()
{
if (!client.connected()) {
reconnect();
}
client.loop();
}
libraries/MySensors/drivers/pubsubclient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino
-60
View File
@@ -1,60 +0,0 @@
/*
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 <SPI.h>
#include <Ethernet.h>
#include <PubSubClient.h>
// 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();
}
libraries/MySensors/drivers/pubsubclient/examples/mqtt_stream/mqtt_stream.ino
-57
View File
@@ -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 <SPI.h>
#include <Ethernet.h>
#include <PubSubClient.h>
#include <SRAM.h>
// 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<length; i++) {
Serial.write(sram.read());
}
Serial.println();
// Reset position for the next message to be stored
sram.seek(1);
}
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient, sram);
void setup()
{
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
sram.begin();
sram.seek(1);
Serial.begin(9600);
}
void loop()
{
client.loop();
}
libraries/MySensors/drivers/pubsubclient/keywords.txt
-30
View File
@@ -1,30 +0,0 @@
#######################################
# Syntax Coloring Map For PubSubClient
#######################################
#######################################
# Datatypes (KEYWORD1)
#######################################
PubSubClient KEYWORD1
#######################################
# Methods and Functions (KEYWORD2)
#######################################
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)
#######################################
libraries/MySensors/drivers/pubsubclient/.gitignore → libraries/PubSubClient/.gitignore
View File
libraries/MySensors/drivers/pubsubclient/.travis.yml → libraries/PubSubClient/.travis.yml
View File
libraries/PubSubClient/CHANGES.txt
+24 -3
View File
@@ -1,7 +1,29 @@
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
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
libraries/PubSubClient/LICENSE.txt
+1 -1
View File
@@ -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
libraries/PubSubClient/PubSubClient.cpp
-427
View File
@@ -1,427 +0,0 @@
/*
PubSubClient.cpp - A simple client for MQTT.
Nicholas O'Leary
http://knolleary.net
*/
#include "PubSubClient.h"
#include <string.h>
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;i<length;i++) {
digit = readByte();
if (this->stream) {
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;i<tl;i++) {
topic[i] = buffer[llen+3+i];
}
topic[tl] = 0;
// 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;
}
}
}
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<plength;i++) {
buffer[length++] = payload[i];
}
uint8_t header = MQTTPUBLISH;
if (retained) {
header |= 1;
}
return write(header,buffer,length-5);
}
return false;
}
boolean PubSubClient::publish_P(char* topic, uint8_t* payload, unsigned int plength, boolean retained) {
uint8_t llen = 0;
uint8_t digit;
unsigned int rc = 0;
uint16_t tlen;
unsigned int pos = 0;
unsigned int i;
uint8_t header;
unsigned int len;
if (!connected()) {
return false;
}
tlen = strlen(topic);
header = MQTTPUBLISH;
if (retained) {
header |= 1;
}
buffer[pos++] = header;
len = plength + 2 + tlen;
do {
digit = len % 128;
len = len / 128;
if (len > 0) {
digit |= 0x80;
}
buffer[pos++] = digit;
llen++;
} while(len>0);
pos = writeString(topic,buffer,pos);
rc += _client->write(buffer,pos);
for (i=0;i<plength;i++) {
rc += _client->write((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;i<llen;i++) {
buf[5-llen+i] = lenBuf[i];
}
rc = _client->write(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;
}
libraries/PubSubClient/PubSubClient.h
-81
View File
@@ -1,81 +0,0 @@
/*
PubSubClient.h - A simple client for MQTT.
Nicholas O'Leary
http://knolleary.net
*/
#ifndef PubSubClient_h
#define PubSubClient_h
#include <Arduino.h>
#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
libraries/PubSubClient/README.md
+47
View File
@@ -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.
libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino
+8 -4
View File
@@ -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();
}
libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino
+52 -13
View File
@@ -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 <SPI.h>
@@ -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<length;i++) {
Serial.print((char)payload[i]);
}
Serial.println();
}
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient);
PubSubClient client(ethClient);
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect("arduinoClient")) {
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()
{
Serial.begin(57600);
client.setServer(server, 1883);
client.setCallback(callback);
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
// Allow the hardware to sort itself out
delay(1500);
}
void loop()
{
if (!client.connected()) {
reconnect();
}
client.loop();
}
libraries/MySensors/drivers/pubsubclient/examples/mqtt_esp8266/mqtt_esp8266.ino → libraries/PubSubClient/examples/mqtt_esp8266/mqtt_esp8266.ino
+9 -8
View File
@@ -38,14 +38,6 @@ long lastMsg = 0;
char msg[50];
int value = 0;
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 setup_wifi() {
delay(10);
@@ -107,6 +99,15 @@ void reconnect() {
}
}
}
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()) {
libraries/PubSubClient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino
+9 -10
View File
@@ -1,17 +1,17 @@
/*
Publishing in the callback
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
be declared before the PubSubClient constructor and the
actual callback defined afterwards.
This ensures the client reference in the callback function
is valid.
*/
#include <SPI.h>
@@ -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();
}
libraries/MySensors/drivers/pubsubclient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino → libraries/PubSubClient/examples/mqtt_reconnect_nonblocking/mqtt_reconnect_nonblocking.ino
View File
libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino
+4 -5
View File
@@ -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();
}
libraries/PubSubClient/keywords.txt
+7 -2
View File
@@ -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)
#######################################
libraries/MySensors/drivers/pubsubclient/library.properties → libraries/PubSubClient/library.properties
+1 -1
View File
@@ -1,5 +1,5 @@
name=PubSubClient
version=2.3
version=2.6
author=Nick O'Leary <nick.oleary@gmail.com>
maintainer=Nick O'Leary <nick.oleary@gmail.com>
sentence=A client library for MQTT messaging.
libraries/PubSubClient/src/PubSubClient.cpp
+591
View File
@@ -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<MQTT_HEADER_VERSION_LENGTH;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 >= ((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;i<length;i++) {
if(!readByte(&digit)) return 0;
if (this->stream) {
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;i<tl;i++) {
topic[i] = buffer[llen+3+i];
}
topic[tl] = 0;
// 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;
}
}
}
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<plength;i++) {
buffer[length++] = payload[i];
}
uint8_t header = MQTTPUBLISH;
if (retained) {
header |= 1;
}
return write(header,buffer,length-5);
}
return false;
}
boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
uint8_t llen = 0;
uint8_t digit;
unsigned int rc = 0;
uint16_t tlen;
unsigned int pos = 0;
unsigned int i;
uint8_t header;
unsigned int len;
if (!connected()) {
return false;
}
tlen = strlen(topic);
header = MQTTPUBLISH;
if (retained) {
header |= 1;
}
buffer[pos++] = header;
len = plength + 2 + tlen;
do {
digit = len % 128;
len = len / 128;
if (len > 0) {
digit |= 0x80;
}
buffer[pos++] = digit;
llen++;
} while(len>0);
pos = writeString(topic,buffer,pos);
rc += _client->write(buffer,pos);
for (i=0;i<plength;i++) {
rc += _client->write((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<llen;i++) {
buf[5-llen+i] = lenBuf[i];
}
#ifdef MQTT_MAX_TRANSFER_SIZE
uint8_t* writeBuf = buf+(4-llen);
uint16_t bytesRemaining = length+1+llen; //Match the length type
uint8_t bytesToWrite;
boolean result = true;
while((bytesRemaining > 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;
}
libraries/PubSubClient/src/PubSubClient.h
+145
View File
@@ -0,0 +1,145 @@
/*
PubSubClient.h - A simple client for MQTT.
Nick O'Leary
http://knolleary.net
*/
#ifndef PubSubClient_h
#define PubSubClient_h
#include <Arduino.h>
#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 <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> 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
libraries/MySensors/drivers/pubsubclient/tests/.gitignore → libraries/PubSubClient/tests/.gitignore
View File
libraries/MySensors/drivers/pubsubclient/tests/Makefile → libraries/PubSubClient/tests/Makefile
View File
libraries/MySensors/drivers/pubsubclient/tests/README.md → libraries/PubSubClient/tests/README.md
View File
libraries/MySensors/drivers/pubsubclient/tests/src/connect_spec.cpp → libraries/PubSubClient/tests/src/connect_spec.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/keepalive_spec.cpp → libraries/PubSubClient/tests/src/keepalive_spec.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Arduino.h → libraries/PubSubClient/tests/src/lib/Arduino.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.cpp → libraries/PubSubClient/tests/src/lib/BDDTest.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/BDDTest.h → libraries/PubSubClient/tests/src/lib/BDDTest.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.cpp → libraries/PubSubClient/tests/src/lib/Buffer.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Buffer.h → libraries/PubSubClient/tests/src/lib/Buffer.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Client.h → libraries/PubSubClient/tests/src/lib/Client.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.cpp → libraries/PubSubClient/tests/src/lib/IPAddress.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/IPAddress.h → libraries/PubSubClient/tests/src/lib/IPAddress.h
+14 -14
View File
@@ -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;
libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.cpp → libraries/PubSubClient/tests/src/lib/ShimClient.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/ShimClient.h → libraries/PubSubClient/tests/src/lib/ShimClient.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.cpp → libraries/PubSubClient/tests/src/lib/Stream.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/Stream.h → libraries/PubSubClient/tests/src/lib/Stream.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/lib/trace.h → libraries/PubSubClient/tests/src/lib/trace.h
View File
libraries/MySensors/drivers/pubsubclient/tests/src/publish_spec.cpp → libraries/PubSubClient/tests/src/publish_spec.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/receive_spec.cpp → libraries/PubSubClient/tests/src/receive_spec.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/src/subscribe_spec.cpp → libraries/PubSubClient/tests/src/subscribe_spec.cpp
View File
libraries/MySensors/drivers/pubsubclient/tests/testcases/__init__.py → libraries/PubSubClient/tests/testcases/__init__.py
View File
libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_basic.py → libraries/PubSubClient/tests/testcases/mqtt_basic.py
View File
libraries/MySensors/drivers/pubsubclient/tests/testcases/mqtt_publish_in_callback.py → libraries/PubSubClient/tests/testcases/mqtt_publish_in_callback.py
View File
libraries/MySensors/drivers/pubsubclient/tests/testcases/settings.py → libraries/PubSubClient/tests/testcases/settings.py
View File
libraries/MySensors/drivers/pubsubclient/tests/testsuite.py → libraries/PubSubClient/tests/testsuite.py
View File