8 Commits
v1.5 ... v1.7

7 changed files with 157 additions and 69 deletions

View File

@ -1,41 +0,0 @@
/*
PubSubClient.h - A simple client for MQTT.
Nicholas O'Leary
http://knolleary.net
*/
#ifndef PubSubClient_h
#define PubSubClient_h
#include "Client.h"
#define MAX_PACKET_SIZE 128
#define KEEPALIVE 15000 // max value = 255000
class PubSubClient {
private:
Client _client;
uint8_t buffer[MAX_PACKET_SIZE];
uint8_t nextMsgId;
long lastActivity;
void (*callback)(char*,uint8_t*,int);
uint8_t readPacket();
uint8_t readByte();
int write(uint8_t header, uint8_t* buf, uint8_t length);
uint8_t writeString(char* string, uint8_t* buf, uint8_t pos);
public:
PubSubClient();
PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,int));
int connect(char *);
int connect(char*, char*, uint8_t, uint8_t, char*);
void disconnect();
int publish(char *, char *);
int publish(char *, uint8_t *, uint8_t);
void subscribe(char *);
int loop();
int connected();
};
#endif

View File

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

View File

@ -5,18 +5,16 @@
*/
#include "PubSubClient.h"
#include "Client.h"
#include "EthernetClient.h"
#include "string.h"
#define MQTTCONNECT 1<<4
#define MQTTPUBLISH 3<<4
#define MQTTSUBSCRIBE 8<<4
PubSubClient::PubSubClient() : _client(0) {
PubSubClient::PubSubClient() : _client() {
}
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,int)) : _client(ip,port) {
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,int)) : _client() {
this->callback = callback;
this->ip = ip;
this->port = port;
}
int PubSubClient::connect(char *id) {
return connect(id,0,0,0,0);
@ -24,9 +22,9 @@ int PubSubClient::connect(char *id) {
int PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) {
if (!connected()) {
if (_client.connect()) {
if (_client.connect(this->ip, this->port)) {
nextMsgId = 1;
uint8_t d[9] = {0x00,0x06,0x4d,0x51,0x49,0x73,0x64,0x70,0x03};
uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p',MQTTPROTOCOLVERSION};
uint8_t length = 0;
int j;
for (j = 0;j<9;j++) {
@ -45,11 +43,20 @@ int PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t wi
length = writeString(willMessage,buffer,length);
}
write(MQTTCONNECT,buffer,length);
while (!_client.available()) {}
lastOutActivity = millis();
lastInActivity = millis();
while (!_client.available()) {
long t= millis();
if (t-lastInActivity > KEEPALIVE) {
_client.stop();
return 0;
}
}
uint8_t len = readPacket();
if (len == 4 && buffer[3] == 0) {
lastActivity = millis();
lastInActivity = millis();
pingOutstanding = false;
return 1;
}
}
@ -92,16 +99,24 @@ uint8_t PubSubClient::readPacket() {
int PubSubClient::loop() {
if (connected()) {
long t = millis();
if (t - lastActivity > KEEPALIVE) {
_client.write(192);
if ((t - lastInActivity > KEEPALIVE) || (t - lastOutActivity > KEEPALIVE)) {
if (pingOutstanding) {
_client.stop();
return 0;
} else {
_client.write(MQTTPINGREQ);
_client.write((uint8_t)0);
lastActivity = t;
lastOutActivity = t;
lastInActivity = t;
pingOutstanding = true;
}
}
if (_client.available()) {
uint8_t len = readPacket();
if (len > 0) {
uint8_t type = buffer[0]>>4;
if (type == 3) { // PUBLISH
lastInActivity = t;
uint8_t type = buffer[0]&0xF0;
if (type == MQTTPUBLISH) {
if (callback) {
uint8_t tl = (buffer[2]<<3)+buffer[3];
char topic[tl+1];
@ -113,10 +128,11 @@ int PubSubClient::loop() {
uint8_t *payload = buffer+4+tl;
callback(topic,payload,len-4-tl);
}
} else if (type == 12) { // PINGREG
_client.write(208);
} else if (type == MQTTPINGREQ) {
_client.write(MQTTPINGRESP);
_client.write((uint8_t)0);
lastActivity = t;
} else if (type == MQTTPINGRESP) {
pingOutstanding = false;
}
}
}
@ -129,16 +145,22 @@ int PubSubClient::publish(char* topic, char* payload) {
return publish(topic,(uint8_t*)payload,strlen(payload));
}
int PubSubClient::publish(char* topic, uint8_t* payload, uint8_t plength) {
return publish(topic, payload, plength, 0);
}
int PubSubClient::publish(char* topic, uint8_t* payload, uint8_t plength, uint8_t retained) {
if (connected()) {
uint8_t length = writeString(topic,buffer,0);
int i;
for (i=0;i<plength;i++) {
buffer[length++] = payload[i];
}
//header |= 1; retain
write(MQTTPUBLISH,buffer,length);
uint8_t header = MQTTPUBLISH;
if (retained != 0) {
header |= 1;
}
write(header,buffer,length);
return 1;
}
return 0;
@ -148,9 +170,8 @@ int PubSubClient::publish(char* topic, uint8_t* payload, uint8_t plength) {
int PubSubClient::write(uint8_t header, uint8_t* buf, uint8_t length) {
_client.write(header);
_client.write(length);
for (int i=0;i<length;i++) {
_client.write(buf[i]);
}
_client.write(buf,length);
lastOutActivity = millis();
return 0;
}
@ -168,10 +189,11 @@ void PubSubClient::subscribe(char* topic) {
}
void PubSubClient::disconnect() {
_client.write(224);
_client.write(MQTTDISCONNECT);
_client.write((uint8_t)0);
_client.stop();
lastActivity = millis();
lastInActivity = millis();
lastOutActivity = millis();
}
uint8_t PubSubClient::writeString(char* string, uint8_t* buf, uint8_t pos) {

View File

@ -0,0 +1,64 @@
/*
PubSubClient.h - A simple client for MQTT.
Nicholas O'Leary
http://knolleary.net
*/
#ifndef PubSubClient_h
#define PubSubClient_h
#include "Ethernet.h"
#include "EthernetClient.h"
#define MAX_PACKET_SIZE 128
#define KEEPALIVE 15000 // max value = 255000
// from mqtt-v3r1
#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
class PubSubClient {
private:
EthernetClient _client;
uint8_t buffer[MAX_PACKET_SIZE];
uint8_t nextMsgId;
long lastOutActivity;
long lastInActivity;
bool pingOutstanding;
void (*callback)(char*,uint8_t*,int);
uint8_t readPacket();
uint8_t readByte();
int write(uint8_t header, uint8_t* buf, uint8_t length);
uint8_t writeString(char* string, uint8_t* buf, uint8_t pos);
uint8_t *ip;
uint16_t port;
public:
PubSubClient();
PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,int));
int connect(char *);
int connect(char*, char*, uint8_t, uint8_t, char*);
void disconnect();
int publish(char *, char *);
int publish(char *, uint8_t *, uint8_t);
int publish(char *, uint8_t *, uint8_t, uint8_t);
void subscribe(char *);
int loop();
int connected();
};
#endif

View File

@ -0,0 +1,37 @@
/*
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,int length) {
// handle message arrived
}
PubSubClient client(server, 1883, callback);
void setup()
{
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic","hello world");
client.subscribe("inTopic");
}
}
void loop()
{
client.loop();
}