Compare commits
10 Commits
Author | SHA1 | Date | |
---|---|---|---|
e4fdabf4db | |||
4ad0d5f4ac | |||
fb5f8de2a4 | |||
fca33c7f5a | |||
632c44a4d1 | |||
ce9fff9349 | |||
31f3742040 | |||
ef565d8d98 | |||
ef75340e6e | |||
07547e2ec0 |
@ -1,40 +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(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
|
|
@ -1,3 +1,13 @@
|
|||||||
|
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
|
1.4
|
||||||
* Fixed connection lost handling
|
* Fixed connection lost handling
|
||||||
|
|
@ -4,17 +4,17 @@
|
|||||||
http://knolleary.net
|
http://knolleary.net
|
||||||
*/
|
*/
|
||||||
|
|
||||||
#include "WConstants.h"
|
|
||||||
#include "PubSubClient.h"
|
#include "PubSubClient.h"
|
||||||
#include "Client.h"
|
#include "EthernetClient.h"
|
||||||
#include "string.h"
|
#include "string.h"
|
||||||
|
|
||||||
#define MQTTCONNECT 1<<4
|
PubSubClient::PubSubClient() : _client() {
|
||||||
#define MQTTPUBLISH 3<<4
|
}
|
||||||
#define MQTTSUBSCRIBE 8<<4
|
|
||||||
|
|
||||||
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->callback = callback;
|
||||||
|
this->ip = ip;
|
||||||
|
this->port = port;
|
||||||
}
|
}
|
||||||
int PubSubClient::connect(char *id) {
|
int PubSubClient::connect(char *id) {
|
||||||
return connect(id,0,0,0,0);
|
return connect(id,0,0,0,0);
|
||||||
@ -22,9 +22,9 @@ int PubSubClient::connect(char *id) {
|
|||||||
|
|
||||||
int PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) {
|
int PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) {
|
||||||
if (!connected()) {
|
if (!connected()) {
|
||||||
if (_client.connect()) {
|
if (_client.connect(this->ip, this->port)) {
|
||||||
nextMsgId = 1;
|
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;
|
uint8_t length = 0;
|
||||||
int j;
|
int j;
|
||||||
for (j = 0;j<9;j++) {
|
for (j = 0;j<9;j++) {
|
||||||
@ -43,11 +43,20 @@ int PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t wi
|
|||||||
length = writeString(willMessage,buffer,length);
|
length = writeString(willMessage,buffer,length);
|
||||||
}
|
}
|
||||||
write(MQTTCONNECT,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();
|
uint8_t len = readPacket();
|
||||||
|
|
||||||
if (len == 4 && buffer[3] == 0) {
|
if (len == 4 && buffer[3] == 0) {
|
||||||
lastActivity = millis();
|
lastInActivity = millis();
|
||||||
|
pingOutstanding = false;
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -90,16 +99,24 @@ uint8_t PubSubClient::readPacket() {
|
|||||||
int PubSubClient::loop() {
|
int PubSubClient::loop() {
|
||||||
if (connected()) {
|
if (connected()) {
|
||||||
long t = millis();
|
long t = millis();
|
||||||
if (t - lastActivity > KEEPALIVE) {
|
if ((t - lastInActivity > KEEPALIVE) || (t - lastOutActivity > KEEPALIVE)) {
|
||||||
_client.write(192);
|
if (pingOutstanding) {
|
||||||
|
_client.stop();
|
||||||
|
return 0;
|
||||||
|
} else {
|
||||||
|
_client.write(MQTTPINGREQ);
|
||||||
_client.write((uint8_t)0);
|
_client.write((uint8_t)0);
|
||||||
lastActivity = t;
|
lastOutActivity = t;
|
||||||
|
lastInActivity = t;
|
||||||
|
pingOutstanding = true;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if (_client.available()) {
|
if (_client.available()) {
|
||||||
uint8_t len = readPacket();
|
uint8_t len = readPacket();
|
||||||
if (len > 0) {
|
if (len > 0) {
|
||||||
uint8_t type = buffer[0]>>4;
|
lastInActivity = t;
|
||||||
if (type == 3) { // PUBLISH
|
uint8_t type = buffer[0]&0xF0;
|
||||||
|
if (type == MQTTPUBLISH) {
|
||||||
if (callback) {
|
if (callback) {
|
||||||
uint8_t tl = (buffer[2]<<3)+buffer[3];
|
uint8_t tl = (buffer[2]<<3)+buffer[3];
|
||||||
char topic[tl+1];
|
char topic[tl+1];
|
||||||
@ -111,10 +128,11 @@ int PubSubClient::loop() {
|
|||||||
uint8_t *payload = buffer+4+tl;
|
uint8_t *payload = buffer+4+tl;
|
||||||
callback(topic,payload,len-4-tl);
|
callback(topic,payload,len-4-tl);
|
||||||
}
|
}
|
||||||
} else if (type == 12) { // PINGREG
|
} else if (type == MQTTPINGREQ) {
|
||||||
_client.write(208);
|
_client.write(MQTTPINGRESP);
|
||||||
_client.write((uint8_t)0);
|
_client.write((uint8_t)0);
|
||||||
lastActivity = t;
|
} else if (type == MQTTPINGRESP) {
|
||||||
|
pingOutstanding = false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -127,16 +145,22 @@ int PubSubClient::publish(char* topic, char* payload) {
|
|||||||
return publish(topic,(uint8_t*)payload,strlen(payload));
|
return publish(topic,(uint8_t*)payload,strlen(payload));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
int PubSubClient::publish(char* topic, uint8_t* payload, uint8_t plength) {
|
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()) {
|
if (connected()) {
|
||||||
uint8_t length = writeString(topic,buffer,0);
|
uint8_t length = writeString(topic,buffer,0);
|
||||||
int i;
|
int i;
|
||||||
for (i=0;i<plength;i++) {
|
for (i=0;i<plength;i++) {
|
||||||
buffer[length++] = payload[i];
|
buffer[length++] = payload[i];
|
||||||
}
|
}
|
||||||
//header |= 1; retain
|
uint8_t header = MQTTPUBLISH;
|
||||||
write(MQTTPUBLISH,buffer,length);
|
if (retained != 0) {
|
||||||
|
header |= 1;
|
||||||
|
}
|
||||||
|
write(header,buffer,length);
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
return 0;
|
return 0;
|
||||||
@ -146,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) {
|
int PubSubClient::write(uint8_t header, uint8_t* buf, uint8_t length) {
|
||||||
_client.write(header);
|
_client.write(header);
|
||||||
_client.write(length);
|
_client.write(length);
|
||||||
for (int i=0;i<length;i++) {
|
_client.write(buf,length);
|
||||||
_client.write(buf[i]);
|
lastOutActivity = millis();
|
||||||
}
|
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -166,10 +189,11 @@ void PubSubClient::subscribe(char* topic) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
void PubSubClient::disconnect() {
|
void PubSubClient::disconnect() {
|
||||||
_client.write(224);
|
_client.write(MQTTDISCONNECT);
|
||||||
_client.write((uint8_t)0);
|
_client.write((uint8_t)0);
|
||||||
_client.stop();
|
_client.stop();
|
||||||
lastActivity = millis();
|
lastInActivity = millis();
|
||||||
|
lastOutActivity = millis();
|
||||||
}
|
}
|
||||||
|
|
||||||
uint8_t PubSubClient::writeString(char* string, uint8_t* buf, uint8_t pos) {
|
uint8_t PubSubClient::writeString(char* string, uint8_t* buf, uint8_t pos) {
|
64
PubSubClient/PubSubClient.h
Normal file
64
PubSubClient/PubSubClient.h
Normal 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
|
37
PubSubClient/examples/mqtt_basic/mqtt_basic.pde
Normal file
37
PubSubClient/examples/mqtt_basic/mqtt_basic.pde
Normal 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();
|
||||||
|
}
|
||||||
|
|
Reference in New Issue
Block a user