12 Commits
v1.8 ... v1.9

Author SHA1 Message Date
ecedcb804f Update CHANGES.txt for 1.9 2012-10-28 23:07:06 +00:00
569ada9937 Write each packet with a single call to _client->write() 2012-10-28 00:14:33 +01:00
97e9614780 Unable to set keep alive to over 32 seconds (signed int vs unsigned long) 2012-10-27 22:10:38 +01:00
6f23967ee1 License update 2012-09-13 22:38:08 +01:00
a971ed4a2c Allow any instance of Client to be passed in.
Added new example for publishing in callback.
2012-09-13 22:35:48 +01:00
2d43044338 Merge pull request #11 from jobytaffey/9607eefa0fbdda698efe538c5537340b58eb3812
Allow setting will, allow message to be stored in flash
2012-09-13 12:40:33 -07:00
9607eefa0f Added publish_P, for publishing a message from PROGMEM.
Message can be any size.
2012-07-13 16:41:31 +01:00
62693d406c Allow setting of a will without also setting a username and password. 2012-07-13 16:34:45 +01:00
7e53f612f2 Merge pull request #8 from WilHall/master
Implemented user/pass authentication
2012-04-04 13:52:49 -07:00
Wil
87fb3a9895 Implemented user/pass auth as per MQTT v3.1 specs 2012-04-04 03:44:45 -04:00
18fe49c070 Merge pull request #7 from dpslwk/master
Minor Update for example to 1.8 api
2012-03-13 12:24:50 -07:00
cc7e1c45c7 Update example to be 1.8 compatible 2012-03-13 18:29:43 +00:00
7 changed files with 244 additions and 59 deletions

8
PubSubClient/CHANGES.txt Normal file → Executable file
View File

@ -1,3 +1,11 @@
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

2
PubSubClient/LICENSE.txt Normal file → Executable file
View File

@ -1,4 +1,4 @@
Copyright (c) 2008-2009 Nicholas O'Leary
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

179
PubSubClient/PubSubClient.cpp Normal file → Executable file
View File

