[PubSubClient] Update to new fork PubSubClient3

This commit is contained in:
TD-er
2025-06-28 23:20:27 +02:00
parent 8275eb743d
commit 7fb4d4e821
67 changed files with 3563 additions and 2818 deletions
+8
View File
@@ -0,0 +1,8 @@
BasedOnStyle: Google
Language: Cpp
IndentWidth: 4
ColumnLimit: 155
AllowShortFunctionsOnASingleLine: Empty
AlignConsecutiveMacros: Consecutive
SkipMacroDefinitionBody: true
LineEnding: LF
+17
View File
@@ -0,0 +1,17 @@
root = true
[*]
charset = utf-8
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true
[*.{c,cpp}]
indent_style = space
indent_size = 4
max_line_length = 155
[*.py]
indent_style = space
indent_size = 4
max_line_length = 155
+53
View File
@@ -0,0 +1,53 @@
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Enable the globstar shell option
shopt -s globstar
# Make sure we are inside the github workspace
cd $GITHUB_WORKSPACE
# Create directories
mkdir $HOME/Arduino
mkdir $HOME/Arduino/libraries
# Install Arduino IDE
export PATH=$PATH:$GITHUB_WORKSPACE/bin
curl -fsSL https://raw.githubusercontent.com/arduino/arduino-cli/master/install.sh | sh
arduino-cli config init
arduino-cli config add board_manager.additional_urls http://arduino.esp8266.com/stable/package_esp8266com_index.json
arduino-cli config set library.enable_unsafe_install true
arduino-cli core update-index
#arduino-cli core search
# Install Arduino cores
arduino-cli core install arduino:avr
arduino-cli core install esp8266:esp8266
arduino-cli core install esp32:esp32
#arduino-cli board listall
# Install Arduino libs
arduino-cli lib install Ethernet
arduino-cli lib install --git-url https://github.com/ennui2342/arduino-sram
# Link the project to the Arduino library
ln -s $GITHUB_WORKSPACE $HOME/Arduino/libraries/CI_Test_Library
# Compile all *.ino files for the Arduino
for f in **/*.ino ; do
if [[ "$f" != *mqtt_esp8266.ino && "$f" != *mqtt_large_message.ino ]]; then
echo "################################################################"
echo "Arduino Uno compiling file ${f}"
arduino-cli compile -b arduino:avr:uno $f
fi
done
# Compile all *.ino files for the ESP32
for f in **/*.ino ; do
if [[ "$f" != *"mqtt_stream.ino" ]]; then
echo "################################################################"
echo "ESP32 compiling file ${f}"
arduino-cli compile -b esp32:esp32:esp32 $f
fi
done
# Compile all *.ino files for the ESP8266
for f in **/*.ino ; do
echo "################################################################"
echo "ESP8266 compiling file ${f}"
arduino-cli compile -b esp8266:esp8266:generic $f
done
+21
View File
@@ -0,0 +1,21 @@
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Enable the globstar shell option
shopt -s globstar
# Make sure we are inside the github workspace
cd $GITHUB_WORKSPACE
# Install clang-format (if not already installed)
#command -v clang-format >/dev/null 2>&1 || sudo apt-get -y install clang-format
# need Ubuntu clang-format version 19.1.1 (1ubuntu1~24.04.2)
# default Ubuntu clang-format version 18.1.3 (1ubuntu1) is not working
sudo apt-get -y install clang-format-19
# Check clang-format output
for f in **/*.{h,c,hpp,cpp,ino} ; do
#if [ -f "$f" ] && [[ "$f" != "tests/"* ]]; then
if [ -f "$f" ]; then
echo "################################################################"
echo "Checking file ${f}"
diff $f <(clang-format-19 -assume-filename=main.cpp $f) 1>&2
fi
done
+31
View File
@@ -0,0 +1,31 @@
name: build
on:
push:
branches: [ "master" ]
pull_request:
branches: [ "master" ]
jobs:
build:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v2
- name: Check clang-format conformity
run: bash .github/clang-lint.sh
- name: Build tests
# Build / Execute tests defined in tests folder
working-directory: ${{ github.workspace }}/tests
run: make
- name: Execute tests
# Build / Execute tests defined in tests folder
working-directory: ${{ github.workspace }}/tests
#run: make test
run: bin/connect_spec && bin/publish_spec && bin/receive_spec && bin/subscribe_spec
- name: Build on Arduino CLI
run: bash .github/build-arduino.sh
+5
View File
@@ -0,0 +1,5 @@
tests/bin
.pioenvs
.piolibdeps
.clang_complete
.gcc-flags.json
+7
View File
@@ -0,0 +1,7 @@
sudo: false
language: cpp
compiler:
- g++
script: cd tests && make && make test
os:
- linux
+136
View File
@@ -0,0 +1,136 @@
3.1.0
* Fix setServer(domain, ...) by copy domain string to own buffer (prevent potential dangling string pointer) #10
* Connect to broker only if port != 0 (e.g. in case of incorrect PubSubClient initialization) #10
* Refactored different buffer length types to size_t #11
* Added a bunch of warnings and -Werror to tests Makefile #11
* Reformatted tests code using Google sytle (see .clang-format) #11
* Fix keepalive handling, thanks to @uschnindler providing this to thingsboard #14
* Use initializer lists instead of assigning global members in constructor #15
* Fix DEBUG_PSC_PRINTF outputs #18
* Added ERROR_PSC_PRINTF and ERROR_PSC_PRINTF_P outputs #25
* Introduce MQTTRETAINED flag instead of hard coding #21
* Added documentation on readByte() and readPacket() functions #23
* Refactored buildHeader() (change return type, use MQTT_MAX_HEADER_SIZE, locals) #22, #34, #37
* Refactored readPacket() (changed return type to size_t, used MQTT_MAX_HEADER_SIZE, locals) #23
* Refactored readByte() #23
* Refactored beginPublish() and endPublish() #28
* Refactored publish() #31
* Refactored publish_P() #30
* Refactored connect() #40
* Refactored connected() (mostly rewritten and simplified, added setting pingOutstanding, don't need flush()) #44
* Refactored subscribe() and unsubscribe() #44
* Refactored internal socketTimeout and keepAlive to millis to improve performance #33, #38, #39
* Fix potential memory corruption in MQTTPUBLISH callback preparation #25
* Fix potential error in readByte() (missing failure test on _client->read()) #32
* Switch from constructor init-lists to class member initialization, see C++ core guideline C.45 #42
3.0.2
* Added github workflow to execute tests
* Added github workflow to compile examples using arduino-cli
* Examples: Use a better source of random #9
* Deactivate use of std::function on demand #7
3.0.1
* Reformatted code using Google sytle (see .clang-format)
* Added .editorconfig
* Added github workflow to check code sytle on push and pull request
* Fix-tests: missing def for strlen_P & Python libs
3.0
* Maintenance taken over by Holger Mueller
* Adds in-source documentation to PubSubClient header.
* Add flag for enabling debugging of library
* Always use PubSubClient() constructor
* Use bool instead of boolean
* Add yield() calls in connect() and write() to avoid wdt resets if either blocks for too long
* Fix bug in publish_P which will always treat the payload length as 0 when used with PROGMEM
* Fix increase `bytesToWrite` to uint16_t to prevent overflow
* Remove compiler warning `expectedLength` should be unsigned
* Extend usage of std::function on all the platforms where it's available
* Fix for keep alive zero
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
* Add MQTT_SOCKET_TIMEOUT to prevent it blocking indefinitely
whilst waiting for inbound data
* Fixed return code when publishing >256 bytes
2.3
* Add publish(topic,payload,retained) function
2.2
* Change code layout to match Arduino Library reqs
2.1
* Add MAX_TRANSFER_SIZE def to chunk messages if needed
* Reject topic/payloads that exceed MQTT_MAX_PACKET_SIZE
2.0
* Add (and default to) MQTT 3.1.1 support
* Fix PROGMEM handling for Intel Galileo/ESP8266
* Add overloaded constructors for convenience
* Add chainable setters for server/callback/client/stream
* Add state function to return connack return code
1.9
* Do not split MQTT packets over multiple calls to _client->write()
* API change: All constructors now require an instance of Client
to be passed in.
* Fixed example to match 1.8 api changes - dpslwk
* Added username/password support - WilHall
* Added publish_P - publishes messages from PROGMEM - jobytaffey
1.8
* KeepAlive interval is configurable in PubSubClient.h
* Maximum packet size is configurable in PubSubClient.h
* API change: Return boolean rather than int from various functions
* API change: Length parameter in message callback changed
from int to unsigned int
* Various internal tidy-ups around types
1.7
* Improved keepalive handling
* Updated to the Arduino-1.0 API
1.6
* Added the ability to publish a retained message
1.5
* Added default constructor
* Fixed compile error when used with arduino-0021 or later
1.4
* Fixed connection lost handling
1.3
* Fixed packet reading bug in PubSubClient.readPacket
1.2
* Fixed compile error when used with arduino-0016 or later
1.1
* Reduced size of library
* Added support for Will messages
* Clarified licensing - see LICENSE.txt
1.0
* Only Quality of Service (QOS) 0 messaging is supported
* The maximum message size, including header, is 128 bytes
* The keepalive interval is set to 30 seconds
* No support for Will messages
+21
View File
@@ -0,0 +1,21 @@
Copyright (c) 2008-2020 Nicholas O'Leary
Copyright (c) 2025 Holger Mueller
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+67
View File
@@ -0,0 +1,67 @@
# Arduino Client for MQTT
This library provides a client for doing simple publish/subscribe messaging with
a server that supports MQTT.
## Notes of this fork
This is a fork of the repository [knolleary/pubsubclient v2.8](https://github.com/knolleary/pubsubclient/releases/tag/v2.8), which was last updated in May 20, 2020. There was an update approach in [#1045](https://github.com/knolleary/pubsubclient/issues/1045), but it's also stale.
I tried lot's of different other MQTT libs, but they need more resources than PubSubClient or lacking maintenance as well:
- https://github.com/256dpi/arduino-mqtt
- https://github.com/hideakitai/MQTTPubSubClient
- https://github.com/bertmelis/espMqttClient
- https://github.com/arduino-libraries/ArduinoMqttClient
- https://github.com/thingsboard/pubsubclient
Since there was no progress I decided to merge the most important PRs manually and publish a new major version. I also renamed to PubSubClient3 to have a similar but different name of the library.
I appreciate every contribution to this library.
## Examples
The library comes with a number of example sketches. See File > Examples > PubSubClient
within the Arduino application.
Full API documentation is available here: https://pubsubclient.knolleary.net
## Limitations
- The client is based on the [MQTT Version 3.1.1 specification](https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html) with some limitations.
- It can only publish QoS 0 messages. It can subscribe at QoS 0 or QoS 1.
- The maximum message size, including header, is **256 bytes** by default. This
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
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
changing value of `MQTT_VERSION` in `PubSubClient.h`.
## Compatible Hardware
The library uses the Arduino Ethernet Client api for interacting with the
underlying network hardware. This means it Just Works with a growing number of
boards and shields, including:
- Arduino Ethernet
- Arduino Ethernet Shield
- Arduino YUN use the included `YunClient` in place of `EthernetClient`, and
be sure to do a `Bridge.begin()` first
- Arduino WiFi Shield - if you want to send packets > 90 bytes with this shield,
enable the `MQTT_MAX_TRANSFER_SIZE` define in `PubSubClient.h`.
- Sparkfun WiFly Shield [library](https://github.com/dpslwk/WiFly)
- TI CC3000 WiFi - [library](https://github.com/sparkfun/SFE_CC3000_Library)
- Intel Galileo/Edison
- ESP8266
- ESP32
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
[alternative library](https://github.com/njh/NanodeMQTT) available.
## License
This code is released under the MIT License.
@@ -0,0 +1,41 @@
/*
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 <Ethernet.h>
#include <PubSubClient.h>
#include <SPI.h>
// Update these with values suitable for your network.
byte mac[] = {0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED};
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
void callback(char* topic, uint8_t* payload, size_t length) {
// handle message arrived
}
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient);
void setup() {
Ethernet.begin(mac, ip);
// Note - the default maximum packet size is 128 bytes. If the
// combined length of clientId, username and password exceed this use the
// following to increase the buffer size:
// client.setBufferSize(255);
if (client.connect("arduinoClient", "testuser", "testpass")) {
client.publish("outTopic", "hello world");
client.subscribe("inTopic");
}
}
void loop() {
client.loop();
}
@@ -0,0 +1,74 @@
/*
Basic MQTT example
This sketch demonstrates the basic capabilities of the library.
It connects to an MQTT server then:
- publishes "hello world" to the topic "outTopic"
- subscribes to the topic "inTopic", printing out any messages
it receives. NB - it assumes the received payloads are strings not binary
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
achieve the same result without blocking the main loop.
*/
#include <Ethernet.h>
#include <PubSubClient.h>
#include <SPI.h>
// Update these with values suitable for your network.
byte mac[] = {0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED};
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
void callback(char* topic, uint8_t* payload, size_t length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (size_t i = 0; i < length; i++) {
Serial.print((char)payload[i]);
}
Serial.println();
}
EthernetClient ethClient;
PubSubClient client(ethClient);
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect("arduinoClient")) {
Serial.println("connected");
// Once connected, publish an announcement...
client.publish("outTopic", "hello world");
// ... and resubscribe
client.subscribe("inTopic");
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
// Wait 5 seconds before retrying
delay(5000);
}
}
}
void setup() {
Serial.begin(57600);
client.setServer(server, 1883);
client.setCallback(callback);
Ethernet.begin(mac, ip);
// Allow the hardware to sort itself out
delay(1500);
}
void loop() {
if (!client.connected()) {
reconnect();
}
client.loop();
}
@@ -0,0 +1,135 @@
/*
Basic ESP8266 / ESP32 MQTT example
This sketch demonstrates the capabilities of the pubsub library in combination
with the ESP8266 board/library.
It connects to an MQTT server then:
- publishes "hello world" to the topic "outTopic" every two seconds
- subscribes to the topic "inTopic", printing out any messages
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,
else switch it off
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
achieve the same result without blocking the main loop.
To install the ESP8266 board, (using Arduino 1.6.4+):
- Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs":
http://arduino.esp8266.com/stable/package_esp8266com_index.json
- Open the "Tools -> Board -> Board Manager" and click install for the ESP8266"
- Select your ESP8266 in "Tools -> Board"
*/
#ifdef ESP8266
#include <ESP8266WiFi.h>
#elif defined(ESP32)
#include <WiFi.h>
#include <esp_random.h>
#define BUILTIN_LED A0
#define RANDOM_REG32 esp_random()
#else
#error Platform not supported.
#endif
#include <PubSubClient.h>
// Update these with values suitable for your network.
const char* ssid = "........";
const char* password = "........";
const char* mqtt_server = "broker.mqtt-dashboard.com";
WiFiClient espClient;
PubSubClient client(espClient);
unsigned long lastMsg = 0;
#define MSG_BUFFER_SIZE (50)
char msg[MSG_BUFFER_SIZE];
int value = 0;
void setup_wifi() {
delay(10);
// We start by connecting to a WiFi network
Serial.println();
Serial.print("Connecting to ");
Serial.println(ssid);
WiFi.mode(WIFI_STA);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
randomSeed(RANDOM_REG32);
Serial.println("");
Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
}
void callback(char* topic, uint8_t* payload, size_t length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (size_t i = 0; i < length; i++) {
Serial.print((char)payload[i]);
}
Serial.println();
// Switch on the LED if an 1 was received as first character
if ((char)payload[0] == '1') {
digitalWrite(BUILTIN_LED, LOW); // Turn the LED on (Note that LOW is the voltage level
// but actually the LED is on; this is because
// it is active low on the ESP-01)
} else {
digitalWrite(BUILTIN_LED, HIGH); // Turn the LED off by making the voltage HIGH
}
}
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Create a random client ID
String clientId = "ESP8266Client-";
clientId += String(random(0xffff), HEX);
// Attempt to connect
if (client.connect(clientId.c_str())) {
Serial.println("connected");
// Once connected, publish an announcement...
client.publish("outTopic", "hello world");
// ... and resubscribe
client.subscribe("inTopic");
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
// Wait 5 seconds before retrying
delay(5000);
}
}
}
void setup() {
pinMode(BUILTIN_LED, OUTPUT); // Initialize the BUILTIN_LED pin as an output
Serial.begin(115200);
setup_wifi();
client.setServer(mqtt_server, 1883);
client.setCallback(callback);
}
void loop() {
if (!client.connected()) {
reconnect();
}
client.loop();
unsigned long now = millis();
if (now - lastMsg > 2000) {
lastMsg = now;
++value;
snprintf(msg, MSG_BUFFER_SIZE, "hello world #%ld", value);
Serial.print("Publish message: ");
Serial.println(msg);
client.publish("outTopic", msg);
}
}
@@ -0,0 +1,185 @@
/*
Long message ESP8266 / ESP32 MQTT example
This sketch demonstrates sending arbitrarily large messages in combination
with the ESP8266 board/library.
It connects to an MQTT server then:
- publishes "hello world" to the topic "outTopic"
- subscribes to the topic "greenBottles/#", printing out any messages
it receives. NB - it assumes the received payloads are strings not binary
- If the sub-topic is a number, it publishes a "greenBottles/lyrics" message
with a payload consisting of the lyrics to "10 green bottles", replacing
10 with the number given in the sub-topic.
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
achieve the same result without blocking the main loop.
To install the ESP8266 board, (using Arduino 1.6.4+):
- Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs":
http://arduino.esp8266.com/stable/package_esp8266com_index.json
- Open the "Tools -> Board -> Board Manager" and click install for the ESP8266"
- Select your ESP8266 in "Tools -> Board"
*/
#ifdef ESP8266
#include <ESP8266WiFi.h>
#elif defined(ESP32)
#include <WiFi.h>
#include <esp_random.h>
#define BUILTIN_LED A0
#define RANDOM_REG32 esp_random()
#else
#error Platform not supported.
#endif
#include <PubSubClient.h>
// Update these with values suitable for your network.
const char* ssid = "........";
const char* password = "........";
const char* mqtt_server = "broker.mqtt-dashboard.com";
WiFiClient espClient;
PubSubClient client(espClient);
long lastMsg = 0;
char msg[50];
int value = 0;
void setup_wifi() {
delay(10);
// We start by connecting to a WiFi network
Serial.println();
Serial.print("Connecting to ");
Serial.println(ssid);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
randomSeed(RANDOM_REG32);
Serial.println("");
Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
}
void callback(char* topic, uint8_t* payload, size_t length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (size_t i = 0; i < length; i++) {
Serial.print((char)payload[i]);
}
Serial.println();
// Find out how many bottles we should generate lyrics for
String topicStr(topic);
int bottleCount = 0; // assume no bottles unless we correctly parse a value from the topic
if (topicStr.indexOf('/') >= 0) {
// The topic includes a '/', we'll try to read the number of bottles from just after that
topicStr.remove(0, topicStr.indexOf('/') + 1);
// Now see if there's a number of bottles after the '/'
bottleCount = topicStr.toInt();
}
if (bottleCount > 0) {
// Work out how big our resulting message will be
size_t msgLen = 0;
for (size_t i = bottleCount; i > 0; i--) {
String numBottles(i);
msgLen += 2 * numBottles.length();
if (i == 1) {
msgLen += 2 * String(" green bottle, standing on the wall\n").length();
} else {
msgLen += 2 * String(" green bottles, standing on the wall\n").length();
}
msgLen += String("And if one green bottle should accidentally fall\nThere'll be ").length();
switch (i) {
case 1:
msgLen += String("no green bottles, standing on the wall\n\n").length();
break;
case 2:
msgLen += String("1 green bottle, standing on the wall\n\n").length();
break;
default:
numBottles = i - 1;
msgLen += numBottles.length();
msgLen += String(" green bottles, standing on the wall\n\n").length();
break;
};
}
// Now we can start to publish the message
client.beginPublish("greenBottles/lyrics", msgLen, false);
for (int i = bottleCount; i > 0; i--) {
for (int j = 0; j < 2; j++) {
client.print(i);
if (i == 1) {
client.print(" green bottle, standing on the wall\n");
} else {
client.print(" green bottles, standing on the wall\n");
}
}
client.print("And if one green bottle should accidentally fall\nThere'll be ");
switch (i) {
case 1:
client.print("no green bottles, standing on the wall\n\n");
break;
case 2:
client.print("1 green bottle, standing on the wall\n\n");
break;
default:
client.print(i - 1);
client.print(" green bottles, standing on the wall\n\n");
break;
};
}
// Now we're done!
client.endPublish();
}
}
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Create a random client ID
String clientId = "ESP8266Client-";
clientId += String(random(0xffff), HEX);
// Attempt to connect
if (client.connect(clientId.c_str())) {
Serial.println("connected");
// Once connected, publish an announcement...
client.publish("outTopic", "hello world");
// ... and resubscribe
client.subscribe("greenBottles/#");
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
// Wait 5 seconds before retrying
delay(5000);
}
}
}
void setup() {
pinMode(BUILTIN_LED, OUTPUT); // Initialize the BUILTIN_LED pin as an output
Serial.begin(115200);
setup_wifi();
client.setServer(mqtt_server, 1883);
client.setCallback(callback);
}
void loop() {
if (!client.connected()) {
reconnect();
}
client.loop();
}
@@ -0,0 +1,56 @@
/*
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 <Ethernet.h>
#include <PubSubClient.h>
#include <SPI.h>
// Update these with values suitable for your network.
byte mac[] = {0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED};
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
// Callback function header
void callback(char* topic, uint8_t* payload, size_t length);
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient);
// Callback function
void callback(char* topic, uint8_t* payload, size_t length) {
// In order to republish this payload, a copy must be made
// as the orignal payload buffer will be overwritten whilst
// constructing the PUBLISH packet.
// Allocate the correct amount of memory for the payload copy
byte* p = (byte*)malloc(length);
// Copy the payload to the new buffer
memcpy(p, payload, length);
client.publish("outTopic", p, length);
// Free the memory
free(p);
}
void setup() {
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic", "hello world");
client.subscribe("inTopic");
}
}
void loop() {
client.loop();
}
@@ -0,0 +1,62 @@
/*
Reconnecting MQTT example - non-blocking
This sketch demonstrates how to keep the client connected
using a non-blocking reconnect function. If the client loses
its connection, it attempts to reconnect every 5 seconds
without blocking the main loop.
*/
#include <Ethernet.h>
#include <PubSubClient.h>
#include <SPI.h>
// Update these with values suitable for your hardware/network.
byte mac[] = {0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED};
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
void callback(char* topic, uint8_t* payload, size_t length) {
// handle message arrived
}
EthernetClient ethClient;
PubSubClient client(ethClient);
long lastReconnectAttempt = 0;
boolean reconnect() {
if (client.connect("arduinoClient")) {
// Once connected, publish an announcement...
client.publish("outTopic", "hello world");
// ... and resubscribe
client.subscribe("inTopic");
}
return client.connected();
}
void setup() {
client.setServer(server, 1883);
client.setCallback(callback);
Ethernet.begin(mac, ip);
delay(1500);
lastReconnectAttempt = 0;
}
void loop() {
if (!client.connected()) {
unsigned long now = millis();
if (now - lastReconnectAttempt > 5000) {
lastReconnectAttempt = now;
// Attempt to reconnect
if (reconnect()) {
lastReconnectAttempt = 0;
}
}
} else {
// Client connected
client.loop();
}
}
@@ -0,0 +1,55 @@
/*
Example of using a Stream object to store the message payload
Uses SRAM library: https://github.com/ennui2342/arduino-sram
but could use any Stream based class such as SD
- connects to an MQTT server
- publishes "hello world" to the topic "outTopic"
- subscribes to the topic "inTopic"
*/
#include <Ethernet.h>
#include <PubSubClient.h>
#include <SPI.h>
#include <SRAM.h>
// Update these with values suitable for your network.
byte mac[] = {0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED};
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
SRAM sram(4, SRAM_1024);
void callback(char* topic, uint8_t* payload, size_t length) {
sram.seek(1);
// do something with the message
for (size_t i = 0; i < length; i++) {
Serial.write(sram.read());
}
Serial.println();
// Reset position for the next message to be stored
sram.seek(1);
}
EthernetClient ethClient;
PubSubClient client(server, 1883, callback, ethClient, sram);
void setup() {
Ethernet.begin(mac, ip);
if (client.connect("arduinoClient")) {
client.publish("outTopic", "hello world");
client.subscribe("inTopic");
}
sram.begin();
sram.seek(1);
Serial.begin(9600);
}
void loop() {
client.loop();
}
+36
View File
@@ -0,0 +1,36 @@
#######################################
# Syntax Coloring Map For PubSubClient
#######################################
#######################################
# Datatypes (KEYWORD1)
#######################################
PubSubClient KEYWORD1
#######################################
# Methods and Functions (KEYWORD2)
#######################################
connect KEYWORD2
disconnect KEYWORD2
publish KEYWORD2
publish_P KEYWORD2
beginPublish KEYWORD2
endPublish KEYWORD2
write KEYWORD2
subscribe KEYWORD2
unsubscribe KEYWORD2
loop KEYWORD2
connected KEYWORD2
setServer KEYWORD2
setCallback KEYWORD2
setClient KEYWORD2
setStream KEYWORD2
setKeepAlive KEYWORD2
setBufferSize KEYWORD2
setSocketTimeout KEYWORD2
#######################################
# Constants (LITERAL1)
#######################################
+19
View File
@@ -0,0 +1,19 @@
{
"name": "PubSubClient3",
"keywords": "ethernet, mqtt, m2m, iot",
"description": "A client library for MQTT messaging. MQTT is a lightweight messaging protocol ideal for small devices. This library allows you to send and receive MQTT messages. It supports the latest MQTT 3.1.1 protocol and can be configured to use the older MQTT 3.1 if needed. It supports all Arduino Ethernet Client compatible hardware, including the Intel Galileo/Edison, ESP8266 and TI CC3000.",
"repository": {
"type": "git",
"url": "https://github.com/hmueller01/pubsubclient3.git"
},
"version": "3.1.0",
"license": "MIT",
"exclude": "tests",
"examples": "examples/*/*.ino",
"frameworks": "arduino",
"platforms": [
"atmelavr",
"espressif8266",
"espressif32"
]
}
+10
View File
@@ -0,0 +1,10 @@
name=PubSubClient3
version=3.1.0
author=Nick O'Leary <nick.oleary@gmail.com>
maintainer=Holger Mueller <github@euhm.de>
license=MIT
sentence=A client library for MQTT messaging.
paragraph=MQTT is a lightweight messaging protocol ideal for small devices. This library allows you to send and receive MQTT messages. It supports the latest MQTT 3.1.1 protocol and can be configured to use the older MQTT 3.1 if needed. It supports all Arduino Ethernet Client compatible hardware, including the Intel Galileo/Edison, ESP8266 and TI CC3000.
category=Communication
url=https://github.com/hmueller01/pubsubclient3.git
architectures=*
+772
View File
@@ -0,0 +1,772 @@
/*
PubSubClient.cpp - A simple client for MQTT.
Nick O'Leary, Holger Mueller
http://knolleary.net
https://github.com/hmueller01/pubsubclient3
*/
#include "PubSubClient.h"
PubSubClient::PubSubClient() {
setBufferSize(MQTT_MAX_PACKET_SIZE);
setKeepAlive(MQTT_KEEPALIVE);
setSocketTimeout(MQTT_SOCKET_TIMEOUT);
}
PubSubClient::PubSubClient(Client& client) : PubSubClient() {
setClient(client);
}
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) : PubSubClient() {
setServer(addr, port);
setClient(client);
}
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) : PubSubClient() {
setServer(addr, port);
setClient(client);
setStream(stream);
}
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) : PubSubClient() {
setServer(addr, port);
setCallback(callback);
setClient(client);
}
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) : PubSubClient() {
setServer(addr, port);
setCallback(callback);
setClient(client);
setStream(stream);
}
PubSubClient::PubSubClient(uint8_t* ip, uint16_t port, Client& client) : PubSubClient() {
setServer(ip, port);
setClient(client);
}
PubSubClient::PubSubClient(uint8_t* ip, uint16_t port, Client& client, Stream& stream) : PubSubClient() {
setServer(ip, port);
setClient(client);
setStream(stream);
}
PubSubClient::PubSubClient(uint8_t* ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) : PubSubClient() {
setServer(ip, port);
setCallback(callback);
setClient(client);
}
PubSubClient::PubSubClient(uint8_t* ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) : PubSubClient() {
setServer(ip, port);
setCallback(callback);
setClient(client);
setStream(stream);
}
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) : PubSubClient() {
setServer(domain, port);
setClient(client);
}
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) : PubSubClient() {
setServer(domain, port);
setClient(client);
setStream(stream);
}
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) : PubSubClient() {
setServer(domain, port);
setCallback(callback);
setClient(client);
}
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) : PubSubClient() {
setServer(domain, port);
setCallback(callback);
setClient(client);
setStream(stream);
}
PubSubClient::~PubSubClient() {
free(this->domain);
free(this->buffer);
}
bool PubSubClient::connect(const char* id) {
return connect(id, nullptr, nullptr, 0, 0, 0, 0, 1);
}
bool PubSubClient::connect(const char* id, const char* user, const char* pass) {
return connect(id, user, pass, 0, 0, 0, 0, 1);
}
bool PubSubClient::connect(const char* id, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage) {
return connect(id, nullptr, nullptr, willTopic, willQos, willRetain, willMessage, 1);
}
bool PubSubClient::connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain,
const char* willMessage) {
return connect(id, user, pass, willTopic, willQos, willRetain, willMessage, 1);
}
bool PubSubClient::connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain,
const char* willMessage, bool cleanSession) {
if (!connected()) {
int result = 0;
if (_client->connected()) {
result = 1;
} else if (this->port != 0) {
if (this->domain) {
result = _client->connect(this->domain, this->port);
} else {
result = _client->connect(this->ip, this->port);
}
}
if (result == 1) {
nextMsgId = 1; // init msgId (packet identifier)
#if MQTT_VERSION == MQTT_VERSION_3_1
const uint8_t protocol[9] = {0x00, 0x06, 'M', 'Q', 'I', 's', 'd', 'p', MQTT_VERSION};
#elif MQTT_VERSION == MQTT_VERSION_3_1_1
const uint8_t protocol[7] = {0x00, 0x04, 'M', 'Q', 'T', 'T', MQTT_VERSION};
#endif
// Leave room in the buffer for header and variable length field
memcpy(this->buffer + MQTT_MAX_HEADER_SIZE, protocol, sizeof(protocol));
size_t length = MQTT_MAX_HEADER_SIZE + sizeof(protocol);
uint8_t flags = 0x00;
if (willTopic) {
flags = (0x01 << 2) | (willQos << 3) | (willRetain << 5); // set will flag bit 2, will QoS and will retain bit 5
}
if (cleanSession) {
flags = flags | (0x01 << 1); // set clean session bit 1
}
if (user) {
flags = flags | (0x01 << 7); // set user name flag bit 7
if (pass) {
flags = flags | (0x01 << 6); // set password flag bit 6
}
}
const uint16_t keepAlive = this->keepAliveMillis / 1000;
this->buffer[length++] = flags;
this->buffer[length++] = keepAlive >> 8;
this->buffer[length++] = keepAlive & 0xFF;
CHECK_STRING_LENGTH(length, id)
length = writeString(id, this->buffer, length);
if (willTopic) {
CHECK_STRING_LENGTH(length, willTopic)
length = writeString(willTopic, this->buffer, length);
CHECK_STRING_LENGTH(length, willMessage)
length = writeString(willMessage, this->buffer, length);
}
if (user) {
CHECK_STRING_LENGTH(length, user)
length = writeString(user, this->buffer, length);
if (pass) {
CHECK_STRING_LENGTH(length, pass)
length = writeString(pass, this->buffer, length);
}
}
write(MQTTCONNECT, this->buffer, length - MQTT_MAX_HEADER_SIZE);
lastInActivity = lastOutActivity = millis();
pingOutstanding = false;
while (!_client->available()) {
yield();
unsigned long t = millis();
if (t - lastInActivity >= this->socketTimeoutMillis) {
DEBUG_PSC_PRINTF("connect aborting due to timeout\n");
_state = MQTT_CONNECTION_TIMEOUT;
_client->stop();
return false;
}
}
uint8_t hdrLen;
size_t len = readPacket(&hdrLen);
if (len == 4) {
if (buffer[3] == 0) {
lastInActivity = millis();
_state = MQTT_CONNECTED;
return true;
} else {
_state = buffer[3];
}
}
DEBUG_PSC_PRINTF("connect aborting due to protocol error\n");
_client->stop();
} else {
_state = MQTT_CONNECT_FAILED;
}
return false;
}
return true;
}
bool PubSubClient::connected() {
if (!_client) return false;
if (_client->connected()) {
return (_state == MQTT_CONNECTED);
} else if (_state == MQTT_CONNECTED) {
DEBUG_PSC_PRINTF("lost connection (client may have more details)\n");
_state = MQTT_CONNECTION_LOST;
_client->stop();
pingOutstanding = false;
}
return false;
}
void PubSubClient::disconnect() {
DEBUG_PSC_PRINTF("disconnect called\n");
this->buffer[0] = MQTTDISCONNECT;
this->buffer[1] = 0;
_client->write(this->buffer, 2);
_state = MQTT_DISCONNECTED;
_client->flush();
_client->stop();
lastInActivity = lastOutActivity = millis();
pingOutstanding = false;
}
/**
* @brief Reads a byte into result.
*
* @param result Pointer to result buffer.
* @return true if byte was read, false if socketTimeout occurred.
*/
bool PubSubClient::readByte(uint8_t* result) {
unsigned long previousMillis = millis();
while (!_client->available()) {
yield();
unsigned long currentMillis = millis();
if (currentMillis - previousMillis >= this->socketTimeoutMillis) {
return false;
}
}
int rc = _client->read();
if (rc < 0) {
return false;
}
*result = (uint8_t)rc;
return true;
}
/**
* @brief Reads a byte into result[*pos] and increments *pos.
* Note: *pos may go out of bounds of result. This must be checked outside of this function!
*
* @return true if a byte was read, otherwise false (socketTimeout).
*/
bool PubSubClient::readByte(uint8_t* result, size_t* pos) {
uint8_t* write_address = &(result[*pos]);
if (readByte(write_address)) {
(*pos)++;
return true;
}
return false;
}
/**
* @brief Reads a complete packet (header, topic, payload) into this->buffer.
*
* @param *hdrLen Returns the variable header length send by MQTT broker (1 .. MQTT_MAX_HEADER_SIZE - 1)
* @return Number of read bytes, 0 in case of an error (socketTimeout, buffer overflow)
*/
size_t PubSubClient::readPacket(uint8_t* hdrLen) {
size_t len = 0;
if (!readByte(this->buffer, &len)) return 0;
bool isPublish = (this->buffer[0] & 0xF0) == MQTTPUBLISH;
uint32_t multiplier = 1;
size_t length = 0;
uint8_t digit = 0;
uint16_t skip = 0;
uint8_t start = 0;
do {
if (len == MQTT_MAX_HEADER_SIZE) {
// Invalid remaining length encoding - kill the connection
DEBUG_PSC_PRINTF("readPacket detected packet of invalid length\n");
_state = MQTT_DISCONNECTED;
_client->stop();
return 0;
}
if (!readByte(&digit)) return 0;
this->buffer[len++] = digit;
length += (digit & 0x7F) * multiplier; // length is coded in the lower 7 bits
multiplier <<= 7; // multiplier *= 128
} while ((digit & 0x80) != 0); // do while 8th continuation bit is set
*hdrLen = (uint8_t)(len - 1);
DEBUG_PSC_PRINTF("readPacket received packet of length %zu (isPublish = %u)\n", length, isPublish);
if (isPublish) {
// Read in topic length to calculate bytes to skip over for Stream writing
if (!readByte(this->buffer, &len)) return 0;
if (!readByte(this->buffer, &len)) return 0;
skip = (this->buffer[*hdrLen + 1] << 8) + this->buffer[*hdrLen + 2];
start = 2;
if (this->buffer[0] & MQTTQOS1) {
// skip message id
skip += 2;
}
}
size_t idx = len;
for (size_t i = start; i < length; i++) {
if (!readByte(&digit)) return 0;
if (this->stream) {
if (isPublish && idx - *hdrLen - 2 > skip) {
this->stream->write(digit);
}
}
if (len < this->bufferSize) {
this->buffer[len++] = digit;
}
idx++;
}
if (!this->stream && idx > this->bufferSize) {
DEBUG_PSC_PRINTF("readPacket ignoring packet of size %zu exceeding buffer of size %zu\n", length, this->bufferSize);
len = 0; // This will cause the packet to be ignored.
}
return len;
}
/**
* @brief After a packet is read handle the content here (call the callback, handle pings).
*
* @param hdrLen Variable header length send by MQTT broker (1 .. MQTT_MAX_HEADER_SIZE - 1).
* @param length Number of read bytes in this->buffer.
* @return true if packet was successfully processed, false if a buffer over or underflow occurred.
*/
bool PubSubClient::handlePacket(uint8_t hdrLen, size_t length) {
uint8_t type = this->buffer[0] & 0xF0;
DEBUG_PSC_PRINTF("received message of type %u\n", type);
switch (type) {
case MQTTPUBLISH:
if (callback) {
// MQTT Publish packet: See section 3.3 MQTT v3.1 protocol specification:
// - Header: 1 byte
// - Remaining header length: hdrLen bytes, multibyte field (1 .. MQTT_MAX_HEADER_SIZE - 1)
// - Topic length: 2 bytes (starts at buffer[hdrLen + 1])
// - Topic: topicLen bytes (starts at buffer[hdrLen + 3])
// - Packet Identifier (msgId): 0 bytes for QoS 0, 2 bytes for QoS 1 and 2 (starts at buffer[hdrLen + 3 + topicLen])
// - Payload (for QoS = 0): length - (hdrLen + 3 + topicLen) bytes (starts at buffer[hdrLen + 3 + topicLen])
// - Payload (for QoS > 0): length - (hdrLen + 5 + topicLen) bytes (starts at buffer[hdrLen + 5 + topicLen])
// To get a null reminated 'C' topic string we move the topic 1 byte to the front (overwriting the LSB of the topic lenght)
uint16_t topicLen = (this->buffer[hdrLen + 1] << 8) + this->buffer[hdrLen + 2]; // topic length in bytes
char* topic = (char*)(this->buffer + hdrLen + 3 - 1); // set the topic in the LSB of the topic lenght, as we move it there
uint16_t payloadOffset = hdrLen + 3 + topicLen; // payload starts after header and topic (if there is no packet identifier)
size_t payloadLen = length - payloadOffset; // this might change by 2 if we have a QoS 1 or 2 message
uint8_t* payload = this->buffer + payloadOffset;
if (length < payloadOffset) { // do not move outside the max bufferSize
ERROR_PSC_PRINTF_P("handlePacket(): Suspicious topicLen (%u) points outside of received buffer length (%zu)\n", topicLen, length);
return false;
}
memmove(topic, topic + 1, topicLen); // move topic inside buffer 1 byte to front
topic[topicLen] = '\0'; // end the topic as a 'C' string with \x00
if ((this->buffer[0] & 0x06) == MQTTQOS0) {
// No msgId for QOS == 0
callback(topic, payload, payloadLen);
} else {
// For QOS 1 and 2 we have a msgId (packet identifier) after the topic at the current payloadOffset
if (payloadLen < 2) { // payload must be >= 2, as we have the msgId before
ERROR_PSC_PRINTF_P("handlePacket(): Missing msgId in QoS 1/2 message\n");
return false;
}
uint16_t msgId = (this->buffer[payloadOffset] << 8) + this->buffer[payloadOffset + 1];
callback(topic, payload + 2, payloadLen - 2); // remove the msgId from the callback payload
this->buffer[0] = MQTTPUBACK;
this->buffer[1] = 2;
this->buffer[2] = (msgId >> 8);
this->buffer[3] = (msgId & 0xFF);
if (_client->write(this->buffer, 4) == 4) {
lastOutActivity = millis();
}
}
}
break;
case MQTTPINGREQ:
this->buffer[0] = MQTTPINGRESP;
this->buffer[1] = 0;
if (_client->write(this->buffer, 2) == 2) {
lastOutActivity = millis();
}
break;
case MQTTPINGRESP:
pingOutstanding = false;
break;
default:
break;
}
return true;
}
bool PubSubClient::loop() {
if (!connected()) {
return false;
}
bool ret = true;
const unsigned long t = millis();
if (keepAliveMillis && ((t - lastInActivity > this->keepAliveMillis) || (t - lastOutActivity > this->keepAliveMillis))) {
if (pingOutstanding) {
DEBUG_PSC_PRINTF("loop aborting due to timeout\n");
_state = MQTT_CONNECTION_TIMEOUT;
_client->stop();
pingOutstanding = false;
return false;
} else {
this->buffer[0] = MQTTPINGREQ;
this->buffer[1] = 0;
if (_client->write(this->buffer, 2) == 2) {
lastInActivity = lastOutActivity = t;
pingOutstanding = true;
}
}
}
if (_client->available()) {
uint8_t hdrLen;
size_t len = readPacket(&hdrLen);
if (len > 0) {
lastInActivity = t;
ret = handlePacket(hdrLen, len);
if (!ret) {
_state = MQTT_DISCONNECTED;
_client->stop();
}
} else if (!connected()) {
// readPacket has closed the connection
return false;
}
}
return ret;
}
bool PubSubClient::publish(const char* topic, const char* payload) {
return publish(topic, (const uint8_t*)payload, payload ? strnlen(payload, MQTT_MAX_POSSIBLE_PACKET_SIZE) : 0, false);
}
bool PubSubClient::publish(const char* topic, const char* payload, bool retained) {
return publish(topic, (const uint8_t*)payload, payload ? strnlen(payload, MQTT_MAX_POSSIBLE_PACKET_SIZE) : 0, retained);
}
bool PubSubClient::publish(const char* topic, const uint8_t* payload, size_t plength) {
return publish(topic, payload, plength, false);
}
bool PubSubClient::publish(const char* topic, const uint8_t* payload, size_t plength, bool retained) {
if (beginPublish(topic, plength, retained)) {
size_t rc = write(payload, plength);
lastOutActivity = millis();
return endPublish() && (rc == plength);
}
return false;
}
bool PubSubClient::publish_P(const char* topic, const char* payload, bool retained) {
return publish_P(topic, (const uint8_t*)payload, payload ? strnlen_P(payload, MQTT_MAX_POSSIBLE_PACKET_SIZE) : 0, retained);
}
bool PubSubClient::publish_P(const char* topic, const uint8_t* payload, size_t plength, bool retained) {
if (beginPublish(topic, plength, retained)) {
size_t rc = 0;
for (size_t i = 0; i < plength; i++) {
rc += _client->write((uint8_t)pgm_read_byte_near(payload + i));
}
lastOutActivity = millis();
return endPublish() && (rc == plength);
}
return false;
}
bool PubSubClient::beginPublish(const char* topic, size_t plength, bool retained) {
if (!topic) return false;
// check if the header and the topic (including 2 length bytes) fit into the buffer
if (connected() && MQTT_MAX_HEADER_SIZE + strlen(topic) + 2 <= this->bufferSize) {
// first write the topic at the end of the maximal variable header (MQTT_MAX_HEADER_SIZE) to the buffer
size_t topicLen = writeString(topic, this->buffer, MQTT_MAX_HEADER_SIZE) - MQTT_MAX_HEADER_SIZE;
// we now know the length of the topic string (lenght + 2 bytes signalling the length) and can build the variable header information
const uint8_t header = MQTTPUBLISH | (retained ? MQTTRETAINED : 0);
uint8_t hdrLen = buildHeader(header, this->buffer, topicLen + plength);
if (hdrLen == 0) return false; // exit here in case of header generation failure
// as the header length is variable, it starts at MQTT_MAX_HEADER_SIZE - hdrLen (see buildHeader() documentation)
size_t rc = _client->write(this->buffer + (MQTT_MAX_HEADER_SIZE - hdrLen), hdrLen + topicLen);
lastOutActivity = millis();
return (rc == (hdrLen + topicLen));
}
return false;
}
bool PubSubClient::endPublish() {
return connected();
}
/**
* @brief Build up the header ready to send.
* Note: the header is built at the end of the first MQTT_MAX_HEADER_SIZE bytes, so will start
* (MQTT_MAX_HEADER_SIZE - <returned size>) bytes into the buffer.
*
* @param header Header byte, e.g. MQTTCONNECT, MQTTPUBLISH, MQTTSUBSCRIBE, MQTTUNSUBSCRIBE.
* @param buf Buffer to write header to.
* @param length Length to encode in the header.
* @return Returns the size of the header (1 .. MQTT_MAX_HEADER_SIZE), or 0 in case of a failure (e.g. length to big).
*/
uint8_t PubSubClient::buildHeader(uint8_t header, uint8_t* buf, size_t length) {
uint8_t hdrBuf[MQTT_MAX_HEADER_SIZE - 1];
uint8_t hdrLen = 0;
uint8_t digit;
size_t len = length;
do {
digit = len & 0x7F; // digit = len % 128
len >>= 7; // len = len / 128
if (len > 0) {
digit |= 0x80;
}
hdrBuf[hdrLen++] = digit;
} while (len > 0 && hdrLen < MQTT_MAX_HEADER_SIZE - 1);
if (len > 0) {
ERROR_PSC_PRINTF_P("buildHeader() length too big %zu, left %zu\n", length, len);
return 0;
}
buf[MQTT_MAX_HEADER_SIZE - 1 - hdrLen] = header;
memcpy(buf + MQTT_MAX_HEADER_SIZE - hdrLen, hdrBuf, hdrLen);
return hdrLen + 1; // Full header size is variable length bit plus the 1-byte fixed header
}
size_t PubSubClient::write(uint8_t data) {
lastOutActivity = millis();
return _client->write(data);
}
size_t PubSubClient::write(const uint8_t* buffer, size_t size) {
lastOutActivity = millis();
return _client->write(buffer, size);
}
/**
* @brief Send the header and the prepared data to the client / MQTT broker.
*
* @param header Header byte, e.g. MQTTCONNECT, MQTTPUBLISH, MQTTSUBSCRIBE, MQTTUNSUBSCRIBE.
* @param buf Buffer of data to write.
* @param length Length of buf to write.
* @return True if successfully sent, otherwise false if buildHeader() failed or buf could not be written.
*/
bool PubSubClient::write(uint8_t header, uint8_t* buf, size_t length) {
bool result = true;
size_t rc;
uint8_t hdrLen = buildHeader(header, buf, length);
if (hdrLen == 0) return false; // exit here in case of header generation failure
#ifdef MQTT_MAX_TRANSFER_SIZE
uint8_t* writeBuf = buf + (MQTT_MAX_HEADER_SIZE - hdrLen);
size_t bytesRemaining = length + hdrLen; // Match the length type
size_t bytesToWrite;
while ((bytesRemaining > 0) && result) {
yield();
bytesToWrite = (bytesRemaining > MQTT_MAX_TRANSFER_SIZE) ? MQTT_MAX_TRANSFER_SIZE : bytesRemaining;
rc = _client->write(writeBuf, bytesToWrite);
result = (rc == bytesToWrite);
bytesRemaining -= rc;
writeBuf += rc;
if (result) {
lastOutActivity = millis();
}
}
#else
rc = _client->write(buf + (MQTT_MAX_HEADER_SIZE - hdrLen), length + hdrLen);
result = (rc == length + hdrLen);
if (result) {
lastOutActivity = millis();
}
#endif
return result;
}
/**
* @brief Write an UTF-8 encoded string to the give buffer and position. The string can have a length of 0 to 65535 bytes. The buffer is prefixed with two
* bytes representing the length of the string. See section 1.5.3 of MQTT v3.1.1 protocol specification.
* @note If the string does not fit in the buffer (bufferSize) or is longer than 65535 bytes nothing is written to the buffer and the returned position
* is unchanged.
*
* @param string 'C' string of the data that shall be written in the buffer.
* @param buf Buffer to write the string into.
* @param pos Position in the buffer to write the string.
* @return New position in the buffer (pos + 2 + string length), or pos if a buffer overrun would occur.
*/
size_t PubSubClient::writeString(const char* string, uint8_t* buf, size_t pos) {
return writeString(string, buf, pos, this->bufferSize);
}
/**
* @brief Write an UTF-8 encoded string to the give buffer and position. The string can have a length of 0 to 65535 bytes. The buffer is prefixed with two
* bytes representing the length of the string. See section 1.5.3 of MQTT v3.1.1 protocol specification.
* @note If the string does not fit in the buffer or is longer than 65535 bytes nothing is written to the buffer and the returned position is unchanged.
*
* @param string 'C' string of the data that shall be written in the buffer.
* @param buf Buffer to write the string into.
* @param pos Position in the buffer to write the string.
* @param size Maximal size of the buffer.
* @return New position in the buffer (pos + 2 + string length), or pos if a buffer overrun would occur or the string is a nullptr.
*/
size_t PubSubClient::writeString(const char* string, uint8_t* buf, size_t pos, size_t size) {
if (!string) return pos;
size_t sLen = strlen(string);
if (pos + 2 + sLen <= size && sLen <= 0xFFFF) {
buf[pos++] = (uint8_t)(sLen >> 8);
buf[pos++] = (uint8_t)(sLen & 0xFF);
memcpy(buf + pos, string, sLen);
pos += sLen;
} else {
ERROR_PSC_PRINTF_P("writeString(): string (%zu) does not fit into buf (%zu)\n", pos + 2 + sLen, size);
}
return pos;
}
bool PubSubClient::subscribe(const char* topic) {
return subscribe(topic, 0);
}
bool PubSubClient::subscribe(const char* topic, uint8_t qos) {
if (!topic) return false;
if (qos > 1) return false; // only QoS 0 and 1 supported
size_t topicLen = strnlen(topic, this->bufferSize);
if (this->bufferSize < MQTT_MAX_HEADER_SIZE + 2 + 2 + topicLen) {
// Too long: header + nextMsgId (2) + topic length bytes (2) + topicLen
return false;
}
if (connected()) {
// Leave room in the buffer for header and variable length field
uint16_t length = MQTT_MAX_HEADER_SIZE;
nextMsgId++;
if (nextMsgId == 0) {
nextMsgId = 1;
}
this->buffer[length++] = (nextMsgId >> 8);
this->buffer[length++] = (nextMsgId & 0xFF);
length = writeString(topic, this->buffer, length);
this->buffer[length++] = qos;
return write(MQTTSUBSCRIBE | MQTTQOS1, this->buffer, length - MQTT_MAX_HEADER_SIZE);
}
return false;
}
bool PubSubClient::unsubscribe(const char* topic) {
if (!topic) return false;
size_t topicLen = strnlen(topic, this->bufferSize);
if (this->bufferSize < MQTT_MAX_HEADER_SIZE + 2 + 2 + topicLen) {
// Too long: header + nextMsgId (2) + topic length bytes (2) + topicLen
return false;
}
if (connected()) {
uint16_t length = MQTT_MAX_HEADER_SIZE;
nextMsgId++;
if (nextMsgId == 0) {
nextMsgId = 1;
}
this->buffer[length++] = (nextMsgId >> 8);
this->buffer[length++] = (nextMsgId & 0xFF);
length = writeString(topic, this->buffer, length);
return write(MQTTUNSUBSCRIBE | MQTTQOS1, this->buffer, length - MQTT_MAX_HEADER_SIZE);
}
return false;
}
PubSubClient& PubSubClient::setServer(uint8_t* ip, uint16_t port) {
IPAddress addr(ip[0], ip[1], ip[2], ip[3]);
return setServer(addr, port);
}
PubSubClient& PubSubClient::setServer(IPAddress ip, uint16_t port) {
this->ip = ip;
this->port = port;
free(this->domain);
this->domain = nullptr;
return *this;
}
PubSubClient& PubSubClient::setServer(const char* domain, uint16_t port) {
char* newDomain = nullptr;
if (domain) {
newDomain = (char*)realloc(this->domain, strlen(domain) + 1);
}
if (newDomain) {
strcpy(newDomain, domain);
this->domain = newDomain;
this->port = port;
} else {
free(this->domain);
this->domain = nullptr;
this->port = 0;
}
return *this;
}
PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE) {
this->callback = callback;
return *this;
}
PubSubClient& PubSubClient::setClient(Client& client) {
this->_client = &client;
return *this;
}
PubSubClient& PubSubClient::setStream(Stream& stream) {
this->stream = &stream;
return *this;
}
bool PubSubClient::setBufferSize(size_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) {
this->buffer = newBuffer;
} else {
return false;
}
}
this->bufferSize = size;
return (this->buffer != nullptr);
}
size_t PubSubClient::getBufferSize() {
return this->bufferSize;
}
PubSubClient& PubSubClient::setKeepAlive(uint16_t keepAlive) {
this->keepAliveMillis = keepAlive * 1000UL;
return *this;
}
PubSubClient& PubSubClient::setSocketTimeout(uint16_t timeout) {
this->socketTimeoutMillis = timeout * 1000UL;
return *this;
}
int PubSubClient::state() {
return this->_state;
}
+612
View File
@@ -0,0 +1,612 @@
/*
PubSubClient.h - A simple client for MQTT.
Nick O'Leary, Holger Mueller
http://knolleary.net
https://github.com/hmueller01/pubsubclient3
*/
#ifndef PubSubClient_h
#define PubSubClient_h
#include <Arduino.h>
#include "Client.h"
#include "IPAddress.h"
#include "Stream.h"
#define MQTT_VERSION_3_1 3
#define MQTT_VERSION_3_1_1 4
//< @note The following #define directives can be used to configure the library.
/**
* @brief Sets the version of the MQTT protocol to use.
* @note Default value is MQTT_VERSION_3_1_1 for MQTT 3.1.1.
*/
#ifndef MQTT_VERSION
#define MQTT_VERSION MQTT_VERSION_3_1_1
#endif
// MQTT_MAX_POSSIBLE_PACKET_SIZE : Maximum packet size defined by MQTT protocol.
#ifndef MQTT_MAX_POSSIBLE_PACKET_SIZE
#define MQTT_MAX_POSSIBLE_PACKET_SIZE 268435455
#endif
/**
* @brief Sets the largest packet size, in bytes, the client will handle.
* Any packet received that exceeds this size will be ignored.
* This value can be overridden by calling setBufferSize.
* @note Default value is 256 bytes.
*/
#ifndef MQTT_MAX_PACKET_SIZE
#define MQTT_MAX_PACKET_SIZE 256
#endif
/**
* @brief Sets the keepalive interval, in seconds, the client will use.
* This is used to maintain the connection when no other packets are being sent or received.
* This value can be overridden by calling setKeepAlive.
* @note Default value is 15 seconds.
*/
#ifndef MQTT_KEEPALIVE
#define MQTT_KEEPALIVE 15
#endif
/**
* @brief Sets the timeout, in seconds, when reading from the network.
* This also applies as the timeout for calls to connect.
* This value can be overridden by calling setSocketTimeout.
* @note Default value is 15 seconds.
*/
#ifndef MQTT_SOCKET_TIMEOUT
#define MQTT_SOCKET_TIMEOUT 15
#endif
/**
* @brief Sets the maximum number of bytes passed to the network client in each write call.
* Some hardware has a limit to how much data can be passed to them in one go,
* such as the Arduino Wifi Shield.
* @note Defaults to undefined, which passes the entire packet in each write call.
*/
// #define MQTT_MAX_TRANSFER_SIZE 80
// Possible values for client.state()
#define MQTT_CONNECTION_TIMEOUT -4
#define MQTT_CONNECTION_LOST -3
#define MQTT_CONNECT_FAILED -2
#define MQTT_DISCONNECTED -1
#define MQTT_CONNECTED 0
#define MQTT_CONNECT_BAD_PROTOCOL 1
#define MQTT_CONNECT_BAD_CLIENT_ID 2
#define MQTT_CONNECT_UNAVAILABLE 3
#define MQTT_CONNECT_BAD_CREDENTIALS 4
#define MQTT_CONNECT_UNAUTHORIZED 5
#define MQTTRETAINED 1 // Retained flag in the header
#define MQTTCONNECT 1 << 4 // Client request to connect to Server
#define MQTTCONNACK 2 << 4 // Connect Acknowledgment
#define MQTTPUBLISH 3 << 4 // Publish message
#define MQTTPUBACK 4 << 4 // Publish Acknowledgment
#define MQTTPUBREC 5 << 4 // Publish Received (assured delivery part 1)
#define MQTTPUBREL 6 << 4 // Publish Release (assured delivery part 2)
#define MQTTPUBCOMP 7 << 4 // Publish Complete (assured delivery part 3)
#define MQTTSUBSCRIBE 8 << 4 // Client Subscribe request
#define MQTTSUBACK 9 << 4 // Subscribe Acknowledgment
#define MQTTUNSUBSCRIBE 10 << 4 // Client Unsubscribe request
#define MQTTUNSUBACK 11 << 4 // Unsubscribe Acknowledgment
#define MQTTPINGREQ 12 << 4 // PING Request
#define MQTTPINGRESP 13 << 4 // PING Response
#define MQTTDISCONNECT 14 << 4 // Client is Disconnecting
#define MQTTRESERVED 15 << 4 // Reserved
#define MQTTQOS0 (0 << 1)
#define MQTTQOS1 (1 << 1)
#define MQTTQOS2 (2 << 1)
// Maximum size of fixed header and variable length size header
#define MQTT_MAX_HEADER_SIZE 5
#if defined(__has_include) && __has_include(<functional>) && !defined(NOFUNCTIONAL)
#include <functional>
/**
* @brief Define the signature required by any callback function.
* @note The parameters are TOPIC, PAYLOAD, and LENGTH, respectively.
*/
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, size_t)> callback
#else
#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, size_t)
#endif
#define CHECK_STRING_LENGTH(l, s) \
if (l + 2 + strnlen(s, this->bufferSize) > this->bufferSize) { \
_client->stop(); \
return false; \
}
#ifdef DEBUG_ESP_PORT
#ifdef DEBUG_PUBSUBCLIENT
#define DEBUG_PSC_PRINTF(fmt, ...) DEBUG_ESP_PORT.printf(("PubSubClient: " fmt), ##__VA_ARGS__)
#else
#define DEBUG_PSC_PRINTF(...)
#endif
#define ERROR_PSC_PRINTF(fmt, ...) DEBUG_ESP_PORT.printf(("PubSubClient error: " fmt), ##__VA_ARGS__)
#define ERROR_PSC_PRINTF_P(fmt, ...) DEBUG_ESP_PORT.printf_P(PSTR("PubSubClient error: " fmt), ##__VA_ARGS__)
#else // DEBUG_ESP_PORT
#ifndef DEBUG_PSC_PRINTF
#define DEBUG_PSC_PRINTF(...)
#endif
#ifndef ERROR_PSC_PRINTF
#define ERROR_PSC_PRINTF(fmt, ...)
#endif
#ifndef ERROR_PSC_PRINTF_P
#define ERROR_PSC_PRINTF_P(fmt, ...)
#endif
#endif
/**
* @class PubSubClient
* @brief This class provides a client for doing simple publish and subscribe messaging with a server that supports MQTT.
*/
class PubSubClient : public Print {
private:
Client* _client{};
uint8_t* buffer{};
size_t bufferSize{};
unsigned long keepAliveMillis{};
unsigned long socketTimeoutMillis{};
uint16_t nextMsgId{};
unsigned long lastOutActivity{};
unsigned long lastInActivity{};
bool pingOutstanding{};
MQTT_CALLBACK_SIGNATURE{};
IPAddress ip{};
char* domain{};
uint16_t port{};
Stream* stream{};
int _state{MQTT_DISCONNECTED};
size_t readPacket(uint8_t* hdrLen);
bool handlePacket(uint8_t hdrLen, size_t len);
bool readByte(uint8_t* result);
bool readByte(uint8_t* result, size_t* pos);
uint8_t buildHeader(uint8_t header, uint8_t* buf, size_t length);
bool write(uint8_t header, uint8_t* buf, size_t length);
size_t writeString(const char* string, uint8_t* buf, size_t pos);
size_t writeString(const char* string, uint8_t* buf, size_t pos, size_t size);
public:
/**
* @brief Creates an uninitialised client instance.
* @note Before it can be used,
* it must be configured using the property setters setClient and setServer.
*/
PubSubClient();
/**
* @brief Creates a partially initialised client instance.
* @param client The network client to use, for example WiFiClient.
* @note Before it can be used,
* it must be configured with the property setter setServer.
*/
PubSubClient(Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(IPAddress, uint16_t, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(IPAddress, uint16_t, Client& client, Stream&);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(uint8_t*, uint16_t, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(uint8_t*, uint16_t, Client& client, Stream&);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(uint8_t*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(uint8_t*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(const char*, uint16_t, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(const char*, uint16_t, Client& client, Stream&);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
*/
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
/**
* @brief Creates a fully configured client instance.
* @param domain The address of the server.
* @param port The port to connect to.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @param client The network client to use, for example WiFiClient.
* @param stream A stream to write received messages to.
*/
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);
/**
* @brief Destructor for the PubSubClient class.
*/
~PubSubClient();
/**
* @brief Sets the server details.
* @param ip The address of the server.
* @param port The port to connect to.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setServer(IPAddress ip, uint16_t port);
/**
* @brief Sets the server details.
* @param ip The address of the server.
* @param port The port to connect to.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setServer(uint8_t* ip, uint16_t port);
/**
* @brief Sets the server details.
* @param ip The address of the server.
* @param port The port to connect to.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setServer(const char* domain, uint16_t port);
/**
* @brief Sets the message callback function.
* @param callback Pointer to a message callback function.
* Called when a message arrives for a subscription created by this client.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE);
/**
* @brief Sets the network client instance to use.
* @param client The network client to use, for example WiFiClient.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setClient(Client& client);
/**
* @brief Sets the stream to write received messages to.
* @param stream A stream to write received messages to.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setStream(Stream& stream);
/**
* @brief Sets the keep alive interval used by the client.
* This value should only be changed when the client is not connected.
* Set keepAlive to zero (0) to turn off the keep alive mechanism.
* @param keepAlive The keep alive interval, in seconds.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setKeepAlive(uint16_t keepAlive);
/**
* @brief Sets the socket timeout used by the client.
* This determines how long the client will wait for incoming data when it expects data to arrive - for example,
* whilst it is in the middle of reading a MQTT packet.
* @param timeout The socket timeout, in seconds.
* @return The client instance, allowing the function to be chained.
*/
PubSubClient& setSocketTimeout(uint16_t timeout);
/**
* @brief Sets the size, in bytes, of the internal send and receive buffer.
* This must be large enough to contain the full MQTT packet.
* When sending or receiving messages, the packet will contain the full topic string,
* the payload data, and a small number of header bytes.
* @param size The size, in bytes, for the internal buffer.
* @return true If the buffer was resized.
* false If the buffer could not be resized.
*/
bool setBufferSize(size_t size);
/**
* @brief Gets the current size of the internal buffer.
* @return The size of the internal buffer.
*/
size_t getBufferSize();
/**
* @brief Connects the client.
* @param id The client ID to use when connecting to the server.
* @return true If client succeeded in establishing a connection to the broker.
* false If client failed to establish a connection to the broker.
*/
bool connect(const char* id);
/**
* @brief Connects the client.
* @param id The client ID to use when connecting to the server.
* @param user The username to use. If NULL, no username or password is used.
* @param pass The password to use. If NULL, no password is used.
* @return true If client succeeded in establishing a connection to the broker.
* false If client failed to establish a connection to the broker.
*/
bool connect(const char* id, const char* user, const char* pass);
/**
* @brief Connects the client.
* @param id The client ID to use when connecting to the server.
* @param willTopic The topic to be used by the will message.
* @param willQos The quality of service to be used by the will message. [0, 1, 2].
* @param willRetain Publish the will message with the retain flag.
* @return true If client succeeded in establishing a connection to the broker.
* false If client failed to establish a connection to the broker.
*/
bool connect(const char* id, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage);
/**
* @brief Connects the client.
* @param id The client ID to use when connecting to the server.
* @param user The username to use. If NULL, no username or password is used.
* @param pass The password to use. If NULL, no password is used.
* @param willTopic The topic to be used by the will message.
* @param willQos The quality of service to be used by the will message. [0, 1, 2].
* @param willRetain Publish the will message with the retain flag.
* @return true If client succeeded in establishing a connection to the broker.
* false If client failed to establish a connection to the broker.
*/
bool connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage);
/**
* @brief Connects the client.
* @param id The client ID to use when connecting to the server.
* @param user The username to use. If NULL, no username or password is used.
* @param pass The password to use. If NULL, no password is used.
* @param willTopic The topic to be used by the will message.
* @param willQos The quality of service to be used by the will message. [0, 1, 2].
* @param willRetain Publish the will message with the retain flag.
* @param cleanSession Connect with a clean session.
* @return true If client succeeded in establishing a connection to the broker.
* false If client failed to establish a connection to the broker.
*/
bool connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage,
bool cleanSession);
/**
* @brief Disconnects the client.
*/
void disconnect();
/**
* @brief Publishes a non retained message to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish(const char* topic, const char* payload);
/**
* @brief Publishes a message to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @param retained Publish the message with the retain flag.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish(const char* topic, const char* payload, bool retained);
/**
* @brief Publishes a non retained message to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @param plength The length of the payload.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish(const char* topic, const uint8_t* payload, size_t plength);
/**
* @brief Publishes a message to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @param plength The length of the payload.
* @param retained Publish the message with the retain flag.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish(const char* topic, const uint8_t* payload, size_t plength, bool retained);
/**
* @brief Publishes a message stored in PROGMEM to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @param retained Publish the message with the retain flag.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish_P(const char* topic, const char* payload, bool retained);
/**
* @brief Publishes a message stored in PROGMEM to the specified topic.
* @param topic The topic to publish to.
* @param payload The message to publish.
* @param plength The length of the payload.
* @param retained Publish the message with the retain flag.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool publish_P(const char* topic, const uint8_t* payload, size_t plength, bool retained);
/**
* @brief Start to publish a message.
* This API:
* beginPublish(...)
* one or more calls to write(...)
* endPublish()
* Allows for arbitrarily large payloads to be sent without them having to be copied into
* a new buffer and held in memory at one time.
* @param topic The topic to publish to.
* @param plength The length of the payload.
* @param retained Publish the message with the retain flag.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool beginPublish(const char* topic, size_t plength, bool retained);
/**
* @brief Finish sending a message that was started with a call to beginPublish.
* @return true If the publish succeeded.
* false If the publish failed, either connection lost or message too large.
*/
bool endPublish();
/**
* @brief Writes a single byte as a component of a publish started with a call to beginPublish.
* @param byte A byte to write to the publish payload.
* @return The number of bytes written.
*/
virtual size_t write(uint8_t);
/**
* @brief Writes an array of bytes as a component of a publish started with a call to beginPublish.
* @param buffer The bytes to write.
* @param size The length of the payload to be sent.
* @return The number of bytes written.
*/
virtual size_t write(const uint8_t* buffer, size_t size);
/**
* @brief Subscribes to messages published to the specified topic.
* @param topic The topic to subscribe to.
* @return true If sending the subscribe succeeded.
* false If sending the subscribe failed, either connection lost or message too large.
*/
bool subscribe(const char* topic);
/**
* @brief Subscribes to messages published to the specified topic.
* @param topic The topic to subscribe to.
* @param qos The qos to subscribe at. [0, 1].
* @return true If sending the subscribe succeeded.
* false If sending the subscribe failed, either connection lost or message too large.
*/
bool subscribe(const char* topic, uint8_t qos);
/**
* @brief Unsubscribes from the specified topic.
* @param topic The topic to unsubscribe from.
* @return true If sending the unsubscribe succeeded.
* false If sending the unsubscribe failed, either connection lost or message too large.
*/
bool unsubscribe(const char* topic);
/**
* @brief This should be called regularly to allow the client to process incoming messages and maintain its connection to the server.
* @return true If the client is still connected.
* false If the client is no longer connected.
*/
bool loop();
/**
* @brief Checks whether the client is connected to the server.
* @return true If the client is connected.
* false If the client is not connected.
*/
bool connected();
/**
* @brief Returns the current state of the client.
* If a connection attempt fails, this can be used to get more information about the failure.
* @note All of the values have corresponding constants defined in PubSubClient.h.
* @return -4 : MQTT_CONNECTION_TIMEOUT - The server didn't respond within the keepalive time.
* -3 : MQTT_CONNECTION_LOST - The network connection was broken.
* -2 : MQTT_CONNECT_FAILED - The network connection failed.
* -1 : MQTT_DISCONNECTED - The client is disconnected cleanly.
* 0 : MQTT_CONNECTED - The client is connected.
* 1 : MQTT_CONNECT_BAD_PROTOCOL - The server doesn't support the requested version of MQTT.
* 2 : MQTT_CONNECT_BAD_CLIENT_ID - The server rejected the client identifier.
* 3 : MQTT_CONNECT_UNAVAILABLE - The server was unable to accept the connection.
* 4 : MQTT_CONNECT_BAD_CREDENTIALS - The username/password were rejected.
* 5 : MQTT_CONNECT_UNAUTHORIZED - The client was not authorized to connect.
*/
int state();
};
#endif
+4
View File
@@ -0,0 +1,4 @@
.build
tmpbin
logs
*.pyc
+29
View File
@@ -0,0 +1,29 @@
SRC_PATH=./src
OUT_PATH=./bin
TEST_SRC=$(wildcard ${SRC_PATH}/*_spec.cpp)
TEST_BIN= $(TEST_SRC:${SRC_PATH}/%.cpp=${OUT_PATH}/%)
VPATH=${SRC_PATH}
SHIM_FILES=${SRC_PATH}/lib/*.cpp
PSC_FILE=../src/PubSubClient.cpp
CC=g++
CFLAGS=-std=c++11 -I${SRC_PATH}/lib -I../src
CFLAGS+=-pedantic -Werror -Wall -Wextra -Wcast-align -Wcast-qual -Wctor-dtor-privacy -Wdisabled-optimization -Wformat=2
CFLAGS+=-Winit-self -Wmissing-declarations -Wmissing-include-dirs -Wno-unused -Wnull-dereference -Woverloaded-virtual
CFLAGS+=-Wredundant-decls -Wreorder -Wsign-conversion -Wsign-promo -Wstrict-overflow=5 -Wswitch-default -Wundef
#CFLAGS+=-DDEBUG_PUBSUBCLIENT
all: $(TEST_BIN)
${OUT_PATH}/%: ${SRC_PATH}/%.cpp ${PSC_FILE} ${SHIM_FILES}
mkdir -p ${OUT_PATH}
${CC} ${CFLAGS} $^ -o $@
clean:
@rm -rf ${OUT_PATH}
test: all
@bin/connect_spec
@bin/publish_spec
@bin/receive_spec
@bin/subscribe_spec
@bin/keepalive_spec
+88
View File
@@ -0,0 +1,88 @@
# Arduino Client for MQTT Test Suite
This is a regression test suite for the `PubSubClient` library.
There are two parts:
- Tests that can be compiled and run on any machine
- Tests that build the example sketches using the Arduino IDE
It is a work-in-progress and is subject to complete refactoring as the whim takes me.
## Local tests
These are a set of executables that can be run to test specific areas of functionality.
They do not require a real Arduino to be attached, nor the use of the Arduino IDE.
The tests include a set of mock files to stub out the parts of the Arduino environment the library depends on.
### Dependencies
- g++
### Running
Build the tests using the provided `Makefile`:
$ make
This will create a set of executables in `./bin/`. Run each of these executables to test the corresponding functionality.
$ make test
*Note:* the `connect_spec` and `keepalive_spec` tests involve testing keepalive timers so naturally take a few minutes to run through.
## Arduino tests
*Note:* INO Tool doesn't currently play nicely with Arduino 1.5. This has broken this test suite.
Without a suitable arduino plugged in, the test suite will only check the
example sketches compile cleanly against the library.
With an arduino plugged in, each sketch that has a corresponding python
test case is built, uploaded and then the tests run.
### Dependencies
- Python 2.7+
- [INO Tool](http://inotool.org/) - this provides command-line build/upload of Arduino sketches
### Running
The test suite _does not_ run an MQTT server - it is assumed to be running already.
$ python testsuite.py
A summary of activity is printed to the console. More comprehensive logs are written
to the `logs` directory.
### What it does
For each sketch in the library's `examples` directory, e.g. `mqtt_basic.ino`, the suite looks for a matching test case
`testcases/mqtt_basic.py`.
The test case must follow these conventions:
- sub-class `unittest.TestCase`
- provide the class methods `setUpClass` and `tearDownClass` (TODO: make this optional)
- all test method names begin with `test_`
The suite will call the `setUpClass` method _before_ uploading the sketch. This
allows any test setup to be performed before the sketch runs - such as connecting
a client and subscribing to topics.
### Settings
The file `testcases/settings.py` is used to config the test environment.
- `server_ip` - the IP address of the broker the client should connect to (the broker port is assumed to be 1883).
- `arduino_ip` - the IP address the arduino should use (when not testing DHCP).
Before each sketch is compiled, these values are automatically substituted in. To
do this, the suite looks for lines that _start_ with the following:
byte server[] = {
byte ip[] = {
and replaces them with the appropriate values.
@@ -0,0 +1,361 @@
#include "BDDTest.h"
#include "Buffer.h"
#include "PubSubClient.h"
#include "ShimClient.h"
#include "trace.h"
byte server[] = {172, 16, 0, 2};
// function declarations
void callback(char* topic, uint8_t* payload, size_t length);
int test_connect_fails_no_network();
int test_connect_fails_on_no_response();
int test_connect_properly_formatted();
int test_connect_properly_formatted_hostname();
int test_connect_fails_on_bad_rc();
int test_connect_non_clean_session();
int test_connect_accepts_username_password();
int test_connect_accepts_username_no_password();
int test_connect_accepts_username_blank_password();
int test_connect_ignores_password_no_username();
int test_connect_with_will();
int test_connect_with_will_username_password();
int test_connect_disconnect_connect();
int test_connect_custom_keepalive();
void callback(_UNUSED_ char* topic, _UNUSED_ uint8_t* payload, _UNUSED_ size_t length) {
// handle message arrived
}
int test_connect_fails_no_network() {
IT("fails to connect if underlying client doesn't connect");
ShimClient shimClient;
shimClient.setAllowConnect(false);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_FALSE(rc);
int state = client.state();
IS_TRUE(state == MQTT_CONNECT_FAILED);
END_IT
}
int test_connect_fails_on_no_response() {
IT("fails to connect if no response received after 15 seconds");
ShimClient shimClient;
shimClient.setAllowConnect(true);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_FALSE(rc);
int state = client.state();
IS_TRUE(state == MQTT_CONNECTION_TIMEOUT);
END_IT
}
int test_connect_properly_formatted() {
IT("sends a properly formatted 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, 0x2, 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);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
END_IT
}
int test_connect_properly_formatted_hostname() {
IT("accepts a hostname");
ShimClient shimClient;
shimClient.setAllowConnect(true);
shimClient.expectConnect("localhost", 1883);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client("localhost", 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_fails_on_bad_rc() {
IT("fails to connect if a bad return code is received");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x01};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_FALSE(rc);
int state = client.state();
IS_TRUE(state == 0x01);
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);
bool rc = client.connect("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() {
IT("accepts a username and password");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x24, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0xc2, 0x0, 0xf, 0x0, 0xc, 0x63, 0x6c, 0x69, 0x65, 0x6e,
0x74, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x31, 0x0, 0x4, 0x75, 0x73, 0x65, 0x72, 0x0, 0x4, 0x70, 0x61, 0x73, 0x73};
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.expect(connect, 0x26);
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1", "user", "pass");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_accepts_username_no_password() {
IT("accepts a username but no password");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x1e, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0x82, 0x0, 0xf, 0x0, 0xc, 0x63, 0x6c,
0x69, 0x65, 0x6e, 0x74, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x31, 0x0, 0x4, 0x75, 0x73, 0x65, 0x72};
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.expect(connect, 0x20);
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1", "user", 0);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_accepts_username_blank_password() {
IT("accepts a username and blank password");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x20, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0xc2, 0x0, 0xf, 0x0, 0xc, 0x63, 0x6c, 0x69,
0x65, 0x6e, 0x74, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x31, 0x0, 0x4, 0x75, 0x73, 0x65, 0x72, 0x0, 0x0};
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.expect(connect, 0x26);
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1", "user", "pass");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_ignores_password_no_username() {
IT("ignores a password but no username");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x18, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0x2, 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);
bool rc = client.connect("client_test1", 0, "pass");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_with_will() {
IT("accepts a will");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x30, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0xe, 0x0, 0xf, 0x0, 0xc, 0x63, 0x6c, 0x69,
0x65, 0x6e, 0x74, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x31, 0x0, 0x9, 0x77, 0x69, 0x6c, 0x6c, 0x54, 0x6f,
0x70, 0x69, 0x63, 0x0, 0xb, 0x77, 0x69, 0x6c, 0x6c, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65};
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.expect(connect, 0x32);
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1", "willTopic", 1, 0, "willMessage");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_with_will_username_password() {
IT("accepts a will, username and password");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connect[] = {0x10, 0x40, 0x0, 0x4, 0x4d, 0x51, 0x54, 0x54, 0x4, 0xce, 0x0, 0xf, 0x0, 0xc, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x74,
0x65, 0x73, 0x74, 0x31, 0x0, 0x9, 0x77, 0x69, 0x6c, 0x6c, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x0, 0xb, 0x77, 0x69, 0x6c, 0x6c, 0x4d,
0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x0, 0x4, 0x75, 0x73, 0x65, 0x72, 0x0, 0x8, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64};
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.expect(connect, 0x42);
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1", "user", "password", "willTopic", 1, 0, "willMessage");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_connect_disconnect_connect() {
IT("connects, disconnects and connects again");
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, 0x2, 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);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
byte disconnect[] = {0xE0, 0x00};
shimClient.expect(disconnect, 2);
client.disconnect();
IS_FALSE(client.connected());
IS_FALSE(shimClient.connected());
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_DISCONNECTED);
shimClient.expect(connect, 28);
shimClient.respond(connack, 4);
rc = client.connect("client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
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);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
END_IT
}
int main() {
SUITE("Connect");
test_connect_fails_no_network();
test_connect_fails_on_no_response();
test_connect_properly_formatted();
test_connect_non_clean_session();
test_connect_accepts_username_password();
test_connect_fails_on_bad_rc();
test_connect_properly_formatted_hostname();
test_connect_accepts_username_no_password();
test_connect_ignores_password_no_username();
test_connect_with_will();
test_connect_with_will_username_password();
test_connect_disconnect_connect();
test_connect_custom_keepalive();
FINISH
}
@@ -0,0 +1,192 @@
#include <unistd.h>
#include "BDDTest.h"
#include "Buffer.h"
#include "PubSubClient.h"
#include "ShimClient.h"
#include "trace.h"
byte server[] = {172, 16, 0, 2};
// function declarations
void callback(char* topic, uint8_t* payload, size_t length);
int test_keepalive_pings_idle();
int test_keepalive_pings_with_outbound_qos0();
int test_keepalive_pings_with_inbound_qos0();
int test_keepalive_no_pings_inbound_qos1();
int test_keepalive_disconnects_hung();
void callback(_UNUSED_ char* topic, _UNUSED_ uint8_t* payload, _UNUSED_ size_t length) {
// handle message arrived
}
int test_keepalive_pings_idle() {
IT("keeps an idle connection alive (takes 1 minute)");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte pingreq[] = {0xC0, 0x0};
shimClient.expect(pingreq, 2);
byte pingresp[] = {0xD0, 0x0};
shimClient.respond(pingresp, 2);
for (int i = 0; i < 50; i++) {
sleep(1);
if (i == 15 || i == 31 || i == 47) {
shimClient.expect(pingreq, 2);
shimClient.respond(pingresp, 2);
}
rc = client.loop();
IS_TRUE(rc);
}
IS_FALSE(shimClient.error());
END_IT
}
int test_keepalive_pings_with_outbound_qos0() {
IT("keeps a connection alive that only sends qos0 (takes 1 minute)");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xe, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
for (int i = 0; i < 50; i++) {
TRACE(i << ":");
shimClient.expect(publish, 16);
rc = client.publish("topic", "payload");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
sleep(1);
if (i == 15 || i == 31 || i == 47) {
byte pingreq[] = {0xC0, 0x0};
shimClient.expect(pingreq, 2);
byte pingresp[] = {0xD0, 0x0};
shimClient.respond(pingresp, 2);
}
rc = client.loop();
IS_TRUE(rc);
IS_FALSE(shimClient.error());
}
END_IT
}
int test_keepalive_pings_with_inbound_qos0() {
IT("keeps a connection alive that only receives qos0 (takes 1 minute)");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xe, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
for (int i = 0; i < 50; i++) {
TRACE(i << ":");
sleep(1);
if (i == 15 || i == 31 || i == 47) {
byte pingreq[] = {0xC0, 0x0};
shimClient.expect(pingreq, 2);
byte pingresp[] = {0xD0, 0x0};
shimClient.respond(pingresp, 2);
}
shimClient.respond(publish, 16);
rc = client.loop();
IS_TRUE(rc);
IS_FALSE(shimClient.error());
}
END_IT
}
int test_keepalive_no_pings_inbound_qos1() {
IT("does not send pings for connections with inbound qos1 (takes 1 minute)");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x32, 0x10, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x34, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
byte puback[] = {0x40, 0x2, 0x12, 0x34};
for (int i = 0; i < 50; i++) {
shimClient.respond(publish, 18);
shimClient.expect(puback, 4);
sleep(1);
rc = client.loop();
IS_TRUE(rc);
IS_FALSE(shimClient.error());
}
END_IT
}
int test_keepalive_disconnects_hung() {
IT("disconnects a hung connection (takes 30 seconds)");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte pingreq[] = {0xC0, 0x0};
shimClient.expect(pingreq, 2);
for (int i = 0; i < 32; i++) {
sleep(1);
rc = client.loop();
}
IS_FALSE(rc);
int state = client.state();
IS_TRUE(state == MQTT_CONNECTION_TIMEOUT);
IS_FALSE(shimClient.error());
END_IT
}
int main() {
SUITE("Keep-alive");
test_keepalive_pings_idle();
test_keepalive_pings_with_outbound_qos0();
test_keepalive_pings_with_inbound_qos0();
test_keepalive_no_pings_inbound_qos1();
test_keepalive_disconnects_hung();
FINISH
}
+38
View File
@@ -0,0 +1,38 @@
#ifndef Arduino_h
#define Arduino_h
#include <math.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <cstdio>
#include "Print.h"
extern "C" {
typedef uint8_t byte;
typedef uint8_t boolean;
/* sketch */
extern void setup(void);
extern void loop(void);
unsigned long millis(void);
}
#define PROGMEM
#define strnlen_P strnlen
#define pgm_read_byte_near(x) *(x)
#define yield(x) {}
#pragma GCC system_header
#define ERROR_PSC_PRINTF(fmt, ...) printf(("PubSubClient error: " fmt), ##__VA_ARGS__)
#define ERROR_PSC_PRINTF_P(fmt, ...) printf(("PubSubClient error: " fmt), ##__VA_ARGS__)
#ifdef DEBUG_PUBSUBCLIENT
#define DEBUG_PSC_PRINTF(fmt, ...) printf(("PubSubClient: " fmt), ##__VA_ARGS__)
#endif
#define _UNUSED_ __attribute__((unused))
#endif // Arduino_h
@@ -0,0 +1,52 @@
#include "BDDTest.h"
#include <iostream>
#include <list>
#include <sstream>
#include <string>
#include "trace.h"
int testCount = 0;
int testPasses = 0;
const char* testDescription;
std::list<std::string> failureList;
void bddtest_suite(const char* name) {
LOG(name << "\n");
}
int bddtest_test(const char* file, int line, const char* assertion, int result) {
if (!result) {
LOG("\n");
std::ostringstream os;
os << " ! " << testDescription << "\n " << file << ":" << line << " : " << assertion << " [" << result << "]";
failureList.push_back(os.str());
}
return result;
}
void bddtest_start(const char* description) {
LOG(" - " << description << " ");
testDescription = description;
testCount++;
}
void bddtest_end() {
LOG("\n");
testPasses++;
}
int bddtest_summary() {
for (std::list<std::string>::iterator it = failureList.begin(); it != failureList.end(); it++) {
LOG("\n");
LOG(*it);
LOG("\n");
}
LOG(std::dec << testPasses << "/" << testCount << " tests passed\n\n");
if (testPasses == testCount) {
return 0;
}
return 1;
}
+23
View File
@@ -0,0 +1,23 @@
#ifndef bddtest_h
#define bddtest_h
void bddtest_suite(const char* name);
int bddtest_test(const char*, int, const char*, int);
void bddtest_start(const char*);
void bddtest_end();
int bddtest_summary();
#define SUITE(x) { bddtest_suite(x); }
#define TEST(x) { if (!bddtest_test(__FILE__, __LINE__, #x, (x))) return false; }
#define IT(x) { bddtest_start(x); }
#define END_IT { bddtest_end();return true;}
#define FINISH { return bddtest_summary(); }
#define IS_TRUE(x) TEST(x)
#define IS_FALSE(x) TEST(!(x))
#define IS_EQUAL(x, y) TEST(x==y)
#define IS_NOT_EQUAL(x, y) TEST(x!=y)
#endif
@@ -0,0 +1,35 @@
#include "Buffer.h"
#include "Arduino.h"
Buffer::Buffer() {
this->pos = 0;
this->length = 0;
}
Buffer::Buffer(uint8_t* buf, size_t size) {
this->pos = 0;
this->length = 0;
this->add(buf, size);
}
bool Buffer::available() {
return this->pos < this->length;
}
uint8_t Buffer::next() {
if (this->available()) {
return this->buffer[this->pos++];
}
return 0;
}
void Buffer::reset() {
this->pos = 0;
}
void Buffer::add(uint8_t* buf, size_t size) {
uint16_t i = 0;
for (; i < size; i++) {
this->buffer[this->length++] = buf[i];
}
}
+23
View File
@@ -0,0 +1,23 @@
#ifndef buffer_h
#define buffer_h
#include "Arduino.h"
class Buffer {
private:
uint8_t buffer[2048];
uint16_t pos;
uint16_t length;
public:
Buffer();
Buffer(uint8_t* buf, size_t size);
virtual bool available();
virtual uint8_t next();
virtual void reset();
virtual void add(uint8_t* buf, size_t size);
};
#endif
+21
View File
@@ -0,0 +1,21 @@
#ifndef client_h
#define client_h
#include "IPAddress.h"
class Client {
public:
virtual int connect(IPAddress ip, uint16_t port) = 0;
virtual int connect(const char *host, uint16_t port) = 0;
virtual size_t write(uint8_t) = 0;
virtual size_t write(const uint8_t *buf, size_t size) = 0;
virtual int available() = 0;
virtual int read() = 0;
virtual int read(uint8_t *buf, size_t size) = 0;
virtual int peek() = 0;
virtual void flush() = 0;
virtual void stop() = 0;
virtual uint8_t connected() = 0;
virtual operator bool() = 0;
};
#endif
@@ -0,0 +1,36 @@
#include <Arduino.h>
#include <IPAddress.h>
IPAddress::IPAddress() {
memset(_address, 0, sizeof(_address));
}
IPAddress::IPAddress(uint8_t first_octet, uint8_t second_octet, uint8_t third_octet, uint8_t fourth_octet) {
_address[0] = first_octet;
_address[1] = second_octet;
_address[2] = third_octet;
_address[3] = fourth_octet;
}
IPAddress::IPAddress(uint32_t address) {
memcpy(_address, &address, sizeof(_address));
}
IPAddress::IPAddress(const uint8_t* address) {
memcpy(_address, address, sizeof(_address));
}
IPAddress& IPAddress::operator=(const uint8_t* address) {
memcpy(_address, address, sizeof(_address));
return *this;
}
IPAddress& IPAddress::operator=(uint32_t address) {
memcpy(_address, (const uint8_t*)&address, sizeof(_address));
return *this;
}
bool IPAddress::operator==(const uint8_t* addr) {
return memcmp(addr, _address, sizeof(_address)) == 0;
}
@@ -0,0 +1,79 @@
/*
*
* MIT License:
* Copyright (c) 2011 Adrian McEwen
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*
* adrianm@mcqn.com 1/1/2011
*/
#ifndef IPAddress_h
#define IPAddress_h
// A class to make it easier to handle and pass around IP addresses
class IPAddress {
private:
alignas(4) uint8_t _address[4]; // IPv4 address
// Access the raw byte array containing the address. Because this returns a pointer
// to the internal structure rather than a copy of the address this function should only
// be used when you know that the usage of the returned uint8_t* will be transient and not
// stored.
uint8_t* raw_address() {
return _address;
};
public:
// Constructors
IPAddress();
IPAddress(uint8_t first_octet, uint8_t second_octet, uint8_t third_octet, uint8_t fourth_octet);
IPAddress(uint32_t address);
IPAddress(const uint8_t* address);
// Overloaded cast operator to allow IPAddress objects to be used where a pointer
// to a four-byte uint8_t array is expected
operator uint32_t() {
return *((uint32_t*)_address);
};
bool operator==(IPAddress& addr) {
return (*((uint32_t*)_address)) == (*(reinterpret_cast<uint32_t*>(addr._address)));
};
bool operator==(const uint8_t* addr);
// Overloaded index operator to allow getting and setting individual octets of the address
uint8_t operator[](int index) const {
return _address[index];
};
uint8_t& operator[](int index) {
return _address[index];
};
// Overloaded copy operators to allow initialisation of IPAddress objects from other types
IPAddress& operator=(const uint8_t* address);
IPAddress& operator=(uint32_t address);
friend class EthernetClass;
friend class UDP;
friend class Client;
friend class Server;
friend class DhcpClass;
friend class DNSClient;
};
#endif
+28
View File
@@ -0,0 +1,28 @@
/*
Print.h - Base class that provides print() and println()
Copyright (c) 2008 David A. Mellis. All right reserved.
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.
This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef Print_h
#define Print_h
class Print {
public:
virtual size_t write(uint8_t) = 0;
};
#endif
@@ -0,0 +1,162 @@
#include "ShimClient.h"
#include <Arduino.h>
#include <ctime>
#include <iostream>
#include "trace.h"
extern "C" {
unsigned long millis(void) {
return (unsigned long)time(0) * 1000UL;
}
}
ShimClient::ShimClient() {
this->responseBuffer = new Buffer();
this->expectBuffer = new Buffer();
this->_allowConnect = true;
this->_connected = false;
this->_error = false;
this->expectAnything = true;
this->_received = 0;
this->_expectedPort = 0;
}
int ShimClient::connect(IPAddress ip, uint16_t port) {
if (this->_allowConnect) {
this->_connected = true;
}
if (this->_expectedPort != 0) {
if (memcmp(&ip, &this->_expectedIP, 4) != 0) {
TRACE("ip mismatch\n");
this->_error = true;
}
if (port != this->_expectedPort) {
TRACE("port mismatch\n");
this->_error = true;
}
}
return this->_connected;
}
int ShimClient::connect(const char *host, uint16_t port) {
if (this->_allowConnect) {
this->_connected = true;
}
if (this->_expectedPort != 0) {
if (strcmp(host, this->_expectedHost) != 0) {
TRACE("host mismatch\n");
this->_error = true;
}
if (port != this->_expectedPort) {
TRACE("port mismatch\n");
this->_error = true;
}
}
return this->_connected;
}
size_t ShimClient::write(uint8_t b) {
this->_received += 1;
TRACE(std::hex << (unsigned int)b);
if (!this->expectAnything) {
if (this->expectBuffer->available()) {
uint8_t expected = this->expectBuffer->next();
if (expected != b) {
this->_error = true;
TRACE("!=" << (unsigned int)expected);
}
} else {
this->_error = true;
}
}
TRACE("\n" << std::dec);
return 1;
}
size_t ShimClient::write(const uint8_t *buf, size_t size) {
this->_received += size;
TRACE("[" << std::dec << (unsigned int)(size) << "] ");
uint16_t i = 0;
for (; i < size; i++) {
if (i > 0) {
TRACE(":");
}
TRACE(std::hex << (unsigned int)(buf[i]));
if (!this->expectAnything) {
if (this->expectBuffer->available()) {
uint8_t expected = this->expectBuffer->next();
if (expected != buf[i]) {
this->_error = true;
TRACE("!=" << (unsigned int)expected);
}
} else {
this->_error = true;
}
}
}
TRACE("\n" << std::dec);
return size;
}
int ShimClient::available() {
return this->responseBuffer->available();
}
int ShimClient::read() {
return this->responseBuffer->next();
}
int ShimClient::read(uint8_t *buf, size_t size) {
uint16_t i = 0;
for (; i < size; i++) {
buf[i] = this->read();
}
return size;
}
int ShimClient::peek() {
return 0;
}
void ShimClient::flush() {}
void ShimClient::stop() {
this->setConnected(false);
}
uint8_t ShimClient::connected() {
return this->_connected;
}
ShimClient::operator bool() {
return true;
}
ShimClient *ShimClient::respond(uint8_t *buf, size_t size) {
this->responseBuffer->add(buf, size);
return this;
}
ShimClient *ShimClient::expect(uint8_t *buf, size_t size) {
this->expectAnything = false;
this->expectBuffer->add(buf, size);
return this;
}
void ShimClient::setConnected(bool b) {
this->_connected = b;
}
void ShimClient::setAllowConnect(bool b) {
this->_allowConnect = b;
}
bool ShimClient::error() {
return this->_error;
}
uint16_t ShimClient::received() {
return this->_received;
}
void ShimClient::expectConnect(IPAddress ip, uint16_t port) {
this->_expectedIP = ip;
this->_expectedPort = port;
}
void ShimClient::expectConnect(const char *host, uint16_t port) {
this->_expectedHost = host;
this->_expectedPort = port;
}
@@ -0,0 +1,50 @@
#ifndef shimclient_h
#define shimclient_h
#include "Arduino.h"
#include "Buffer.h"
#include "Client.h"
#include "IPAddress.h"
class ShimClient : public Client {
private:
Buffer *responseBuffer;
Buffer *expectBuffer;
bool _allowConnect;
bool _connected;
bool expectAnything;
bool _error;
uint16_t _received;
IPAddress _expectedIP;
uint16_t _expectedPort;
const char *_expectedHost;
public:
ShimClient();
virtual int connect(IPAddress ip, uint16_t port);
virtual int connect(const char *host, uint16_t port);
virtual size_t write(uint8_t);
virtual size_t write(const uint8_t *buf, size_t size);
virtual int available();
virtual int read();
virtual int read(uint8_t *buf, size_t size);
virtual int peek();
virtual void flush();
virtual void stop();
virtual uint8_t connected();
virtual operator bool();
virtual ShimClient *respond(uint8_t *buf, size_t size);
virtual ShimClient *expect(uint8_t *buf, size_t size);
virtual void expectConnect(IPAddress ip, uint16_t port);
virtual void expectConnect(const char *host, uint16_t port);
virtual uint16_t received();
virtual bool error();
virtual void setAllowConnect(bool b);
virtual void setConnected(bool b);
};
#endif
@@ -0,0 +1,41 @@
#include "Stream.h"
#include <Arduino.h>
#include <iostream>
#include "trace.h"
Stream::Stream() {
this->expectBuffer = new Buffer();
this->_error = false;
this->_written = 0;
}
size_t Stream::write(uint8_t b) {
this->_written++;
TRACE(std::hex << (unsigned int)b);
if (this->expectBuffer->available()) {
uint8_t expected = this->expectBuffer->next();
if (expected != b) {
this->_error = true;
TRACE("!=" << (unsigned int)expected);
}
} else {
this->_error = true;
}
TRACE("\n" << std::dec);
return 1;
}
bool Stream::error() {
return this->_error;
}
void Stream::expect(uint8_t *buf, size_t size) {
this->expectBuffer->add(buf, size);
}
uint16_t Stream::length() {
return this->_written;
}
+22
View File
@@ -0,0 +1,22 @@
#ifndef Stream_h
#define Stream_h
#include "Arduino.h"
#include "Buffer.h"
class Stream {
private:
Buffer* expectBuffer;
bool _error;
uint16_t _written;
public:
Stream();
virtual size_t write(uint8_t);
virtual bool error();
virtual void expect(uint8_t* buf, size_t size);
virtual uint16_t length();
};
#endif
+10
View File
@@ -0,0 +1,10 @@
#ifndef trace_h
#define trace_h
#include <stdlib.h>
#include <iostream>
#define LOG(x) {std::cout << x << std::flush; }
#define TRACE(x) {if (getenv("TRACE")) { std::cout << x << std::flush; }}
#endif
@@ -0,0 +1,228 @@
#include "BDDTest.h"
#include "Buffer.h"
#include "PubSubClient.h"
#include "ShimClient.h"
#include "trace.h"
byte server[] = {172, 16, 0, 2};
// function declarations
void callback(char* topic, uint8_t* payload, size_t length);
int test_publish();
int test_publish_bytes();
int test_publish_retained();
int test_publish_retained_2();
int test_publish_not_connected();
int test_publish_too_long();
int test_publish_P();
int test_publish_P_too_long();
void callback(_UNUSED_ char* topic, _UNUSED_ uint8_t* payload, _UNUSED_ size_t length) {
// handle message arrived
}
int test_publish() {
IT("publishes a null-terminated string");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xe, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
shimClient.expect(publish, 16);
rc = client.publish("topic", "payload");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_bytes() {
IT("publishes a byte array");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte payload[] = {0x01, 0x02, 0x03, 0x0, 0x05};
size_t length = 5;
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xc, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x1, 0x2, 0x3, 0x0, 0x5};
shimClient.expect(publish, 14);
rc = client.publish("topic", payload, length);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_retained() {
IT("publishes retained - 1");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte payload[] = {0x01, 0x02, 0x03, 0x0, 0x05};
size_t length = 5;
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x31, 0xc, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x1, 0x2, 0x3, 0x0, 0x5};
shimClient.expect(publish, 14);
rc = client.publish("topic", payload, length, true);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_retained_2() {
IT("publishes retained - 2");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x31, 0xc, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 'A', 'B', 'C', 'D', 'E'};
shimClient.expect(publish, 14);
rc = client.publish("topic", "ABCDE", true);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_not_connected() {
IT("publish fails when not connected");
ShimClient shimClient;
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.publish("topic", "payload");
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_too_long() {
IT("publish fails when topic/payload are too long");
ShimClient shimClient;
shimClient.setAllowConnect(true);
// 0 1 2 3 4 5 6 7 8 9 0 1 2
char topic[] = "1234567890123456789012345678901234567890123456789012345678901234";
// 0 1 2 3 4 5 6 7 8 9 0 1 2
char payload[] = "123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890";
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(64);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
rc = client.publish(topic, payload);
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_P() {
IT("publishes using PROGMEM");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte payload[] = {0x01, 0x02, 0x03, 0x0, 0x05};
size_t length = 5;
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x31, 0xc, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x1, 0x2, 0x3, 0x0, 0x5};
shimClient.expect(publish, 14);
rc = client.publish_P("topic", payload, length, true);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_publish_P_too_long() {
IT("publish using PROGMEM fails when topic are too long");
ShimClient shimClient;
shimClient.setAllowConnect(true);
// 0 1 2 3 4 5 6 7 8 9 0 1 2
char topic[] = "1234567890123456789012345678901234567890123456789012345678901234";
// 0 1 2 3 4 5 6 7 8 9 0 1 2
char payload[] = "123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890";
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(64);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
rc = client.publish_P(topic, payload, false);
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int main() {
SUITE("Publish");
test_publish();
test_publish_bytes();
test_publish_retained();
test_publish_retained_2();
test_publish_not_connected();
test_publish_too_long();
test_publish_P();
test_publish_P_too_long();
FINISH
}
@@ -0,0 +1,345 @@
#include "BDDTest.h"
#include "Buffer.h"
#include "PubSubClient.h"
#include "ShimClient.h"
#include "trace.h"
// 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.
#define PUBLISH_LEN 80
byte server[] = {172, 16, 0, 2};
bool callback_called = false;
char lastTopic[1024];
char lastPayload[1024];
size_t lastLength;
// function declarations
void callback(char* topic, uint8_t* payload, size_t length);
void reset_callback();
int test_receive_callback();
int test_receive_stream();
int test_receive_max_sized_message();
int test_drop_invalid_remaining_length_message();
int test_receive_oversized_message();
int test_resize_buffer();
int test_receive_oversized_stream_message();
int test_receive_qos1();
void reset_callback() {
callback_called = false;
lastTopic[0] = '\0';
lastPayload[0] = '\0';
lastLength = 0;
}
void callback(char* topic, uint8_t* payload, size_t length) {
TRACE("Callback received topic=[" << topic << "] length=" << length << "\n")
callback_called = true;
strcpy(lastTopic, topic);
memcpy(lastPayload, payload, length);
lastLength = length;
}
int test_receive_callback() {
IT("receives a callback message");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xe, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
shimClient.respond(publish, 16);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(memcmp(lastPayload, "payload", 7) == 0);
IS_TRUE(lastLength == 7);
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_stream() {
IT("receives a streamed callback message");
reset_callback();
Stream stream;
uint8_t payload[] = "payload";
stream.expect(payload, sizeof(payload) - 1);
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient, stream);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, 0xe, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
shimClient.respond(publish, 16);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(lastLength == 7);
IS_FALSE(stream.error());
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_max_sized_message() {
IT("receives an max-sized message");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(PUBLISH_LEN);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, PUBLISH_LEN - 2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
byte bigPublish[PUBLISH_LEN];
memset(bigPublish, 'A', PUBLISH_LEN);
bigPublish[PUBLISH_LEN - 1] = 'B';
memcpy(bigPublish, publish, 16);
shimClient.respond(bigPublish, PUBLISH_LEN);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(lastLength == PUBLISH_LEN - 9);
IS_TRUE(memcmp(lastPayload, bigPublish + 9, lastLength) == 0);
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_oversized_message() {
IT("drops an oversized message");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(PUBLISH_LEN - 1);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, PUBLISH_LEN - 2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
byte bigPublish[PUBLISH_LEN];
memset(bigPublish, 'A', PUBLISH_LEN);
bigPublish[PUBLISH_LEN - 1] = 'B';
memcpy(bigPublish, publish, 16);
shimClient.respond(bigPublish, PUBLISH_LEN);
rc = client.loop();
IS_TRUE(rc);
IS_FALSE(callback_called);
IS_FALSE(shimClient.error());
END_IT
}
int test_drop_invalid_remaining_length_message() {
IT("drops invalid remaining length message");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
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, 20);
rc = client.loop();
IS_FALSE(rc);
IS_FALSE(callback_called);
IS_FALSE(shimClient.error());
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);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(PUBLISH_LEN - 1);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, PUBLISH_LEN - 2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
byte bigPublish[PUBLISH_LEN];
memset(bigPublish, 'A', PUBLISH_LEN);
bigPublish[PUBLISH_LEN - 1] = 'B';
memcpy(bigPublish, publish, 16);
// Send it twice
shimClient.respond(bigPublish, PUBLISH_LEN);
shimClient.respond(bigPublish, PUBLISH_LEN);
rc = client.loop();
IS_TRUE(rc);
// First message fails as it is too big
IS_FALSE(callback_called);
// Resize the buffer
client.setBufferSize(PUBLISH_LEN);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(lastLength == PUBLISH_LEN - 9);
IS_TRUE(memcmp(lastPayload, bigPublish + 9, lastLength) == 0);
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_oversized_stream_message() {
IT("receive an oversized streamed message");
reset_callback();
Stream stream;
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient, stream);
client.setBufferSize(PUBLISH_LEN - 1);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x30, PUBLISH_LEN - 2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
byte bigPublish[PUBLISH_LEN];
memset(bigPublish, 'A', PUBLISH_LEN);
bigPublish[PUBLISH_LEN - 1] = 'B';
memcpy(bigPublish, publish, 16);
shimClient.respond(bigPublish, PUBLISH_LEN);
stream.expect(bigPublish + 9, PUBLISH_LEN - 9);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(lastLength == PUBLISH_LEN - 10);
IS_FALSE(stream.error());
IS_FALSE(shimClient.error());
END_IT
}
int test_receive_qos1() {
IT("receives a qos1 message");
reset_callback();
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte publish[] = {0x32, 0x10, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x34, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
shimClient.respond(publish, 18);
byte puback[] = {0x40, 0x2, 0x12, 0x34};
shimClient.expect(puback, 4);
rc = client.loop();
IS_TRUE(rc);
IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic, "topic") == 0);
IS_TRUE(memcmp(lastPayload, "payload", 7) == 0);
IS_TRUE(lastLength == 7);
IS_FALSE(shimClient.error());
END_IT
}
int main() {
SUITE("Receive");
test_receive_callback();
test_receive_stream();
test_receive_max_sized_message();
test_drop_invalid_remaining_length_message();
test_receive_oversized_message();
test_resize_buffer();
test_receive_oversized_stream_message();
test_receive_qos1();
FINISH
}
@@ -0,0 +1,185 @@
#include "BDDTest.h"
#include "Buffer.h"
#include "PubSubClient.h"
#include "ShimClient.h"
#include "trace.h"
byte server[] = {172, 16, 0, 2};
// function declarations
void callback(char* topic, uint8_t* payload, size_t length);
int test_subscribe_no_qos();
int test_subscribe_qos_1();
int test_subscribe_not_connected();
int test_subscribe_invalid_qos();
int test_subscribe_too_long();
int test_unsubscribe();
int test_unsubscribe_not_connected();
void callback(_UNUSED_ char* topic, _UNUSED_ uint8_t* payload, _UNUSED_ size_t length) {
// handle message arrived
}
int test_subscribe_no_qos() {
IT("subscribe without qos defaults to 0");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte subscribe[] = {0x82, 0xa, 0x0, 0x2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x0};
shimClient.expect(subscribe, 12);
byte suback[] = {0x90, 0x3, 0x0, 0x2, 0x0};
shimClient.respond(suback, 5);
rc = client.subscribe("topic");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_subscribe_qos_1() {
IT("subscribes qos 1");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte subscribe[] = {0x82, 0xa, 0x0, 0x2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x1};
shimClient.expect(subscribe, 12);
byte suback[] = {0x90, 0x3, 0x0, 0x2, 0x1};
shimClient.respond(suback, 5);
rc = client.subscribe("topic", 1);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_subscribe_not_connected() {
IT("subscribe fails when not connected");
ShimClient shimClient;
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.subscribe("topic");
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_subscribe_invalid_qos() {
IT("subscribe fails with invalid qos values");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
rc = client.subscribe("topic", 2);
IS_FALSE(rc);
rc = client.subscribe("topic", 254);
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_subscribe_too_long() {
IT("subscribe fails with too long topic");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
client.setBufferSize(128);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
// max length should be allowed
// 0 1 2 3 4 5 6 7 8 9 0 1 2
rc = client.subscribe("12345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789");
IS_TRUE(rc);
// 0 1 2 3 4 5 6 7 8 9 0 1 2
rc = client.subscribe("123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890");
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_unsubscribe() {
IT("unsubscribes");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte connack[] = {0x20, 0x02, 0x00, 0x00};
shimClient.respond(connack, 4);
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.connect("client_test1");
IS_TRUE(rc);
byte unsubscribe[] = {0xA2, 0x9, 0x0, 0x2, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63};
shimClient.expect(unsubscribe, 12);
byte unsuback[] = {0xB0, 0x2, 0x0, 0x2};
shimClient.respond(unsuback, 4);
rc = client.unsubscribe("topic");
IS_TRUE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int test_unsubscribe_not_connected() {
IT("unsubscribe fails when not connected");
ShimClient shimClient;
PubSubClient client(server, 1883, callback, shimClient);
bool rc = client.unsubscribe("topic");
IS_FALSE(rc);
IS_FALSE(shimClient.error());
END_IT
}
int main() {
SUITE("Subscribe");
test_subscribe_no_qos();
test_subscribe_qos_1();
test_subscribe_not_connected();
test_subscribe_invalid_qos();
test_subscribe_too_long();
test_unsubscribe();
test_unsubscribe_not_connected();
FINISH
}
@@ -0,0 +1,36 @@
import unittest
import settings
import time
import paho.mqtt.client as mqttclinet
class mqtt_basic(unittest.TestCase):
def on_message(self, client, userdata, message: mqttclinet.MQTTMessage):
self.message_queue.append(message)
message_queue = []
@classmethod
def setUpClass(self):
self.client = mqttclinet.Client("pubsubclient_ut", clean_session=True)
self.client.connect(settings.server_ip)
self.client.on_message = self.on_message
self.client.subscribe("outTopic", 0)
@classmethod
def tearDownClass(self):
self.client.disconnect()
def test_one(self):
i = 30
while len(self.message_queue) == 0 and i > 0:
self.client.loop()
time.sleep(0.5)
i -= 1
self.assertTrue(i > 0, "message receive timed-out")
self.assertEqual(len(self.message_queue), 1, "unexpected number of messages received")
msg = self.message_queue[0]
self.assertEqual(msg.mid, 0, "message id not 0")
self.assertEqual(msg.topic, "outTopic", "message topic incorrect")
self.assertEqual(msg.payload, "hello world")
self.assertEqual(msg.qos, 0, "message qos not 0")
self.assertEqual(msg.retain, False, "message retain flag incorrect")
@@ -0,0 +1,57 @@
import unittest
import settings
import time
import paho.mqtt.client as mqttclinet
class mqtt_publish_in_callback(unittest.TestCase):
def on_message(self, client, userdata, message: mqttclinet.MQTTMessage):
self.message_queue.append(message)
message_queue = []
@classmethod
def setUpClass(self):
self.client = mqttclinet.Client("pubsubclient_ut", clean_session=True)
self.client.connect(settings.server_ip)
self.client.on_message = self.on_message
self.client.subscribe("outTopic", 0)
@classmethod
def tearDownClass(self):
self.client.disconnect()
def test_connect(self):
i = 30
while len(self.message_queue) == 0 and i > 0:
self.client.loop()
time.sleep(0.5)
i -= 1
self.assertTrue(i > 0, "message receive timed-out")
self.assertEqual(len(self.message_queue), 1, "unexpected number of messages received")
msg = self.message_queue.pop(0)
self.assertEqual(msg.mid, 0, "message id not 0")
self.assertEqual(msg.topic, "outTopic", "message topic incorrect")
self.assertEqual(msg.payload, "hello world")
self.assertEqual(msg.qos, 0, "message qos not 0")
self.assertEqual(msg.retain, False, "message retain flag incorrect")
def test_publish(self):
self.assertEqual(len(self.message_queue), 0, "message queue not empty")
payload = "abcdefghij"
self.client.publish("inTopic", payload)
i = 30
while len(self.message_queue) == 0 and i > 0:
self.client.loop()
time.sleep(0.5)
i -= 1
self.assertTrue(i > 0, "message receive timed-out")
self.assertEqual(len(self.message_queue), 1, "unexpected number of messages received")
msg = self.message_queue.pop(0)
self.assertEqual(msg.mid, 0, "message id not 0")
self.assertEqual(msg.topic, "outTopic", "message topic incorrect")
self.assertEqual(msg.payload, payload)
self.assertEqual(msg.qos, 0, "message qos not 0")
self.assertEqual(msg.retain, False, "message retain flag incorrect")
@@ -0,0 +1,2 @@
server_ip = "172.16.0.2"
arduino_ip = "172.16.0.100"
+181
View File
@@ -0,0 +1,181 @@
#!/usr/bin/env python
import os
import os.path
import sys
import shutil
from subprocess import call
import importlib
import unittest
import re
from testcases import settings
class Workspace(object):
def __init__(self):
self.root_dir = os.getcwd()
self.build_dir = os.path.join(self.root_dir, "tmpbin")
self.log_dir = os.path.join(self.root_dir, "logs")
self.tests_dir = os.path.join(self.root_dir, "testcases")
self.examples_dir = os.path.join(self.root_dir, "../PubSubClient/examples")
self.examples = []
self.tests = []
if not os.path.isdir("../src"):
raise Exception("Cannot find PubSubClient library")
try:
return __import__('ino')
except ImportError:
raise Exception("ino tool not installed")
def init(self):
if os.path.isdir(self.build_dir):
shutil.rmtree(self.build_dir)
os.mkdir(self.build_dir)
if os.path.isdir(self.log_dir):
shutil.rmtree(self.log_dir)
os.mkdir(self.log_dir)
os.chdir(self.build_dir)
call(["ino", "init"])
shutil.copytree("../../PubSubClient", "lib/PubSubClient")
filenames = []
for root, dirs, files in os.walk(self.examples_dir):
filenames += [os.path.join(root, f) for f in files if f.endswith(".ino")]
filenames.sort()
for e in filenames:
self.examples.append(Sketch(self, e))
filenames = []
for root, dirs, files in os.walk(self.tests_dir):
filenames += [os.path.join(root, f) for f in files if f.endswith(".ino")]
filenames.sort()
for e in filenames:
self.tests.append(Sketch(self, e))
def clean(self):
shutil.rmtree(self.build_dir)
class Sketch(object):
def __init__(self, wksp, fn):
self.w = wksp
self.filename = fn
self.basename = os.path.basename(self.filename)
self.build_log = os.path.join(self.w.log_dir, "%s.log" % (os.path.basename(self.filename),))
self.build_err_log = os.path.join(self.w.log_dir, "%s.err.log" % (os.path.basename(self.filename),))
self.build_upload_log = os.path.join(self.w.log_dir, "%s.upload.log" % (os.path.basename(self.filename),))
def build(self):
sys.stdout.write(" Build: ")
sys.stdout.flush()
# Copy sketch over, replacing IP addresses as necessary
fin = open(self.filename, "r")
lines = fin.readlines()
fin.close()
fout = open(os.path.join(self.w.build_dir, "src", "sketch.ino"), "w")
for l in lines:
if re.match(r"^byte server\[\] = {", l):
fout.write("byte server[] = { %s };\n" % (settings.server_ip.replace(".", ", "),))
elif re.match(r"^byte ip\[\] = {", l):
fout.write("byte ip[] = { %s };\n" % (settings.arduino_ip.replace(".", ", "),))
else:
fout.write(l)
fout.flush()
fout.close()
# Run build
fout = open(self.build_log, "w")
ferr = open(self.build_err_log, "w")
rc = call(["ino", "build"], stdout=fout, stderr=ferr)
fout.close()
ferr.close()
if rc == 0:
sys.stdout.write("pass")
sys.stdout.write("\n")
return True
else:
sys.stdout.write("fail")
sys.stdout.write("\n")
with open(self.build_err_log) as f:
for line in f:
print(" " + line)
return False
def upload(self):
sys.stdout.write(" Upload: ")
sys.stdout.flush()
fout = open(self.build_upload_log, "w")
rc = call(["ino", "upload"], stdout=fout, stderr=fout)
fout.close()
if rc == 0:
sys.stdout.write("pass")
sys.stdout.write("\n")
return True
else:
sys.stdout.write("fail")
sys.stdout.write("\n")
with open(self.build_upload_log) as f:
for line in f:
print(" " + line)
return False
def test(self):
# import the matching test case, if it exists
try:
basename = os.path.basename(self.filename)[:-4]
i = importlib.import_module("testcases." + basename)
except:
sys.stdout.write(" Test: no tests found")
sys.stdout.write("\n")
return
c = getattr(i, basename)
testmethods = [m for m in dir(c) if m.startswith("test_")]
testmethods.sort()
tests = []
for m in testmethods:
tests.append(c(m))
result = unittest.TestResult()
c.setUpClass()
if self.upload():
sys.stdout.write(" Test: ")
sys.stdout.flush()
for t in tests:
t.run(result)
print(str(result.testsRun - len(result.failures) - len(result.errors)) + "/" + str(result.testsRun))
if not result.wasSuccessful():
if len(result.failures) > 0:
for f in result.failures:
print("-- " + str(f[0]))
print(f[1])
if len(result.errors) > 0:
print(" Errors:")
for f in result.errors:
print("-- " + str(f[0]))
print(f[1])
c.tearDownClass()
if __name__ == '__main__':
run_tests = True
w = Workspace()
w.init()
for e in w.examples:
print("--------------------------------------")
print("[" + e.basename + "]")
if e.build() and run_tests:
e.test()
for e in w.tests:
print("--------------------------------------")
print("[" + e.basename + "]")
if e.build() and run_tests:
e.test()
w.clean()