wn/pubsubclient
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into master

Browse Source
This commit is contained in:
Nick O'Leary 2018-11-02 01:07:33 +00:00 committed by GitHub
commit fa3c4362ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 392 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -8,7 +8,7 @@ a server that supports MQTT.
The library comes with a number of example sketches. See File > Examples > PubSubClient
within the Arduino application.
Full API documentation is available here: http://pubsubclient.knolleary.net
Full API documentation is available here: https://pubsubclient.knolleary.net
## Limitations

4
examples/mqtt_esp8266/mqtt_esp8266.ino
View File

@ -74,7 +74,7 @@ void callback(char* topic, byte* payload, unsigned int length) {
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 acive low on the ESP-01)
// it is active low on the ESP-01)
} else {
digitalWrite(BUILTIN_LED, HIGH); // Turn the LED off by making the voltage HIGH
}
@ -124,7 +124,7 @@ void loop() {
if (now - lastMsg > 2000) {
lastMsg = now;
++value;
snprintf (msg, 75, "hello world #%ld", value);
snprintf (msg, 50, "hello world #%ld", value);
Serial.print("Publish message: ");
Serial.println(msg);
client.publish("outTopic", msg);

179
examples/mqtt_large_message/mqtt_large_message.ino Normal file
View File

@ -0,0 +1,179 @@
/*
Long message ESP8266 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"
*/
#include <ESP8266WiFi.h>
#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(micros());
Serial.println("");
Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
}
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (int 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
int msgLen = 0;
for (int 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();
}

3
keywords.txt
View File

@ -16,6 +16,9 @@ connect KEYWORD2
disconnect KEYWORD2
publish KEYWORD2
publish_P KEYWORD2
beginPublish KEYWORD2
endPublish KEYWORD2
write KEYWORD2
subscribe KEYWORD2
unsubscribe KEYWORD2
loop KEYWORD2

109
src/PubSubClient.cpp
View File

@ -102,18 +102,22 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
}
boolean PubSubClient::connect(const char *id) {
return connect(id,NULL,NULL,0,0,0,0);
return connect(id,NULL,NULL,0,0,0,0,1);
}
boolean PubSubClient::connect(const char *id, const char *user, const char *pass) {
return connect(id,user,pass,0,0,0,0);
return connect(id,user,pass,0,0,0,0,1);
}
boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage);
return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage,1);
}
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1);
}
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession) {
if (!connected()) {
int result = 0;
@ -125,7 +129,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
if (result == 1) {
nextMsgId = 1;
// Leave room in the buffer for header and variable length field
uint16_t length = 5;
uint16_t length = MQTT_MAX_HEADER_SIZE;
unsigned int j;
#if MQTT_VERSION == MQTT_VERSION_3_1
@ -141,9 +145,12 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
uint8_t v;
if (willTopic) {
v = 0x06|(willQos<<3)|(willRetain<<5);
v = 0x04|(willQos<<3)|(willRetain<<5);
} else {
v = 0x02;
v = 0x00;
}
if (cleanSession) {
v = v|0x02;
}
if(user != NULL) {
@ -158,20 +165,26 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
buffer[length++] = ((MQTT_KEEPALIVE) >> 8);
buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF);
CHECK_STRING_LENGTH(length,id)
length = writeString(id,buffer,length);
if (willTopic) {
CHECK_STRING_LENGTH(length,willTopic)
length = writeString(willTopic,buffer,length);
CHECK_STRING_LENGTH(length,willMessage)
length = writeString(willMessage,buffer,length);
}
if(user != NULL) {
CHECK_STRING_LENGTH(length,user)
length = writeString(user,buffer,length);
if(pass != NULL) {
CHECK_STRING_LENGTH(length,pass)
length = writeString(pass,buffer,length);
}
}
write(MQTTCONNECT,buffer,length-5);
write(MQTTCONNECT,buffer,length-MQTT_MAX_HEADER_SIZE);
lastInActivity = lastOutActivity = millis();
@ -209,6 +222,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
boolean PubSubClient::readByte(uint8_t * result) {
uint32_t previousMillis = millis();
while(!_client->available()) {
yield();
uint32_t currentMillis = millis();
if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){
return false;
@ -240,6 +254,12 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) {
uint8_t start = 0;
do {
if (len == 5) {
// Invalid remaining length encoding - kill the connection
_state = MQTT_DISCONNECTED;
_client->stop();
return 0;
}
if(!readByte(&digit)) return 0;
buffer[len++] = digit;
length += (digit & 127) * multiplier;
@ -335,6 +355,9 @@ boolean PubSubClient::loop() {
} else if (type == MQTTPINGRESP) {
pingOutstanding = false;
}
} else if (!connected()) {
// readPacket has closed the connection
return false;
}
}
return true;
@ -356,12 +379,12 @@ boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigne
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
if (connected()) {
if (MQTT_MAX_PACKET_SIZE < 5 + 2+strlen(topic) + plength) {
if (MQTT_MAX_PACKET_SIZE < MQTT_MAX_HEADER_SIZE + 2+strlen(topic) + plength) {
// Too long
return false;
}
// Leave room in the buffer for header and variable length field
uint16_t length = 5;
uint16_t length = MQTT_MAX_HEADER_SIZE;
length = writeString(topic,buffer,length);
uint16_t i;
for (i=0;i<plength;i++) {
@ -371,11 +394,15 @@ boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigne
if (retained) {
header |= 1;
}
return write(header,buffer,length-5);
return write(header,buffer,length-MQTT_MAX_HEADER_SIZE);
}
return false;
}
boolean PubSubClient::publish_P(const char* topic, const char* payload, boolean retained) {
return publish_P(topic, (const uint8_t*)payload, strlen(payload), retained);
}
boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
uint8_t llen = 0;
uint8_t digit;
@ -421,12 +448,43 @@ boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsig
return rc == tlen + 4 + plength;
}
boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
boolean PubSubClient::beginPublish(const char* topic, unsigned int plength, boolean retained) {
if (connected()) {
// Send the header and variable length field
uint16_t length = MQTT_MAX_HEADER_SIZE;
length = writeString(topic,buffer,length);
uint16_t i;
uint8_t header = MQTTPUBLISH;
if (retained) {
header |= 1;
}
size_t hlen = buildHeader(header, buffer, plength+length-MQTT_MAX_HEADER_SIZE);
uint16_t rc = _client->write(buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen));
lastOutActivity = millis();
return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen)));
}
return false;
}
int PubSubClient::endPublish() {
return 1;
}
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);
}
size_t PubSubClient::buildHeader(uint8_t header, uint8_t* buf, uint16_t length) {
uint8_t lenBuf[4];
uint8_t llen = 0;
uint8_t digit;
uint8_t pos = 0;
uint16_t rc;
uint16_t len = length;
do {
digit = len % 128;
@ -440,12 +498,18 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
buf[4-llen] = header;
for (int i=0;i<llen;i++) {
buf[5-llen+i] = lenBuf[i];
buf[MQTT_MAX_HEADER_SIZE-llen+i] = lenBuf[i];
}
return llen+1; // Full header size is variable length bit plus the 1-byte fixed header
}
boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
uint16_t rc;
uint8_t hlen = buildHeader(header, buf, length);
#ifdef MQTT_MAX_TRANSFER_SIZE
uint8_t* writeBuf = buf+(4-llen);
uint16_t bytesRemaining = length+1+llen; //Match the length type
uint8_t* writeBuf = buf+(MQTT_MAX_HEADER_SIZE-hlen);
uint16_t bytesRemaining = length+hlen; //Match the length type
uint8_t bytesToWrite;
boolean result = true;
while((bytesRemaining > 0) && result) {
@ -457,9 +521,9 @@ boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
}
return result;
#else
rc = _client->write(buf+(4-llen),length+1+llen);
rc = _client->write(buf+(MQTT_MAX_HEADER_SIZE-hlen),length+hlen);
lastOutActivity = millis();
return (rc == 1+llen+length);
return (rc == hlen+length);
#endif
}
@ -468,7 +532,7 @@ boolean PubSubClient::subscribe(const char* topic) {
}
boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
if (qos < 0 || qos > 1) {
if (qos > 1) {
return false;
}
if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) {
@ -477,7 +541,7 @@ boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
}
if (connected()) {
// Leave room in the buffer for header and variable length field
uint16_t length = 5;
uint16_t length = MQTT_MAX_HEADER_SIZE;
nextMsgId++;
if (nextMsgId == 0) {
nextMsgId = 1;
@ -486,7 +550,7 @@ boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
buffer[length++] = (nextMsgId & 0xFF);
length = writeString((char*)topic, buffer,length);
buffer[length++] = qos;
return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5);
return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
}
return false;
}
@ -497,7 +561,7 @@ boolean PubSubClient::unsubscribe(const char* topic) {
return false;
}
if (connected()) {
uint16_t length = 5;
uint16_t length = MQTT_MAX_HEADER_SIZE;
nextMsgId++;
if (nextMsgId == 0) {
nextMsgId = 1;
@ -505,7 +569,7 @@ boolean PubSubClient::unsubscribe(const char* topic) {
buffer[length++] = (nextMsgId >> 8);
buffer[length++] = (nextMsgId & 0xFF);
length = writeString(topic, buffer,length);
return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5);
return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
}
return false;
}
@ -515,6 +579,7 @@ void PubSubClient::disconnect() {
buffer[1] = 0;
_client->write(buffer,2);
_state = MQTT_DISCONNECTED;
_client->flush();
_client->stop();
lastInActivity = lastOutActivity = millis();
}

