Adafruit_MQTT_Library/Adafruit_MQTT_CC3000.cpp

319 lines
8.2 KiB
C++
Raw Normal View History

2015-06-01 00:38:33 +02:00
#include "Adafruit_MQTT.h"
#include "Adafruit_MQTT_CC3000.h"
2015-06-03 01:21:49 +02:00
#include <Adafruit_SleepyDog.h>
2015-06-01 00:38:33 +02:00
2015-06-03 00:33:03 +02:00
static void printBuffer(uint8_t *buffer, uint8_t len) {
for (uint8_t i=0; i<len; i++) {
if (isprint(buffer[i]))
Serial.write(buffer[i]);
else
Serial.print(" ");
Serial.print(F(" [0x"));
if (buffer[i] < 0x10)
Serial.print("0");
Serial.print(buffer[i],HEX);
Serial.print("], ");
if (i % 8 == 7) Serial.println();
}
Serial.println();
}
Adafruit_MQTT_CC3000::Adafruit_MQTT_CC3000(Adafruit_CC3000 *cc3k, const char *server, uint16_t port, const char *cid, const char *user, const char *pass) : Adafruit_MQTT(server, port, cid, user, pass), cc3000(cc3k)
2015-06-01 00:38:33 +02:00
{
// nothin doin
}
int8_t Adafruit_MQTT_CC3000::connect(void) {
uint32_t ip = 0;
Watchdog.reset();
// look up IP address
if (serverip == 0) {
// Try looking up the website's IP address using CC3K's built in getHostByName
strcpy_P((char *)buffer, servername);
Serial.print((char *)buffer); Serial.print(F(" -> "));
2015-06-01 00:38:33 +02:00
uint8_t dnsretries = 5;
Watchdog.reset();
while (ip == 0) {
if (! cc3000->getHostByName((char *)buffer, &ip)) {
2015-06-01 00:38:33 +02:00
Serial.println(F("Couldn't resolve!"));
dnsretries--;
Watchdog.reset();
}
//Serial.println("OK"); Serial.println(ip, HEX);
if (!dnsretries) return -1;
delay(500);
}
serverip = ip;
cc3000->printIPdotsRev(serverip);
Serial.println();
}
Watchdog.reset();
// connect to server
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_CONNECT
Serial.println(F("Connecting to TCP"));
2015-06-03 00:33:03 +02:00
#endif
2015-06-01 00:38:33 +02:00
mqttclient = cc3000->connectTCP(serverip, portnum);
uint8_t len = connectPacket(buffer);
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_CONNECT
Serial.println(F("MQTT connection packet:")); printBuffer(buffer, len);
#endif
2015-06-01 00:38:33 +02:00
if (mqttclient.connected()) {
uint16_t ret = mqttclient.write(buffer, len);
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_CONNECT
Serial.print("returned: "); Serial.println(ret);
#endif
2015-06-01 00:38:33 +02:00
if (ret != len) return -1;
} else {
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_CONNECT
Serial.println(F("Connection failed"));
#endif
2015-06-01 00:38:33 +02:00
return -1;
}
len = readPacket(buffer, 4, CONNECT_TIMEOUT_MS);
if (len != 4) return -1;
2015-06-03 00:33:03 +02:00
if ((buffer[0] != (MQTT_CTRL_CONNECTACK << 4)) || (buffer[1] != 2)) {
return -1;
2015-06-01 00:38:33 +02:00
}
2015-06-03 00:33:03 +02:00
if (buffer[3] != 0) return buffer[3];
/**************** subscription time! */
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == 0) continue;
#ifdef DEBUG_MQTT_CONNECT
Serial.print(F("Subscribing..."));
#endif
uint8_t len = subscribePacket(buffer, subscriptions[i]->topic, subscriptions[i]->qos);
#ifdef DEBUG_MQTT_CONNECT
Serial.println(F("MQTT subscription packet:")); printBuffer(buffer, len);
#endif
2015-06-01 00:38:33 +02:00
2015-06-03 00:33:03 +02:00
if (mqttclient.connected()) {
uint16_t ret = mqttclient.write(buffer, len);
#ifdef DEBUG_MQTT_CONNECT
Serial.print("returned: "); Serial.println(ret);
#endif
if (ret != len) return -1;
} else {
#ifdef DEBUG_MQTT_CONNECT
Serial.println(F("Connection failed"));
#endif
return -1;
}
// Get SUBACK
len = readPacket(buffer, 5, CONNECT_TIMEOUT_MS);
#ifdef DEBUG_MQTT_CONNECT
Serial.print(F("SUBACK:\t")); printBuffer(buffer, len);
#endif
if ((len != 5) || (buffer[0] != (MQTT_CTRL_SUBACK << 4))) {
return 6; // failure to subscribe
}
}
return 0;
2015-06-01 00:38:33 +02:00
}
2015-06-03 00:33:03 +02:00
uint16_t Adafruit_MQTT_CC3000::readPacket(uint8_t *buffer, uint8_t maxlen, int16_t timeout, boolean checkForValidPubPacket) {
2015-06-01 00:38:33 +02:00
/* Read data until either the connection is closed, or the idle timeout is reached. */
uint16_t len = 0;
int16_t t = timeout;
2015-06-01 00:38:33 +02:00
2015-06-03 00:33:03 +02:00
while (mqttclient.connected() && (timeout >= 0)) {
//Serial.print('.');
2015-06-01 00:38:33 +02:00
while (mqttclient.available()) {
2015-06-03 00:33:03 +02:00
//Serial.print('!');
2015-06-01 00:38:33 +02:00
char c = mqttclient.read();
timeout = t; // reset the timeout
2015-06-01 00:38:33 +02:00
buffer[len] = c;
//Serial.print((uint8_t)c,HEX);
2015-06-01 00:38:33 +02:00
len++;
if (len == maxlen) { // we read all we want, bail
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_PACKETREAD
Serial.print(F("Read packet:\t")); printBuffer(buffer, len);
#endif
return len;
}
2015-06-03 00:33:03 +02:00
// special case where we just one one publication packet at a time
if (checkForValidPubPacket) {
if ((buffer[0] == (MQTT_CTRL_PUBLISH << 4)) && (buffer[1] == len-2)) {
// oooh a valid publish packet!
#ifdef DEBUG_MQTT_PACKETREAD
Serial.print(F("PUBLISH packet:\t")); printBuffer(buffer, len);
#endif
return len;
}
2015-06-01 00:38:33 +02:00
}
}
Watchdog.reset();
2015-06-03 00:33:03 +02:00
timeout -= MQTT_CC3000_INTERAVAILDELAY;
delay(MQTT_CC3000_INTERAVAILDELAY);
2015-06-01 00:38:33 +02:00
}
return len;
}
2015-06-01 01:02:42 +02:00
boolean Adafruit_MQTT_CC3000::ping(uint8_t times) {
while (times) {
uint8_t len = pingPacket(buffer);
2015-06-03 00:33:03 +02:00
Serial.print(F("Sending:\t")); printBuffer(buffer, len);
if (mqttclient.connected()) {
uint16_t ret = mqttclient.write(buffer, len);
//Serial.print("returned: "); Serial.println(ret);
if (ret != len) return false;
} else {
Serial.println(F("Connection failed"));
return false;
}
// process ping reply
len = readPacket(buffer, 2, PING_TIMEOUT_MS);
2015-06-02 04:51:59 +02:00
if (buffer[0] == (MQTT_CTRL_PINGRESP << 4))
return true;
2015-06-02 04:51:59 +02:00
}
return false;
}
2015-06-01 01:02:42 +02:00
int32_t Adafruit_MQTT_CC3000::close(void) {
return mqttclient.close();
}
2015-06-01 05:19:13 +02:00
boolean Adafruit_MQTT_CC3000::publish(const char *topic, char *data, uint8_t qos) {
2015-06-01 05:19:13 +02:00
uint8_t len = publishPacket(buffer, topic, data, qos);
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_PUBLISH
Serial.println(F("MQTT publish packet:")); printBuffer(buffer, len);
#endif
2015-06-01 05:19:13 +02:00
if (mqttclient.connected()) {
uint16_t ret = mqttclient.write(buffer, len);
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_PUBLISH
Serial.print("returned: "); Serial.println(ret);
#endif
2015-06-01 05:19:13 +02:00
if (ret != len) return false;
} else {
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_PUBLISH
2015-06-01 05:19:13 +02:00
Serial.println(F("Connection failed"));
2015-06-03 00:33:03 +02:00
#endif
2015-06-01 05:19:13 +02:00
return false;
}
if (qos > 0) {
len = readPacket(buffer, 4, PUBLISH_TIMEOUT_MS);
2015-06-03 00:33:03 +02:00
#ifdef DEBUG_MQTT_PUBLISH
Serial.print(F("Reply:\t")); printBuffer(buffer, len);
#endif
2015-06-01 05:19:13 +02:00
return true;
} else {
return true;
}
}
2015-06-03 00:33:03 +02:00
boolean Adafruit_MQTT_CC3000::subscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i;
// see if we are already subscribed
for (i=0; i<MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == sub) {
#ifdef DEBUG_MQTT_SUBSCRIBE
Serial.println(F("Already subscribed"));
#endif
break;
}
}
if (i==MAXSUBSCRIPTIONS) { // add to subscriptionlist
for (i=0; i<MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == 0) {
#ifdef DEBUG_MQTT_SUBSCRIBE
Serial.print(F("Added sub ")); Serial.println(i);
#endif
subscriptions[i] = sub;
break;
}
}
}
if (i==MAXSUBSCRIPTIONS) {
#ifdef DEBUG_MQTT_SUBSCRIBE
Serial.println(F("no more space :("));
#endif
return false;
}
}
Adafruit_MQTT_Subscribe *Adafruit_MQTT_CC3000::readSubscription(int16_t timeout) {
uint8_t i, topiclen, datalen;
#ifdef DEBUG_MQTT_READSUB
Serial.println(F("reading..."));
#endif
uint16_t len = readPacket(buffer, MAXBUFFERSIZE, timeout, true); // return one full packet
#ifdef DEBUG_MQTT_READSUB
printBuffer(buffer, len);
#endif
if (!len) return NULL;
topiclen = buffer[3];
#ifdef DEBUG_MQTT_READSUB
Serial.print(F("Looking for subscription len ")); Serial.println(topiclen);
#endif
// figure out what subscription this is!
for (i=0; i<MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i]) {
//Serial.print(i);
boolean flag = true;
// TODO: REPLACE WITH MEMCMP?
for (uint8_t k=0; k<topiclen; k++) {
if ( buffer[4+k] != pgm_read_byte(subscriptions[i]->topic+k) )
flag = false;
}
if (flag) {
#ifdef DEBUG_MQTT_READSUB
Serial.println((char *)buffer+4);
Serial.print(F("Found sub #")); Serial.println(i);
#endif
break;
}
}
}
if (i==MAXSUBSCRIPTIONS) return NULL; // matching sub not found ???
// zero out the old data
memset(subscriptions[i]->lastread, 0, SUBSCRIPTIONDATALEN);
datalen = len - topiclen - 4;
if (datalen > SUBSCRIPTIONDATALEN) {
datalen = SUBSCRIPTIONDATALEN-1; // cut it off
}
// extract out just the data, into the subscription object itself
memcpy(subscriptions[i]->lastread, buffer+4+topiclen, datalen);
#ifdef DEBUG_MQTT_READSUB
Serial.print(F("Data len: ")); Serial.println(datalen);
Serial.print("Data: "); Serial.println((char *)subscriptions[i]->lastread);
#endif
// return the valid matching subscription
return subscriptions[i];
}