@ -5,65 +5,99 @@
*/
#include "PubSubClient.h"
#include <EthernetClient.h>
#include <string.h>
PubSubClient::PubSubClient() : _client() {
PubSubClient::PubSubClient(Client& client) {
this->_client = &client;
}
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int)) : _client() {
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;
}
PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int)) : _client() {
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;
}
boolean PubSubClient::connect(char *id) {
return connect(id,0,0,0,0);
return connect(id,NULL,NULL,0,0,0,0);
}
boolean PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) {
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);
result = _client->connect(this->domain, this->port);
} else {
result = _client.connect(this->ip, this->port);
result = _client->connect(this->ip, this->port);
}
if (result) {
nextMsgId = 1;
uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p',MQTTPROTOCOLVERSION};
uint8_t length = 0;
// 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) {
buffer[length++] = 0x06|(willQos<<3)|(willRetain<<5);
v = 0x06|(willQos<<3)|(willRetain<<5);
} else {
buffer[length++] = 0x02;
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);
buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF);
length = writeString(id,buffer,length);
if (willTopic) {
length = writeString(willTopic,buffer,length);
length = writeString(willMessage,buffer,length);
}
write(MQTTCONNECT,buffer,length);
lastOutActivity = millis();
lastInActivity = millis();
while (!_client.available()) {
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*1000) {
_client.stop();
if (t-lastInActivity > MQTT_KEEPALIVE*1000UL) {
_client->stop();
return false;
}
}
@ -75,14 +109,14 @@ boolean PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_
return true;
}
}
_client.stop();
_client->stop();
}
return false;
}
uint8_t PubSubClient::readByte() {
while(!_client.available()) {}
return _client.read();
while(!_client->available()) {}
return _client->read();
}
uint16_t PubSubClient::readPacket() {
@ -114,19 +148,20 @@ uint16_t PubSubClient::readPacket() {
boolean PubSubClient::loop() {
if (connected()) {
unsigned long t = millis();
if ((t - lastInActivity > MQTT_KEEPALIVE*1000) || (t - lastOutActivity > MQTT_KEEPALIVE*1000)) {
if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) {
if (pingOutstanding) {
_client.stop();
_client->stop();
return false;
} else {
_client.write(MQTTPINGREQ);
_client.write((uint8_t)0);
buffer[0] = MQTTPINGREQ;
buffer[1] = 0;
_client->write(buffer,2);
lastOutActivity = t;
lastInActivity = t;
pingOutstanding = true;
}
}
if (_client.available()) {
if (_client->available()) {
uint16_t len = readPacket();
if (len > 0) {
lastInActivity = t;
@ -144,8 +179,9 @@ boolean PubSubClient::loop() {
callback(topic,payload,len-4-tl);
}
} else if (type == MQTTPINGREQ) {
_client.write(MQTTPINGRESP);
_client.write((uint8_t)0);
buffer[0] = MQTTPINGRESP;
buffer[1] = 0;
_client->write(buffer,2);
} else if (type == MQTTPINGRESP) {
pingOutstanding = false;
}
@ -166,7 +202,9 @@ boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plengt
boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plength, boolean retained) {
if (connected()) {
uint16_t length = writeString(topic,buffer,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];
@ -175,11 +213,54 @@ boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plengt
if (retained) {
header |= 1;
}
return write(header,buffer,length);
return write(header,buffer,length-5);
}
return false;
}
boolean PubSubClient::publish_P(char* topic, uint8_t* PROGMEM payload, unsigned int plength, boolean retained) {
uint8_t llen = 0;
uint8_t digit;
int rc;
uint16_t tlen;
int pos = 0;
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 == len + 1 + plength;
}
boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
uint8_t lenBuf[4];
@ -198,9 +279,12 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
llen++;
} while(len>0);
rc = _client.write(header);
rc += _client.write(lenBuf,llen);
rc += _client.write(buf,length);
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);
}
@ -208,26 +292,27 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
boolean PubSubClient::subscribe(char* topic) {
if (connected()) {
uint16_t length = 2;
// Leave room in the buffer for header and variable length field
uint16_t length = 7;
nextMsgId++;
if (nextMsgId == 0) {
nextMsgId = 1;
}
buffer[0] = nextMsgId >> 8;
buffer[1] = nextMsgId - (buffer[0]<<8);
buffer[0] = (nextMsgId >> 8);
buffer[1] = (nextMsgId & 0xFF);
length = writeString(topic, buffer,length);
buffer[length++] = 0; // Only do QoS 0 subs
return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length);
return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5);
}
return false;
}
void PubSubClient::disconnect() {
_client.write(MQTTDISCONNECT);
_client.write((uint8_t)0);
_client.stop();
lastInActivity = millis();
lastOutActivity = millis();
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) {
@ -238,17 +323,15 @@ uint16_t PubSubClient::writeString(char* string, uint8_t* buf, uint16_t pos) {
buf[pos++] = *idp++;
i++;
}
buf[pos-i-2] = 0;
buf[pos-i-1] = i;
buf[pos-i-2] = (i >> 8);
buf[pos-i-1] = (i & 0xFF);
return pos;
}
boolean PubSubClient::connected() {
int rc = (int)_client.connected();
if (!rc) _client.stop();
int rc = (int)_client->connected();
if (!rc) _client->stop();
return rc;
}

15
PubSubClient/PubSubClient.h Normal file → Executable file
View File

@ -7,8 +7,8 @@
#ifndef PubSubClient_h
#define PubSubClient_h
#include "Ethernet.h"
#include "EthernetClient.h"
#include <Arduino.h>
#include "Client.h"
// MQTT_MAX_PACKET_SIZE : Maximum packet size
#define MQTT_MAX_PACKET_SIZE 128
@ -39,7 +39,7 @@
class PubSubClient {
private:
EthernetClient _client;
Client* _client;
uint8_t buffer[MQTT_MAX_PACKET_SIZE];
uint16_t nextMsgId;
unsigned long lastOutActivity;
@ -54,15 +54,18 @@ private:
char* domain;
uint16_t port;
public:
PubSubClient();
PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int));
PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int));
PubSubClient(Client& client);
PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client);
PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client);
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 PROGMEM *, unsigned int, boolean);
boolean subscribe(char *);
boolean loop();
boolean connected();

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();
}

View File

@ -15,11 +15,12 @@ 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,int length) {
void callback(char* topic, byte* payload, unsigned int length) {
// handle message arrived
}
PubSubClient client(server, 1883, callback);
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient);
void setup()
{

View File

@ -0,0 +1,51 @@
/*
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) {
client.publish("outTopic", payload, length);
}
void setup()
{
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
}
void loop()
{
client.loop();
}