Compare commits

..

63 Commits

Author SHA1 Message Date
8b3a05fec2
Arduino abstraction layer continued 2020-11-12 12:29:42 +01:00
673f2815d5
fixes 2020-11-11 20:03:44 +01:00
121cef72e6
start implementing Arduino abstraction layer and add Makefile 2020-11-11 19:52:21 +01:00
Nick O'Leary
2d228f2f86 Publish v2.8 2020-05-20 01:40:39 +01:00
Nick O'Leary
a69b18a4cb
Merge pull request #518 from ivankravets/patch-3
Declare support for ESP32
2020-05-20 01:37:47 +01:00
Nick O'Leary
24510271ff Add setSocketTimeout function 2020-05-20 01:33:31 +01:00
Nick O'Leary
77c4e445ea Add setKeepAlive function and tests 2020-05-20 01:18:02 +01:00
Nick O'Leary
64e981190b Merge branch 'master' into pr_282 2020-05-20 00:49:38 +01:00
Nick O'Leary
baad92dd0c Fix merge error with strnlen 2020-05-09 23:20:44 +01:00
Nick O'Leary
aca970d2d4
Merge pull request #655 from abderraouf-adjal/strnlen-patch-1
Using strnlen() instead of strlen()
2020-05-09 22:30:23 +01:00
Nick O'Leary
98098ede85
Merge branch 'master' into strnlen-patch-1 2020-05-09 22:29:59 +01:00
Nick O'Leary
289b64ce89
Merge pull request #693 from leojz/master
declare wifi mode prior to connect
2020-05-09 22:21:53 +01:00
Nick O'Leary
25977bef8a
Merge pull request #727 from sn0kerbuzz/fix-subscribe-empty-topic
Added check to prevent subscribe/unsubscribe to empty topics
2020-05-09 22:19:35 +01:00
Andrei Balasescu
6099ee028f Added check to prevent subscribe/unsubscribe to empty topics 2020-04-12 01:20:08 +03:00
Leo Zimmermann
d88909db3e declare wifi mode prior to connect
must have statement for current Arduino Framework, see also https://github.com/knolleary/pubsubclient/issues/138#issuecomment-326113915
2019-12-24 11:08:59 +01:00
Nick O'Leary
7dd2ff90a9
Merge pull request #279 from PhirePhly/esp8266_example_fixes
ESP8266 example fixes
2019-11-25 11:37:11 +00:00
Nick O'Leary
d64528b31b
Merge branch 'master' into esp8266_example_fixes 2019-11-25 11:36:58 +00:00
Nick O'Leary
cff1fc7bdd
Merge pull request #568 from turgu1/patch-1
Allow streaming of large messages
2019-11-25 11:34:25 +00:00
Nick O'Leary
8a5b51803d
Merge pull request #600 from sensidev/master
Fix publish_P return value
2019-11-25 11:32:18 +00:00
Nick O'Leary
1f4011142b
Merge pull request #528 from lexszero/patch-1
Fix publish() when payload == NULL
2019-11-25 11:29:38 +00:00
Nick O'Leary
7cad688df3
Merge pull request #363 from eykamp/documentation_update
Updated documentation
2019-11-25 11:28:19 +00:00
Nick O'Leary
b1cb57208c
Merge pull request #361 from eykamp/comments
Add some comments
2019-11-25 11:28:00 +00:00
Nick O'Leary
299c8293cb
Merge pull request #365 from czaraugust/master
Adding some speed improvments
2019-11-25 11:27:24 +00:00
Nick O'Leary
2b83aa0804
Merge pull request #571 from kevinresol/patch-1
Support pre-connected Client
2019-11-25 11:21:09 +00:00
Abderraouf Adjal
f13ad2af3b
PubSubClient.cpp: Using strnlen() 2019-08-22 17:19:38 +01:00
Abderraouf Adjal
719b90eafd
PubSubClient.h: Using strnlen() 2019-08-22 17:14:02 +01:00
Nick O'Leary
d2ff771f4a
Merge pull request #618 from Brandon2255p/patch-1
Removed an unused variable
2019-05-22 14:58:57 +01:00
Brandon Piner
bc65fe5244
Removed an unused variable 2019-05-09 22:10:56 +02:00
lucian
1b9f52b1fc Fix publish_P return value 2019-04-07 22:19:47 +03:00
Kevin Leung
a4f7821a70
Support pre-connected Client 2019-02-04 00:07:05 +08:00
Guy Turcotte
b25040a0d7
Update receive_spec.cpp 2019-01-29 09:05:26 -05:00
Guy Turcotte
7d6e409b59
Update receive_spec.cpp 2019-01-29 08:59:35 -05:00
Guy Turcotte
e7d9688ca5
Update receive_spec.cpp 2019-01-29 08:43:30 -05:00
Guy Turcotte
373c7d3569
Update PubSubClient.h 2019-01-28 18:47:18 -05:00
Guy Turcotte
b6239823fb
Allow streaming of large messages
These changes are required to allow for the transmission of large messages through a connected stream. The changes do not have an impact on the class interface and habitual behavior. In particular, it will enable the use of OTA through a stream hooked through the setStream() class method. I've designed such a stream to demonstrate the functionality: https://github.com/turgu1/mqtt_ota_example.git

Guy
2019-01-28 18:02:37 -05:00
Alex Ignatov
2dca84a776
Fix publish() when payload == NULL
Messages with empty payload are completely valid and are actually useful (e.g., to delete retained value).
This adds a check before calling `strlen()` to prevent crash when using shortcut methods taking `const char*`
2018-11-21 03:09:08 +02:00
Ivan Kravets
3f34227911
Declare support for ESP32 2018-11-06 13:34:26 +02:00
Nick O'Leary
26ce89fa47 Update for 2.7 2018-11-02 11:42:44 +00:00
Nick O'Leary
a29d0c3d72
Merge pull request #336 from jaecktec/master
Enable ESP32 to use PubSubClient (make MQTT_CALLBACK_SIGNATURE functional)
2018-11-02 01:07:40 +00:00
Nick O'Leary
fa3c4362ea
Merge branch 'master' into master 2018-11-02 01:07:33 +00:00
Nick O'Leary
b381728998 Add yield to mock test framework 2018-11-02 01:05:02 +00:00
Nick O'Leary
2bcd9b074a
Merge pull request #472 from apicquot/master
Add yield to resolve connection failures
2018-11-02 01:01:44 +00:00
Nick O'Leary
2ed03ad522
Merge pull request #362 from eykamp/new_sig
Create new signature to permit cleaner user code
2018-11-02 00:53:02 +00:00
Nick O'Leary
2d053d2df0 Add buffer overflow protection to connect
Closes #492
2018-11-02 00:50:52 +00:00
Nick O'Leary
ee30733e24 Fixup bad revert of 500 2018-11-02 00:12:44 +00:00
Nick O'Leary
4daba0ae5c Fix remaining length protection 2018-11-02 00:06:32 +00:00
Nick O'Leary
af860133e8 Merge branch 'pr_500' 2018-11-01 23:47:15 +00:00
Nick O'Leary
a0f09681f5 Add separate connect function for clean session + test 2018-11-01 23:46:09 +00:00
Nick O'Leary
05a601cc55
Merge pull request #512 from knolleary/revert-500-master
Revert "Added support of CleanSession flag during connect"
2018-11-01 23:35:56 +00:00
apicquot
9e1a6e6479
Update PubSubClient.cpp
added yield to resolve random connection failure
2018-08-07 17:29:29 -04:00
César Augusto
3b0775a91d Excluding nada.txt which was empty and improving indentation 2017-12-07 15:31:24 -03:00
César Augusto
6bc3b76a9b Adding improvments 2017-12-07 12:20:23 -03:00
Unknown
1174d642ab Updated documentation. 2017-11-16 16:37:41 -08:00
Unknown
8795fdf0f5 Create new signature to permit cleaner user code 2017-11-16 12:49:01 -08:00
Unknown
4fa0226cce Add some comments 2017-11-16 11:45:33 -08:00
César Augusto
54043f5469 Start 2017-11-14 22:34:33 -03:00
constantin
8498284792 Add ESP32 to documentation 2017-09-19 18:10:09 +02:00
constantin
49f307506b Add ESP32 callback signature to be functional like ESP8266 2017-09-18 19:27:44 +02:00
duncan law
5b23a97fb0 Update documentation for dynamic buffer resizing. 2017-04-30 10:43:18 +01:00
duncan law
e21ac7baae Add test for dynamic buffer resize. 2017-04-30 10:32:48 +01:00
duncan law
a257083a66 Dynamic buffer size. 2017-04-30 10:29:25 +01:00
Kenneth Finnegan
98ad16eff8 Correct buffer overflow and remove magic constants from esp8266 example
The msg[] array is defined as 50 bytes long, but then the snprintf is
protected at 75 bytes long, which is incorrectly 25 bytes longer than
the underlying array.

