mathieu/Adafruit_MQTT_Library
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Moved to actions, no doxygen

Browse Source
This commit is contained in:
dherrada 2020-06-25 09:56:46 -04:00
parent 3693eb8800
commit 148cfb3591
No known key found for this signature in database
GPG Key ID: CE2ADBAB8775CE81
10 changed files with 414 additions and 359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
.github/workflows/githubci.yml vendored Normal file
View File

@ -0,0 +1,26 @@
name: Arduino Library CI
on: [pull_request, push, repository_dispatch]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/setup-python@v1
with:
python-version: '3.x'
- uses: actions/checkout@v2
- uses: actions/checkout@v2
with:
repository: adafruit/ci-arduino
path: ci
- name: pre-install
run: bash ci/actions_install.sh
- name: test platforms
run: python3 ci/build_platform.py esp8266 zero
- name: clang
run: python3 ci/run-clang-format.py -e "ci/*" -e "bin/*" -r .

14
.travis.yml
View File

@ -1,14 +0,0 @@
language: c
sudo: false
before_install:
- source <(curl -SLs https://raw.githubusercontent.com/adafruit/travis-ci-arduino/master/install.sh)
install:
- arduino --install-library "Adafruit SleepyDog Library","Adafruit FONA Library"
script:
- build_platform esp8266
- arduino --install-library "WiFi101"
- build_platform zero
notifications:
email:
on_success: change
on_failure: change

325
Adafruit_MQTT.cpp
View File

@ -9,8 +9,8 @@
// copies of the Software, and to permit persons to whom the Software is // copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions: // furnished to do so, subject to the following conditions:
// //
// The above copyright notice and this permission notice shall be included in all // The above copyright notice and this permission notice shall be included in
// copies or substantial portions of the Software. // all copies or substantial portions of the Software.
// //
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
@ -21,8 +21,10 @@
// SOFTWARE. // SOFTWARE.
#include "Adafruit_MQTT.h" #include "Adafruit_MQTT.h"
#if defined(ARDUINO_SAMD_ZERO) || defined(ARDUINO_SAMD_MKR1000) || defined(ARDUINO_ARCH_SAMD) #if defined(ARDUINO_SAMD_ZERO) || defined(ARDUINO_SAMD_MKR1000) || \
static char *dtostrf (double val, signed char width, unsigned char prec, char *sout) { defined(ARDUINO_ARCH_SAMD)
static char *dtostrf(double val, signed char width, unsigned char prec,
char *sout) {
char fmt[20]; char fmt[20];
sprintf(fmt, "%%%d.%df", width, prec); sprintf(fmt, "%%%d.%df", width, prec);
sprintf(sout, fmt, val); sprintf(sout, fmt, val);
@ -31,12 +33,12 @@ static char *dtostrf (double val, signed char width, unsigned char prec, char *s
#endif #endif
#if defined(ESP8266) #if defined(ESP8266)
int strncasecmp(const char * str1, const char * str2, int len) { int strncasecmp(const char *str1, const char *str2, int len) {
int d = 0; int d = 0;
while(len--) { while (len--) {
int c1 = tolower(*str1++); int c1 = tolower(*str1++);
int c2 = tolower(*str2++); int c2 = tolower(*str2++);
if(((d = c1 - c2) != 0) || (c2 == '\0')) { if (((d = c1 - c2) != 0) || (c2 == '\0')) {
return d; return d;
} }
} }
@ -44,10 +46,9 @@ int strncasecmp(const char * str1, const char * str2, int len) {
} }
#endif #endif
void printBuffer(uint8_t *buffer, uint16_t len) { void printBuffer(uint8_t *buffer, uint16_t len) {
DEBUG_PRINTER.print('\t'); DEBUG_PRINTER.print('\t');
for (uint16_t i=0; i<len; i++) { for (uint16_t i = 0; i < len; i++) {
if (isprint(buffer[i])) if (isprint(buffer[i]))
DEBUG_PRINTER.write(buffer[i]); DEBUG_PRINTER.write(buffer[i]);
else else
@ -55,7 +56,7 @@ void printBuffer(uint8_t *buffer, uint16_t len) {
DEBUG_PRINTER.print(F(" [0x")); DEBUG_PRINTER.print(F(" [0x"));
if (buffer[i] < 0x10) if (buffer[i] < 0x10)
DEBUG_PRINTER.print("0"); DEBUG_PRINTER.print("0");
DEBUG_PRINTER.print(buffer[i],HEX); DEBUG_PRINTER.print(buffer[i], HEX);
DEBUG_PRINTER.print("], "); DEBUG_PRINTER.print("], ");
if (i % 8 == 7) { if (i % 8 == 7) {
DEBUG_PRINTER.print("\n\t"); DEBUG_PRINTER.print("\n\t");
@ -74,7 +75,7 @@ static uint8_t *stringprint(uint8_t *p, char *s) {
} }
*/ */
static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen=0) { static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen = 0) {
// If maxlen is specified (has a non-zero value) then use it as the maximum // If maxlen is specified (has a non-zero value) then use it as the maximum
// length of the source string to write to the buffer. Otherwise write // length of the source string to write to the buffer. Otherwise write
// the entire source string. // the entire source string.
@ -87,20 +88,18 @@ static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen=0) {
Serial.write(pgm_read_byte(s+i)); Serial.write(pgm_read_byte(s+i));
} }
*/ */
p[0] = len >> 8; p++; p[0] = len >> 8;
p[0] = len & 0xFF; p++; p++;
p[0] = len & 0xFF;
p++;
strncpy((char *)p, s, len); strncpy((char *)p, s, len);
return p+len; return p + len;
} }
// Adafruit_MQTT Definition //////////////////////////////////////////////////// // Adafruit_MQTT Definition ////////////////////////////////////////////////////
Adafruit_MQTT::Adafruit_MQTT(const char *server, Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port, const char *cid,
uint16_t port, const char *user, const char *pass) {
const char *cid,
const char *user,
const char *pass) {
servername = server; servername = server;
portnum = port; portnum = port;
clientid = cid; clientid = cid;
@ -108,7 +107,7 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
password = pass; password = pass;
// reset subscriptions // reset subscriptions
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
subscriptions[i] = 0; subscriptions[i] = 0;
} }
@ -118,14 +117,10 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
will_retain = 0; will_retain = 0;
packet_id_counter = 0; packet_id_counter = 0;
} }
Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port,
Adafruit_MQTT::Adafruit_MQTT(const char *server, const char *user, const char *pass) {
uint16_t port,
const char *user,
const char *pass) {
servername = server; servername = server;
portnum = port; portnum = port;
clientid = ""; clientid = "";
@ -133,7 +128,7 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
password = pass; password = pass;
// reset subscriptions // reset subscriptions
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
subscriptions[i] = 0; subscriptions[i] = 0;
} }
@ -143,7 +138,6 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
will_retain = 0; will_retain = 0;
packet_id_counter = 0; packet_id_counter = 0;
} }
int8_t Adafruit_MQTT::connect() { int8_t Adafruit_MQTT::connect() {
@ -166,66 +160,70 @@ int8_t Adafruit_MQTT::connect() {
return buffer[3]; return buffer[3];
// Setup subscriptions once connected. // Setup subscriptions once connected.
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
// Ignore subscriptions that aren't defined. // Ignore subscriptions that aren't defined.
if (subscriptions[i] == 0) continue; if (subscriptions[i] == 0)
continue;
boolean success = false; boolean success = false;
for (uint8_t retry=0; (retry<3) && !success; retry++) { // retry until we get a suback for (uint8_t retry = 0; (retry < 3) && !success;
retry++) { // retry until we get a suback
// Construct and send subscription packet. // Construct and send subscription packet.
uint8_t len = subscribePacket(buffer, subscriptions[i]->topic, subscriptions[i]->qos); uint8_t len = subscribePacket(buffer, subscriptions[i]->topic,
subscriptions[i]->qos);
if (!sendPacket(buffer, len)) if (!sendPacket(buffer, len))
return -1; return -1;
if(MQTT_PROTOCOL_LEVEL < 3) // older versions didn't suback if (MQTT_PROTOCOL_LEVEL < 3) // older versions didn't suback
break; break;
// Check for SUBACK if using MQTT 3.1.1 or higher // Check for SUBACK if using MQTT 3.1.1 or higher
// TODO: The Server is permitted to start sending PUBLISH packets matching the // TODO: The Server is permitted to start sending PUBLISH packets matching
// Subscription before the Server sends the SUBACK Packet. (will really need to use callbacks - ada) // the Subscription before the Server sends the SUBACK Packet. (will
// really need to use callbacks - ada)
//Serial.println("\t**looking for suback"); // Serial.println("\t**looking for suback");
if (processPacketsUntil(buffer, MQTT_CTRL_SUBACK, SUBACK_TIMEOUT_MS)) { if (processPacketsUntil(buffer, MQTT_CTRL_SUBACK, SUBACK_TIMEOUT_MS)) {
success = true; success = true;
break; break;
} }
} }
if (! success) return -2; // failed to sub for some reason if (!success)
return -2; // failed to sub for some reason
} }
return 0; return 0;
} }
int8_t Adafruit_MQTT::connect(const char *user, const char *pass) int8_t Adafruit_MQTT::connect(const char *user, const char *pass) {
{
username = user; username = user;
password = pass; password = pass;
return connect(); return connect();
} }
uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer, uint8_t waitforpackettype, uint16_t timeout) { uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer,
uint8_t waitforpackettype,
uint16_t timeout) {
uint16_t len; uint16_t len;
while(true) { while (true) {
len = readFullPacket(buffer, MAXBUFFERSIZE, timeout); len = readFullPacket(buffer, MAXBUFFERSIZE, timeout);
if(len == 0){ if (len == 0) {
break; break;
} }
if ((buffer[0] >> 4) == waitforpackettype) if ((buffer[0] >> 4) == waitforpackettype) {
{
return len; return len;
} } else {
else
{
ERROR_PRINTLN(F("Dropped a packet")); ERROR_PRINTLN(F("Dropped a packet"));
} }
} }
return 0; return 0;
} }
uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16_t timeout) { uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize,
uint16_t timeout) {
// will read a packet and Do The Right Thing with length // will read a packet and Do The Right Thing with length
uint8_t *pbuff = buffer; uint8_t *pbuff = buffer;
@ -233,9 +231,11 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16
// read the packet type: // read the packet type:
rlen = readPacket(pbuff, 1, timeout); rlen = readPacket(pbuff, 1, timeout);
if (rlen != 1) return 0; if (rlen != 1)
return 0;
DEBUG_PRINT(F("Packet Type:\t")); DEBUG_PRINTBUFFER(pbuff, rlen); DEBUG_PRINT(F("Packet Type:\t"));
DEBUG_PRINTBUFFER(pbuff, rlen);
pbuff++; pbuff++;
uint32_t value = 0; uint32_t value = 0;
@ -244,44 +244,59 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16
do { do {
rlen = readPacket(pbuff, 1, timeout); rlen = readPacket(pbuff, 1, timeout);
if (rlen != 1) return 0; if (rlen != 1)
return 0;
encodedByte = pbuff[0]; // save the last read val encodedByte = pbuff[0]; // save the last read val
pbuff++; // get ready for reading the next byte pbuff++; // get ready for reading the next byte
uint32_t intermediate = encodedByte & 0x7F; uint32_t intermediate = encodedByte & 0x7F;
intermediate *= multiplier; intermediate *= multiplier;
value += intermediate; value += intermediate;
multiplier *= 128; multiplier *= 128;
if (multiplier > (128UL*128UL*128UL)) { if (multiplier > (128UL * 128UL * 128UL)) {
DEBUG_PRINT(F("Malformed packet len\n")); DEBUG_PRINT(F("Malformed packet len\n"));
return 0; return 0;
} }
} while (encodedByte & 0x80); } while (encodedByte & 0x80);
DEBUG_PRINT(F("Packet Length:\t")); DEBUG_PRINTLN(value); DEBUG_PRINT(F("Packet Length:\t"));
DEBUG_PRINTLN(value);
if (value > (maxsize - (pbuff-buffer) - 1)) { if (value > (maxsize - (pbuff - buffer) - 1)) {
DEBUG_PRINTLN(F("Packet too big for buffer")); DEBUG_PRINTLN(F("Packet too big for buffer"));
rlen = readPacket(pbuff, (maxsize - (pbuff-buffer) - 1), timeout); rlen = readPacket(pbuff, (maxsize - (pbuff - buffer) - 1), timeout);
} else { } else {
rlen = readPacket(pbuff, value, timeout); rlen = readPacket(pbuff, value, timeout);
} }
//DEBUG_PRINT(F("Remaining packet:\t")); DEBUG_PRINTBUFFER(pbuff, rlen); // DEBUG_PRINT(F("Remaining packet:\t")); DEBUG_PRINTBUFFER(pbuff, rlen);
return ((pbuff - buffer)+rlen); return ((pbuff - buffer) + rlen);
} }
const __FlashStringHelper* Adafruit_MQTT::connectErrorString(int8_t code) { const __FlashStringHelper *Adafruit_MQTT::connectErrorString(int8_t code) {
switch (code) { switch (code) {
case 1: return F("The Server does not support the level of the MQTT protocol requested"); case 1:
case 2: return F("The Client identifier is correct UTF-8 but not allowed by the Server"); return F(
case 3: return F("The MQTT service is unavailable"); "The Server does not support the level of the MQTT protocol requested");
case 4: return F("The data in the user name or password is malformed"); case 2:
case 5: return F("Not authorized to connect"); return F(
case 6: return F("Exceeded reconnect rate limit. Please try again later."); "The Client identifier is correct UTF-8 but not allowed by the Server");
case 7: return F("You have been banned from connecting. Please contact the MQTT server administrator for more details."); case 3:
case -1: return F("Connection failed"); return F("The MQTT service is unavailable");
case -2: return F("Failed to subscribe"); case 4:
default: return F("Unknown error"); return F("The data in the user name or password is malformed");
case 5:
return F("Not authorized to connect");
case 6:
return F("Exceeded reconnect rate limit. Please try again later.");
case 7:
return F("You have been banned from connecting. Please contact the MQTT "
"server administrator for more details.");
case -1:
return F("Connection failed");
case -2:
return F("Failed to subscribe");
default:
return F("Unknown error");
} }
} }
@ -289,19 +304,18 @@ bool Adafruit_MQTT::disconnect() {
// Construct and send disconnect packet. // Construct and send disconnect packet.
uint8_t len = disconnectPacket(buffer); uint8_t len = disconnectPacket(buffer);
if (! sendPacket(buffer, len)) if (!sendPacket(buffer, len))
DEBUG_PRINTLN(F("Unable to send disconnect packet")); DEBUG_PRINTLN(F("Unable to send disconnect packet"));
return disconnectServer(); return disconnectServer();
} }
bool Adafruit_MQTT::publish(const char *topic, const char *data, uint8_t qos) { bool Adafruit_MQTT::publish(const char *topic, const char *data, uint8_t qos) {
return publish(topic, (uint8_t*)(data), strlen(data), qos); return publish(topic, (uint8_t *)(data), strlen(data), qos);
} }
bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen, uint8_t qos) { bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen,
uint8_t qos) {
// Construct and send publish packet. // Construct and send publish packet.
uint16_t len = publishPacket(buffer, topic, data, bLen, qos); uint16_t len = publishPacket(buffer, topic, data, bLen, qos);
if (!sendPacket(buffer, len)) if (!sendPacket(buffer, len))
@ -320,7 +334,8 @@ bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen, uin
packnum <<= 8; packnum <<= 8;
packnum |= buffer[3]; packnum |= buffer[3];
// we increment the packet_id_counter right after publishing so inc here too to match // we increment the packet_id_counter right after publishing so inc here too
// to match
packnum++; packnum++;
if (packnum != packet_id_counter) if (packnum != packet_id_counter)
return false; return false;
@ -329,7 +344,8 @@ bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen, uin
return true; return true;
} }
bool Adafruit_MQTT::will(const char *topic, const char *payload, uint8_t qos, uint8_t retain) { bool Adafruit_MQTT::will(const char *topic, const char *payload, uint8_t qos,
uint8_t retain) {
if (connected()) { if (connected()) {
DEBUG_PRINT(F("Will defined after connect")); DEBUG_PRINT(F("Will defined after connect"));
@ -342,22 +358,22 @@ bool Adafruit_MQTT::will(const char *topic, const char *payload, uint8_t qos, ui
will_retain = retain; will_retain = retain;
return true; return true;
} }
bool Adafruit_MQTT::subscribe(Adafruit_MQTT_Subscribe *sub) { bool Adafruit_MQTT::subscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i; uint8_t i;
// see if we are already subscribed // see if we are already subscribed
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == sub) { if (subscriptions[i] == sub) {
DEBUG_PRINTLN(F("Already subscribed")); DEBUG_PRINTLN(F("Already subscribed"));
return true; return true;
} }
} }
if (i==MAXSUBSCRIPTIONS) { // add to subscriptionlist if (i == MAXSUBSCRIPTIONS) { // add to subscriptionlist
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == 0) { if (subscriptions[i] == 0) {
DEBUG_PRINT(F("Added sub ")); DEBUG_PRINTLN(i); DEBUG_PRINT(F("Added sub "));
DEBUG_PRINTLN(i);
subscriptions[i] = sub; subscriptions[i] = sub;
return true; return true;
} }
@ -372,22 +388,23 @@ bool Adafruit_MQTT::unsubscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i; uint8_t i;
// see if we are already subscribed // see if we are already subscribed
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == sub) { if (subscriptions[i] == sub) {
DEBUG_PRINTLN(F("Found matching subscription and attempting to unsubscribe.")); DEBUG_PRINTLN(
F("Found matching subscription and attempting to unsubscribe."));
// Construct and send unsubscribe packet. // Construct and send unsubscribe packet.
uint8_t len = unsubscribePacket(buffer, subscriptions[i]->topic); uint8_t len = unsubscribePacket(buffer, subscriptions[i]->topic);
// sending unsubscribe failed // sending unsubscribe failed
if (! sendPacket(buffer, len)) if (!sendPacket(buffer, len))
return false; return false;
// if QoS for this subscription is 1 or 2, we need // if QoS for this subscription is 1 or 2, we need
// to wait for the unsuback to confirm unsubscription // to wait for the unsuback to confirm unsubscription
if(subscriptions[i]->qos > 0 && MQTT_PROTOCOL_LEVEL > 3) { if (subscriptions[i]->qos > 0 && MQTT_PROTOCOL_LEVEL > 3) {
// wait for UNSUBACK // wait for UNSUBACK
len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS); len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS);
@ -402,12 +419,10 @@ bool Adafruit_MQTT::unsubscribe(Adafruit_MQTT_Subscribe *sub) {
subscriptions[i] = 0; subscriptions[i] = 0;
return true; return true;
} }
} }
// subscription not found, so we are unsubscribed // subscription not found, so we are unsubscribed
return true; return true;
} }
void Adafruit_MQTT::processPackets(int16_t timeout) { void Adafruit_MQTT::processPackets(int16_t timeout) {
@ -417,30 +432,32 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
while (elapsed < (uint32_t)timeout) { while (elapsed < (uint32_t)timeout) {
Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed); Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed);
if (sub) { if (sub) {
//Serial.println("**** sub packet received"); // Serial.println("**** sub packet received");
if (sub->callback_uint32t != NULL) { if (sub->callback_uint32t != NULL) {
// huh lets do the callback in integer mode // huh lets do the callback in integer mode
uint32_t data = 0; uint32_t data = 0;
data = atoi((char *)sub->lastread); data = atoi((char *)sub->lastread);
//Serial.print("*** calling int callback with : "); Serial.println(data); // Serial.print("*** calling int callback with : ");
// Serial.println(data);
sub->callback_uint32t(data); sub->callback_uint32t(data);
} } else if (sub->callback_double != NULL) {
else if (sub->callback_double != NULL) {
// huh lets do the callback in doublefloat mode // huh lets do the callback in doublefloat mode
double data = 0; double data = 0;
data = atof((char *)sub->lastread); data = atof((char *)sub->lastread);
//Serial.print("*** calling double callback with : "); Serial.println(data); // Serial.print("*** calling double callback with : ");
// Serial.println(data);
sub->callback_double(data); sub->callback_double(data);
} } else if (sub->callback_buffer != NULL) {
else if (sub->callback_buffer != NULL) {
// huh lets do the callback in buffer mode // huh lets do the callback in buffer mode
//Serial.print("*** calling buffer callback with : "); Serial.println((char *)sub->lastread); // Serial.print("*** calling buffer callback with : ");
// Serial.println((char *)sub->lastread);
sub->callback_buffer((char *)sub->lastread, sub->datalen); sub->callback_buffer((char *)sub->lastread, sub->datalen);
} } else if (sub->callback_io != NULL) {
else if (sub->callback_io != NULL) {
// huh lets do the callback in io mode // huh lets do the callback in io mode
//Serial.print("*** calling io instance callback with : "); Serial.println((char *)sub->lastread); // Serial.print("*** calling io instance callback with : ");
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread, sub->datalen); // Serial.println((char *)sub->lastread);
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread,
sub->datalen);
} }
} }
@ -457,21 +474,26 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
uint16_t i, topiclen, datalen; uint16_t i, topiclen, datalen;
// Check if data is available to read. // Check if data is available to read.
uint16_t len = readFullPacket(buffer, MAXBUFFERSIZE, timeout); // return one full packet uint16_t len =
readFullPacket(buffer, MAXBUFFERSIZE, timeout); // return one full packet
if (!len) if (!len)
return NULL; // No data available, just quit. return NULL; // No data available, just quit.
DEBUG_PRINT("Packet len: "); DEBUG_PRINTLN(len); DEBUG_PRINT("Packet len: ");
DEBUG_PRINTLN(len);
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
if (len<3) return NULL; if (len < 3)
if ((buffer[0] & 0xF0) != (MQTT_CTRL_PUBLISH) << 4) return NULL; return NULL;
if ((buffer[0] & 0xF0) != (MQTT_CTRL_PUBLISH) << 4)
return NULL;
// Parse out length of packet. // Parse out length of packet.
topiclen = buffer[3]; topiclen = buffer[3];
DEBUG_PRINT(F("Looking for subscription len ")); DEBUG_PRINTLN(topiclen); DEBUG_PRINT(F("Looking for subscription len "));
DEBUG_PRINTLN(topiclen);
// Find subscription associated with this packet. // Find subscription associated with this packet.
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i]) { if (subscriptions[i]) {
// Skip this subscription if its name length isn't the same as the // Skip this subscription if its name length isn't the same as the
// received topic name. // received topic name.
@ -479,22 +501,25 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
continue; continue;
// Stop if the subscription topic matches the received topic. Be careful // Stop if the subscription topic matches the received topic. Be careful
// to make comparison case insensitive. // to make comparison case insensitive.
if (strncasecmp((char*)buffer+4, subscriptions[i]->topic, topiclen) == 0) { if (strncasecmp((char *)buffer + 4, subscriptions[i]->topic, topiclen) ==
DEBUG_PRINT(F("Found sub #")); DEBUG_PRINTLN(i); 0) {
DEBUG_PRINT(F("Found sub #"));
DEBUG_PRINTLN(i);
break; break;
} }
} }
} }
if (i==MAXSUBSCRIPTIONS) return NULL; // matching sub not found ??? if (i == MAXSUBSCRIPTIONS)
return NULL; // matching sub not found ???
uint8_t packet_id_len = 0; uint8_t packet_id_len = 0;
uint16_t packetid = 0; uint16_t packetid = 0;
// Check if it is QoS 1, TODO: we dont support QoS 2 // Check if it is QoS 1, TODO: we dont support QoS 2
if ((buffer[0] & 0x6) == 0x2) { if ((buffer[0] & 0x6) == 0x2) {
packet_id_len = 2; packet_id_len = 2;
packetid = buffer[topiclen+4]; packetid = buffer[topiclen + 4];
packetid <<= 8; packetid <<= 8;
packetid |= buffer[topiclen+5]; packetid |= buffer[topiclen + 5];
} }
// zero out the old data // zero out the old data
@ -502,15 +527,18 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
datalen = len - topiclen - packet_id_len - 4; datalen = len - topiclen - packet_id_len - 4;
if (datalen > SUBSCRIPTIONDATALEN) { if (datalen > SUBSCRIPTIONDATALEN) {
datalen = SUBSCRIPTIONDATALEN-1; // cut it off datalen = SUBSCRIPTIONDATALEN - 1; // cut it off
} }
// extract out just the data, into the subscription object itself // extract out just the data, into the subscription object itself
memmove(subscriptions[i]->lastread, buffer+4+topiclen+packet_id_len, datalen); memmove(subscriptions[i]->lastread, buffer + 4 + topiclen + packet_id_len,
datalen);
subscriptions[i]->datalen = datalen; subscriptions[i]->datalen = datalen;
DEBUG_PRINT(F("Data len: ")); DEBUG_PRINTLN(datalen); DEBUG_PRINT(F("Data len: "));
DEBUG_PRINT(F("Data: ")); DEBUG_PRINTLN((char *)subscriptions[i]->lastread); DEBUG_PRINTLN(datalen);
DEBUG_PRINT(F("Data: "));
DEBUG_PRINTLN((char *)subscriptions[i]->lastread);
if ((MQTT_PROTOCOL_LEVEL > 3) &&(buffer[0] & 0x6) == 0x2) { if ((MQTT_PROTOCOL_LEVEL > 3) && (buffer[0] & 0x6) == 0x2) {
uint8_t ackpacket[4]; uint8_t ackpacket[4];
// Construct and send puback packet. // Construct and send puback packet.
@ -526,11 +554,12 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
void Adafruit_MQTT::flushIncoming(uint16_t timeout) { void Adafruit_MQTT::flushIncoming(uint16_t timeout) {
// flush input! // flush input!
DEBUG_PRINTLN(F("Flushing input buffer")); DEBUG_PRINTLN(F("Flushing input buffer"));
while (readPacket(buffer, MAXBUFFERSIZE, timeout)); while (readPacket(buffer, MAXBUFFERSIZE, timeout))
;
} }
bool Adafruit_MQTT::ping(uint8_t num) { bool Adafruit_MQTT::ping(uint8_t num) {
//flushIncoming(100); // flushIncoming(100);
while (num--) { while (num--) {
// Construct and send ping packet. // Construct and send ping packet.
@ -560,7 +589,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
// fixed header, connection messsage no flags // fixed header, connection messsage no flags
p[0] = (MQTT_CTRL_CONNECT << 4) | 0x0; p[0] = (MQTT_CTRL_CONNECT << 4) | 0x0;
p+=2; p += 2;
// fill in packet[1] last // fill in packet[1] last
#if MQTT_PROTOCOL_LEVEL == 3 #if MQTT_PROTOCOL_LEVEL == 3
@ -568,7 +597,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
#elif MQTT_PROTOCOL_LEVEL == 4 #elif MQTT_PROTOCOL_LEVEL == 4
p = stringprint(p, "MQTT"); p = stringprint(p, "MQTT");
#else #else
#error "MQTT level not supported" #error "MQTT level not supported"
#endif #endif
p[0] = MQTT_PROTOCOL_LEVEL; p[0] = MQTT_PROTOCOL_LEVEL;
@ -582,14 +611,13 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
p[0] |= MQTT_CONN_WILLFLAG; p[0] |= MQTT_CONN_WILLFLAG;
if(will_qos == 1) if (will_qos == 1)
p[0] |= MQTT_CONN_WILLQOS_1; p[0] |= MQTT_CONN_WILLQOS_1;
else if(will_qos == 2) else if (will_qos == 2)
p[0] |= MQTT_CONN_WILLQOS_2; p[0] |= MQTT_CONN_WILLQOS_2;
if(will_retain == 1) if (will_retain == 1)
p[0] |= MQTT_CONN_WILLRETAIN; p[0] |= MQTT_CONN_WILLRETAIN;
} }
if (pgm_read_byte(username) != 0) if (pgm_read_byte(username) != 0)
@ -603,7 +631,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
p[0] = MQTT_CONN_KEEPALIVE & 0xFF; p[0] = MQTT_CONN_KEEPALIVE & 0xFF;
p++; p++;
if(MQTT_PROTOCOL_LEVEL == 3) { if (MQTT_PROTOCOL_LEVEL == 3) {
p = stringprint(p, clientid, 23); // Limit client ID to first 23 characters. p = stringprint(p, clientid, 23); // Limit client ID to first 23 characters.
} else { } else {
if (pgm_read_byte(clientid) != 0) { if (pgm_read_byte(clientid) != 0) {
@ -631,23 +659,24 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT connect packet:")); DEBUG_PRINTLN(F("MQTT connect packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
} }
// as per
// as per http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718040 // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718040
uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic, uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
uint8_t *data, uint16_t bLen, uint8_t qos) { uint8_t *data, uint16_t bLen,
uint8_t qos) {
uint8_t *p = packet; uint8_t *p = packet;
uint16_t len=0; uint16_t len = 0;
// calc length of non-header data // calc length of non-header data
len += 2; // two bytes to set the topic size len += 2; // two bytes to set the topic size
len += strlen(topic); // topic length len += strlen(topic); // topic length
if(qos > 0) { if (qos > 0) {
len += 2; // qos packet id len += 2; // qos packet id
} }
len += bLen; // payload length len += bLen; // payload length
@ -661,28 +690,28 @@ uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
uint8_t encodedByte = len % 128; uint8_t encodedByte = len % 128;
len /= 128; len /= 128;
// if there are more data to encode, set the top bit of this byte // if there are more data to encode, set the top bit of this byte
if ( len > 0 ) { if (len > 0) {
encodedByte |= 0x80; encodedByte |= 0x80;
} }
p[0] = encodedByte; p[0] = encodedByte;
p++; p++;
} while ( len > 0 ); } while (len > 0);
// topic comes before packet identifier // topic comes before packet identifier
p = stringprint(p, topic); p = stringprint(p, topic);
// add packet identifier. used for checking PUBACK in QOS > 0 // add packet identifier. used for checking PUBACK in QOS > 0
if(qos > 0) { if (qos > 0) {
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
} }
memmove(p, data, bLen); memmove(p, data, bLen);
p+= bLen; p += bLen;
len = p - packet; len = p - packet;
DEBUG_PRINTLN(F("MQTT publish packet:")); DEBUG_PRINTLN(F("MQTT publish packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
@ -696,12 +725,12 @@ uint8_t Adafruit_MQTT::subscribePacket(uint8_t *packet, const char *topic,
p[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1; p[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1;
// fill in packet[1] last // fill in packet[1] last
p+=2; p += 2;
// packet identifier. used for checking SUBACK // packet identifier. used for checking SUBACK
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
@ -712,14 +741,12 @@ uint8_t Adafruit_MQTT::subscribePacket(uint8_t *packet, const char *topic,
p++; p++;
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT subscription packet:")); DEBUG_PRINTLN(F("MQTT subscription packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
} }
uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) { uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
uint8_t *p = packet; uint8_t *p = packet;
@ -727,12 +754,12 @@ uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
p[0] = MQTT_CTRL_UNSUBSCRIBE << 4 | 0x1; p[0] = MQTT_CTRL_UNSUBSCRIBE << 4 | 0x1;
// fill in packet[1] last // fill in packet[1] last
p+=2; p += 2;
// packet identifier. used for checking UNSUBACK // packet identifier. used for checking UNSUBACK
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
@ -740,11 +767,10 @@ uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
p = stringprint(p, topic); p = stringprint(p, topic);
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT unsubscription packet:")); DEBUG_PRINTLN(F("MQTT unsubscription packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
} }
uint8_t Adafruit_MQTT::pingPacket(uint8_t *packet) { uint8_t Adafruit_MQTT::pingPacket(uint8_t *packet) {
@ -794,7 +820,8 @@ bool Adafruit_MQTT_Publish::publish(uint32_t i) {
} }
bool Adafruit_MQTT_Publish::publish(double f, uint8_t precision) { bool Adafruit_MQTT_Publish::publish(double f, uint8_t precision) {
char payload[41]; // Need to technically hold float max, 39 digits and minus sign. char payload[41]; // Need to technically hold float max, 39 digits and minus
// sign.
dtostrf(f, 0, precision, payload); dtostrf(f, 0, precision, payload);
return mqtt->publish(topic, payload, qos); return mqtt->publish(topic, payload, qos);
} }
@ -803,13 +830,12 @@ bool Adafruit_MQTT_Publish::publish(const char *payload) {
return mqtt->publish(topic, payload, qos); return mqtt->publish(topic, payload, qos);
} }
//publish buffer of arbitrary length // publish buffer of arbitrary length
bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen) { bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen) {
return mqtt->publish(topic, payload, bLen, qos); return mqtt->publish(topic, payload, bLen, qos);
} }
// Adafruit_MQTT_Subscribe Definition ////////////////////////////////////////// // Adafruit_MQTT_Subscribe Definition //////////////////////////////////////////
Adafruit_MQTT_Subscribe::Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver, Adafruit_MQTT_Subscribe::Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver,
@ -837,9 +863,10 @@ void Adafruit_MQTT_Subscribe::setCallback(SubscribeCallbackBufferType cb) {
callback_buffer = cb; callback_buffer = cb;
} }
void Adafruit_MQTT_Subscribe::setCallback(AdafruitIO_MQTT *io, SubscribeCallbackIOType cb) { void Adafruit_MQTT_Subscribe::setCallback(AdafruitIO_MQTT *io,
SubscribeCallbackIOType cb) {
callback_io = cb; callback_io = cb;
io_mqtt= io; io_mqtt = io;
} }
void Adafruit_MQTT_Subscribe::removeCallback(void) { void Adafruit_MQTT_Subscribe::removeCallback(void) {

121
Adafruit_MQTT.h
View File

@ -9,8 +9,8 @@
// copies of the Software, and to permit persons to whom the Software is // copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions: // furnished to do so, subject to the following conditions:
// //
// The above copyright notice and this permission notice shall be included in all // The above copyright notice and this permission notice shall be included in
// copies or substantial portions of the Software. // all copies or substantial portions of the Software.
// //
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
@ -44,23 +44,35 @@
// Define actual debug output functions when necessary. // Define actual debug output functions when necessary.
#ifdef MQTT_DEBUG #ifdef MQTT_DEBUG
#define DEBUG_PRINT(...) { DEBUG_PRINTER.print(__VA_ARGS__); } #define DEBUG_PRINT(...) \
#define DEBUG_PRINTLN(...) { DEBUG_PRINTER.println(__VA_ARGS__); } { DEBUG_PRINTER.print(__VA_ARGS__); }
#define DEBUG_PRINTBUFFER(buffer, len) { printBuffer(buffer, len); } #define DEBUG_PRINTLN(...) \
{ DEBUG_PRINTER.println(__VA_ARGS__); }
#define DEBUG_PRINTBUFFER(buffer, len) \
{ printBuffer(buffer, len); }
#else #else
#define DEBUG_PRINT(...) {} #define DEBUG_PRINT(...) \
#define DEBUG_PRINTLN(...) {} {}
#define DEBUG_PRINTBUFFER(buffer, len) {} #define DEBUG_PRINTLN(...) \
{}
#define DEBUG_PRINTBUFFER(buffer, len) \
{}
#endif #endif
#ifdef MQTT_ERROR #ifdef MQTT_ERROR
#define ERROR_PRINT(...) { DEBUG_PRINTER.print(__VA_ARGS__); } #define ERROR_PRINT(...) \
#define ERROR_PRINTLN(...) { DEBUG_PRINTER.println(__VA_ARGS__); } { DEBUG_PRINTER.print(__VA_ARGS__); }
#define ERROR_PRINTBUFFER(buffer, len) { printBuffer(buffer, len); } #define ERROR_PRINTLN(...) \
{ DEBUG_PRINTER.println(__VA_ARGS__); }
#define ERROR_PRINTBUFFER(buffer, len) \
{ printBuffer(buffer, len); }
#else #else
#define ERROR_PRINT(...) {} #define ERROR_PRINT(...) \
#define ERROR_PRINTLN(...) {} {}
#define ERROR_PRINTBUFFER(buffer, len) {} #define ERROR_PRINTLN(...) \
{}
#define ERROR_PRINTBUFFER(buffer, len) \
{}
#endif #endif
// Use 3 (MQTT 3.0) or 4 (MQTT 3.1.1) // Use 3 (MQTT 3.0) or 4 (MQTT 3.1.1)
@ -107,40 +119,36 @@
// how much data we save in a subscription object // how much data we save in a subscription object
// and how many subscriptions we want to be able to track. // and how many subscriptions we want to be able to track.
#if defined (__AVR_ATmega32U4__) || defined(__AVR_ATmega328P__) #if defined(__AVR_ATmega32U4__) || defined(__AVR_ATmega328P__)
#define MAXSUBSCRIPTIONS 5 #define MAXSUBSCRIPTIONS 5
#define SUBSCRIPTIONDATALEN 20 #define SUBSCRIPTIONDATALEN 20
#else #else
#define MAXSUBSCRIPTIONS 15 #define MAXSUBSCRIPTIONS 15
#define SUBSCRIPTIONDATALEN 100 #define SUBSCRIPTIONDATALEN 100
#endif #endif
class AdafruitIO_MQTT; // forward decl class AdafruitIO_MQTT; // forward decl
//Function pointer that returns an int // Function pointer that returns an int
typedef void (*SubscribeCallbackUInt32Type)(uint32_t); typedef void (*SubscribeCallbackUInt32Type)(uint32_t);
// returns a double // returns a double
typedef void (*SubscribeCallbackDoubleType)(double); typedef void (*SubscribeCallbackDoubleType)(double);
// returns a chunk of raw data // returns a chunk of raw data
typedef void (*SubscribeCallbackBufferType)(char *str, uint16_t len); typedef void (*SubscribeCallbackBufferType)(char *str, uint16_t len);
// returns an io data wrapper instance // returns an io data wrapper instance
typedef void (AdafruitIO_MQTT::*SubscribeCallbackIOType)(char *str, uint16_t len); typedef void (AdafruitIO_MQTT::*SubscribeCallbackIOType)(char *str,
uint16_t len);
extern void printBuffer(uint8_t *buffer, uint16_t len); extern void printBuffer(uint8_t *buffer, uint16_t len);
class Adafruit_MQTT_Subscribe; // forward decl class Adafruit_MQTT_Subscribe; // forward decl
class Adafruit_MQTT { class Adafruit_MQTT {
public: public:
Adafruit_MQTT(const char *server, Adafruit_MQTT(const char *server, uint16_t port, const char *cid,
uint16_t port, const char *user, const char *pass);
const char *cid,
const char *user,
const char *pass);
Adafruit_MQTT(const char *server, Adafruit_MQTT(const char *server, uint16_t port, const char *user = "",
uint16_t port,
const char *user = "",
const char *pass = ""); const char *pass = "");
virtual ~Adafruit_MQTT() {} virtual ~Adafruit_MQTT() {}
@ -162,7 +170,7 @@ class Adafruit_MQTT {
// connect(). This returns a __FlashStringHelper*, which points to a // connect(). This returns a __FlashStringHelper*, which points to a
// string stored in flash, but can be directly passed to e.g. // string stored in flash, but can be directly passed to e.g.
// Serial.println without any further processing. // Serial.println without any further processing.
const __FlashStringHelper* connectErrorString(int8_t code); const __FlashStringHelper *connectErrorString(int8_t code);
// Sends MQTT disconnect packet and calls disconnectServer() // Sends MQTT disconnect packet and calls disconnectServer()
bool disconnect(); bool disconnect();
@ -173,12 +181,14 @@ class Adafruit_MQTT {
// Set MQTT last will topic, payload, QOS, and retain. This needs // Set MQTT last will topic, payload, QOS, and retain. This needs
// to be called before connect() because it is sent as part of the // to be called before connect() because it is sent as part of the
// connect control packet. // connect control packet.
bool will(const char *topic, const char *payload, uint8_t qos = 0, uint8_t retain = 0); bool will(const char *topic, const char *payload, uint8_t qos = 0,
uint8_t retain = 0);
// Publish a message to a topic using the specified QoS level. Returns true // Publish a message to a topic using the specified QoS level. Returns true
// if the message was published, false otherwise. // if the message was published, false otherwise.
bool publish(const char *topic, const char *payload, uint8_t qos = 0); bool publish(const char *topic, const char *payload, uint8_t qos = 0);
bool publish(const char *topic, uint8_t *payload, uint16_t bLen, uint8_t qos = 0); bool publish(const char *topic, uint8_t *payload, uint16_t bLen,
uint8_t qos = 0);
// Add a subscription to receive messages for a topic. Returns true if the // Add a subscription to receive messages for a topic. Returns true if the
// subscription could be added or was already present, false otherwise. // subscription could be added or was already present, false otherwise.
@ -190,23 +200,25 @@ class Adafruit_MQTT {
bool unsubscribe(Adafruit_MQTT_Subscribe *sub); bool unsubscribe(Adafruit_MQTT_Subscribe *sub);
// Check if any subscriptions have new messages. Will return a reference to // Check if any subscriptions have new messages. Will return a reference to
// an Adafruit_MQTT_Subscribe object which has a new message. Should be called // an Adafruit_MQTT_Subscribe object which has a new message. Should be
// in the sketch's loop function to ensure new messages are recevied. Note // called in the sketch's loop function to ensure new messages are recevied.
// that subscribe should be called first for each topic that receives messages! // Note that subscribe should be called first for each topic that receives
Adafruit_MQTT_Subscribe *readSubscription(int16_t timeout=0); // messages!
Adafruit_MQTT_Subscribe *readSubscription(int16_t timeout = 0);
void processPackets(int16_t timeout); void processPackets(int16_t timeout);
// Ping the server to ensure the connection is still alive. // Ping the server to ensure the connection is still alive.
bool ping(uint8_t n = 1); bool ping(uint8_t n = 1);
protected: protected:
// Interface that subclasses need to implement: // Interface that subclasses need to implement:
// Connect to the server and return true if successful, false otherwise. // Connect to the server and return true if successful, false otherwise.
virtual bool connectServer() = 0; virtual bool connectServer() = 0;
// Disconnect from the MQTT server. Returns true if disconnected, false otherwise. // Disconnect from the MQTT server. Returns true if disconnected, false
// otherwise.
virtual bool disconnectServer() = 0; // Subclasses need to fill this in! virtual bool disconnectServer() = 0; // Subclasses need to fill this in!
// Send data to the server specified by the buffer and length of data. // Send data to the server specified by the buffer and length of data.
@ -215,12 +227,14 @@ class Adafruit_MQTT {
// Read MQTT packet from the server. Will read up to maxlen bytes and store // Read MQTT packet from the server. Will read up to maxlen bytes and store
// the data in the provided buffer. Waits up to the specified timeout (in // the data in the provided buffer. Waits up to the specified timeout (in
// milliseconds) for data to be available. // milliseconds) for data to be available.
virtual uint16_t readPacket(uint8_t *buffer, uint16_t maxlen, int16_t timeout) = 0; virtual uint16_t readPacket(uint8_t *buffer, uint16_t maxlen,
int16_t timeout) = 0;
// Read a full packet, keeping note of the correct length // Read a full packet, keeping note of the correct length
uint16_t readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16_t timeout); uint16_t readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16_t timeout);
// Properly process packets until you get to one you want // Properly process packets until you get to one you want
uint16_t processPacketsUntil(uint8_t *buffer, uint8_t waitforpackettype, uint16_t timeout); uint16_t processPacketsUntil(uint8_t *buffer, uint8_t waitforpackettype,
uint16_t timeout);
// Shared state that subclasses can use: // Shared state that subclasses can use:
const char *servername; const char *servername;
@ -235,7 +249,7 @@ class Adafruit_MQTT {
uint8_t buffer[MAXBUFFERSIZE]; // one buffer, used for all incoming/outgoing uint8_t buffer[MAXBUFFERSIZE]; // one buffer, used for all incoming/outgoing
uint16_t packet_id_counter; uint16_t packet_id_counter;
private: private:
Adafruit_MQTT_Subscribe *subscriptions[MAXSUBSCRIPTIONS]; Adafruit_MQTT_Subscribe *subscriptions[MAXSUBSCRIPTIONS];
void flushIncoming(uint16_t timeout); void flushIncoming(uint16_t timeout);
@ -243,26 +257,29 @@ class Adafruit_MQTT {
// Functions to generate MQTT packets. // Functions to generate MQTT packets.
uint8_t connectPacket(uint8_t *packet); uint8_t connectPacket(uint8_t *packet);
uint8_t disconnectPacket(uint8_t *packet); uint8_t disconnectPacket(uint8_t *packet);
uint16_t publishPacket(uint8_t *packet, const char *topic, uint8_t *payload, uint16_t bLen, uint8_t qos); uint16_t publishPacket(uint8_t *packet, const char *topic, uint8_t *payload,
uint16_t bLen, uint8_t qos);
uint8_t subscribePacket(uint8_t *packet, const char *topic, uint8_t qos); uint8_t subscribePacket(uint8_t *packet, const char *topic, uint8_t qos);
uint8_t unsubscribePacket(uint8_t *packet, const char *topic); uint8_t unsubscribePacket(uint8_t *packet, const char *topic);
uint8_t pingPacket(uint8_t *packet); uint8_t pingPacket(uint8_t *packet);
uint8_t pubackPacket(uint8_t *packet, uint16_t packetid); uint8_t pubackPacket(uint8_t *packet, uint16_t packetid);
}; };
class Adafruit_MQTT_Publish { class Adafruit_MQTT_Publish {
public: public:
Adafruit_MQTT_Publish(Adafruit_MQTT *mqttserver, const char *feed, uint8_t qos = 0); Adafruit_MQTT_Publish(Adafruit_MQTT *mqttserver, const char *feed,
uint8_t qos = 0);
bool publish(const char *s); bool publish(const char *s);
bool publish(double f, uint8_t precision=2); // Precision controls the minimum number of digits after decimal. bool publish(
double f,
uint8_t precision =
2); // Precision controls the minimum number of digits after decimal.
// This might be ignored and a higher precision value sent. // This might be ignored and a higher precision value sent.
bool publish(int32_t i); bool publish(int32_t i);
bool publish(uint32_t i); bool publish(uint32_t i);
bool publish(uint8_t *b, uint16_t bLen); bool publish(uint8_t *b, uint16_t bLen);
private: private:
Adafruit_MQTT *mqtt; Adafruit_MQTT *mqtt;
const char *topic; const char *topic;
@ -270,8 +287,9 @@ private:
}; };
class Adafruit_MQTT_Subscribe { class Adafruit_MQTT_Subscribe {
public: public:
Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver, const char *feedname, uint8_t q=0); Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver, const char *feedname,
uint8_t q = 0);
void setCallback(SubscribeCallbackUInt32Type callb); void setCallback(SubscribeCallbackUInt32Type callb);
void setCallback(SubscribeCallbackDoubleType callb); void setCallback(SubscribeCallbackDoubleType callb);
@ -294,9 +312,8 @@ class Adafruit_MQTT_Subscribe {
AdafruitIO_MQTT *io_mqtt; AdafruitIO_MQTT *io_mqtt;
private: private:
Adafruit_MQTT *mqtt; Adafruit_MQTT *mqtt;
}; };
#endif #endif

26
Adafruit_MQTT_Client.cpp
View File

@ -9,8 +9,8 @@
// copies of the Software, and to permit persons to whom the Software is // copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions: // furnished to do so, subject to the following conditions:
// //
// The above copyright notice and this permission notice shall be included in all // The above copyright notice and this permission notice shall be included in
// copies or substantial portions of the Software. // all copies or substantial portions of the Software.
// //
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
@ -21,15 +21,16 @@
// SOFTWARE. // SOFTWARE.
#include "Adafruit_MQTT_Client.h" #include "Adafruit_MQTT_Client.h"
bool Adafruit_MQTT_Client::connectServer() { bool Adafruit_MQTT_Client::connectServer() {
// Grab server name from flash and copy to buffer for name resolution. // Grab server name from flash and copy to buffer for name resolution.
memset(buffer, 0, sizeof(buffer)); memset(buffer, 0, sizeof(buffer));
strcpy((char *)buffer, servername); strcpy((char *)buffer, servername);
DEBUG_PRINT(F("Connecting to: ")); DEBUG_PRINTLN((char *)buffer); DEBUG_PRINT(F("Connecting to: "));
DEBUG_PRINTLN((char *)buffer);
// Connect and check for success (0 result). // Connect and check for success (0 result).
int r = client->connect((char *)buffer, portnum); int r = client->connect((char *)buffer, portnum);
DEBUG_PRINT(F("Connect result: ")); DEBUG_PRINTLN(r); DEBUG_PRINT(F("Connect result: "));
DEBUG_PRINTLN(r);
return r != 0; return r != 0;
} }
@ -49,19 +50,19 @@ bool Adafruit_MQTT_Client::connected() {
uint16_t Adafruit_MQTT_Client::readPacket(uint8_t *buffer, uint16_t maxlen, uint16_t Adafruit_MQTT_Client::readPacket(uint8_t *buffer, uint16_t maxlen,
int16_t timeout) { int16_t timeout) {
/* Read data until either the connection is closed, or the idle timeout is reached. */ /* Read data until either the connection is closed, or the idle timeout is
* reached. */
uint16_t len = 0; uint16_t len = 0;
int16_t t = timeout; int16_t t = timeout;
while (client->connected() && (timeout >= 0)) { while (client->connected() && (timeout >= 0)) {
//DEBUG_PRINT('.'); // DEBUG_PRINT('.');
while (client->available()) { while (client->available()) {
//DEBUG_PRINT('!'); // DEBUG_PRINT('!');
char c = client->read(); char c = client->read();
timeout = t; // reset the timeout timeout = t; // reset the timeout
buffer[len] = c; buffer[len] = c;
//DEBUG_PRINTLN((uint8_t)c, HEX); // DEBUG_PRINTLN((uint8_t)c, HEX);
len++; len++;
if (maxlen == 0) { // handle zero-length packets if (maxlen == 0) { // handle zero-length packets
@ -88,9 +89,10 @@ bool Adafruit_MQTT_Client::sendPacket(uint8_t *buffer, uint16_t len) {
// send 250 bytes at most at a time, can adjust this later based on Client // send 250 bytes at most at a time, can adjust this later based on Client
uint16_t sendlen = len > 250 ? 250 : len; uint16_t sendlen = len > 250 ? 250 : len;
//Serial.print("Sending: "); Serial.println(sendlen); // Serial.print("Sending: "); Serial.println(sendlen);
ret = client->write(buffer, sendlen); ret = client->write(buffer, sendlen);
DEBUG_PRINT(F("Client sendPacket returned: ")); DEBUG_PRINTLN(ret); DEBUG_PRINT(F("Client sendPacket returned: "));
DEBUG_PRINTLN(ret);
len -= ret; len -= ret;
if (ret != sendlen) { if (ret != sendlen) {

27
Adafruit_MQTT_Client.h
View File

@ -9,8 +9,8 @@
// copies of the Software, and to permit persons to whom the Software is // copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions: // furnished to do so, subject to the following conditions:
// //
// The above copyright notice and this permission notice shall be included in all // The above copyright notice and this permission notice shall be included in
// copies or substantial portions of the Software. // all copies or substantial portions of the Software.
// //
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
@ -22,30 +22,24 @@
#ifndef _ADAFRUIT_MQTT_CLIENT_H_ #ifndef _ADAFRUIT_MQTT_CLIENT_H_
#define _ADAFRUIT_MQTT_CLIENT_H_ #define _ADAFRUIT_MQTT_CLIENT_H_
#include "Client.h"
#include "Adafruit_MQTT.h" #include "Adafruit_MQTT.h"
#include "Client.h"
// How long to delay waiting for new data to be available in readPacket. // How long to delay waiting for new data to be available in readPacket.
#define MQTT_CLIENT_READINTERVAL_MS 10 #define MQTT_CLIENT_READINTERVAL_MS 10
// MQTT client implementation for a generic Arduino Client interface. Can work // MQTT client implementation for a generic Arduino Client interface. Can work
// with almost all Arduino network hardware like ethernet shield, wifi shield, // with almost all Arduino network hardware like ethernet shield, wifi shield,
// and even other platforms like ESP8266. // and even other platforms like ESP8266.
class Adafruit_MQTT_Client : public Adafruit_MQTT { class Adafruit_MQTT_Client : public Adafruit_MQTT {
public: public:
Adafruit_MQTT_Client(Client *client, const char *server, uint16_t port, Adafruit_MQTT_Client(Client *client, const char *server, uint16_t port,
const char *cid, const char *user, const char *pass): const char *cid, const char *user, const char *pass)
Adafruit_MQTT(server, port, cid, user, pass), : Adafruit_MQTT(server, port, cid, user, pass), client(client) {}
client(client)
{}
Adafruit_MQTT_Client(Client *client, const char *server, uint16_t port, Adafruit_MQTT_Client(Client *client, const char *server, uint16_t port,
const char *user="", const char *pass=""): const char *user = "", const char *pass = "")
Adafruit_MQTT(server, port, user, pass), : Adafruit_MQTT(server, port, user, pass), client(client) {}
client(client)
{}
bool connectServer(); bool connectServer();
bool disconnectServer(); bool disconnectServer();
@ -53,9 +47,8 @@ class Adafruit_MQTT_Client : public Adafruit_MQTT {
uint16_t readPacket(uint8_t *buffer, uint16_t maxlen, int16_t timeout); uint16_t readPacket(uint8_t *buffer, uint16_t maxlen, int16_t timeout);
bool sendPacket(uint8_t *buffer, uint16_t len); bool sendPacket(uint8_t *buffer, uint16_t len);
private: private:
Client* client; Client *client;
}; };
#endif #endif

52
Adafruit_MQTT_FONA.h
View File

@ -9,8 +9,8 @@
// copies of the Software, and to permit persons to whom the Software is // copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions: // furnished to do so, subject to the following conditions:
// //
// The above copyright notice and this permission notice shall be included in all // The above copyright notice and this permission notice shall be included in
// copies or substantial portions of the Software. // all copies or substantial portions of the Software.
// //
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
@ -22,30 +22,25 @@
#ifndef _ADAFRUIT_MQTT_FONA_H_ #ifndef _ADAFRUIT_MQTT_FONA_H_
#define _ADAFRUIT_MQTT_FONA_H_ #define _ADAFRUIT_MQTT_FONA_H_
#include <Adafruit_FONA.h>
#include "Adafruit_MQTT.h" #include "Adafruit_MQTT.h"
#include <Adafruit_FONA.h>
#define MQTT_FONA_INTERAVAILDELAY 100 #define MQTT_FONA_INTERAVAILDELAY 100
#define MQTT_FONA_QUERYDELAY 500 #define MQTT_FONA_QUERYDELAY 500
// FONA-specific version of the Adafruit_MQTT class. // FONA-specific version of the Adafruit_MQTT class.
// Note that this is defined as a header-only class to prevent issues with using // Note that this is defined as a header-only class to prevent issues with using
// the library on non-FONA platforms (since Arduino will include all .cpp files // the library on non-FONA platforms (since Arduino will include all .cpp files
// in the compilation of the library). // in the compilation of the library).
class Adafruit_MQTT_FONA : public Adafruit_MQTT { class Adafruit_MQTT_FONA : public Adafruit_MQTT {
public: public:
Adafruit_MQTT_FONA(Adafruit_FONA *f, const char *server, uint16_t port, Adafruit_MQTT_FONA(Adafruit_FONA *f, const char *server, uint16_t port,
const char *cid, const char *user, const char *pass): const char *cid, const char *user, const char *pass)
Adafruit_MQTT(server, port, cid, user, pass), : Adafruit_MQTT(server, port, cid, user, pass), fona(f) {}
fona(f)
{}
Adafruit_MQTT_FONA(Adafruit_FONA *f, const char *server, uint16_t port, Adafruit_MQTT_FONA(Adafruit_FONA *f, const char *server, uint16_t port,
const char *user="", const char *pass=""): const char *user = "", const char *pass = "")
Adafruit_MQTT(server, port, user, pass), : Adafruit_MQTT(server, port, user, pass), fona(f) {}
fona(f)
{}
bool connectServer() { bool connectServer() {
char server[40]; char server[40];
@ -59,9 +54,7 @@ class Adafruit_MQTT_FONA : public Adafruit_MQTT {
return fona->TCPconnect(server, portnum); return fona->TCPconnect(server, portnum);
} }
bool disconnectServer() { bool disconnectServer() { return fona->TCPclose(); }
return fona->TCPclose();
}
bool connected() { bool connected() {
// Return true if connected, false if not connected. // Return true if connected, false if not connected.
@ -72,33 +65,36 @@ class Adafruit_MQTT_FONA : public Adafruit_MQTT {
uint8_t *buffp = buffer; uint8_t *buffp = buffer;
DEBUG_PRINTLN(F("Reading data..")); DEBUG_PRINTLN(F("Reading data.."));
if (!fona->TCPconnected()) return 0; if (!fona->TCPconnected())
return 0;
/* Read data until either the connection is closed, or the idle timeout is
/* Read data until either the connection is closed, or the idle timeout is reached. */ * reached. */
uint16_t len = 0; uint16_t len = 0;
int16_t t = timeout; int16_t t = timeout;
uint16_t avail; uint16_t avail;
while (fona->TCPconnected() && (timeout >= 0)) { while (fona->TCPconnected() && (timeout >= 0)) {
//DEBUG_PRINT('.'); // DEBUG_PRINT('.');
while (avail = fona->TCPavailable()) { while (avail = fona->TCPavailable()) {
//DEBUG_PRINT('!'); // DEBUG_PRINT('!');
if (len + avail > maxlen) { if (len + avail > maxlen) {
avail = maxlen - len; avail = maxlen - len;
if (avail == 0) return len; if (avail == 0)
return len;
} }
// try to read the data into the end of the pointer // try to read the data into the end of the pointer
if (! fona->TCPread(buffp, avail)) return len; if (!fona->TCPread(buffp, avail))
return len;
// read it! advance pointer // read it! advance pointer
buffp += avail; buffp += avail;
len += avail; len += avail;
timeout = t; // reset the timeout timeout = t; // reset the timeout
//DEBUG_PRINTLN((uint8_t)c, HEX); // DEBUG_PRINTLN((uint8_t)c, HEX);
if (len == maxlen) { // we read all we want, bail if (len == maxlen) { // we read all we want, bail
DEBUG_PRINT(F("Read:\t")); DEBUG_PRINT(F("Read:\t"));
@ -110,7 +106,8 @@ class Adafruit_MQTT_FONA : public Adafruit_MQTT {
Watchdog.reset(); Watchdog.reset();
#endif #endif
timeout -= MQTT_FONA_INTERAVAILDELAY; timeout -= MQTT_FONA_INTERAVAILDELAY;
timeout -= MQTT_FONA_QUERYDELAY; // this is how long it takes to query the FONA for avail() timeout -= MQTT_FONA_QUERYDELAY; // this is how long it takes to query the
// FONA for avail()
delay(MQTT_FONA_INTERAVAILDELAY); delay(MQTT_FONA_INTERAVAILDELAY);
} }
@ -121,7 +118,7 @@ class Adafruit_MQTT_FONA : public Adafruit_MQTT {
DEBUG_PRINTLN(F("Writing packet")); DEBUG_PRINTLN(F("Writing packet"));
if (fona->TCPconnected()) { if (fona->TCPconnected()) {
boolean ret = fona->TCPsend((char *)buffer, len); boolean ret = fona->TCPsend((char *)buffer, len);
//DEBUG_PRINT(F("sendPacket returned: ")); DEBUG_PRINTLN(ret); // DEBUG_PRINT(F("sendPacket returned: ")); DEBUG_PRINTLN(ret);
if (!ret) { if (!ret) {
DEBUG_PRINTLN("Failed to send packet."); DEBUG_PRINTLN("Failed to send packet.");
return false; return false;
@ -133,10 +130,9 @@ class Adafruit_MQTT_FONA : public Adafruit_MQTT {
return true; return true;
} }
private: private:
uint32_t serverip; uint32_t serverip;
Adafruit_FONA *fona; Adafruit_FONA *fona;
}; };
#endif #endif

2
README.md
View File

@ -1,4 +1,4 @@
# Adafruit MQTT Library [![Build Status](https://travis-ci.com/adafruit/Adafruit_MQTT_Library.svg?branch=master)](https://travis-ci.com/adafruit/Adafruit_MQTT_Library) # Adafruit MQTT Library [![Build Status](https://github.com/adafruit/Adafruit_MQTT_Library/workflows/Arduino%20Library%20CI/badge.svg)](https://github.com/adafruit/Adafruit_MQTT_Library/actions)
Arduino library for MQTT support, including access to Adafruit IO. Works with Arduino library for MQTT support, including access to Adafruit IO. Works with
the Adafruit FONA, Arduino Yun, ESP8266 Arduino platforms, and anything that supports the Adafruit FONA, Arduino Yun, ESP8266 Arduino platforms, and anything that supports

15
examples/mqtt_fona/fonahelper.cpp
View File

@ -1,20 +1,27 @@
#include "Adafruit_FONA.h"
#include <Adafruit_SleepyDog.h> #include <Adafruit_SleepyDog.h>
#include <SoftwareSerial.h> #include <SoftwareSerial.h>
#include "Adafruit_FONA.h"
#define halt(s) { Serial.println(F( s )); while(1); } #define halt(s) \
{ \
Serial.println(F(s)); \
while (1) \
; \
}
extern Adafruit_FONA fona; extern Adafruit_FONA fona;
extern SoftwareSerial fonaSS; extern SoftwareSerial fonaSS;
boolean FONAconnect(const __FlashStringHelper *apn, const __FlashStringHelper *username, const __FlashStringHelper *password) { boolean FONAconnect(const __FlashStringHelper *apn,
const __FlashStringHelper *username,
const __FlashStringHelper *password) {
Watchdog.reset(); Watchdog.reset();
Serial.println(F("Initializing FONA....(May take 3 seconds)")); Serial.println(F("Initializing FONA....(May take 3 seconds)"));
fonaSS.begin(4800); // if you're using software serial fonaSS.begin(4800); // if you're using software serial
if (! fona.begin(fonaSS)) { // can also try fona.begin(Serial1) if (!fona.begin(fonaSS)) { // can also try fona.begin(Serial1)
Serial.println(F("Couldn't find FONA")); Serial.println(F("Couldn't find FONA"));
return false; return false;
} }

3
library.properties
View File

@ -1,5 +1,5 @@
name=Adafruit MQTT Library name=Adafruit MQTT Library
version=1.0.3 version=1.1.0
author=Adafruit author=Adafruit
maintainer=Adafruit <info@adafruit.com> maintainer=Adafruit <info@adafruit.com>
sentence=MQTT library that supports the FONA, ESP8266, Yun, and generic Arduino Client hardware. sentence=MQTT library that supports the FONA, ESP8266, Yun, and generic Arduino Client hardware.
@ -7,3 +7,4 @@ paragraph=Simple MQTT library that supports the bare minimum to publish and subs
category=Communication category=Communication
url=https://github.com/adafruit/Adafruit_MQTT_Library url=https://github.com/adafruit/Adafruit_MQTT_Library
architectures=* architectures=*
depends=Adafruit SleepyDog Library, Adafruit FONA Library, WiFi101