mgillespie/TinyMqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #6 from hsaturn/AsyncAndWifi

Browse Source 
AsyncTcp can be activated by removing the command on TCP_ASYNC in TinyMqtt.h
But the code is not bug free yet.
This commit is contained in:
hsaturn
2021-04-11 23:33:29 +02:00
committed by GitHub
parent 1cfb5cfab1 d3210c3c93
commit ce896f02c4
4 changed files with 499 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
examples/client-with-wifi/client-with-wifi.ino
View File

@@ -1,7 +1,21 @@
#include <TinyMqtt.h> // https://github.com/hsaturn/TinyMqtt
/**
* Local broker that accept connections and two local clients
*
* +-----------------------------+
* | ESP |
* | +--------+ | 1883 <--- External client/s
* | +-------->| broker | | 1883 <--- External client/s
* | | +--------+ |
* | | ^ |
* | | | |
* | v v |
* | +----------+ +----------+ |
* | | internal | | internal | |
* | | client | | client | |
* | +----------+ +----------+ |
* | |
* +-----------------------------+
*
* pros - Reduces internal latency (when publish is received by the same ESP)
* - Reduces wifi traffic
@@ -12,11 +26,6 @@
* cons - Takes more memory
* - a bit hard to understand
*
* This sounds crazy: a mqtt mqtt that do not need a broker !
* The use case arise when one ESP wants to publish topics and subscribe to them at the same time.
* Without broker, the ESP won't react to its own topics.
*
* TinyMqtt mqtt allows this use case to work.
*/
#include <my_credentials.h>
@@ -58,7 +67,7 @@ void setup()
void loop()
{
broker.loop();
broker.loop(); // Don't forget to add loop for every broker and clients
mqtt_a.loop();
mqtt_b.loop();

224
examples/tinymqtt-test/tinymqtt-test.ino
View File

@@ -138,7 +138,7 @@ std::set<std::string> commands = {
"auto", "broker", "blink", "client", "connect",
"create", "delete", "help", "interval",
"ls", "ip", "off", "on", "set",
"publish", "reset", "subscribe", "unsubscribe", "view"
"publish", "reset", "subscribe", "unsubscribe", "view", "every"
};
void getCommand(std::string& search)
@@ -316,83 +316,46 @@ std::map<MqttClient*, automatic*> automatic::autos;
bool compare(std::string s, const char* cmd)
{
if (s.length()==0 or s.length()>strlen(cmd)) return false;
return strncmp(cmd, s.c_str(), s.length())==0;
uint8_t p=0;
while(s[p++]==*cmd++)
{
if (*cmd==0 or s[p]==0) return true;
if (s[p]==' ') return true;
}
return false;
}
using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
struct Every
{
std::string cmd;
uint32_t ms;
uint32_t next;
void dump()
{
auto mill=millis();
Serial << ms << "ms [" << cmd << "] next in ";
if (mill > next)
Serial << "now";
else
Serial << next-mill << "ms";
}
};
uint32_t blink_ms_on[16];
uint32_t blink_ms_off[16];
uint32_t blink_next[16];
bool blink_state[16];
int16_t blink;
void loop()
std::vector<Every> everies;
void eval(std::string& cmd)
{
auto ms=millis();
int8_t out=0;
int16_t blink_bits = blink;
while(blink_bits)
{
if (blink_ms_on[out] and ms > blink_next[out])
{
if (blink_state[out])
{
blink_next[out] += blink_ms_on[out];
digitalWrite(out, LOW);
}
else
{
blink_next[out] += blink_ms_off[out];
digitalWrite(abs(out), HIGH);
}
blink_state[out] = not blink_state[out];
}
blink_bits >>=1;
out++;
}
static long count;
MDNS.update();
if (MqttClient::counter != count)
{
Serial << "# " << MqttClient::counter << endl;
count = MqttClient::counter;
}
for(auto it: brokers)
it.second->loop();
for(auto it: clients)
it.second->loop();
automatic::loop();
if (Serial.available())
{
static std::string cmd;
char c=Serial.read();
if (c==10 or c==14)
{
Serial << "----------------[ " << cmd.c_str() << " ]--------------" << endl;
static std::string last_cmd;
if (cmd=="!")
cmd=last_cmd;
else
last_cmd=cmd;
if (cmd.substr(0,3)!="set") replaceVars(cmd);
while(cmd.length())
{
std::string cmd_end;
if (cmd.find(';') != std::string::npos)
{
cmd_end = cmd;
cmd = getword(cmd_end,"",';');
}
MqttError retval = MqttOk;
std::string s;
@@ -508,6 +471,60 @@ void loop()
client->dump();
}
}
else if (compare(s, "on"))
{
uint8_t pin=getint(cmd, 2);
pinMode(pin, OUTPUT);
digitalWrite(pin, HIGH);
}
else if (compare(s, "off"))
{
uint8_t pin=getint(cmd, 2);
pinMode(pin, OUTPUT);
digitalWrite(pin, LOW);
}
else if (compare(s, "every"))
{
uint32_t ms = getint(cmd, 0);
if (ms and cmd.length())
{
Every every;
every.ms=ms;
every.cmd=cmd;
every.next=millis()+ms;
everies.push_back(every);
every.dump();
Serial << endl;
cmd="";
}
else if (ms==0 and compare(cmd, "list"))
{
getword(cmd);
Serial << "List of everies (ms=" << millis() << ")" << endl;
uint8_t count=0;
for(auto& every: everies)
{
Serial << count << ": ";
every.dump();
Serial << endl;
count++;
}
}
else if (ms==0 and compare(cmd, "remove"))
{
getword(cmd);
int8_t every=getint(cmd, -1);
if (every==-1 and compare(cmd, "all"))
{
getword(cmd);
everies.clear();
}
else if (everies.size() > (uint8_t)every)
{
everies.erase(everies.begin()+every);
}
}
}
else if (compare(s, "blink"))
{
int8_t blink_nr = getint(cmd, -1);
@@ -626,7 +643,7 @@ void loop()
Serial << "IP: " << WiFi.localIP() << endl;
else if (compare(s,"help"))
{
Serial << "syntax: instr; instr; ..." << endl;
Serial << "syntax:" << endl;
Serial << " MqttBroker:" << endl;
Serial << " broker {name} {port} : create a new broker" << endl;
Serial << endl;
@@ -646,6 +663,8 @@ void loop()
Serial << " set [name][value]" << endl;
Serial << " ! repeat last command" << endl;
Serial << endl;
Serial << " every ms [command]; every list; every remove [nr|all]" << endl;
Serial << " on {output}; off {output}" << endl;
Serial << " $id : name of the client." << endl;
Serial << " default topic is '" << topic.c_str() << "'" << endl;
Serial << endl;
@@ -661,12 +680,77 @@ void loop()
{
Serial << "## ERROR " << retval << endl;
}
if (cmd_end.length())
}
}
void loop()
{
auto ms=millis();
int8_t out=0;
int16_t blink_bits = blink;
for(auto& every: everies)
{
cmd = cmd_end;
cmd_end = "";
if (every.ms && every.cmd.length() && ms > every.next)
{
std::string cmd(every.cmd);
eval(cmd);
every.next += every.ms;
}
}
while(blink_bits)
{
if (blink_ms_on[out] and ms > blink_next[out])
{
if (blink_state[out])
{
blink_next[out] += blink_ms_on[out];
digitalWrite(out, LOW);
}
else
{
blink_next[out] += blink_ms_off[out];
digitalWrite(abs(out), HIGH);
}
blink_state[out] = not blink_state[out];
}
blink_bits >>=1;
out++;
}
static long count;
MDNS.update();
if (MqttClient::counter != count)
{
Serial << "# " << MqttClient::counter << endl;
count = MqttClient::counter;
}
for(auto it: brokers)
it.second->loop();
for(auto it: clients)
it.second->loop();
automatic::loop();
if (Serial.available())
{
static std::string cmd;
char c=Serial.read();
if (c==10 or c==14)
{
Serial << "----------------[ " << cmd.c_str() << " ]--------------" << endl;
static std::string last_cmd;
if (cmd=="!")
cmd=last_cmd;
else
last_cmd=cmd;
if (cmd.substr(0,3)!="set") replaceVars(cmd);
eval(cmd);
}
else
{

60
src/TinyMqtt.cpp
View File

@@ -11,8 +11,10 @@ void outstring(const char* prefix, const char*p, uint16_t len)
MqttBroker::MqttBroker(uint16_t port)
{
server = new AsyncServer(port);
server = new TcpServer(port);
#ifdef TCP_ASYNC
server->onClient(onClient, this);
#endif
}
MqttBroker::~MqttBroker()
@@ -25,12 +27,17 @@ MqttBroker::~MqttBroker()
}
// private constructor used by broker only
MqttClient::MqttClient(MqttBroker* parent, AsyncClient* new_client)
: parent(parent), client(new_client)
MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client)
: parent(parent)
{
#ifdef TCP_ASYNC
client = new_client;
client->onData(onData, this);
// client->onConnect() TODO
// client->onDisconnect() TODO
#else
client = new WiFiClient(*new_client);
#endif
alive = millis()+5000; // client expires after 5s if no CONNECT msg
}
@@ -75,11 +82,19 @@ void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
keep_alive = ka;
close();
if (client) delete client;
client = new AsyncClient;
client = new TcpClient;
debug("Trying to connect to " << broker.c_str() << ':' << port);
#ifdef TCP_ASYNC
client->onData(onData, this);
client->onConnect(onConnect, this);
debug("Trying to connect to " << broker.c_str() << ':' << port);
client->connect(broker.c_str(), port);
#else
if (client->connect(broker.c_str(), port))
{
onConnect(this, client);
}
#endif
}
void MqttBroker::addClient(MqttClient* client)
@@ -115,16 +130,24 @@ void MqttBroker::removeClient(MqttClient* remove)
debug("Error cannot remove client"); // TODO should not occur
}
void MqttBroker::onClient(void* broker_ptr, AsyncClient* client)
void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
{
MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
broker->addClient(new MqttClient(broker, client));
debug("New client #" << broker->clients.size());
debug("New client");
}
void MqttBroker::loop()
{
#ifndef TCP_ASYNC
WiFiClient client = server->available();
if (client)
{
onClient(this, &client);
}
#endif
if (broker)
{
// TODO should monitor broker's activity.
@@ -144,7 +167,7 @@ void MqttBroker::loop()
}
else
{
debug("Client " << client->id().c_str() << " Disconnected, parent=" << (int32_t)client->parent);
debug("Client " << client->id().c_str() << " Disconnected, parent=" << (dbg_ptr)client->parent);
// Note: deleting a client not added by the broker itself will probably crash later.
delete client;
break;
@@ -248,9 +271,20 @@ void MqttClient::loop()
// there is no need to send one PingReq per instance.
}
}
#ifndef TCP_ASYNC
while(client && client->available()>0)
{
message.incoming(client->read());
if (message.type())
{
processMessage(&message);
message.reset();
}
}
#endif
}
void MqttClient::onConnect(void *mqttclient_ptr, AsyncClient*)
void MqttClient::onConnect(void *mqttclient_ptr, TcpClient*)
{
MqttClient* mqtt = static_cast<MqttClient*>(mqttclient_ptr);
debug("cnx: connecting");
@@ -265,12 +299,13 @@ void MqttClient::onConnect(void *mqttclient_ptr, AsyncClient*)
debug("cnx: mqtt connecting");
msg.sendTo(mqtt);
msg.reset();
debug("cnx: mqtt sent " << (int32_t)mqtt->parent);
debug("cnx: mqtt sent " << (dbg_ptr)mqtt->parent);
mqtt->clientAlive(0);
}
void MqttClient::onData(void* client_ptr, AsyncClient*, void* data, size_t len)
#ifdef TCP_ASYNC
void MqttClient::onData(void* client_ptr, TcpClient*, void* data, size_t len)
{
char* char_ptr = static_cast<char*>(data);
MqttClient* client=static_cast<MqttClient*>(client_ptr);
@@ -285,6 +320,7 @@ void MqttClient::onData(void* client_ptr, AsyncClient*, void* data, size_t len)
len--;
}
}
#endif
void MqttClient::resubscribe()
{
@@ -361,7 +397,7 @@ void MqttClient::processMessage(const MqttMessage* mesg)
#ifdef TINY_MQTT_DEBUG
if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp)
{
Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
// mesg->hexdump("Incoming");
}
#endif