To correct this, define a new macro MSG_BUFFER_SIZE and use it in both
places:
 * defining msg[]
 * writing to msg[] with snprintf()
2017-04-25 20:43:54 -07:00
Kenneth Finnegan
b85f4dc35c Corrected time data types in esp8266 example
The millis() function returns unsigned longs, not longs.
2017-04-25 20:39:09 -07:00
29 changed files with 745 additions and 133 deletions

6
.gitignore vendored
View File

@ -1 +1,7 @@
build
pubsub.a
tests/bin tests/bin
.pioenvs
.piolibdeps
.clang_complete
.gcc-flags.json

12
AAL/Arduino.cpp Normal file
View File

@ -0,0 +1,12 @@
#include <Arduino.h>
#include <stdint.h>
millis_t millis() {
return HAL_GetTick();
}
void yield() {
// does nothing
}

25
AAL/Arduino.h Normal file
View File

@ -0,0 +1,25 @@
#ifndef _ARDUINO_H_
#define _ARDUINO_H_
#include <stdint.h>
#include <stdbool.h>
#include <string.h>
#include <Print.h>
#include <IPAddress.h>
#include <Stream.h>
#include <Client.h>
#include <stubs.h>
typedef uint32_t millis_t;
typedef bool boolean;
millis_t millis();
void yield();
#define pgm_read_byte(addr) (*(const unsigned char *)(addr))
#define pgm_read_byte_near(addr) pgm_read_byte(addr)
#endif // _ARDUINO_H_

122
AAL/Client.cpp Normal file
View File

@ -0,0 +1,122 @@
#include <socket.h>
#include <Arduino.h>
#include <Client.h>
#include <stdlib.h>
#include <stdint.h>
/*
class Client {
public:
Client();
int connect(const char *host, uint16_t port);
int connect(IPAddress ip, uint16_t port);
int available();
void stop();
int read();
size_t write(const uint8_t *buf, size_t size);
size_t write(uint8_t b);
void flush();
uint8_t connected();
};
*/
Client::Client(uint8_t sockNum) : sockNum(sockNum) {
}
int Client::connect(const char *host, uint16_t) {
// DNS request required
return 0;
}
int Client::connect(IPAddress ip, uint16_t port) {
int8_t res = socket(this->sockNum, Sn_MR_TCP, port, SF_IO_NONBLOCK);
if (res != this->sockNum) {
close(this->sockNum);
return INVALID_RESPONSE;
}
logMsg("Client::connect: socket initialized");
res = ::connect(this->sockNum, ip.raw_address(), port);
if (res != SOCK_BUSY) {
close(this->sockNum);
return INVALID_RESPONSE;
}
uint32_t startTime = HAL_GetTick();
while (startTime + TIMEOUT_MS > HAL_GetTick()) {
uint8_t sockState = getSn_SR(this->sockNum);
if (sockState == SOCK_ESTABLISHED) {
logMsg("Client::connect: connection established");
return SUCCESS;
}
}
return TIMED_OUT;
}
int Client::available() {
return getSn_RX_RSR(this->sockNum);
}
void Client::stop() {
int8_t res = disconnect(this->sockNum);
if (res != SOCK_BUSY) {
close(this->sockNum);
logMsg("Client::stop: disconnect returns 0x%02x, invalid response, ignore it", res);
} else {
bool successfullyClosed = false;
uint32_t startTime = HAL_GetTick();
while (startTime + TIMEOUT_MS > HAL_GetTick()) {
uint8_t sockState = getSn_SR(this->sockNum);
if (sockState == SOCK_CLOSED) {
logMsg("Client::stop: connection closed");
successfullyClosed = true;
break;
}
}
if (successfullyClosed) {
logMsg("Client::stop: done");
close(this->sockNum);
} else {
logMsg("Client::stop: timeout when closing, ignore");
close(this->sockNum);
}
}
}
int Client::read() {
int res = -1;
if (this->available() >= 1) {
uint8_t buf;
int32_t res = recv(this->sockNum, &buf, 1);
if (res == 1) {
res = (int) buf;
}
}
return res;
}
size_t Client::write(const uint8_t *buf, size_t size) {
int32_t res = send(this->sockNum, (uint8_t*) buf, size);
return (res == size) ? size : 0;
}
size_t Client::write(uint8_t b) {
return this->write(&b, 1);
}
void Client::flush() {
// does nothing
}
uint8_t Client::connected() {
return (getSn_SR(this->sockNum) == SOCK_ESTABLISHED) ? 1 : 0;
}

35
AAL/Client.h Normal file
View File

@ -0,0 +1,35 @@
#ifndef _CLIENT_H_
#define _CLIENT_H_
#include <stdlib.h>
#include <stdint.h>
#include <Arduino.h>
#include <IPAddress.h>
const int SUCCESS = 1;
const int TIMED_OUT = -1;
const int INVALID_SERVER = -2;
const int TRUNCATED = -3;
const int INVALID_RESPONSE = -4;
const uint32_t TIMEOUT_MS = 1000;
class Client {
private:
const uint8_t sockNum;
public:
Client(const uint8_t sockNum);
int connect(const char *host, uint16_t port);
int connect(IPAddress ip, uint16_t port);
int available();
void stop();
int read();
size_t write(const uint8_t *buf, size_t size);
size_t write(uint8_t b);
void flush();
uint8_t connected();
};
#endif // _CLIENT_H_

16
AAL/IPAddress.cpp Normal file
View File

@ -0,0 +1,16 @@
#include <IPAddress.h>
#include <stdint.h>
#include <string.h>
IPAddress::IPAddress() {
memset(_address, 0, sizeof(_address));
}
IPAddress::IPAddress(uint8_t o1, uint8_t o2, uint8_t o3, uint8_t o4) {
_address[0] = o1;
_address[1] = o2;
_address[2] = o3;
_address[3] = o4;
}

18
AAL/IPAddress.h Normal file
View File

@ -0,0 +1,18 @@
#ifndef _IPADDRESS_H_
#define _IPADDRESS_H_
#include <stdint.h>
class IPAddress {
private:
uint8_t _address[4];
uint8_t *raw_address() { return _address; };
public:
IPAddress(uint8_t o1, uint8_t o2, uint8_t o3, uint8_t o4);
IPAddress();
friend class Client;
};
#endif // _IPADDRESS_H_

