On Node-Red and MQTT
Dependencies: mbed ESP8266Interface MbedJSONValue MQTT JSON
Revision 22:518ae39b6d77, committed 2019-05-02
- Comitter:
- sbetrabet
- Date:
- Thu May 02 06:05:52 2019 +0000
- Parent:
- 21:f08f17225c7c
- Commit message:
- On Node-Red and MQTT with K64
Changed in this revision
diff -r f08f17225c7c -r 518ae39b6d77 Linux-example/LinuxIPStack.h --- a/Linux-example/LinuxIPStack.h Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,119 +0,0 @@ -#if !defined(LINUX_IPSTACK_H) -#define LINUX_IPSTACK_H - -class IPStack -{ -public: - IPStack() - { - - } - - int Socket_error(const char* aString) - { - - if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) - { - if (strcmp(aString, "shutdown") != 0 || (errno != ENOTCONN && errno != ECONNRESET)) - printf("Socket error %s in %s for socket %d\n", strerror(errno), aString, mysock); - } - return errno; - } - - int connect(const char* hostname, int port) - { - int type = SOCK_STREAM; - struct sockaddr_in address; - int rc = -1; - sa_family_t family = AF_INET; - struct addrinfo *result = NULL; - struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL}; - - if ((rc = getaddrinfo(hostname, NULL, &hints, &result)) == 0) - { - struct addrinfo* res = result; - - /* prefer ip4 addresses */ - while (res) - { - if (res->ai_family == AF_INET) - { - result = res; - break; - } - res = res->ai_next; - } - - if (result->ai_family == AF_INET) - { - address.sin_port = htons(port); - address.sin_family = family = AF_INET; - address.sin_addr = ((struct sockaddr_in*)(result->ai_addr))->sin_addr; - } - else - rc = -1; - - freeaddrinfo(result); - } - - if (rc == 0) - { - mysock = socket(family, type, 0); - if (mysock != -1) - { - int opt = 1; - - //if (setsockopt(mysock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0) - // printf("Could not set SO_NOSIGPIPE for socket %d", mysock); - - rc = ::connect(mysock, (struct sockaddr*)&address, sizeof(address)); - } - } - - return rc; - } - - int read(char* buffer, int len, int timeout_ms) - { - struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000}; - if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0)) - { - interval.tv_sec = 0; - interval.tv_usec = 100; - } - - setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval)); - - //printf("reading %d bytes\n", len); - int rc = ::recv(mysock, buffer, (size_t)len, 0); - if (rc == -1) - Socket_error("read"); - //printf("read %d bytes\n", rc); - return rc; - } - - int write(char* buffer, int len, int timeout) - { - struct timeval tv; - - tv.tv_sec = 0; /* 30 Secs Timeout */ - tv.tv_usec = timeout * 1000; // Not init'ing this can cause strange errors - - setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv,sizeof(struct timeval)); - int rc = ::write(mysock, buffer, len); - //printf("write rc %d\n", rc); - return rc; - } - - int disconnect() - { - return ::close(mysock); - } - -private: - - int mysock; - -}; - -#endif \ No newline at end of file
diff -r f08f17225c7c -r 518ae39b6d77 Linux-example/LinuxMQTT.h --- a/Linux-example/LinuxMQTT.h Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,64 +0,0 @@ -#if !defined(LINUXMQTT_H) -#define LINUXMQTT_H - -class Countdown -{ -public: - Countdown() - { - - } - - Countdown(int ms) - { - countdown_ms(ms); - } - - - bool expired() - { - struct timeval now, res; - gettimeofday(&now, NULL); - timersub(&end_time, &now, &res); - //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); - //if (res.tv_sec > 0 || res.tv_usec > 0) - // printf("expired %d %d\n", res.tv_sec, res.tv_usec); - return res.tv_sec < 0 || (res.tv_sec == 0 && res.tv_usec <= 0); - } - - - void countdown_ms(int ms) - { - struct timeval now; - gettimeofday(&now, NULL); - struct timeval interval = {ms / 1000, (ms % 1000) * 1000}; - //printf("interval %d %d\n", interval.tv_sec, interval.tv_usec); - timeradd(&now, &interval, &end_time); - } - - - void countdown(int seconds) - { - struct timeval now; - gettimeofday(&now, NULL); - struct timeval interval = {seconds, 0}; - timeradd(&now, &interval, &end_time); - } - - - int left_ms() - { - struct timeval now, res; - gettimeofday(&now, NULL); - timersub(&end_time, &now, &res); - //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); - return (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000; - } - -private: - - struct timeval end_time; -}; - - -#endif \ No newline at end of file
diff -r f08f17225c7c -r 518ae39b6d77 Linux-example/linux-main.cpp --- a/Linux-example/linux-main.cpp Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,149 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2014 IBM Corp. - * - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Eclipse Distribution License v1.0 which accompany this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * and the Eclipse Distribution License is available at - * http://www.eclipse.org/org/documents/edl-v10.php. - * - * Contributors: - * Ian Craggs - initial API and implementation and/or initial documentation - *******************************************************************************/ - - /** - This is a sample program to illustrate the use of the MQTT Client library - on Linux. The Client class requires two classes which mediate - access to system interfaces for networking and timing. As long as these two - classes provide the required public programming interfaces, it does not matter - what facilities they use underneath. In this program, they use the Linux - system libraries. - - */ - -#if defined(LINUX) - -#include "LinuxMQTT.h" -#include "LinuxIPStack.h" -#include "MQTTClient.h" - -#include <sys/types.h> -#include <sys/socket.h> -#include <sys/param.h> -#include <sys/time.h> -#include <sys/select.h> -#include <netinet/in.h> -#include <netinet/tcp.h> -#include <arpa/inet.h> -#include <netdb.h> -#include <stdio.h> -#include <unistd.h> -#include <errno.h> -#include <fcntl.h> - -#include <stdlib.h> -#include <string.h> -#include <signal.h> - -#define DEFAULT_STACK_SIZE -1 - - -int arrivedcount = 0; - -void messageArrived(MQTT::Message* message) -{ - printf("Message %d arrived: qos %d, retained %d, dup %d, packetid %d\n", - ++arrivedcount, message->qos, message->retained, message->dup, message->id); - printf("Payload %.*s\n", message->payloadlen, (char*)message->payload); -} - - -int connect(MQTT::Client<IPStack, Countdown>::connectionLostInfo* info) -{ - const char* hostname = "localhost"; //"m2m.eclipse.org"; - int port = 1883; - printf("Connecting to %s:%d\n", hostname, port); - int rc = info->network->connect(hostname, port); - if (rc != 0) - printf("rc from TCP connect is %d\n", rc); - - MQTTPacket_connectData data = MQTTPacket_connectData_initializer; - data.MQTTVersion = 3; - data.clientID.cstring = (char*)"mbed-icraggs"; - rc = info->client->connect(&data); - if (rc != 0) - printf("rc from MQTT connect is %d\n", rc); - - return rc; -} - - -int main(int argc, char* argv[]) -{ - IPStack ipstack = IPStack(); - float version = 0.3; - const char* topic = "mbed-sample"; - - printf("Version is %f\n", version); - - MQTT::Client<IPStack, Countdown> client = MQTT::Client<IPStack, Countdown>(ipstack); - - client.setConnectionLostHandler(connect); - - MQTT::Client<IPStack, Countdown>::connectionLostInfo info = {&client, &ipstack}; - int rc = connect(&info); - - rc = client.subscribe(topic, MQTT::QOS2, messageArrived); - if (rc != 0) - printf("rc from MQTT subscribe is %d\n", rc); - - MQTT::Message message; - - // QoS 0 - char buf[100]; - sprintf(buf, "Hello World! QoS 0 message from app version %f", version); - message.qos = MQTT::QOS0; - message.retained = false; - message.dup = false; - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, &message); - while (arrivedcount == 0) - client.yield(100); - - // QoS 1 - printf("Now QoS 1\n"); - sprintf(buf, "Hello World! QoS 1 message from app version %f", version); - message.qos = MQTT::QOS1; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, &message); - while (arrivedcount == 1) - client.yield(100); - - // QoS 2 - sprintf(buf, "Hello World! QoS 2 message from app version %f", version); - message.qos = MQTT::QOS2; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, &message); - while (arrivedcount == 2) - client.yield(100); - - rc = client.unsubscribe(topic); - if (rc != 0) - printf("rc from unsubscribe was %d\n", rc); - - rc = client.disconnect(); - if (rc != 0) - printf("rc from disconnect was %d\n", rc); - - ipstack.disconnect(); - - printf("Finishing with %d messages received\n", arrivedcount); - - return 0; -} - -#endif
diff -r f08f17225c7c -r 518ae39b6d77 main.cpp --- a/main.cpp Fri Mar 08 07:05:46 2019 +0000 +++ b/main.cpp Thu May 02 06:05:52 2019 +0000 @@ -1,228 +1,55 @@ #include "MQTTESP8266.h" #include "MQTTClient.h" #include "string.h" -#include "rgbled.h" #include "MbedJSONValue.h" char buf[100]; - -MQTTESP8266 ipstack(D1, D0, D10, "Sidsap10","Arduino111"); // change to match your wifi access point +//MQTTESP8266 ipstack(D1, D0, D10, "Sidsap10","Arduino111"); // change to match your wifi access point +MQTTESP8266 ipstack(D1, D0, D10, "BHNTG1682GFFE2","220960b5"); // change to match your wifi access point float version = 0.47; -char* topic = "k64"; -char* topic2 = "k64rec"; -char* topic3 = "bulb"; -char* key2 = "key"; -char* hostname = "192.168.43.87"; // Ip +char* topic = "toast"; +char* topic2 = "bread"; +char* hostname = "52.28.141.81";//"192.168.0.9";//";"192.168.43.87"; // Ip int port = 1883; int rc = ipstack.connect(hostname, port); -unsigned int previousTime = 0; // will store last time LED was updated - -// constants won't change: -const long interval = 20; -int flag =0; +AnalogIn ain(A0); MQTT::Client<MQTTESP8266, Countdown> client = MQTT::Client<MQTTESP8266, Countdown>(ipstack); //MbedJSONValue demo; -int arrivedcount = 0; //using namespace std; -std::string my_str; -std::string my_str2; std::string someString; -std::string clearsky ("clear"); -std::string clouds ("few clouds"); -std::string snow ("light snow"); -std::string automatic ("automatic"); -std::string manual ("manual"); -int j=0; -rgbled rgb(PTB22, PTE26, PTB21); // create rgb object with pin definitions for FRDM-K64F board +std::string substrings ("senddata"); // callback for subscribe topic void subscribeCallback(MQTT::MessageData& md) { - MQTT::Message &message = md.message; - printf("Message received: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id); - - printf("%s",(char*)message.payload); - printf("\r\n"); - - printf("Payload %.*s\n", message.payloadlen, (char*)message.payload); - printf("nonedited payload is %s and payload is %d\r\n",(char*)message.payload,message.payloadlen); - - // int f = int(message.payload); - + MQTT::Message &message = md.message; std::string someString((char*)message.payload); someString.resize (message.payloadlen); - printf("Edited payload iss %s \r\n",someString.c_str()); - //j= someString.find(clouds); - //printf("j = %d \r\n", j); - - if ( !(someString.find(clouds)) ) - { - sprintf(buf, "dCglDR-G9WoQUiig0XMNhOV4ozIEIVu3rWdBh5IlQ7d,cloud %s\n", key2); - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic3, message); - printf("if condition checked, its cloudy\r\n"); - //rgb.set(red); - - } - else if ( !(someString.find(clearsky)) ) - { - sprintf(buf, "dCglDR-G9WoQUiig0XMNhOV4ozIEIVu3rWdBh5IlQ7d,clearsky %s\r\n", key2); + if ( !(someString.find(substrings)) ) + { sprintf(buf, "%6.4lf", ain.read()*100.0f); message.payload = (void*)buf; message.payloadlen = strlen(buf)+1; - rc = client.publish(topic3, message); - printf("if condition checked, its clear\r\n"); - - //rgb.set(yellow); - } - else if ( !(someString.find(snow)) ) - { - - sprintf(buf, "dCglDR-G9WoQUiig0XMNhOV4ozIEIVu3rWdBh5IlQ7d,snow %s\r\n",key2); - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic3, message); - printf("if condition checked, its snowy\r\n"); - - // rgb.set(blue); + rc = client.publish(topic, message); + printf("Publishing data to MQTT \r\n"); } - else if ( !(someString.find(automatic)) ) - { - - printf("Automatic, i.e Decide based on weather \r\n"); - flag = 0; - // rgb.set(blue); - } - else if ( !(someString.find(manual)) ) - { - - printf("Manual Flag set \r\n"); - flag = 1; - // rgb.set(blue); - } - else - { - printf("went into else\r\n"); - } - } - int main(int argc, char* argv[]) { wait (1); - printf("Starting\r\n"); - - set_time(1256729737); - time_t seconds = time(NULL); - previousTime = (unsigned int)seconds; - - + printf("Starting\r\n"); printf("Version is %f\r\n", version); - - rgb.active(false); // set led output to active low - rgb.set(none); // set led output of all off - - - if (rc != 0) printf("rc from TCP connect is %d\r\n", rc); - + if (rc == 0) + printf("K64 is online \r\n"); MQTTPacket_connectData data = MQTTPacket_connectData_initializer; - //data.MQTTVersion = 3; - //data.clientID.cstring = "mbed-clientID"; - //data.username.cstring = "testuser"; - //data.password.cstring = "testpassword"; - if ((rc = client.connect(data)) != 0) printf("rc from MQTT connect is %d\r\n", rc); - if ((rc = client.subscribe(topic2, MQTT::QOS1, subscribeCallback)) != 0) printf("Recv'd from MQTT subscribe is %d\r\n", rc); - - MQTT::Message message; // QoS 0 - - - sprintf(buf, "fa02d31610ce9ea0de15b22bb2fe279d, %s\r\n", key2); - message.qos = MQTT::QOS0; - message.retained = false; - message.dup = false; - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - - - rc = client.publish(topic, message); - wait(5); - - /* - sprintf(buf, "dCglDR-G9WoQUiig0XMNhOV4ozIEIVu3rWdBh5IlQ7d, %s\r\n", key2); - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic2, message); - */ - - while (arrivedcount < 10) - { client.yield(100); - - time_t seconds = time(NULL); - //printf("%u\r\n", (unsigned int)seconds); - wait(1); - if ( seconds - previousTime >= interval ) - { - if (flag ==0) - { - //printf("its been a few seconds \r\n"); - printf("Checking the weather\r\n"); - previousTime =seconds; - sprintf(buf, "fa02d31610ce9ea0de15b22bb2fe279d, %s\r\n", key2); - message.qos = MQTT::QOS0; - message.retained = false; - message.dup = false; - message.payload = (void*)buf; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, message); - } - } + while (true) + { + client.yield(100); } - - - /* - // QoS 1 - sprintf(buf, "Hello World! QoS 1 message from app version %f\r\n", version); - message.qos = MQTT::QOS1; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, message); - while (arrivedcount < 10) - client.yield(100); - - // QoS 2 - sprintf(buf, "Hello World! QoS 2 message from app version %f\r\n", version); - message.qos = MQTT::QOS2; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, message); - while (arrivedcount < 3) - client.yield(100); - - // n * QoS 2 - - for (int i = 1; i <= 10; ++i) { - sprintf(buf, "Hello World! QoS 2 message number %d from app version %f\r\n", i, version); - message.qos = MQTT::QOS2; - message.payloadlen = strlen(buf)+1; - rc = client.publish(topic, message); - while (arrivedcount < i + 3) - client.yield(100); - } - */ - - if ((rc = client.unsubscribe(topic)) != 0) - printf("rc from unsubscribe was %d\r\n", rc); - - if ((rc = client.disconnect()) != 0) - printf("rc from disconnect was %d\r\n", rc); - - //ipstack.disconnect(); - //printf("Finishing with %d messages received\r\n", arrivedcount); - - return 0; }
diff -r f08f17225c7c -r 518ae39b6d77 mbed-rtos.lib --- a/mbed-rtos.lib Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,1 +0,0 @@ -http://mbed.org/users/mbed_official/code/mbed-rtos/#e45e4ac7c3c8
diff -r f08f17225c7c -r 518ae39b6d77 rgbled.cpp --- a/rgbled.cpp Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,84 +0,0 @@ -#include "mbed.h" -#include "rgbled.h" - -rgbled::rgbled(PinName pin_red, PinName pin_green, PinName pin_blue) - : _pin_red(pin_red), _pin_green(pin_green), _pin_blue(pin_blue){ - _on = false; - _off = !_on; - _pin_red = _off; - _pin_green = _off; - _pin_blue = _off; -} - -void rgbled::active(bool a){ - _on = a; - _off = !_on; -} - -void rgbled::_none(void){ - _pin_red = _off; - _pin_green = _off; - _pin_blue = _off; -} - -void rgbled::_red(void){ - _pin_red = _on; - _pin_green = _off; - _pin_blue = _off; -} - -void rgbled::_green(void){ - _pin_red = _off; - _pin_green = _on; - _pin_blue = _off; -} - -void rgbled::_blue(void){ - _pin_red = _off; - _pin_green = _off; - _pin_blue = _on; -} - -void rgbled::_cyan(void){ - _pin_red = _off; - _pin_green = _on; - _pin_blue = _on; -} - -void rgbled::_yellow(void){ - _pin_red = _on; - _pin_green = _on; - _pin_blue = _off; -} - -void rgbled::_magenta(void){ - _pin_red = _on; - _pin_green = _off; - _pin_blue = _on; -} - -void rgbled::set(color c){ - switch (c) { - case none : - _none(); - break; - case red : - _red(); - break; - case yellow : - _yellow(); - break; - case green : - _green(); - break; - case cyan: - _cyan(); - break; - case blue : - _blue(); - break; - case magenta : - _magenta(); - break; - } -}
diff -r f08f17225c7c -r 518ae39b6d77 rgbled.h --- a/rgbled.h Fri Mar 08 07:05:46 2019 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,31 +0,0 @@ -#ifndef __RGBLED_H__ -#define __RGBLED_H__ - -// includes -#include "mbed.h" - -// defines -enum color {none, red, yellow, green, cyan, blue, magenta}; - -// rgb led -class rgbled { -private: - DigitalOut _pin_red; - DigitalOut _pin_green; - DigitalOut _pin_blue; - bool _on; - bool _off; - void _none(void); - void _red(void); - void _yellow(void); - void _green(void); - void _cyan(void); - void _blue(void); - void _magenta(void); -public: - rgbled(PinName pin_red, PinName pin_green, PinName pin_blue); - void active(bool a); - void set(color c); -}; - -#endif \ No newline at end of file