45
src/TinyMqtt.h
View File

@@ -1,14 +1,28 @@
#pragma once
#ifdef ESP8266
// TODO Should add a AUnit with both TCP_ASYNC and not TCP_ASYNC
// #define TCP_ASYNC // Uncomment this to use ESPAsyncTCP instead of normal cnx
#if defined(ESP8266) || defined(EPOXY_DUINO)
#ifdef TCP_ASYNC
#include <ESPAsyncTCP.h>
#else
#include <ESP8266WiFi.h>
#endif
#elif defined(ESP32)
#include <WiFi.h>
#ifdef TCP_ASYNC
#include <AsyncTCP.h> // https://github.com/me-no-dev/AsyncTCP
#elif defined(EPOXY_DUINO)
#include <ESPAsyncTCP.h>
#else
#include <WiFi.h>
#endif
#else
#error "Unsupported platform"
#endif
#ifdef EPOXY_DUINO
#define dbg_ptr uint64_t
#else
#define dbg_ptr uint32_t
#endif
#include <vector>
#include <set>
#include <string>
@@ -23,6 +37,14 @@
#define debug(what) {}
#endif
#ifdef TCP_ASYNC
using TcpClient = AsyncClient;
using TcpServer = AsyncServer;
#else
using TcpClient = WiFiClient;
using TcpServer = WiFiServer;
#endif
enum MqttError
{
MqttOk = 0,
@@ -192,14 +214,15 @@ class MqttClient
static long counter;
private:
static void onConnect(void * client_ptr, AsyncClient*);
static void onData(void* client_ptr, AsyncClient*, void* data, size_t len);
static void onConnect(void * client_ptr, TcpClient*);
#ifdef TCP_ASYNC
static void onData(void* client_ptr, TcpClient*, void* data, size_t len);
#endif
MqttError sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos);
void resubscribe();
friend class MqttBroker;
MqttClient(MqttBroker* parent, AsyncClient* client);
MqttClient(MqttBroker* parent, TcpClient* client);
// republish a received publish if topic matches any in subscriptions
MqttError publishIfSubscribed(const Topic& topic, const MqttMessage& msg);
@@ -217,7 +240,7 @@ class MqttClient
// (this is the case when MqttBroker isn't used except here)
MqttBroker* parent=nullptr; // connection to local broker
AsyncClient* client=nullptr; // connection to mqtt client or to remote broker
TcpClient* client=nullptr; // connection to mqtt client or to remote broker
std::set<Topic> subscriptions;
std::string clientId;
CallBack callback = nullptr;
@@ -257,7 +280,7 @@ class MqttBroker
private:
friend class MqttClient;
static void onClient(void*, AsyncClient*);
static void onClient(void*, TcpClient*);
bool checkUser(const char* user, uint8_t len) const
{ return compareString(auth_user, user, len); }
@@ -275,7 +298,7 @@ class MqttBroker
bool compareString(const char* good, const char* str, uint8_t str_len) const;
std::vector<MqttClient*> clients;
AsyncServer* server;
TcpServer* server;
const char* auth_user = "guest";
const char* auth_password = "guest";