7
AAL/Print.cpp Normal file
View File

@ -0,0 +1,7 @@
#include <Arduino.h>
#include <Print.h>
Print::Print() {
}

9
AAL/Print.h Normal file
View File

@ -0,0 +1,9 @@
#ifndef _PRINT_H_
#define _PRINT_H_
class Print {
public:
Print();
};
#endif // _PRINT_H_

12
AAL/Stream.cpp Normal file
View File

@ -0,0 +1,12 @@
#include <Stream.h>
#include <stdint.h>
#include <stdlib.h>
Stream::Stream() {
}
size_t Stream::write(uint8_t c) {
return 0;
}

13
AAL/Stream.h Normal file
View File

@ -0,0 +1,13 @@
#ifndef _STREAM_H_
#define _STREAM_H_
#include <stdint.h>
#include <stdlib.h>
class Stream {
public:
Stream();
size_t write(uint8_t c);
};
#endif // _STREAM_H_

9
AAL/stubs.h Normal file
View File

@ -0,0 +1,9 @@
#ifndef _STUBS_H_
#define _STUBS_H_
uint32_t HAL_GetTick(void);
int logMsg(const char *format, ...);
#endif // _STUBS_H_

View File

@ -1,3 +1,20 @@
2.8
* Add setBufferSize() to override MQTT_MAX_PACKET_SIZE
* Add setKeepAlive() to override MQTT_KEEPALIVE
* Add setSocketTimeout() to overide MQTT_SOCKET_TIMEOUT
* Added check to prevent subscribe/unsubscribe to empty topics
* Declare wifi mode prior to connect in ESP example
* Use `strnlen` to avoid overruns
* Support pre-connected Client objects
2.7
* Fix remaining-length handling to prevent buffer overrun
* Add large-payload API - beginPublish/write/publish/endPublish
* Add yield call to improve reliability on ESP
* Add Clean Session flag to connect options
* Add ESP32 support for functional callback signature
* Various other fixes
2.4 2.4
* Add MQTT_SOCKET_TIMEOUT to prevent it blocking indefinitely * Add MQTT_SOCKET_TIMEOUT to prevent it blocking indefinitely
whilst waiting for inbound data whilst waiting for inbound data

View File

