From 40dbcb1e37ef481e6404343f75e824ca40d66fe9 Mon Sep 17 00:00:00 2001 From: Thomas Krebs Date: Sun, 30 Aug 2015 20:08:02 +0200 Subject: [PATCH] Add MQTTClientGateway --- .../MQTTClientGateway/MQTTClientGateway.ino | 605 ++++++++++++++++++ .../examples/MQTTClientGateway/MyMQTTClient.h | 122 ++++ libraries/PubSubClient/CHANGES.txt | 47 ++ libraries/PubSubClient/LICENSE.txt | 20 + libraries/PubSubClient/PubSubClient.cpp | 427 ++++++++++++ libraries/PubSubClient/PubSubClient.h | 81 +++ .../examples/mqtt_auth/mqtt_auth.ino | 39 ++ .../examples/mqtt_basic/mqtt_basic.ino | 38 ++ .../mqtt_publish_in_callback.ino | 61 ++ .../examples/mqtt_stream/mqtt_stream.ino | 58 ++ libraries/PubSubClient/keywords.txt | 25 + 11 files changed, 1523 insertions(+) create mode 100644 libraries/MySensors/examples/MQTTClientGateway/MQTTClientGateway.ino create mode 100644 libraries/MySensors/examples/MQTTClientGateway/MyMQTTClient.h create mode 100755 libraries/PubSubClient/CHANGES.txt create mode 100755 libraries/PubSubClient/LICENSE.txt create mode 100755 libraries/PubSubClient/PubSubClient.cpp create mode 100755 libraries/PubSubClient/PubSubClient.h create mode 100755 libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino create mode 100755 libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino create mode 100755 libraries/PubSubClient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino create mode 100755 libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino create mode 100755 libraries/PubSubClient/keywords.txt diff --git a/libraries/MySensors/examples/MQTTClientGateway/MQTTClientGateway.ino b/libraries/MySensors/examples/MQTTClientGateway/MQTTClientGateway.ino new file mode 100644 index 0000000..ece4e0c --- /dev/null +++ b/libraries/MySensors/examples/MQTTClientGateway/MQTTClientGateway.ino @@ -0,0 +1,605 @@ +/* MyMQTT Client Gateway 0.1b + Created by Norbert Truchsess + Based on MyMQTT-broker gateway created by Daniel Wiegert + Based on MySensors Ethernet Gateway by Henrik Ekblad + http://www.mysensors.org + + Changes by Thomas Krebs + - Add signing support from MySensors 1.5 and update for MySensors 1.5; + - Restructured code back to a C like implementation following the existing MQTTGateway + Requires MySensors lib 1.5 + + * Change below; TCP_IP, TCP_PORT, TCP_MAC + This will listen on your selected TCP_IP:TCP_PORT below, Please change TCP_MAC your liking also. + *1 -> NOTE: Keep first byte at x2, x6, xA or xE (replace x with any hex value) for using Local Ranges. + *2 You can use standard pin set-up as MySensors recommends or if you own a IBOARD you may change + the radio-pins below if you hardware mod your iBoard. see [URL BELOW] for more details. + http://forum.mysensors.org/topic/224/iboard-cheap-single-board-ethernet-arduino-with-radio/5 + * Don't forget to look at the definitions in libraries\MySensors\MyMQTT.h! + define TCPDUMP and connect serial interface if you have problems, please write on + http://forum.mysensors.org/ and explain your problem, include serial output. Don't forget to + turn on DEBUG in libraries\MySensors\MyConfig.h also. + MQTT_FIRST_SENSORID is for 'DHCP' server in MyMQTT. You may limit the ID's with FIRST and LAST definition. + If you want your manually configured below 20 set MQTT_FIRST_SENSORID to 20. + To disable: set MQTT_FIRST_SENSORID to 255. + MQTT_BROKER_PREFIX is the leading prefix for your nodes. This can be only one char if like. + MQTT_SEND_SUBSCRIPTION is if you want the MyMQTT to send a empty payload message to your nodes. + This can be useful if you want to send latest state back to the MQTT client. Just check if incoming + message has any length or not. + Example: if (msg.type==V_LIGHT && strlen(msg.getString())>0) otherwise the code might do strange things. + * Address-layout is : [MQTT_BROKER_PREFIX]/[NodeID]/[SensorID]/V_[SensorType][/set | /get] + NodeID and SensorID is uint8 (0-255) number. + Segment after SensorID is translation of the sensor type, look inside MyMQTT.cpp for the definitions. + User can change this to their needs. We have also left some space for custom types. + Special: (sensor 255 reserved for special commands) + + The last seqment is a command to to be send down to the sensor. The gateway is subscribed to topics which have + a command present! This is to avoid that a publish by the gateway will end up in the gateway via it subsription. + You can receive a node sketch name with MyMQTT/20/255/V_Sketch_name (or version with _version) + + So far tested in openhab and MyMQTT for Android (Not my creation) + - http://www.openhab.org/ + * How to set-up Openhab and MQTTGateway: + http://forum.mysensors.org/topic/303/mqtt-broker-gateway + + */ + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "MyMQTTClient.h" + +#ifdef MY_SIGNING_FEATURE +#include +#include +#include +#endif + +//#define DSRTC +#ifdef DSRTC +#include +#include // a basic DS1307 library that returns time as a time_t +#endif + +/* + * To configure MQTTClientGateway.ino to use an ENC28J60 based board include + * 'UIPEthernet.h' (SPI.h required for MySensors anyway). The UIPEthernet-library can be downloaded + * from: https://github.com/ntruchsess/arduino_uip + */ + +//#include +/* + * To execute MQTTClientGateway.ino on Yun uncomment Bridge.h and YunClient.h. + * Do not include Ethernet.h or SPI.h in this case. + * On Yun there's no need to configure local_ip and mac in the sketch + * as this is configured on the linux-side of Yun. + */ + +//#include +//#include +// * Use this for IBOARD modded to use standard MISO/MOSI/SCK, see note *1 above! +/* + #define RADIO_CE_PIN 3 // radio chip enable + #define RADIO_SPI_SS_PIN 8 // radio SPI serial select + #define RADIO_ERROR_LED_PIN A2 // Error led pin + #define RADIO_RX_LED_PIN A1 // Receive led pin + #define RADIO_TX_LED_PIN A0 // the PCB, on board LED*/ + +// * Use this for default configured pro mini / nano etc : +// + +//#define RADIO_CE_PIN 5 // radio chip enable +//#define RADIO_SPI_SS_PIN 6 // radio SPI serial select +//#define RADIO_ERROR_LED_PIN 7 // Error led pin +//#define RADIO_RX_LED_PIN 8 // Receive led pin +//#define RADIO_TX_LED_PIN 9 // the PCB, on board LED*/ + +// CE_PIN and SPI_SS_PIN for Mega +#define RADIO_CE_PIN 48 // radio chip enable +#define RADIO_SPI_SS_PIN 49 // radio SPI serial select +#define RADIO_ERROR_LED_PIN A2 // Error led pin +#define RADIO_RX_LED_PIN A1 // Receive led pin +#define RADIO_TX_LED_PIN A0 // the PCB, on board LED*/ + +//replace with ip of server you want to connect to, comment out if using 'remote_host' +uint8_t remote_ip[] = { 192, 168, 178, 74 }; // Mosquitto broker + +//replace with hostname of server you want to connect to, comment out if using 'remote_ip' +//char* remote_ip = "server.local"; +//replace with the port that your server is listening on +#define remote_port 1883 +//replace with arduinos ip-address. Comment out if Ethernet-startup should use dhcp. Is ignored on Yun +uint8_t local_ip[] = {192, 168, 178, 11}; +//replace with ethernet shield mac. It's mandatory every device is assigned a unique mac. Is ignored on Yun +uint8_t mac[] = { 0xA2, 0xAE, 0xAD, 0xA0, 0xA0, 0xA2 }; + +/////////////////////////////////////////////////////////////////// +#define MQTT_AUTH_REQUIRED // if your broker has anonymous access uncomment +#define MQTT_USER "username" // username on MQTT broker +#define MQTT_PWD "pwd" // password for MQTT username above +////////////////////////////////////////////////////////////////// + +#if defined remote_ip && defined remote_host +#error "cannot define both remote_ip and remote_host at the same time!" +#endif + +#ifdef _YUN_CLIENT_H_ +YunClient ethClient; +#else +EthernetClient ethClient; +#endif + +//////////////////////////////////////////////////////////////// +// NRFRF24L01 radio driver (set low transmit power by default) +MyTransportNRF24 transport(RADIO_CE_PIN, RADIO_SPI_SS_PIN, RF24_PA_LEVEL_GW); +//MyTransportRFM69 transport; + +// Message signing driver (signer needed if MY_SIGNING_FEATURE is turned on in MyConfig.h) +//MySigningNone signer; +//MySigningAtsha204Soft signer(true); +//MySigningAtsha204 signer; + +// Hardware profile +MyHwATMega328 hw; + +MyMessage msg; +char convBuf[MAX_PAYLOAD * 2 + 1]; +uint8_t buffsize; +char buffer[MQTT_MAX_PACKET_SIZE]; + +//////////////////////////////////////////////////////////////// + +volatile uint8_t countRx; +volatile uint8_t countTx; +volatile uint8_t countErr; + + +//////////////////////////////////////////////////////////////// + +void processMQTTMessages(char* topic, byte* payload, unsigned int length); +PubSubClient client(remote_ip, remote_port, processMQTTMessages, ethClient); + +//////////////////////////////////////////////////////////////// + + +// Declare and initialize MySensor instance +// Construct MyMQTTClient (signer needed if MY_SIGNING_FEATURE is turned on in MyConfig.h, if signing +// feature not to be used, uncomment) +// To use LEDs blinking, uncomment WITH_LEDS_BLINKING in MyConfig.h +MySensor gw(transport, hw +#ifdef MY_SIGNING_FEATURE + , signer +#endif +#ifdef WITH_LEDS_BLINKING + , RADIO_RX_LED_PIN, RADIO_TX_LED_PIN, RADIO_ERROR_LED_PIN +#endif + ); + +/* + * setup + */ +void setup() +{ + countRx = 0; + countTx = 0; + countErr = 0; + + Ethernet.begin(mac, local_ip); + //Bridge.begin(); + delay(1000); // Wait for Ethernet to get configured. + + begin(); +} + +/* + * loop + */ +void loop() +{ + if (!client.connected()) + { + client.connect("MySensor" +#ifdef MQTT_AUTH_REQUIRED + ,MQTT_USER,MQTT_PWD +#endif + ); + client.subscribe(MQTT_TOPIC_MASK); + } + client.loop(); + gw.process(); +} + +/* + * processRadioMessage + * + * Receives radio message, parses it and forwards it to the MQTT broker + */ +void processRadioMessage(const MyMessage &message) +{ + rxBlink(1); + sendMQTT(message); +} + +/* + * sendMQTT + * Handles processing of radio messages and eventually publishes it to the MQTT broker + */ +void sendMQTT(const MyMessage &inMsg) +{ + MyMessage msg = inMsg; + buffsize = 0; + if (!client.connected()) + return; //We have no connections - return + if (msg.isAck()) + { +#ifdef DEBUG + Serial.println("msg is ack!"); +#endif + if (msg.sender == 255 && mGetCommand(msg) == C_INTERNAL + && msg.type == I_ID_REQUEST) + { + // TODO: sending ACK request on id_response fucks node up. doesn't work. + // The idea was to confirm id and save to EEPROM_LATEST_NODE_ADDRESS. + } + } + else + { + // we have to check every message if its a newly assigned id or not. + // Ack on I_ID_RESPONSE does not work, and checking on C_PRESENTATION isn't reliable. + uint8_t newNodeID = gw.loadState(EEPROM_LATEST_NODE_ADDRESS) + 1; + if (newNodeID <= MQTT_FIRST_SENSORID) + newNodeID = MQTT_FIRST_SENSORID; + if (msg.sender == newNodeID) + { + gw.saveState(EEPROM_LATEST_NODE_ADDRESS, newNodeID); + } + if (mGetCommand(msg) == C_INTERNAL) + { + if (msg.type == I_CONFIG) + { + txBlink(1); + if (!gw.sendRoute( + build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, + I_CONFIG, 0).set("M"))) + errBlink(1); + } + else if (msg.type == I_TIME) + { +#ifdef DEBUG + Serial.println("I_TIME requested!"); +#endif + txBlink(1); + if (!gw.sendRoute( + build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, + I_TIME, 0).set(now()))) + errBlink(1); + } + else if (msg.type == I_ID_REQUEST && msg.sender == 255) + { + uint8_t newNodeID = gw.loadState(EEPROM_LATEST_NODE_ADDRESS) + 1; + if (newNodeID <= MQTT_FIRST_SENSORID) + newNodeID = MQTT_FIRST_SENSORID; + if (newNodeID >= MQTT_LAST_SENSORID) + return; // Sorry no more id's left :( + txBlink(1); + if (!gw.sendRoute( + build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, + I_ID_RESPONSE, 0).set(newNodeID))) + errBlink(1); + } + else if (msg.type == I_BATTERY_LEVEL) + { + strcpy_P(buffer, mqtt_prefix); + buffsize += strlen_P(mqtt_prefix); + buffsize += sprintf(&buffer[buffsize], "/%i/255/BATTERY_LEVEL\0", msg.sender ); + msg.getString(convBuf); +#ifdef DEBUG + Serial.print("publish: "); + Serial.print((char*) buffer); + Serial.print(" "); + Serial.println((char*) convBuf); +#endif + client.publish(buffer, convBuf); + } + else if (msg.type == I_SKETCH_NAME) + { + strcpy_P(buffer, mqtt_prefix); + buffsize += strlen_P(mqtt_prefix); + buffsize += sprintf(&buffer[buffsize], "/%i/255/SKETCH_NAME\0", msg.sender ); + msg.getString(convBuf); +#ifdef DEBUG + Serial.print("publish: "); + Serial.print((char*) buffer); + Serial.print(" "); + Serial.println((char*) convBuf); +#endif + client.publish(buffer, convBuf); + } + else if (msg.type == I_SKETCH_VERSION) + { + strcpy_P(buffer, mqtt_prefix); + buffsize += strlen_P(mqtt_prefix); + buffsize += sprintf(&buffer[buffsize], "/%i/255/SKETCH_VERSION\0", msg.sender ); + msg.getString(convBuf); +#ifdef DEBUG + Serial.print("publish: "); + Serial.print((char*) buffer); + Serial.print(" "); + Serial.println((char*) convBuf); +#endif + client.publish(buffer, convBuf); + } + + } + else if (mGetCommand(msg) != 0) + { + if (mGetCommand(msg) == 3) + msg.type = msg.type + (S_FIRSTCUSTOM - 10); //Special message + + if (msg.type > VAR_TOTAL) + msg.type = VAR_TOTAL; // If type > defined types set to unknown. + strcpy_P(buffer, mqtt_prefix); + buffsize += strlen_P(mqtt_prefix); + buffsize += sprintf(&buffer[buffsize], "/%i/%i/V_%s\0", msg.sender, + msg.sensor, getType(convBuf, &VAR_Type[msg.type])); + msg.getString(convBuf); +#ifdef DEBUG + Serial.print("publish: "); + Serial.print((char*) buffer); + Serial.print(" "); + Serial.println((char*) convBuf); +#endif + client.publish(buffer, convBuf); + } + } +} + +/* + * build + * Constructs a radio message + */ +inline MyMessage& build(MyMessage &msg, uint8_t sender, uint8_t destination, uint8_t sensor, + uint8_t command, uint8_t type, bool enableAck) +{ + msg.destination = destination; + msg.sender = sender; + msg.sensor = sensor; + msg.type = type; + mSetCommand(msg, command); + mSetRequestAck(msg, enableAck); + mSetAck(msg, false); + return msg; +} + +/* + * getType + */ +char *getType(char *b, const char **index) +{ + char *q = b; + char *p = (char *) pgm_read_word(index); + while (*q++ = pgm_read_byte(p++)) + ; + *q = 0; + return b; +} + +/* + * begin + * wraps MySensors begin method; setup of RTC and led timers interrupt + */ +void begin() +{ +#ifdef DEBUG + Serial.begin(BAUD_RATE); +#endif +#ifdef DSRTC + // Attach RTC + setSyncProvider(RTC.get); // the function to get the time from the RTC + setSyncInterval(60); +#endif + + gw.begin(processRadioMessage, 0, true, 0); + + MsTimer2::set(200, ledTimersInterrupt); + MsTimer2::start(); + +#ifdef DEBUG + Serial.print(getType(convBuf, &VAR_Type[S_FIRSTCUSTOM])); +#endif +} + +/* + * processMQTTMessages + * message handler for the PubSubClient + */ +void processMQTTMessages(char* topic, byte* payload, unsigned int length) +{ + processMQTTMessage(topic, payload, length); +} + +/* + * processMQTTMessage + * processes MQTT messages, parses the topic, extracts radio address out of topic and sends them + * to the respective radio + */ +void processMQTTMessage(char* topic, byte* payload, + unsigned int length) +{ + char *str, *p; + uint8_t i = 0; + buffer[0] = 0; + buffsize = 0; + uint8_t cmd = -1; + + for (str = strtok_r(topic, "/", &p); str && i < 5; + str = strtok_r(NULL, "/", &p)) + { + switch (i) + { + case 0: + { + if (strcmp_P(str, mqtt_prefix) != 0) + { //look for MQTT_PREFIX + return; //Message not for us or malformatted! + } + break; + } + case 1: + { + msg.destination = atoi(str); //NodeID + break; + } + case 2: + { + msg.sensor = atoi(str); //SensorID + break; + } + case 3: + { + char match = 0; //SensorType + //strcpy(str,(char*)&str[2]); //Strip VAR_ + + for (uint8_t j = 0; + strcpy_P(convBuf, (char*) pgm_read_word(&(VAR_Type[j]))); j++) + { + if (strcmp((char*) &str[2], convBuf) == 0) + { //Strip VAR_ and compare + match = j; + break; + } + if (j >= VAR_TOTAL) + { // No match found! + match = VAR_TOTAL; // last item. + break; + } + } + msg.type = match; + break; + } + case 4: + { + // support the command get and set; get will be mapped to a C_REQ and set to C_SET + if (strcmp(str,MQTT_CMD_SET) == 0) + { + cmd = C_SET; + } + else if (strcmp(str,MQTT_CMD_GET) == 0) { + cmd = C_REQ; + } + else { +#ifdef DEBUG + Serial.print("Received unsupported command - ignore: "); + Serial.println(str); +#endif + } + } + } + i++; + } //Check if packge has payload + + if (cmd != -1) { + char* ca; + ca = (char *)payload; + ca += length; + *ca = '\0'; + + msg.set((const char*)payload); //Payload + + txBlink(1); + // inject time + if ((msg.destination == 0) && (msg.sensor == 199)) { + unsigned long epoch = atol((char*)payload); + if (epoch > 10000) { +#ifdef DSRTC + RTC.set(epoch); // this sets the RTC to the time from controller - which we do want periodically +#endif + setTime(epoch); + } +#ifdef DEBUG + Serial.print("Time recieved "); + Serial.println(epoch); +#endif + } + // + if (!gw.sendRoute( + build(msg, GATEWAY_ADDRESS, msg.destination, msg.sensor, C_SET, msg.type, + 0))) + errBlink(1); + } +} + +////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// Led handling +////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +/* + * ledTimersIntterupt + */ +void ledTimersInterrupt() { + if (countRx && countRx != 255) { + // switch led on + digitalWrite(RADIO_RX_LED_PIN, LOW); + } else if (!countRx) { + // switching off + digitalWrite(RADIO_RX_LED_PIN, HIGH); + } + if (countRx != 255) { + countRx--; + } + + if (countTx && countTx != 255) { + // switch led on + digitalWrite(RADIO_TX_LED_PIN, LOW); + } else if (!countTx) { + // switching off + digitalWrite(RADIO_TX_LED_PIN, HIGH); + } + if (countTx != 255) { + countTx--; + } + + if (countErr && countErr != 255) { + // switch led on + digitalWrite(RADIO_ERROR_LED_PIN, LOW); + } else if (!countErr) { + // switching off + digitalWrite(RADIO_ERROR_LED_PIN, HIGH); + } + if (countErr != 255) { + countErr--; + } +} + + +void rxBlink(uint8_t cnt) +{ + if (countRx == 255) + { + countRx = cnt; + } +} + +void txBlink(uint8_t cnt) +{ + if (countTx == 255) + { + countTx = cnt; + } +} + +void errBlink(uint8_t cnt) +{ + if (countErr == 255) + { + countErr = cnt; + } +} + + diff --git a/libraries/MySensors/examples/MQTTClientGateway/MyMQTTClient.h b/libraries/MySensors/examples/MQTTClientGateway/MyMQTTClient.h new file mode 100644 index 0000000..cd6e4fe --- /dev/null +++ b/libraries/MySensors/examples/MQTTClientGateway/MyMQTTClient.h @@ -0,0 +1,122 @@ +#define MY_HMAC_KEY 0x10,0x23,0x0A,0x78,0x77,0x35,0xB0,0x01,0x70,0x47,0xF1,0xDE,0x21,0x94,0x94,0x67,0xEE,0x36,0x72,0x42,0x97,0xBB,0xA0,0x0A,0x0F,0x09,0x03,0xE2,0x00,0x31,0xE4,0x41/* +The MySensors library adds a new layer on top of the RF24 library. +It handles radio network routing, relaying and ids. +Created by Henrik Ekblad +Modified by Daniel Wiegert +Modified by Norbert Truchsess +This program is free software; you can redistribute it and/or +modify it under the terms of the GNU General Public License +version 2 as published by the Free Software Foundation. +*/ + + +#ifndef MyMQTTClient_h +#define MyMQTTClient_h + +////////////////////////////////////////////////////////////////// + +#ifdef DEBUG +#define TCPDUMP // Dump TCP packages. +#endif +#define MQTT_FIRST_SENSORID 20 // If you want manually configured nodes below this value. 255 = Disable +#define MQTT_LAST_SENSORID 254 // 254 is max! 255 reserved. +#define MQTT_PREFIX "MyMQTT" // First prefix in MQTT tree, keep short! +#define MQTT_TOPIC_MASK "MyMQTT/+/+/+/+" // Topic mask the client will subscribe to. This ensures that it will not receive any sensor data send by sensors + // the following topic structure is assumed: MyMQTT//// +#define MQTT_SEND_SUBSCRIPTION 1 // Send empty payload (request) to node upon MQTT client subscribe request. +#define MQTT_CMD_SET "set" // command (see MQTT_TOPIC_MASK); mapped to C_SET command type of sensor network +#define MQTT_CMD_GET "get" // command (see MQTT_TOPIC_MASK); mapped to C_REQ command type of sensor network + +// NOTE above : Beware to check if there is any length on payload in your incommingMessage code: +// Example: if (msg.type==V_LIGHT && strlen(msg.getString())>0) otherwise the code might do strange things. + +////////////////////////////////////////////////////////////////// + + +char VAR_0[] PROGMEM = "TEMP"; //V_TEMP +char VAR_1[] PROGMEM = "HUM"; //V_HUM +char VAR_2[] PROGMEM = "LIGHT"; //V_LIGHT +char VAR_3[] PROGMEM = "DIMMER"; //V_DIMMER +char VAR_4[] PROGMEM = "PRESSURE"; //V_PRESSURE +char VAR_5[] PROGMEM = "FORECAST"; //V_FORECAST +char VAR_6[] PROGMEM = "RAIN"; //V_RAIN +char VAR_7[] PROGMEM = "RAINRATE"; //V_RAINRATE +char VAR_8[] PROGMEM = "WIND"; //V_WIND +char VAR_9[] PROGMEM = "GUST"; //V_GUST +char VAR_10[] PROGMEM = "DIRECTON"; //V_DIRECTON +char VAR_11[] PROGMEM = "UV"; //V_UV +char VAR_12[] PROGMEM = "WEIGHT"; //V_WEIGHT +char VAR_13[] PROGMEM = "DISTANCE"; //V_DISTANCE +char VAR_14[] PROGMEM = "IMPEDANCE"; //V_IMPEDANCE +char VAR_15[] PROGMEM = "ARMED"; //V_ARMED +char VAR_16[] PROGMEM = "TRIPPED"; //V_TRIPPED +char VAR_17[] PROGMEM = "WATT"; //V_WATT +char VAR_18[] PROGMEM = "KWH"; //V_KWH +char VAR_19[] PROGMEM = "SCENE_ON"; //V_SCENE_ON +char VAR_20[] PROGMEM = "SCENE_OFF"; //V_SCENE_OFF +char VAR_21[] PROGMEM = "HEATER"; //V_HEATER +char VAR_22[] PROGMEM = "HEATER_SW"; //V_HEATER_SW +char VAR_23[] PROGMEM = "LIGHT_LEVEL"; //V_LIGHT_LEVEL +char VAR_24[] PROGMEM = "VAR1"; //V_VAR1 +char VAR_25[] PROGMEM = "VAR2"; //V_VAR2 +char VAR_26[] PROGMEM = "VAR3"; //V_VAR3 +char VAR_27[] PROGMEM = "VAR4"; //V_VAR4 +char VAR_28[] PROGMEM = "VAR5"; //V_VAR5 +char VAR_29[] PROGMEM = "UP"; //V_UP +char VAR_30[] PROGMEM = "DOWN"; //V_DOWN +char VAR_31[] PROGMEM = "STOP"; //V_STOP +char VAR_32[] PROGMEM = "IR_SEND"; //V_IR_SEND +char VAR_33[] PROGMEM = "IR_RECEIVE"; //V_IR_RECEIVE +char VAR_34[] PROGMEM = "FLOW"; //V_FLOW +char VAR_35[] PROGMEM = "VOLUME"; //V_VOLUME +char VAR_36[] PROGMEM = "LOCK_STATUS"; //V_LOCK_STATUS +char VAR_37[] PROGMEM = "LEVEL"; // S_DUST, S_LIGHT_LEVEL, ... +char VAR_38[] PROGMEM = "VOLTAGE"; //V_VOLTAGE +char VAR_39[] PROGMEM = "CURRENT"; //V_CURRENT +char VAR_40[] PROGMEM = "RGB"; //V_RGB +char VAR_41[] PROGMEM = ""; // +char VAR_42[] PROGMEM = ""; // +char VAR_43[] PROGMEM = ""; // +char VAR_44[] PROGMEM = ""; // +char VAR_45[] PROGMEM = ""; // +char VAR_46[] PROGMEM = ""; // +char VAR_47[] PROGMEM = ""; // +char VAR_48[] PROGMEM = ""; // +char VAR_49[] PROGMEM = ""; // +char VAR_50[] PROGMEM = ""; // +char VAR_51[] PROGMEM = ""; // +char VAR_52[] PROGMEM = ""; // +char VAR_53[] PROGMEM = ""; // +char VAR_54[] PROGMEM = ""; // +char VAR_55[] PROGMEM = ""; // +char VAR_56[] PROGMEM = ""; // +char VAR_57[] PROGMEM = ""; // +char VAR_58[] PROGMEM = ""; // +char VAR_59[] PROGMEM = ""; // +char VAR_60[] PROGMEM = "Started!\n"; //Custom for MQTTGateway +char VAR_61[] PROGMEM = "SKETCH_NAME"; //Custom for MQTTGateway +char VAR_62[] PROGMEM = "SKETCH_VERSION"; //Custom for MQTTGateway +char VAR_63[] PROGMEM = "UNKNOWN"; //Custom for MQTTGateway + +////////////////////////////////////////////////////////////////// + +PROGMEM const char *VAR_Type[] = +{ VAR_0, VAR_1, VAR_2, VAR_3, VAR_4, VAR_5, VAR_6, VAR_7, VAR_8, VAR_9, VAR_10, VAR_11, VAR_12, VAR_13, + VAR_14, VAR_15, VAR_16, VAR_17, VAR_18, VAR_19, VAR_20, VAR_21, VAR_22, VAR_23, VAR_24, VAR_25, + VAR_26, VAR_27, VAR_28, VAR_29, VAR_30, VAR_31, VAR_32, VAR_33, VAR_34, VAR_35, VAR_36, VAR_37, + VAR_38, VAR_39, VAR_40, VAR_41, VAR_42, VAR_43, VAR_44, VAR_45, VAR_46, VAR_47, VAR_48, VAR_49, + VAR_50, VAR_51, VAR_52, VAR_53, VAR_54, VAR_55, VAR_56, VAR_57, VAR_58, VAR_59, VAR_60, VAR_61, + VAR_62, VAR_63 +}; + +char mqtt_prefix[] PROGMEM = MQTT_PREFIX; + +#define S_FIRSTCUSTOM 60 +#define TYPEMAXLEN 20 +#define VAR_TOTAL (sizeof(VAR_Type)/sizeof(char *))-1 + + +#define EEPROM_LATEST_NODE_ADDRESS ((uint8_t)EEPROM_LOCAL_CONFIG_ADDRESS) + +//#define DSRTC // RTC Dallas support -> disable debug and use isp programmer to free memory +#endif diff --git a/libraries/PubSubClient/CHANGES.txt b/libraries/PubSubClient/CHANGES.txt new file mode 100755 index 0000000..8688011 --- /dev/null +++ b/libraries/PubSubClient/CHANGES.txt @@ -0,0 +1,47 @@ +1.9 + * Do not split MQTT packets over multiple calls to _client->write() + * API change: All constructors now require an instance of Client + to be passed in. + * Fixed example to match 1.8 api changes - dpslwk + * Added username/password support - WilHall + * Added publish_P - publishes messages from PROGMEM - jobytaffey + +1.8 + * KeepAlive interval is configurable in PubSubClient.h + * Maximum packet size is configurable in PubSubClient.h + * API change: Return boolean rather than int from various functions + * API change: Length parameter in message callback changed + from int to unsigned int + * Various internal tidy-ups around types +1.7 + * Improved keepalive handling + * Updated to the Arduino-1.0 API +1.6 + * Added the ability to publish a retained message + +1.5 + * Added default constructor + * Fixed compile error when used with arduino-0021 or later + +1.4 + * Fixed connection lost handling + +1.3 + * Fixed packet reading bug in PubSubClient.readPacket + +1.2 + * Fixed compile error when used with arduino-0016 or later + + +1.1 + * Reduced size of library + * Added support for Will messages + * Clarified licensing - see LICENSE.txt + + +1.0 + * Only Quality of Service (QOS) 0 messaging is supported + * The maximum message size, including header, is 128 bytes + * The keepalive interval is set to 30 seconds + * No support for Will messages + diff --git a/libraries/PubSubClient/LICENSE.txt b/libraries/PubSubClient/LICENSE.txt new file mode 100755 index 0000000..f512cba --- /dev/null +++ b/libraries/PubSubClient/LICENSE.txt @@ -0,0 +1,20 @@ +Copyright (c) 2008-2012 Nicholas O'Leary + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/libraries/PubSubClient/PubSubClient.cpp b/libraries/PubSubClient/PubSubClient.cpp new file mode 100755 index 0000000..b3f5c0e --- /dev/null +++ b/libraries/PubSubClient/PubSubClient.cpp @@ -0,0 +1,427 @@ +/* + PubSubClient.cpp - A simple client for MQTT. + Nicholas O'Leary + http://knolleary.net +*/ + +#include "PubSubClient.h" +#include + +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) { + this->_client = &client; + this->callback = callback; + 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) { + this->_client = &client; + 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) { + return connect(id,NULL,NULL,0,0,0,0); +} + +boolean PubSubClient::connect(char *id, char *user, char *pass) { + return connect(id,user,pass,0,0,0,0); +} + +boolean PubSubClient::connect(char *id, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) +{ + return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage); +} + +boolean PubSubClient::connect(char *id, char *user, char *pass, char* willTopic, uint8_t willQos, uint8_t willRetain, char* willMessage) { + if (!connected()) { + int result = 0; + + if (domain != NULL) { + result = _client->connect(this->domain, this->port); + } else { + result = _client->connect(this->ip, this->port); + } + + if (result) { + nextMsgId = 1; + uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p',MQTTPROTOCOLVERSION}; + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + unsigned int j; + for (j = 0;j<9;j++) { + buffer[length++] = d[j]; + } + + uint8_t v; + if (willTopic) { + v = 0x06|(willQos<<3)|(willRetain<<5); + } else { + v = 0x02; + } + + if(user != NULL) { + v = v|0x80; + + if(pass != NULL) { + v = v|(0x80>>1); + } + } + + buffer[length++] = v; + + buffer[length++] = ((MQTT_KEEPALIVE) >> 8); + buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF); + length = writeString(id,buffer,length); + if (willTopic) { + length = writeString(willTopic,buffer,length); + length = writeString(willMessage,buffer,length); + } + + if(user != NULL) { + length = writeString(user,buffer,length); + if(pass != NULL) { + length = writeString(pass,buffer,length); + } + } + + write(MQTTCONNECT,buffer,length-5); + + lastInActivity = lastOutActivity = millis(); + + while (!_client->available()) { + unsigned long t = millis(); + if (t-lastInActivity > MQTT_KEEPALIVE*1000UL) { + _client->stop(); + return false; + } + } + uint8_t llen; + uint16_t len = readPacket(&llen); + + if (len == 4 && buffer[3] == 0) { + lastInActivity = millis(); + pingOutstanding = false; + return true; + } + } + _client->stop(); + } + return false; +} + +uint8_t PubSubClient::readByte() { + while(!_client->available()) {} + return _client->read(); +} + +uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { + uint16_t len = 0; + buffer[len++] = readByte(); + bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; + uint32_t multiplier = 1; + uint16_t length = 0; + uint8_t digit = 0; + uint16_t skip = 0; + uint8_t start = 0; + + do { + digit = readByte(); + buffer[len++] = digit; + length += (digit & 127) * multiplier; + multiplier *= 128; + } while ((digit & 128) != 0); + *lengthLength = len-1; + + if (isPublish) { + // Read in topic length to calculate bytes to skip over for Stream writing + buffer[len++] = readByte(); + buffer[len++] = readByte(); + skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; + start = 2; + if (buffer[0]&MQTTQOS1) { + // skip message id + skip += 2; + } + } + + for (uint16_t i = start;istream) { + if (isPublish && len-*lengthLength-2>skip) { + this->stream->write(digit); + } + } + if (len < MQTT_MAX_PACKET_SIZE) { + buffer[len] = digit; + } + len++; + } + + if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { + len = 0; // This will cause the packet to be ignored. + } + + return len; +} + +boolean PubSubClient::loop() { + if (connected()) { + unsigned long t = millis(); + if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) { + if (pingOutstanding) { + _client->stop(); + return false; + } else { + buffer[0] = MQTTPINGREQ; + buffer[1] = 0; + _client->write(buffer,2); + lastOutActivity = t; + lastInActivity = t; + pingOutstanding = true; + } + } + if (_client->available()) { + uint8_t llen; + uint16_t len = readPacket(&llen); + uint16_t msgId = 0; + uint8_t *payload; + if (len > 0) { + lastInActivity = t; + uint8_t type = buffer[0]&0xF0; + if (type == MQTTPUBLISH) { + if (callback) { + uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; + char topic[tl+1]; + for (uint16_t i=0;i0 + if ((buffer[0]&0x06) == MQTTQOS1) { + msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1]; + payload = buffer+llen+3+tl+2; + callback(topic,payload,len-llen-3-tl-2); + + buffer[0] = MQTTPUBACK; + buffer[1] = 2; + buffer[2] = (msgId >> 8); + buffer[3] = (msgId & 0xFF); + _client->write(buffer,4); + lastOutActivity = t; + + } else { + payload = buffer+llen+3+tl; + callback(topic,payload,len-llen-3-tl); + } + } + } else if (type == MQTTPINGREQ) { + buffer[0] = MQTTPINGRESP; + buffer[1] = 0; + _client->write(buffer,2); + } else if (type == MQTTPINGRESP) { + pingOutstanding = false; + } + } + } + return true; + } + return false; +} + +boolean PubSubClient::publish(char* topic, char* payload) { + return publish(topic,(uint8_t*)payload,strlen(payload),false); +} + +boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plength) { + return publish(topic, payload, plength, false); +} + +boolean PubSubClient::publish(char* topic, uint8_t* payload, unsigned int plength, boolean retained) { + if (connected()) { + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + length = writeString(topic,buffer,length); + uint16_t i; + for (i=0;i 0) { + digit |= 0x80; + } + buffer[pos++] = digit; + llen++; + } while(len>0); + + pos = writeString(topic,buffer,pos); + + rc += _client->write(buffer,pos); + + for (i=0;iwrite((char)pgm_read_byte_near(payload + i)); + } + + lastOutActivity = millis(); + + return rc == tlen + 4 + plength; +} + +boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) { + uint8_t lenBuf[4]; + uint8_t llen = 0; + uint8_t digit; + uint8_t pos = 0; + uint8_t rc; + uint8_t len = length; + do { + digit = len % 128; + len = len / 128; + if (len > 0) { + digit |= 0x80; + } + lenBuf[pos++] = digit; + llen++; + } while(len>0); + + buf[4-llen] = header; + for (int i=0;iwrite(buf+(4-llen),length+1+llen); + + lastOutActivity = millis(); + return (rc == 1+llen+length); +} + +boolean PubSubClient::subscribe(char* topic) { + return subscribe(topic, 0); +} + +boolean PubSubClient::subscribe(char* topic, uint8_t qos) { + if (qos < 0 || qos > 1) + return false; + + if (connected()) { + // Leave room in the buffer for header and variable length field + uint16_t length = 5; + nextMsgId++; + if (nextMsgId == 0) { + nextMsgId = 1; + } + buffer[length++] = (nextMsgId >> 8); + buffer[length++] = (nextMsgId & 0xFF); + length = writeString(topic, buffer,length); + buffer[length++] = qos; + return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5); + } + return false; +} + +boolean PubSubClient::unsubscribe(char* topic) { + if (connected()) { + uint16_t length = 5; + nextMsgId++; + if (nextMsgId == 0) { + nextMsgId = 1; + } + buffer[length++] = (nextMsgId >> 8); + buffer[length++] = (nextMsgId & 0xFF); + length = writeString(topic, buffer,length); + return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5); + } + return false; +} + +void PubSubClient::disconnect() { + buffer[0] = MQTTDISCONNECT; + buffer[1] = 0; + _client->write(buffer,2); + _client->stop(); + lastInActivity = lastOutActivity = millis(); +} + +uint16_t PubSubClient::writeString(char* string, uint8_t* buf, uint16_t pos) { + char* idp = string; + uint16_t i = 0; + pos += 2; + while (*idp) { + buf[pos++] = *idp++; + i++; + } + buf[pos-i-2] = (i >> 8); + buf[pos-i-1] = (i & 0xFF); + return pos; +} + + +boolean PubSubClient::connected() { + boolean rc; + if (_client == NULL ) { + rc = false; + } else { + rc = (int)_client->connected(); + if (!rc) _client->stop(); + } + return rc; +} + diff --git a/libraries/PubSubClient/PubSubClient.h b/libraries/PubSubClient/PubSubClient.h new file mode 100755 index 0000000..f2faff8 --- /dev/null +++ b/libraries/PubSubClient/PubSubClient.h @@ -0,0 +1,81 @@ +/* + PubSubClient.h - A simple client for MQTT. + Nicholas O'Leary + http://knolleary.net +*/ + +#ifndef PubSubClient_h +#define PubSubClient_h + +#include +#include "Client.h" +#include "Stream.h" + +// MQTT_MAX_PACKET_SIZE : Maximum packet size +#define MQTT_MAX_PACKET_SIZE 128 + +// MQTT_KEEPALIVE : keepAlive interval in Seconds +#define MQTT_KEEPALIVE 15 + +#define MQTTPROTOCOLVERSION 3 +#define MQTTCONNECT 1 << 4 // Client request to connect to Server +#define MQTTCONNACK 2 << 4 // Connect Acknowledgment +#define MQTTPUBLISH 3 << 4 // Publish message +#define MQTTPUBACK 4 << 4 // Publish Acknowledgment +#define MQTTPUBREC 5 << 4 // Publish Received (assured delivery part 1) +#define MQTTPUBREL 6 << 4 // Publish Release (assured delivery part 2) +#define MQTTPUBCOMP 7 << 4 // Publish Complete (assured delivery part 3) +#define MQTTSUBSCRIBE 8 << 4 // Client Subscribe request +#define MQTTSUBACK 9 << 4 // Subscribe Acknowledgment +#define MQTTUNSUBSCRIBE 10 << 4 // Client Unsubscribe request +#define MQTTUNSUBACK 11 << 4 // Unsubscribe Acknowledgment +#define MQTTPINGREQ 12 << 4 // PING Request +#define MQTTPINGRESP 13 << 4 // PING Response +#define MQTTDISCONNECT 14 << 4 // Client is Disconnecting +#define MQTTReserved 15 << 4 // Reserved + +#define MQTTQOS0 (0 << 1) +#define MQTTQOS1 (1 << 1) +#define MQTTQOS2 (2 << 1) + +class PubSubClient { +private: + Client* _client; + uint8_t buffer[MQTT_MAX_PACKET_SIZE]; + uint16_t nextMsgId; + unsigned long lastOutActivity; + unsigned long lastInActivity; + bool pingOutstanding; + void (*callback)(char*,uint8_t*,unsigned int); + uint16_t readPacket(uint8_t*); + uint8_t readByte(); + boolean write(uint8_t header, uint8_t* buf, uint16_t length); + uint16_t writeString(char* string, uint8_t* buf, uint16_t pos); + 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 *); + boolean connect(char *, char *, char *, char *, uint8_t, uint8_t, char*); + void disconnect(); + boolean publish(char *, char *); + boolean publish(char *, uint8_t *, unsigned int); + boolean publish(char *, uint8_t *, unsigned int, boolean); + boolean publish_P(char *, uint8_t *, unsigned int, boolean); + boolean subscribe(char *); + boolean subscribe(char *, uint8_t qos); + boolean unsubscribe(char *); + boolean loop(); + boolean connected(); +}; + + +#endif diff --git a/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino b/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino new file mode 100755 index 0000000..c8a519a --- /dev/null +++ b/libraries/PubSubClient/examples/mqtt_auth/mqtt_auth.ino @@ -0,0 +1,39 @@ +/* + Basic MQTT example with Authentication + + - connects to an MQTT server, providing username + and password + - publishes "hello world" to the topic "outTopic" + - subscribes to the topic "inTopic" +*/ + +#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 }; + +void callback(char* topic, byte* payload, unsigned int length) { + // handle message arrived +} + +EthernetClient ethClient; +PubSubClient client(server, 1883, callback, ethClient); + +void setup() +{ + Ethernet.begin(mac, ip); + if (client.connect("arduinoClient", "testuser", "testpass")) { + client.publish("outTopic","hello world"); + client.subscribe("inTopic"); + } +} + +void loop() +{ + client.loop(); +} + diff --git a/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino b/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino new file mode 100755 index 0000000..72276cd --- /dev/null +++ b/libraries/PubSubClient/examples/mqtt_basic/mqtt_basic.ino @@ -0,0 +1,38 @@ +/* + Basic MQTT example + + - connects to an MQTT server + - publishes "hello world" to the topic "outTopic" + - subscribes to the topic "inTopic" +*/ + +#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 }; + +void callback(char* topic, byte* payload, unsigned int length) { + // handle message arrived +} + +EthernetClient ethClient; +PubSubClient client(server, 1883, callback, ethClient); + +void setup() +{ + Ethernet.begin(mac, ip); + if (client.connect("arduinoClient")) { + client.publish("outTopic","hello world"); + client.subscribe("inTopic"); + } +} + +void loop() +{ + client.loop(); +} + diff --git a/libraries/PubSubClient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino b/libraries/PubSubClient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino new file mode 100755 index 0000000..ece5c2e --- /dev/null +++ b/libraries/PubSubClient/examples/mqtt_publish_in_callback/mqtt_publish_in_callback.ino @@ -0,0 +1,61 @@ +/* + Publishing in the callback + + - connects to an MQTT server + - subscribes to the topic "inTopic" + - when a message is received, republishes it to "outTopic" + + This example shows how to publish messages within the + callback function. The callback function header needs to + be declared before the PubSubClient constructor and the + actual callback defined afterwards. + This ensures the client reference in the callback function + is valid. + +*/ + +#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 }; + +// Callback function header +void callback(char* topic, byte* payload, unsigned int length); + +EthernetClient ethClient; +PubSubClient client(server, 1883, callback, ethClient); + +// Callback function +void callback(char* topic, byte* payload, unsigned int length) { + // In order to republish this payload, a copy must be made + // as the orignal payload buffer will be overwritten whilst + // constructing the PUBLISH packet. + + // Allocate the correct amount of memory for the payload copy + byte* p = (byte*)malloc(length); + // Copy the payload to the new buffer + memcpy(p,payload,length); + client.publish("outTopic", p, length); + // Free the memory + free(p); +} + +void setup() +{ + + Ethernet.begin(mac, ip); + if (client.connect("arduinoClient")) { + client.publish("outTopic","hello world"); + client.subscribe("inTopic"); + } +} + +void loop() +{ + client.loop(); +} + diff --git a/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino b/libraries/PubSubClient/examples/mqtt_stream/mqtt_stream.ino new file mode 100755 index 0000000..a416ee2 --- /dev/null +++ b/libraries/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