31
src/PubSubClient.h
View File

@ -73,6 +73,9 @@
#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(ESP8266) || defined(ESP32)
#include <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> callback
@ -80,7 +83,9 @@
#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int)
#endif
class PubSubClient {
#define CHECK_STRING_LENGTH(l,s) if (l+2+strlen(s) > MQTT_MAX_PACKET_SIZE) {_client->stop();return false;}
class PubSubClient : public Print {
private:
Client* _client;
uint8_t buffer[MQTT_MAX_PACKET_SIZE];
@ -94,6 +99,11 @@ private:
boolean readByte(uint8_t * result, uint16_t * index);
boolean write(uint8_t header, uint8_t* buf, uint16_t length);
uint16_t writeString(const char* string, uint8_t* buf, uint16_t pos);
// Build up the header ready to send
// Returns the size of the header
// 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
size_t buildHeader(uint8_t header, uint8_t* buf, uint16_t length);
IPAddress ip;
const char* domain;
uint16_t port;
@ -126,12 +136,31 @@ public:
boolean connect(const char* id, const char* user, const char* pass);
boolean connect(const char* id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage);
boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage);
boolean connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession);
void disconnect();
boolean publish(const char* topic, const char* payload);
boolean publish(const char* topic, const char* payload, boolean retained);
boolean publish(const char* topic, const uint8_t * payload, unsigned int plength);
boolean publish(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained);
boolean publish_P(const char* topic, const char* payload, boolean retained);
boolean publish_P(const char* topic, const uint8_t * payload, unsigned int plength, boolean retained);
// 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
// Returns 1 if the message was started successfully, 0 if there was an error
boolean beginPublish(const char* topic, unsigned int plength, boolean retained);
// Finish off this publish message (started with beginPublish)
// Returns 1 if the packet was sent successfully, 0 if there was an error
int endPublish();
// Write a single byte of payload (only to be used with beginPublish/endPublish)
virtual size_t write(uint8_t);
// Write size bytes from buffer into the payload (only to be used with beginPublish/endPublish)
// Returns the number of bytes written
virtual size_t write(const uint8_t *buffer, size_t size);
boolean subscribe(const char* topic);
boolean subscribe(const char* topic, uint8_t qos);
boolean unsubscribe(const char* topic);

