diff --git a/PubSubClient/PubSubClient.cpp b/PubSubClient/PubSubClient.cpp index 27f37a6..fcfd86d 100755 --- a/PubSubClient/PubSubClient.cpp +++ b/PubSubClient/PubSubClient.cpp @@ -9,6 +9,7 @@ PubSubClient::PubSubClient() { this->_client = NULL; + this->stream = NULL; } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client) { @@ -17,6 +18,7 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,ui this->ip = ip; this->port = port; this->domain = NULL; + this->stream = NULL; } PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client) { @@ -24,6 +26,24 @@ PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,u this->callback = callback; this->domain = domain; this->port = port; + this->stream = NULL; +} + +PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client, Stream *stream) { + this->_client = &client; + this->callback = callback; + this->ip = ip; + this->port = port; + this->domain = NULL; + this->stream = stream; +} + +PubSubClient::PubSubClient(char* domain, uint16_t port, void (*callback)(char*,uint8_t*,unsigned int), Client& client, Stream *stream) { + this->_client = &client; + this->callback = callback; + this->domain = domain; + this->port = port; + this->stream = stream; } boolean PubSubClient::connect(char *id) { @@ -134,16 +154,32 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { multiplier *= 128; } while ((digit & 128) != 0); *lengthLength = len-1; - for (uint16_t i = 0;istream && ((buffer[0]&0xF0) == MQTTPUBLISH) && len-*lengthLength-2>skip) { + this->stream->write(digit); + } if (len < MQTT_MAX_PACKET_SIZE) { - buffer[len++] = readByte(); + buffer[len++] = digit; } else { - readByte(); - len = 0; // This will cause the packet to be ignored. + if(!this->stream) len = 0; // This will cause the packet to be ignored. } } + // If a stream has been provided, indicate that we wrote the whole length, + // else return 0 if the length exceed the max packet size return len; } diff --git a/PubSubClient/PubSubClient.h b/PubSubClient/PubSubClient.h index f87d5c1..6d7b1f2 100755 --- a/PubSubClient/PubSubClient.h +++ b/PubSubClient/PubSubClient.h @@ -53,10 +53,13 @@ private: uint8_t *ip; char* domain; uint16_t port; + Stream* stream; public: PubSubClient(); PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client); + PubSubClient(uint8_t *, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client, Stream*); PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client); + PubSubClient(char*, uint16_t, void(*)(char*,uint8_t*,unsigned int),Client& client, Stream*); boolean connect(char *); boolean connect(char *, char *, char *); boolean connect(char *, char *, uint8_t, uint8_t, char *); diff --git a/PubSubClient/examples/mqtt_stream/mqtt_stream.ino b/PubSubClient/examples/mqtt_stream/mqtt_stream.ino new file mode 100644 index 0000000..0bc4279 --- /dev/null +++ b/PubSubClient/examples/mqtt_stream/mqtt_stream.ino @@ -0,0 +1,58 @@ +/* + 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 +#include +#include +#include + +// Update these with values suitable for your network. +byte mac[] = { 0xDE, 0xED, 0xBA, 0xFE, 0xFE, 0xED }; +byte server[] = { 172, 16, 0, 2 }; +byte ip[] = { 172, 16, 0, 100 }; + +SRAM sram(4, SRAM_1024); + +void callback(char* topic, byte* payload, unsigned int length) { + sram.seek(1); + + // do something with the message + for(uint8_t i=0; i