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

Dynamic buffer size.

Browse Source
This commit is contained in:
duncan law 2017-04-29 09:44:44 +01:00
parent f46d0011ee
commit a257083a66
2 changed files with 54 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/PubSubClient.cpp
View File

@ -12,12 +12,16 @@ PubSubClient::PubSubClient() {
this->_client = NULL; this->_client = NULL;
this->stream = NULL; this->stream = NULL;
setCallback(NULL); setCallback(NULL);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(Client& client) { PubSubClient::PubSubClient(Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
@ -25,12 +29,16 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
setServer(addr, port); setServer(addr, port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(addr,port); setServer(addr,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -38,6 +46,8 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATUR
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -45,6 +55,8 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATUR
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
@ -52,12 +64,16 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
setServer(ip, port); setServer(ip, port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(ip,port); setServer(ip,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -65,6 +81,8 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE,
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -72,6 +90,8 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE,
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) { PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
@ -79,12 +99,16 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
setServer(domain,port); setServer(domain,port);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) { PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
setServer(domain,port); setServer(domain,port);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -92,6 +116,8 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
this->stream = NULL; this->stream = NULL;
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
} }
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
this->_state = MQTT_DISCONNECTED; this->_state = MQTT_DISCONNECTED;
@ -99,6 +125,12 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
setCallback(callback); setCallback(callback);
setClient(client); setClient(client);
setStream(stream); setStream(stream);
this->buffer_size = MQTT_MAX_PACKET_SIZE;
this->buffer = (uint8_t*)malloc(MQTT_MAX_PACKET_SIZE);
}
PubSubClient::~PubSubClient() {
free(this->buffer);
} }
boolean PubSubClient::connect(const char *id) { boolean PubSubClient::connect(const char *id) {
@ -266,13 +298,13 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) {
this->stream->write(digit); this->stream->write(digit);
} }
} }
if (len < MQTT_MAX_PACKET_SIZE) { if (len < this->buffer_size) {
buffer[len] = digit; buffer[len] = digit;
} }
len++; len++;
} }
if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { if (!this->stream && len > this->buffer_size) {
len = 0; // This will cause the packet to be ignored. len = 0; // This will cause the packet to be ignored.
} }
@ -356,7 +388,7 @@ boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigne
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) { boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
if (connected()) { if (connected()) {
if (MQTT_MAX_PACKET_SIZE < 5 + 2+strlen(topic) + plength) { if (this->buffer_size < 5 + 2+strlen(topic) + plength) {
// Too long // Too long
return false; return false;
} }
@ -471,7 +503,7 @@ boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
if (qos < 0 || qos > 1) { if (qos < 0 || qos > 1) {
return false; return false;
} }
if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { if (this->buffer_size < 9 + strlen(topic)) {
// Too long // Too long
return false; return false;
} }
@ -492,7 +524,7 @@ boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
} }
boolean PubSubClient::unsubscribe(const char* topic) { boolean PubSubClient::unsubscribe(const char* topic) {
if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) { if (this->buffer_size < 9 + strlen(topic)) {
// Too long // Too long
return false; return false;
} }
@ -586,3 +618,13 @@ PubSubClient& PubSubClient::setStream(Stream& stream){
int PubSubClient::state() { int PubSubClient::state() {
return this->_state; return this->_state;
} }
boolean PubSubClient::setBufferSize(uint16_t size) {
this->buffer = (uint8_t*)realloc(this->buffer, size);
this->buffer_size = size;
return (this->buffer == NULL);
}
uint16_t PubSubClient::getBufferSize() {
return this->buffer_size;
}

9
src/PubSubClient.h
View File

@ -21,7 +21,7 @@
#define MQTT_VERSION MQTT_VERSION_3_1_1 #define MQTT_VERSION MQTT_VERSION_3_1_1
#endif #endif
// MQTT_MAX_PACKET_SIZE : Maximum packet size // MQTT_MAX_PACKET_SIZE : Maximum packet size. Override with setBufferSize().
#ifndef MQTT_MAX_PACKET_SIZE #ifndef MQTT_MAX_PACKET_SIZE
#define MQTT_MAX_PACKET_SIZE 128 #define MQTT_MAX_PACKET_SIZE 128
#endif #endif
@ -83,7 +83,8 @@
class PubSubClient { class PubSubClient {
private: private:
Client* _client; Client* _client;
uint8_t buffer[MQTT_MAX_PACKET_SIZE]; uint8_t* buffer;
uint16_t buffer_size;
uint16_t nextMsgId; uint16_t nextMsgId;
unsigned long lastOutActivity; unsigned long lastOutActivity;
unsigned long lastInActivity; unsigned long lastInActivity;
@ -115,6 +116,8 @@ public:
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client); PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client);
PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&); PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE,Client& client, Stream&);
~PubSubClient();
PubSubClient& setServer(IPAddress ip, uint16_t port); PubSubClient& setServer(IPAddress ip, uint16_t port);
PubSubClient& setServer(uint8_t * ip, uint16_t port); PubSubClient& setServer(uint8_t * ip, uint16_t port);
PubSubClient& setServer(const char * domain, uint16_t port); PubSubClient& setServer(const char * domain, uint16_t port);
@ -138,6 +141,8 @@ public:
boolean loop(); boolean loop();
boolean connected(); boolean connected();
int state(); int state();
boolean setBufferSize(uint16_t size);
uint16_t getBufferSize();
}; };