Add MQTTClientGateway

This commit is contained in:
Thomas Krebs
2015-08-30 20:08:02 +02:00
parent 2e2e61ac64
commit 40dbcb1e37
11 changed files with 1523 additions and 0 deletions
+47
View File
@@ -0,0 +1,47 @@
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.
* Fixed example to match 1.8 api changes - dpslwk
* Added username/password support - WilHall
* Added publish_P - publishes messages from PROGMEM - jobytaffey
1.8
* 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
from int to unsigned int
* Various internal tidy-ups around types
1.7
* Improved keepalive handling
* Updated to the Arduino-1.0 API
1.6
* Added the ability to publish a retained message
1.5
* Added default constructor
* Fixed compile error when used with arduino-0021 or later
1.4
* Fixed connection lost handling
1.3
* Fixed packet reading bug in PubSubClient.readPacket
1.2
* Fixed compile error when used with arduino-0016 or later
1.1
* Reduced size of library
* Added support for Will messages
* Clarified licensing - see LICENSE.txt
1.0
* Only Quality of Service (QOS) 0 messaging is supported
* The maximum message size, including header, is 128 bytes
* The keepalive interval is set to 30 seconds
* No support for Will messages
+20
View File
@@ -0,0 +1,20 @@
Copyright (c) 2008-2012 Nicholas O'Leary
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+427
View File
@@ -0,0 +1,427 @@
/*
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 < 0 || 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;
}
+81
View File
@@ -0,0 +1,81 @@
/*
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
+39
View File
@@ -0,0 +1,39 @@
/*
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 };
byte server[] = { 172, 16, 0, 2 };
byte ip[] = { 172, 16, 0, 100 };
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);
if (client.connect("arduinoClient", "testuser", "testpass")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
}
void loop()
{
client.loop();
}
+38
View File
@@ -0,0 +1,38 @@
/*
Basic MQTT example
- 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>
// 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 };
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);
if (client.connect("arduinoClient")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
}
void loop()
{
client.loop();
}
@@ -0,0 +1,61 @@
/*
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 };
byte server[] = { 172, 16, 0, 2 };
byte ip[] = { 172, 16, 0, 100 };
// 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();
}
@@ -0,0 +1,58 @@
/*
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 };
byte server[] = { 172, 16, 0, 2 };
byte ip[] = { 172, 16, 0, 100 };
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();
}
+25
View File
@@ -0,0 +1,25 @@
#######################################
# Syntax Coloring Map For Ultrasound
#######################################
#######################################
# Datatypes (KEYWORD1)
#######################################
PubSubClient KEYWORD1
#######################################
# Methods and Functions (KEYWORD2)
#######################################
connect KEYWORD2
disconnect KEYWORD2
publish KEYWORD2
subscribe KEYWORD2
loop KEYWORD2
connected KEYWORD2
#######################################
# Constants (LITERAL1)
#######################################