@ -1,4 +1,4 @@
Copyright (c) 2008-2015 Nicholas O'Leary Copyright (c) 2008-2020 Nicholas O'Leary
Permission is hereby granted, free of charge, to any person obtaining Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the a copy of this software and associated documentation files (the

30
Makefile Normal file
View File

@ -0,0 +1,30 @@
CFLAGS?=-mcpu=cortex-m3 -mthumb -Og -fdata-sections -ffunction-sections -g -gdwarf-2
CC=arm-none-eabi-gcc
CXX=arm-none-eabi-g++
AR=arm-none-eabi-ar
CFLAGS+=-I../ioLibrary_Driver/Ethernet -Isrc -IAAL
OBJDIR=build
VPATH=src AAL
OBJS=$(addprefix $(OBJDIR)/,PubSubClient.o IPAddress.o Stream.o Arduino.o Print.o Client.o)
all: $(OBJS)
$(AR) rcs pubsub.a $^
$(OBJDIR)/%.o: %.c
$(CC) $(CFLAGS) -c $< -o $@
$(OBJDIR)/%.o: %.cpp
$(CXX) $(CFLAGS) -c $< -o $@
$(OBJS): | $(OBJDIR)
$(OBJDIR):
mkdir $(OBJDIR)
.PHONY: clean
clean:
-rm -rf $(OBJDIR)

View File

@ -13,10 +13,12 @@ Full API documentation is available here: https://pubsubclient.knolleary.net
## Limitations ## Limitations
- It can only publish QoS 0 messages. It can subscribe at QoS 0 or QoS 1. - 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 - The maximum message size, including header, is **256 bytes** by default. This
is configurable via `MQTT_MAX_PACKET_SIZE` in `PubSubClient.h`. is configurable via `MQTT_MAX_PACKET_SIZE` in `PubSubClient.h` or can be changed
by calling `PubSubClient::setBufferSize(size)`.
- The keepalive interval is set to 15 seconds by default. This is configurable - The keepalive interval is set to 15 seconds by default. This is configurable
via `MQTT_KEEPALIVE` in `PubSubClient.h`. via `MQTT_KEEPALIVE` in `PubSubClient.h` or can be changed by calling
`PubSubClient::setKeepAlive(keepAlive)`.
- The client uses MQTT 3.1.1 by default. It can be changed to use MQTT 3.1 by - 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`. changing value of `MQTT_VERSION` in `PubSubClient.h`.
@ -37,6 +39,7 @@ boards and shields, including:
- TI CC3000 WiFi - [library](https://github.com/sparkfun/SFE_CC3000_Library) - TI CC3000 WiFi - [library](https://github.com/sparkfun/SFE_CC3000_Library)
- Intel Galileo/Edison - Intel Galileo/Edison
- ESP8266 - ESP8266
- ESP32
The library cannot currently be used with hardware based on the ENC28J60 chip 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 such as the Nanode or the Nuelectronics Ethernet Shield. For those, there is an

View File

@ -27,9 +27,9 @@ void setup()
{ {
Ethernet.begin(mac, ip); Ethernet.begin(mac, ip);
// Note - the default maximum packet size is 128 bytes. If the // Note - the default maximum packet size is 128 bytes. If the
// combined length of clientId, username and password exceed this, // combined length of clientId, username and password exceed this use the
// you will need to increase the value of MQTT_MAX_PACKET_SIZE in // following to increase the buffer size:
// PubSubClient.h // client.setBufferSize(255);
if (client.connect("arduinoClient", "testuser", "testpass")) { if (client.connect("arduinoClient", "testuser", "testpass")) {
client.publish("outTopic","hello world"); client.publish("outTopic","hello world");

View File

@ -1,26 +1,21 @@
/* /*
Basic ESP8266 MQTT example Basic ESP8266 MQTT example
This sketch demonstrates the capabilities of the pubsub library in combination This sketch demonstrates the capabilities of the pubsub library in combination
with the ESP8266 board/library. with the ESP8266 board/library.
It connects to an MQTT server then: It connects to an MQTT server then:
- publishes "hello world" to the topic "outTopic" every two seconds - publishes "hello world" to the topic "outTopic" every two seconds
- subscribes to the topic "inTopic", printing out any messages - subscribes to the topic "inTopic", printing out any messages
it receives. NB - it assumes the received payloads are strings not binary it receives. NB - it assumes the received payloads are strings not binary
- If the first character of the topic "inTopic" is an 1, switch ON the ESP Led, - If the first character of the topic "inTopic" is an 1, switch ON the ESP Led,
else switch it off else switch it off
It will reconnect to the server if the connection is lost using a blocking 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 reconnect function. See the 'mqtt_reconnect_nonblocking' example for how to
achieve the same result without blocking the main loop. achieve the same result without blocking the main loop.
To install the ESP8266 board, (using Arduino 1.6.4+): To install the ESP8266 board, (using Arduino 1.6.4+):
- Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs": - Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs":
http://arduino.esp8266.com/stable/package_esp8266com_index.json http://arduino.esp8266.com/stable/package_esp8266com_index.json
- Open the "Tools -> Board -> Board Manager" and click install for the ESP8266" - Open the "Tools -> Board -> Board Manager" and click install for the ESP8266"
- Select your ESP8266 in "Tools -> Board" - Select your ESP8266 in "Tools -> Board"
*/ */
#include <ESP8266WiFi.h> #include <ESP8266WiFi.h>
@ -34,8 +29,9 @@ const char* mqtt_server = "broker.mqtt-dashboard.com";
WiFiClient espClient; WiFiClient espClient;
PubSubClient client(espClient); PubSubClient client(espClient);
long lastMsg = 0; unsigned long lastMsg = 0;
char msg[50]; #define MSG_BUFFER_SIZE (50)
char msg[MSG_BUFFER_SIZE];
int value = 0; int value = 0;
void setup_wifi() { void setup_wifi() {
@ -46,6 +42,7 @@ void setup_wifi() {
Serial.print("Connecting to "); Serial.print("Connecting to ");
Serial.println(ssid); Serial.println(ssid);
WiFi.mode(WIFI_STA);
WiFi.begin(ssid, password); WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) { while (WiFi.status() != WL_CONNECTED) {
@ -120,11 +117,11 @@ void loop() {
} }
client.loop(); client.loop();
long now = millis(); unsigned long now = millis();
if (now - lastMsg > 2000) { if (now - lastMsg > 2000) {
lastMsg = now; lastMsg = now;
++value; ++value;
snprintf (msg, 50, "hello world #%ld", value); snprintf (msg, MSG_BUFFER_SIZE, "hello world #%ld", value);
Serial.print("Publish message: "); Serial.print("Publish message: ");
Serial.println(msg); Serial.println(msg);
client.publish("outTopic", msg); client.publish("outTopic", msg);

View File

@ -27,6 +27,9 @@ setServer KEYWORD2
setCallback KEYWORD2 setCallback KEYWORD2
setClient KEYWORD2 setClient KEYWORD2
setStream KEYWORD2 setStream KEYWORD2
setKeepAlive KEYWORD2
setBufferSize KEYWORD2
setSocketTimeout KEYWORD2
####################################### #######################################
# Constants (LITERAL1) # Constants (LITERAL1)

View File

@ -6,12 +6,13 @@
"type": "git", "type": "git",
"url": "https://github.com/knolleary/pubsubclient.git" "url": "https://github.com/knolleary/pubsubclient.git"
}, },
"version": "2.6", "version": "2.8",
"exclude": "tests", "exclude": "tests",
"examples": "examples/*/*.ino", "examples": "examples/*/*.ino",
"frameworks": "arduino", "frameworks": "arduino",
"platforms": [ "platforms": [
"atmelavr", "atmelavr",
"espressif" "espressif8266",
"espressif32"
] ]
} }

View File

@ -1,5 +1,5 @@
name=PubSubClient name=PubSubClient
version=2.6 version=2.8
author=Nick O'Leary <nick.oleary@gmail.com> author=Nick O'Leary <nick.oleary@gmail.com>
maintainer=Nick O'Leary <nick.oleary@gmail.com> maintainer=Nick O'Leary <nick.oleary@gmail.com>
sentence=A client library for MQTT messaging. sentence=A client library for MQTT messaging.

View File

@ -1,4 +1,5 @@
/* /*
PubSubClient.cpp - A simple client for MQTT. PubSubClient.cpp - A simple client for MQTT.
Nick O'Leary Nick O'Leary
http://knolleary.net http://knolleary.net
@ -12,12 +13,20 @@ PubSubClient::PubSubClient() {
this->_client = NULL; this->_client = NULL;
this->stream = NULL; this->stream = NULL;
setCallback(NULL); setCallback(NULL);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(Client& client) { PubSubClient::PubSubClient(Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
@ -25,12 +34,20 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
setServer(addr, port); setServer(addr, port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(addr,port); setServer(addr,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -38,6 +55,10 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATUR
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -45,6 +66,10 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATUR
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
@ -52,12 +77,20 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
setServer(ip, port); setServer(ip, port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(ip,port); setServer(ip,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -65,6 +98,10 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE,
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -72,6 +109,10 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE,
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) { PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
@ -79,12 +120,20 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
setServer(domain,port); setServer(domain,port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(domain,port); setServer(domain,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -92,6 +141,10 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -99,29 +152,47 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->bufferSize = 0;
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
}
PubSubClient::~PubSubClient() {
free(this->buffer);
} }
boolean PubSubClient::connect(const char *id) { boolean PubSubClient::connect(const char *id) {
return connect(id,NULL,NULL,0,0,0,0); return connect(id,NULL,NULL,0,0,0,0,1);
} }
boolean PubSubClient::connect(const char *id, const char *user, const char *pass) { boolean PubSubClient::connect(const char *id, const char *user, const char *pass) {
return connect(id,user,pass,0,0,0,0); return connect(id,user,pass,0,0,0,0,1);
} }
boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { 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); return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage,1);
} }
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1);
}
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession) {
if (!connected()) { if (!connected()) {
int result = 0; int result = 0;
if (domain != NULL) {
result = _client->connect(this->domain, this->port); if(_client->connected()) {
result = 1;
} else { } else {
result = _client->connect(this->ip, this->port); if (domain != NULL) {
result = _client->connect(this->domain, this->port);
} else {
result = _client->connect(this->ip, this->port);
}
} }
if (result == 1) { if (result == 1) {
nextMsgId = 1; nextMsgId = 1;
// Leave room in the buffer for header and variable length field // Leave room in the buffer for header and variable length field
@ -136,14 +207,17 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
#define MQTT_HEADER_VERSION_LENGTH 7 #define MQTT_HEADER_VERSION_LENGTH 7
#endif #endif
for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) { for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) {
buffer[length++] = d[j]; this->buffer[length++] = d[j];
} }
uint8_t v; uint8_t v;
if (willTopic) { if (willTopic) {
v = 0x06|(willQos<<3)|(willRetain<<5); v = 0x04|(willQos<<3)|(willRetain<<5);
} else { } else {
v = 0x02; v = 0x00;
}
if (cleanSession) {
v = v|0x02;
} }
if(user != NULL) { if(user != NULL) {
@ -153,38 +227,43 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
v = v|(0x80>>1); v = v|(0x80>>1);
} }
} }
this->buffer[length++] = v;
buffer[length++] = v; this->buffer[length++] = ((this->keepAlive) >> 8);
this->buffer[length++] = ((this->keepAlive) & 0xFF);
buffer[length++] = ((MQTT_KEEPALIVE) >> 8); CHECK_STRING_LENGTH(length,id)
buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF); length = writeString(id,this->buffer,length);
length = writeString(id,buffer,length);
if (willTopic) { if (willTopic) {
length = writeString(willTopic,buffer,length); CHECK_STRING_LENGTH(length,willTopic)
length = writeString(willMessage,buffer,length); length = writeString(willTopic,this->buffer,length);
CHECK_STRING_LENGTH(length,willMessage)
length = writeString(willMessage,this->buffer,length);
} }
if(user != NULL) { if(user != NULL) {
length = writeString(user,buffer,length); CHECK_STRING_LENGTH(length,user)
length = writeString(user,this->buffer,length);
if(pass != NULL) { if(pass != NULL) {
length = writeString(pass,buffer,length); CHECK_STRING_LENGTH(length,pass)
length = writeString(pass,this->buffer,length);
} }
} }
write(MQTTCONNECT,buffer,length-MQTT_MAX_HEADER_SIZE); write(MQTTCONNECT,this->buffer,length-MQTT_MAX_HEADER_SIZE);
lastInActivity = lastOutActivity = millis(); lastInActivity = lastOutActivity = millis();
while (!_client->available()) { while (!_client->available()) {
unsigned long t = millis(); unsigned long t = millis();
if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) { if (t-lastInActivity >= ((int32_t) this->socketTimeout*1000UL)) {
_state = MQTT_CONNECTION_TIMEOUT; _state = MQTT_CONNECTION_TIMEOUT;
_client->stop(); _client->stop();
return false; return false;
} }
} }
uint8_t llen; uint8_t llen;
uint16_t len = readPacket(&llen); uint32_t len = readPacket(&llen);
if (len == 4) { if (len == 4) {
if (buffer[3] == 0) { if (buffer[3] == 0) {
@ -209,8 +288,9 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
boolean PubSubClient::readByte(uint8_t * result) { boolean PubSubClient::readByte(uint8_t * result) {
uint32_t previousMillis = millis(); uint32_t previousMillis = millis();
while(!_client->available()) { while(!_client->available()) {
yield();
uint32_t currentMillis = millis(); uint32_t currentMillis = millis();
if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){ if(currentMillis - previousMillis >= ((int32_t) this->socketTimeout * 1000)){
return false; return false;
} }
} }
@ -229,74 +309,76 @@ boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){
return false; return false;
} }
uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { uint32_t PubSubClient::readPacket(uint8_t* lengthLength) {
uint16_t len = 0; uint16_t len = 0;
if(!readByte(buffer, &len)) return 0; if(!readByte(this->buffer, &len)) return 0;
bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; bool isPublish = (this->buffer[0]&0xF0) == MQTTPUBLISH;
uint32_t multiplier = 1; uint32_t multiplier = 1;
uint16_t length = 0; uint32_t length = 0;
uint8_t digit = 0; uint8_t digit = 0;
uint16_t skip = 0; uint16_t skip = 0;
uint8_t start = 0; uint32_t start = 0;
do { do {
if (len == 6) { if (len == 5) {
// Invalid remaining length encoding - kill the connection // Invalid remaining length encoding - kill the connection
_state = MQTT_DISCONNECTED; _state = MQTT_DISCONNECTED;
_client->stop(); _client->stop();
return 0; return 0;
} }
if(!readByte(&digit)) return 0; if(!readByte(&digit)) return 0;
buffer[len++] = digit; this->buffer[len++] = digit;
length += (digit & 127) * multiplier; length += (digit & 127) * multiplier;
multiplier *= 128; multiplier <<=7; //multiplier *= 128
} while ((digit & 128) != 0); } while ((digit & 128) != 0);
*lengthLength = len-1; *lengthLength = len-1;
if (isPublish) { if (isPublish) {
// Read in topic length to calculate bytes to skip over for Stream writing // Read in topic length to calculate bytes to skip over for Stream writing
if(!readByte(buffer, &len)) return 0; if(!readByte(this->buffer, &len)) return 0;
if(!readByte(buffer, &len)) return 0; if(!readByte(this->buffer, &len)) return 0;
skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; skip = (this->buffer[*lengthLength+1]<<8)+this->buffer[*lengthLength+2];
start = 2; start = 2;
if (buffer[0]&MQTTQOS1) { if (this->buffer[0]&MQTTQOS1) {
// skip message id // skip message id
skip += 2; skip += 2;
} }
} }
uint32_t idx = len;
for (uint16_t i = start;i<length;i++) { for (uint32_t i = start;i<length;i++) {
if(!readByte(&digit)) return 0; if(!readByte(&digit)) return 0;
if (this->stream) { if (this->stream) {
if (isPublish && len-*lengthLength-2>skip) { if (isPublish && idx-*lengthLength-2>skip) {
this->stream->write(digit); this->stream->write(digit);
} }
} }
if (len < MQTT_MAX_PACKET_SIZE) {
buffer[len] = digit; if (len < this->bufferSize) {
this->buffer[len] = digit;
len++;
} }
len++; idx++;
} }
if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { if (!this->stream && idx > this->bufferSize) {
len = 0; // This will cause the packet to be ignored. len = 0; // This will cause the packet to be ignored.
} }
return len; return len;
} }
boolean PubSubClient::loop() { boolean PubSubClient::loop() {
if (connected()) { if (connected()) {
unsigned long t = millis(); unsigned long t = millis();
if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) { if ((t - lastInActivity > this->keepAlive*1000UL) || (t - lastOutActivity > this->keepAlive*1000UL)) {
if (pingOutstanding) { if (pingOutstanding) {
this->_state = MQTT_CONNECTION_TIMEOUT; this->_state = MQTT_CONNECTION_TIMEOUT;
_client->stop(); _client->stop();
return false; return false;
} else { } else {
buffer[0] = MQTTPINGREQ; this->buffer[0] = MQTTPINGREQ;
buffer[1] = 0; this->buffer[1] = 0;
_client->write(buffer,2); _client->write(this->buffer,2);
lastOutActivity = t; lastOutActivity = t;
lastInActivity = t; lastInActivity = t;
pingOutstanding = true; pingOutstanding = true;
@ -309,35 +391,35 @@ boolean PubSubClient::loop() {
uint8_t *payload; uint8_t *payload;
if (len > 0) { if (len > 0) {
lastInActivity = t; lastInActivity = t;
uint8_t type = buffer[0]&0xF0; uint8_t type = this->buffer[0]&0xF0;
if (type == MQTTPUBLISH) { if (type == MQTTPUBLISH) {
if (callback) { if (callback) {
uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; /* topic length in bytes */ uint16_t tl = (this->buffer[llen+1]<<8)+this->buffer[llen+2]; /* topic length in bytes */
memmove(buffer+llen+2,buffer+llen+3,tl); /* move topic inside buffer 1 byte to front */ memmove(this->buffer+llen+2,this->buffer+llen+3,tl); /* move topic inside buffer 1 byte to front */
buffer[llen+2+tl] = 0; /* end the topic as a 'C' string with \x00 */ this->buffer[llen+2+tl] = 0; /* end the topic as a 'C' string with \x00 */
char *topic = (char*) buffer+llen+2; char *topic = (char*) this->buffer+llen+2;
// msgId only present for QOS>0 // msgId only present for QOS>0
if ((buffer[0]&0x06) == MQTTQOS1) { if ((this->buffer[0]&0x06) == MQTTQOS1) {
msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1]; msgId = (this->buffer[llen+3+tl]<<8)+this->buffer[llen+3+tl+1];
payload = buffer+llen+3+tl+2; payload = this->buffer+llen+3+tl+2;
callback(topic,payload,len-llen-3-tl-2); callback(topic,payload,len-llen-3-tl-2);
buffer[0] = MQTTPUBACK; this->buffer[0] = MQTTPUBACK;
buffer[1] = 2; this->buffer[1] = 2;
buffer[2] = (msgId >> 8); this->buffer[2] = (msgId >> 8);
buffer[3] = (msgId & 0xFF); this->buffer[3] = (msgId & 0xFF);
_client->write(buffer,4); _client->write(this->buffer,4);
lastOutActivity = t; lastOutActivity = t;
} else { } else {
payload = buffer+llen+3+tl; payload = this->buffer+llen+3+tl;
callback(topic,payload,len-llen-3-tl); callback(topic,payload,len-llen-3-tl);
} }
} }
} else if (type == MQTTPINGREQ) { } else if (type == MQTTPINGREQ) {
buffer[0] = MQTTPINGRESP; this->buffer[0] = MQTTPINGRESP;
buffer[1] = 0; this->buffer[1] = 0;
_client->write(buffer,2); _client->write(this->buffer,2);
} else if (type == MQTTPINGRESP) { } else if (type == MQTTPINGRESP) {
pingOutstanding = false; pingOutstanding = false;
} }
@ -352,11 +434,11 @@ boolean PubSubClient::loop() {
} }
boolean PubSubClient::publish(const char* topic, const char* payload) { boolean PubSubClient::publish(const char* topic, const char* payload) {
return publish(topic,(const uint8_t*)payload,strlen(payload),false); return publish(topic,(const uint8_t*)payload, payload ? strnlen(payload, this->bufferSize) : 0,false);
} }
boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained) { boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained) {
return publish(topic,(const uint8_t*)payload,strlen(payload),retained); return publish(topic,(const uint8_t*)payload, payload ? strnlen(payload, this->bufferSize) : 0,retained);
} }
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength) { boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength) {
@ -365,26 +447,34 @@ boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigne
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) { boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
if (connected()) { if (connected()) {
if (MQTT_MAX_PACKET_SIZE < MQTT_MAX_HEADER_SIZE + 2+strlen(topic) + plength) { if (this->bufferSize < MQTT_MAX_HEADER_SIZE + 2+strnlen(topic, this->bufferSize) + plength) {
// Too long // Too long
return false; return false;
} }
// Leave room in the buffer for header and variable length field // Leave room in the buffer for header and variable length field
uint16_t length = MQTT_MAX_HEADER_SIZE; uint16_t length = MQTT_MAX_HEADER_SIZE;
length = writeString(topic,buffer,length); length = writeString(topic,this->buffer,length);
// Add payload
uint16_t i; uint16_t i;
for (i=0;i<plength;i++) { for (i=0;i<plength;i++) {
buffer[length++] = payload[i]; this->buffer[length++] = payload[i];
} }
// Write the header
uint8_t header = MQTTPUBLISH; uint8_t header = MQTTPUBLISH;
if (retained) { if (retained) {
header |= 1; header |= 1;
} }
return write(header,buffer,length-MQTT_MAX_HEADER_SIZE); return write(header,this->buffer,length-MQTT_MAX_HEADER_SIZE);
} }
return false; return false;
} }
boolean PubSubClient::publish_P(const char* topic, const char* payload, boolean retained) {
return publish_P(topic, (const uint8_t*)payload, payload ? strnlen(payload, this->bufferSize) : 0, retained);
}
boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) { boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
uint8_t llen = 0; uint8_t llen = 0;
uint8_t digit; uint8_t digit;
@ -394,32 +484,33 @@ boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsig
unsigned int i; unsigned int i;
uint8_t header; uint8_t header;
unsigned int len; unsigned int len;
int expectedLength;
if (!connected()) { if (!connected()) {
return false; return false;
} }
tlen = strlen(topic); tlen = strnlen(topic, this->bufferSize);
header = MQTTPUBLISH; header = MQTTPUBLISH;
if (retained) { if (retained) {
header |= 1; header |= 1;
} }
buffer[pos++] = header; this->buffer[pos++] = header;
len = plength + 2 + tlen; len = plength + 2 + tlen;
do { do {
digit = len % 128; digit = len & 127; //digit = len %128
len = len / 128; len >>= 7; //len = len / 128
if (len > 0) { if (len > 0) {
digit |= 0x80; digit |= 0x80;
} }
buffer[pos++] = digit; this->buffer[pos++] = digit;
llen++; llen++;
} while(len>0); } while(len>0);
pos = writeString(topic,buffer,pos); pos = writeString(topic,this->buffer,pos);
rc += _client->write(buffer,pos); rc += _client->write(this->buffer,pos);
for (i=0;i<plength;i++) { for (i=0;i<plength;i++) {
rc += _client->write((char)pgm_read_byte_near(payload + i)); rc += _client->write((char)pgm_read_byte_near(payload + i));
@ -427,21 +518,22 @@ boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsig
lastOutActivity = millis(); lastOutActivity = millis();
return rc == tlen + 4 + plength; expectedLength = 1 + llen + 2 + tlen + plength;
return (rc == expectedLength);
} }
boolean PubSubClient::beginPublish(const char* topic, unsigned int plength, boolean retained) { boolean PubSubClient::beginPublish(const char* topic, unsigned int plength, boolean retained) {
if (connected()) { if (connected()) {
// Send the header and variable length field // Send the header and variable length field
uint16_t length = MQTT_MAX_HEADER_SIZE; uint16_t length = MQTT_MAX_HEADER_SIZE;
length = writeString(topic,buffer,length); length = writeString(topic,this->buffer,length);
uint16_t i;
uint8_t header = MQTTPUBLISH; uint8_t header = MQTTPUBLISH;
if (retained) { if (retained) {
header |= 1; header |= 1;
} }
size_t hlen = buildHeader(header, buffer, plength+length-MQTT_MAX_HEADER_SIZE); size_t hlen = buildHeader(header, this->buffer, plength+length-MQTT_MAX_HEADER_SIZE);
uint16_t rc = _client->write(buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen)); uint16_t rc = _client->write(this->buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen));
lastOutActivity = millis(); lastOutActivity = millis();
return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen))); return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen)));
} }
@ -469,8 +561,9 @@ size_t PubSubClient::buildHeader(uint8_t header, uint8_t* buf, uint16_t length)
uint8_t pos = 0; uint8_t pos = 0;
uint16_t len = length; uint16_t len = length;
do { do {
digit = len % 128;
len = len / 128; digit = len & 127; //digit = len %128
len >>= 7; //len = len / 128
if (len > 0) { if (len > 0) {
digit |= 0x80; digit |= 0x80;
} }
@ -514,10 +607,14 @@ boolean PubSubClient::subscribe(const char* topic) {
} }
boolean PubSubClient::subscribe(const char* topic, uint8_t qos) { boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
size_t topicLength = strnlen(topic, this->bufferSize);
if (topic == 0) {
return false;
}
if (qos > 1) { if (qos > 1) {
return false; return false;
} }
if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { if (this->bufferSize < 9 + topicLength) {
// Too long // Too long
return false; return false;
} }
@ -528,17 +625,21 @@ boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
if (nextMsgId == 0) { if (nextMsgId == 0) {
nextMsgId = 1; nextMsgId = 1;
} }
buffer[length++] = (nextMsgId >> 8); this->buffer[length++] = (nextMsgId >> 8);
buffer[length++] = (nextMsgId & 0xFF); this->buffer[length++] = (nextMsgId & 0xFF);
length = writeString((char*)topic, buffer,length); length = writeString((char*)topic, this->buffer,length);
buffer[length++] = qos; this->buffer[length++] = qos;
return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE); return write(MQTTSUBSCRIBE|MQTTQOS1,this->buffer,length-MQTT_MAX_HEADER_SIZE);
} }
return false; return false;
} }
boolean PubSubClient::unsubscribe(const char* topic) { boolean PubSubClient::unsubscribe(const char* topic) {
if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { size_t topicLength = strnlen(topic, this->bufferSize);
if (topic == 0) {
return false;
}
if (this->bufferSize < 9 + topicLength) {
// Too long // Too long
return false; return false;
} }
@ -548,18 +649,18 @@ boolean PubSubClient::unsubscribe(const char* topic) {
if (nextMsgId == 0) { if (nextMsgId == 0) {
nextMsgId = 1; nextMsgId = 1;
} }
buffer[length++] = (nextMsgId >> 8); this->buffer[length++] = (nextMsgId >> 8);
buffer[length++] = (nextMsgId & 0xFF); this->buffer[length++] = (nextMsgId & 0xFF);
length = writeString(topic, buffer,length); length = writeString(topic, this->buffer,length);
return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE); return write(MQTTUNSUBSCRIBE|MQTTQOS1,this->buffer,length-MQTT_MAX_HEADER_SIZE);
} }
return false; return false;
} }
void PubSubClient::disconnect() { void PubSubClient::disconnect() {
buffer[0] = MQTTDISCONNECT; this->buffer[0] = MQTTDISCONNECT;
buffer[1] = 0; this->buffer[1] = 0;
_client->write(buffer,2); _client->write(this->buffer,2);
_state = MQTT_DISCONNECTED; _state = MQTT_DISCONNECTED;
_client->flush(); _client->flush();
_client->stop(); _client->stop();
@ -592,6 +693,8 @@ boolean PubSubClient::connected() {
_client->flush(); _client->flush();
_client->stop(); _client->stop();
} }
} else {
return this->_state == MQTT_CONNECTED;
} }
} }
return rc; return rc;
@ -633,3 +736,34 @@ PubSubClient& PubSubClient::setStream(Stream& stream){
int PubSubClient::state() { int PubSubClient::state() {
return this->_state; return this->_state;
} }
boolean PubSubClient::setBufferSize(uint16_t size) {
if (size == 0) {
// Cannot set it back to 0
return false;
}
if (this->bufferSize == 0) {
this->buffer = (uint8_t*)malloc(size);
} else {
uint8_t* newBuffer = (uint8_t*)realloc(this->buffer, size);
if (newBuffer != NULL) {
this->buffer = newBuffer;
} else {
return false;
}
}
this->bufferSize = size;
return (this->buffer != NULL);
}
uint16_t PubSubClient::getBufferSize() {
return this->bufferSize;
}
PubSubClient& PubSubClient::setKeepAlive(uint16_t keepAlive) {
this->keepAlive = keepAlive;
return *this;
}
PubSubClient& PubSubClient::setSocketTimeout(uint16_t timeout) {
this->socketTimeout = timeout;
return *this;
}

View File

@ -21,17 +21,17 @@
#define MQTT_VERSION MQTT_VERSION_3_1_1 #define MQTT_VERSION MQTT_VERSION_3_1_1
#endif #endif
// MQTT_MAX_PACKET_SIZE : Maximum packet size // MQTT_MAX_PACKET_SIZE : Maximum packet size. Override with setBufferSize().
#ifndef MQTT_MAX_PACKET_SIZE #ifndef MQTT_MAX_PACKET_SIZE
#define MQTT_MAX_PACKET_SIZE 128 #define MQTT_MAX_PACKET_SIZE 256
#endif #endif
// MQTT_KEEPALIVE : keepAlive interval in Seconds // MQTT_KEEPALIVE : keepAlive interval in Seconds. Override with setKeepAlive()
#ifndef MQTT_KEEPALIVE #ifndef MQTT_KEEPALIVE
#define MQTT_KEEPALIVE 15 #define MQTT_KEEPALIVE 15
#endif #endif
// MQTT_SOCKET_TIMEOUT: socket timeout interval in Seconds // MQTT_SOCKET_TIMEOUT: socket timeout interval in Seconds. Override with setSocketTimeout()
#ifndef MQTT_SOCKET_TIMEOUT #ifndef MQTT_SOCKET_TIMEOUT
#define MQTT_SOCKET_TIMEOUT 15 #define MQTT_SOCKET_TIMEOUT 15
#endif #endif
@ -76,23 +76,28 @@
// Maximum size of fixed header and variable length size header // Maximum size of fixed header and variable length size header
#define MQTT_MAX_HEADER_SIZE 5 #define MQTT_MAX_HEADER_SIZE 5
#ifdef ESP8266 #if defined(ESP8266) || defined(ESP32)
#include <functional> #include <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> callback #define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> callback
#else #else
#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int) #define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int)
#endif #endif
#define CHECK_STRING_LENGTH(l,s) if (l+2+strnlen(s, this->bufferSize) > this->bufferSize) {_client->stop();return false;}
class PubSubClient : public Print { class PubSubClient : public Print {
private: private:
Client* _client; Client* _client;
uint8_t buffer[MQTT_MAX_PACKET_SIZE]; uint8_t* buffer;
uint16_t bufferSize;
uint16_t keepAlive;
uint16_t socketTimeout;
uint16_t nextMsgId; uint16_t nextMsgId;
unsigned long lastOutActivity; unsigned long lastOutActivity;
unsigned long lastInActivity; unsigned long lastInActivity;
bool pingOutstanding; bool pingOutstanding;
MQTT_CALLBACK_SIGNATURE; MQTT_CALLBACK_SIGNATURE;
uint16_t readPacket(uint8_t*); uint32_t readPacket(uint8_t*);
boolean readByte(uint8_t * result); boolean readByte(uint8_t * result);
boolean readByte(uint8_t * result, uint16_t * index); boolean readByte(uint8_t * result, uint16_t * index);
boolean write(uint8_t header, uint8_t* buf, uint16_t length); boolean write(uint8_t header, uint8_t* buf, uint16_t length);
@ -123,22 +128,31 @@ public:
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client);
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&);
~PubSubClient();
PubSubClient& setServer(IPAddress ip, uint16_t port); PubSubClient& setServer(IPAddress ip, uint16_t port);
PubSubClient& setServer(uint8_t * ip, uint16_t port); PubSubClient& setServer(uint8_t * ip, uint16_t port);
PubSubClient& setServer(const char * domain, uint16_t port); PubSubClient& setServer(const char * domain, uint16_t port);
PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE); PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE);
PubSubClient& setClient(Client& client); PubSubClient& setClient(Client& client);
PubSubClient& setStream(Stream& stream); PubSubClient& setStream(Stream& stream);
PubSubClient& setKeepAlive(uint16_t keepAlive);
PubSubClient& setSocketTimeout(uint16_t timeout);
boolean setBufferSize(uint16_t size);
uint16_t getBufferSize();
boolean connect(const char* id); boolean connect(const char* id);
boolean connect(const char* id, const char* user, const char* pass); boolean connect(const char* id, const char* user, const char* pass);
boolean connect(const char* id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); boolean connect(const char* id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage);
boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage); boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage);
boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession);
void disconnect(); void disconnect();
boolean publish(const char* topic, const char* payload); boolean publish(const char* topic, const char* payload);
boolean publish(const char* topic, const char* payload, boolean retained); boolean publish(const char* topic, const char* payload, boolean retained);
boolean publish(const char* topic, const uint8_t * payload, unsigned int plength); boolean publish(const char* topic, const uint8_t * payload, unsigned int plength);
boolean publish(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); boolean publish(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained);
boolean publish_P(const char* topic, const char* payload, boolean retained);
boolean publish_P(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained); boolean publish_P(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained);
// Start to publish a message. // Start to publish a message.
// This API: // This API:
@ -163,6 +177,7 @@ public:
boolean loop(); boolean loop();
boolean connected(); boolean connected();
int state(); int state();
}; };

View File

@ -98,6 +98,33 @@ int test_connect_fails_on_bad_rc() {
END_IT END_IT
} }
int test_connect_non_clean_session() {
IT("sends a properly formatted non-clean session connect packet and succeeds");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte expectServer[] = { 172, 16, 0, 2 };
shimClient.expectConnect(expectServer,1883);
byte connect[] = {0x10,0x18,0x0,0x4,0x4d,0x51,0x54,0x54,0x4,0x0,0x0,0xf,0x0,0xc,0x63,0x6c,0x69,0x65,0x6e,0x74,0x5f,0x74,0x65,0x73,0x74,0x31};
byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.expect(connect,26);
shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient);
int state = client.state();
IS_TRUE(state == MQTT_DISCONNECTED);
int rc = client.connect((char*)"client_test1",0,0,0,0,0,0,0);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
END_IT
}
int test_connect_accepts_username_password() { int test_connect_accepts_username_password() {
IT("accepts a username and password"); IT("accepts a username and password");
ShimClient shimClient; ShimClient shimClient;
@ -253,13 +280,47 @@ int test_connect_disconnect_connect() {
END_IT END_IT
} }
int test_connect_custom_keepalive() {
IT("sends a properly formatted connect packet with custom keepalive value");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte expectServer[] = { 172, 16, 0, 2 };
shimClient.expectConnect(expectServer,1883);
// Set keepalive to 300secs == 0x01 0x2c
byte connect[] = {0x10,0x18,0x0,0x4,0x4d,0x51,0x54,0x54,0x4,0x2,0x01,0x2c,0x0,0xc,0x63,0x6c,0x69,0x65,0x6e,0x74,0x5f,0x74,0x65,0x73,0x74,0x31};
byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.expect(connect,26);
shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient);
int state = client.state();
IS_TRUE(state == MQTT_DISCONNECTED);
client.setKeepAlive(300);
int rc = client.connect((char*)"client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
END_IT
}
int main() int main()
{ {
SUITE("Connect"); SUITE("Connect");
test_connect_fails_no_network(); test_connect_fails_no_network();
test_connect_fails_on_no_response(); test_connect_fails_on_no_response();
test_connect_properly_formatted(); test_connect_properly_formatted();
test_connect_non_clean_session();
test_connect_accepts_username_password(); test_connect_accepts_username_password();
test_connect_fails_on_bad_rc(); test_connect_fails_on_bad_rc();
test_connect_properly_formatted_hostname(); test_connect_properly_formatted_hostname();
@ -269,5 +330,7 @@ int main()
test_connect_with_will(); test_connect_with_will();
test_connect_with_will_username_password(); test_connect_with_will_username_password();
test_connect_disconnect_connect(); test_connect_disconnect_connect();
test_connect_custom_keepalive();
FINISH FINISH
} }

View File

@ -21,4 +21,6 @@ extern "C"{
#define PROGMEM #define PROGMEM
#define pgm_read_byte_near(x) *(x) #define pgm_read_byte_near(x) *(x)
#define yield(x) {}
#endif // Arduino_h #endif // Arduino_h

View File

@ -5,7 +5,7 @@
class Buffer { class Buffer {
private: private:
uint8_t buffer[1024]; uint8_t buffer[2048];
uint16_t pos; uint16_t pos;
uint16_t length; uint16_t length;

View File

@ -134,6 +134,7 @@ int test_publish_too_long() {
shimClient.respond(connack,4); shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient); PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(128);
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);

View File

@ -20,6 +20,7 @@ void reset_callback() {
} }
void callback(char* topic, byte* payload, unsigned int length) { void callback(char* topic, byte* payload, unsigned int length) {
TRACE("Callback received topic=[" << topic << "] length=" << length << "\n")
callback_called = true; callback_called = true;
strcpy(lastTopic,topic); strcpy(lastTopic,topic);
memcpy(lastPayload,payload,length); memcpy(lastPayload,payload,length);
@ -102,10 +103,15 @@ int test_receive_max_sized_message() {
shimClient.respond(connack,4); shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient); PubSubClient client(server, 1883, callback, shimClient);
int length = 80; // If this is changed to > 128 then the publish packet below
// is no longer valid as it assumes the remaining length
// is a single-byte. Don't make that mistake like I just
// did and lose a whole evening tracking down the issue.
client.setBufferSize(length);
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);
int length = MQTT_MAX_PACKET_SIZE;
byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64}; byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64};
byte bigPublish[length]; byte bigPublish[length];
memset(bigPublish,'A',length); memset(bigPublish,'A',length);
@ -137,11 +143,13 @@ int test_receive_oversized_message() {
byte connack[] = { 0x20, 0x02, 0x00, 0x00 }; byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.respond(connack,4); shimClient.respond(connack,4);
int length = 80; // See comment in test_receive_max_sized_message before changing this value
PubSubClient client(server, 1883, callback, shimClient); PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(length-1);
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);
int length = MQTT_MAX_PACKET_SIZE+1;
byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64}; byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64};
byte bigPublish[length]; byte bigPublish[length];
memset(bigPublish,'A',length); memset(bigPublish,'A',length);
@ -174,8 +182,8 @@ int test_drop_invalid_remaining_length_message() {
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);
byte publish[] = {0x30,0x92,0x92,0x92,0x92,0x92,0x92,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64}; byte publish[] = {0x30,0x92,0x92,0x92,0x92,0x01,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64};
shimClient.respond(publish,21); shimClient.respond(publish,20);
rc = client.loop(); rc = client.loop();
@ -188,9 +196,58 @@ int test_drop_invalid_remaining_length_message() {
END_IT END_IT
} }
int test_resize_buffer() {
IT("receives a message larger than the default maximum");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.respond(connack,4);
int length = 80; // See comment in test_receive_max_sized_message before changing this value
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(length-1);
int rc = client.connect((char*)"client_test1");
IS_TRUE(rc);
byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64};
byte bigPublish[length];
memset(bigPublish,'A',length);
bigPublish[length] = 'B';
memcpy(bigPublish,publish,16);
// Send it twice
shimClient.respond(bigPublish,length);
shimClient.respond(bigPublish,length);
rc = client.loop();
IS_TRUE(rc);
// First message fails as it is too big
IS_FALSE(callback_called);
// Resize the buffer
client.setBufferSize(length);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic,"topic")==0);
IS_TRUE(lastLength == length-9);
IS_TRUE(memcmp(lastPayload,bigPublish+9,lastLength)==0);
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_oversized_stream_message() { int test_receive_oversized_stream_message() {
IT("drops an oversized message"); IT("receive an oversized streamed message");
reset_callback(); reset_callback();
Stream stream; Stream stream;
@ -201,11 +258,13 @@ int test_receive_oversized_stream_message() {
byte connack[] = { 0x20, 0x02, 0x00, 0x00 }; byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.respond(connack,4); shimClient.respond(connack,4);
int length = 80; // See comment in test_receive_max_sized_message before changing this value
PubSubClient client(server, 1883, callback, shimClient, stream); PubSubClient client(server, 1883, callback, shimClient, stream);
client.setBufferSize(length-1);
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);
int length = MQTT_MAX_PACKET_SIZE+1;
byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64}; byte publish[] = {0x30,length-2,0x0,0x5,0x74,0x6f,0x70,0x69,0x63,0x70,0x61,0x79,0x6c,0x6f,0x61,0x64};
byte bigPublish[length]; byte bigPublish[length];
@ -222,7 +281,8 @@ int test_receive_oversized_stream_message() {
IS_TRUE(callback_called); IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic,"topic")==0); IS_TRUE(strcmp(lastTopic,"topic")==0);
IS_TRUE(lastLength == length-9);
IS_TRUE(lastLength == length-10);
IS_FALSE(stream.error()); IS_FALSE(stream.error());
IS_FALSE(shimClient.error()); IS_FALSE(shimClient.error());
@ -272,6 +332,7 @@ int main()
test_receive_max_sized_message(); test_receive_max_sized_message();
test_drop_invalid_remaining_length_message(); test_drop_invalid_remaining_length_message();
test_receive_oversized_message(); test_receive_oversized_message();
test_resize_buffer();
test_receive_oversized_stream_message(); test_receive_oversized_stream_message();
test_receive_qos1(); test_receive_qos1();

View File

@ -106,6 +106,7 @@ int test_subscribe_too_long() {
shimClient.respond(connack,4); shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient); PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(128);
int rc = client.connect((char*)"client_test1"); int rc = client.connect((char*)"client_test1");
IS_TRUE(rc); IS_TRUE(rc);