29
tests/src/connect_spec.cpp
View File

@ -98,6 +98,33 @@ int test_connect_fails_on_bad_rc() {
END_IT
}
int test_connect_non_clean_session() {
IT("sends a properly formatted non-clean session connect packet and succeeds");
ShimClient shimClient;
shimClient.setAllowConnect(true);
byte expectServer[] = { 172, 16, 0, 2 };
shimClient.expectConnect(expectServer,1883);
byte connect[] = {0x10,0x18,0x0,0x4,0x4d,0x51,0x54,0x54,0x4,0x0,0x0,0xf,0x0,0xc,0x63,0x6c,0x69,0x65,0x6e,0x74,0x5f,0x74,0x65,0x73,0x74,0x31};
byte connack[] = { 0x20, 0x02, 0x00, 0x00 };
shimClient.expect(connect,26);
shimClient.respond(connack,4);
PubSubClient client(server, 1883, callback, shimClient);
int state = client.state();
IS_TRUE(state == MQTT_DISCONNECTED);
int rc = client.connect((char*)"client_test1",0,0,0,0,0,0,0);
IS_TRUE(rc);
IS_FALSE(shimClient.error());
state = client.state();
IS_TRUE(state == MQTT_CONNECTED);
END_IT
}
int test_connect_accepts_username_password() {
IT("accepts a username and password");
ShimClient shimClient;
@ -256,10 +283,12 @@ int test_connect_disconnect_connect() {
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();

3
tests/src/lib/Arduino.h
View File

@ -5,6 +5,7 @@
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include "Print.h"
extern "C"{
@ -20,4 +21,6 @@ extern "C"{
#define PROGMEM
#define pgm_read_byte_near(x) *(x)
#define yield(x) {}
#endif // Arduino_h

28
tests/src/lib/Print.h Normal file
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

30
tests/src/receive_spec.cpp
View File

@ -160,6 +160,35 @@ int test_receive_oversized_message() {
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);
int rc = client.connect((char*)"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_receive_oversized_stream_message() {
IT("drops an oversized message");
reset_callback();
@ -241,6 +270,7 @@ int main()
test_receive_callback();
test_receive_stream();
test_receive_max_sized_message();
test_drop_invalid_remaining_length_message();
test_receive_oversized_message();
test_receive_oversized_stream_message();
test_receive_qos1();