souti/Arduino
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Jérôme Delacotte
2025-03-06 11:15:32 +01:00
commit 7b30d6e298
5276 changed files with 2108927 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
libraries/ESP8266MQTTClient/LICENSE Normal file
View File

@@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

52
libraries/ESP8266MQTTClient/README.md Normal file
View File

@@ -0,0 +1,52 @@
# MQTT Client library for ESP8266 Arduino
This is MQTT client library for ESP8266, using mqtt_msg package from [MQTT client library for Contiki](https://github.com/esar/contiki-mqtt) and use for ESP8266 NON-OS SDK [esp_mqtt](https://github.com/tuanpmt/esp_mqtt)
Features:
- Support subscribing, publishing, authentication, will messages, keep alive pings and all 3 QoS levels (it should be a fully functional client).
## Requirements
- ESP8266WiFi
- WiFiClientSecure
## Status
- Support 3 type of qos (0, 1, 2) and outbox
- only mqtt over TCP
## MQTT URI Scheme
- `mqtt://[username][:password@]hostname[:port][#clientId]`
+ `mqtt` for MQTT over TCP
+ `ws` for MQTT over Websocket
- Example:
+ **Full** `mqtt://username:password@test.mosquitto.org:1883#my_client_id`
+ **Websocket** `ws://username:password@test.mosquitto.org:1883/mqtt#my_client_id`
+ **Minimal** `mqtt://test.mosquitto.org`, with `user`, `pass` = NULL, port = 1883, client id = "ESP_" + ESP.getChipId()
## API
### Setup
- bool begin(String uri);
- bool begin(String uri, int keepalive, bool clean_session);
- bool begin(String uri, LwtOptions lwt);
- bool begin(String uri, LwtOptions lwt, int keepalive, bool clean_session)
### Events
- void onConnect(THandlerFunction fn);
- void onDisconnect(THandlerFunction fn);
- void onSubscribe(THandlerFunction_PubSub fn);
- void onPublish(THandlerFunction_PubSub fn);
- void onData(THandlerFunction_Data fn);
### Pub/Sub
- int subscribe(String topic);
- int subscribe(String topic, uint8_t qos);
- int publish(String topic, String data);
- int publish(String topic, String data, int qos, int retain);
## License
Copyright (c) 2016 Tuan PM (https://twitter.com/tuanpmt)
ESP8266 port (c) 2016 Ivan Grokhotkov (ivan@esp8266.com)
License Apache License

40
libraries/ESP8266MQTTClient/examples/MQTTClient/MQTTClient.ino Normal file
View File

@@ -0,0 +1,40 @@
#include <ESP8266MQTTClient.h>
#include <ESP8266WiFi.h>
MQTTClient mqtt;
void setup() {
Serial.begin(115200);
// WiFi.begin("ssid", "pass");
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
//topic, data, data is continuing
mqtt.onData([](String topic, String data, bool cont) {
Serial.printf("Data received, topic: %s, data: %s\r\n", topic.c_str(), data.c_str());
mqtt.unSubscribe("/qos0");
});
mqtt.onSubscribe([](int sub_id) {
Serial.printf("Subscribe topic id: %d ok\r\n", sub_id);
mqtt.publish("/qos0", "qos0", 0, 0);
});
mqtt.onConnect([]() {
Serial.printf("MQTT: Connected\r\n");
Serial.printf("Subscribe id: %d\r\n", mqtt.subscribe("/qos0", 0));
// mqtt.subscribe("/qos1", 1);
// mqtt.subscribe("/qos2", 2);
});
mqtt.begin("mqtt://test.mosquitto.org:1883");
// mqtt.begin("mqtt://test.mosquitto.org:1883", {.lwtTopic = "hello", .lwtMsg = "offline", .lwtQos = 0, .lwtRetain = 0});
// mqtt.begin("mqtt://user:pass@mosquito.org:1883");
// mqtt.begin("mqtt://user:pass@mosquito.org:1883#clientId");
}
void loop() {
mqtt.handle();
}

40
libraries/ESP8266MQTTClient/examples/MQTTOverWebsocketClient/MQTTOverWebsocketClient.ino Normal file
View File

@@ -0,0 +1,40 @@
#include <ESP8266MQTTClient.h>
#include <ESP8266WiFi.h>
MQTTClient mqtt;
void setup() {
Serial.begin(115200);
// WiFi.begin("ssid", "pass");
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
//topic, data, data is continuing
mqtt.onData([](String topic, String data, bool cont) {
Serial.printf("Data received, topic: %s, data: %s\r\n", topic.c_str(), data.c_str());
mqtt.unSubscribe("/qos0");
});
mqtt.onSubscribe([](int sub_id) {
Serial.printf("Subscribe topic id: %d ok\r\n", sub_id);
mqtt.publish("/qos0", "qos0", 0, 0);
});
mqtt.onConnect([]() {
Serial.printf("MQTT: Connected\r\n");
Serial.printf("Subscribe id: %d\r\n", mqtt.subscribe("/qos0", 0));
mqtt.subscribe("/qos1", 1);
mqtt.subscribe("/qos2", 2);
});
mqtt.begin("ws://broker.mqttdashboard.com:8000/mqtt");
//mqtt.begin("ws://test.mosquitto.org:8080", {.lwtTopic = "hello", .lwtMsg = "offline", .lwtQos = 0, .lwtRetain = 0});
//mqtt.begin("ws://user:pass@mosquito.org:8080");
//mqtt.begin("ws://user:pass@mosquito.org:8080#clientId");
}
void loop() {
mqtt.handle();
}

46
libraries/ESP8266MQTTClient/examples/MQTTSecureClient/MQTTSecureClient.ino Normal file
View File

@@ -0,0 +1,46 @@
#include <ESP8266MQTTClient.h>
#include <ESP8266WiFi.h>
MQTTClient mqtt;
String fingerprint = "7E 36 22 01 F9 7E 99 2F C5 DB 3D BE AC 48 67 5B 5D 47 94 D2";
void setup() {
Serial.begin(115200);
// WiFi.begin("ssid", "pass");
while(WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
configTime(3 * 3600, 0, "pool.ntp.org", "time.nist.gov");
mqtt.onSecure([](WiFiClientSecure *client, String host) {
Serial.printf("Verify: %s\r\n", host.c_str());
return client->verify(fingerprint.c_str(), host.c_str());
});
//topic, data, data is continuing
mqtt.onData([](String topic, String data, bool cont) {
Serial.printf("Data received, topic: %s, data: %s\r\n", topic.c_str(), data.c_str());
mqtt.unSubscribe("/qos0");
});
mqtt.onSubscribe([](int sub_id) {
Serial.printf("Subscribe topic id: %d ok\r\n", sub_id);
mqtt.publish("/qos0", "qos0", 0, 0);
});
mqtt.onConnect([]() {
Serial.printf("MQTT: Connected\r\n");
mqtt.subscribe("/qos0", 0);
});
mqtt.begin("mqtts://test.mosquitto.org:8883");
//mqtt.begin("mqtts://test.mosquitto.org:8883", {.lwtTopic = "hello", .lwtMsg = "offline", .lwtQos = 0, .lwtRetain = 0});
//mqtt.begin("mqtts://user:pass@mosquito.org:8883");
//mqtt.begin("mqtts://user:pass@mosquito.org:8883#clientId");
}
void loop() {
mqtt.handle();
}

9
libraries/ESP8266MQTTClient/library.properties Normal file
View File

@@ -0,0 +1,9 @@
name=ESP8266MQTTClient
version=1.0.5
author=Tuan PM
maintainer=Tuan PM
sentence=MQTT Client for ESP8266
paragraph=
category=Communication
url=https://github.com/tuanpmt/ESP8266MQTTClient
architectures=esp8266

547
libraries/ESP8266MQTTClient/src/ESP8266MQTTClient.cpp Normal file
View File

@@ -0,0 +1,547 @@
/*
ESP8266 MQTT Client library for ESP8266 Arduino
Version 0.1
Copyright (c) 2016 Tuan PM (tuanpm@live.com)
ESP8266 port (c) 2015 Ivan Grokhotkov (ivan@esp8266.com)
License (MIT license):
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.
*/
#include <Arduino.h>
#include <ESP8266WiFi.h>
#include <WiFiClientSecure.h>
#include <StreamString.h>
#include <base64.h>
#include "ESP8266MQTTClient.h"
#include "MQTTTransport.h"
/**
* constructor
*/
MQTTClient::MQTTClient():
_connected_cb(NULL),
_disconnected_cb(NULL),
_subscribe_cb(NULL),
_publish_cb(NULL),
_data_cb(NULL),
_secure_cb(NULL),
_initialized(false),
_reconnect_tick(0)
{
_outbox = ob_create();
}
/**
* destructor
*/
MQTTClient::~MQTTClient()
{
if(_tcp) {
_tcp->stop();
}
ob_destroy(_outbox);
}
bool MQTTClient::begin(String uri)
{
return begin(uri, {.lwtTopic = "", .lwtMsg = "", .lwtQos = 0, .lwtRetain = 0}, DEFAULT_MQTT_KEEPALIVE, DEFAULT_MQTT_CLEAN_SESSION);
}
bool MQTTClient::begin(String uri, int keepalive, bool clean_session)
{
return begin(uri, {.lwtTopic = "", .lwtMsg = "", .lwtQos = 0, .lwtRetain = 0}, keepalive, clean_session);
}
bool MQTTClient::begin(String uri, LwtOptions lwt)
{
return begin(uri, lwt, DEFAULT_MQTT_KEEPALIVE, DEFAULT_MQTT_CLEAN_SESSION);
}
bool MQTTClient::begin(String uri, LwtOptions lwt, int keepalive, bool clean_session)
{
parsed_uri_t *puri = parse_uri(uri.c_str());
MQTT_CHECK(puri->scheme == NULL, "ERROR: Protocol is not NULL\r\n", false);
MQTT_CHECK(puri->host == NULL, "ERROR: Host is not NULL\r\n", false);
delay(1000);
_scheme = String(puri->scheme);
_host = String(puri->host);
_port = DEFAULT_MQTT_PORT;
_path = "/";
if(puri->fragment) {
_client_id = String(puri->fragment);
} else {
_client_id = String("ESP_") + ESP.getChipId();
}
LOG("MQTT ClientId: %s\r\n", _client_id.c_str());
if(puri->port) {
_port = atoi(puri->port);
}
if(puri->path) {
_path += String(puri->path);
}
if(puri->username)
_username = String(puri->username);
if(puri->password)
_password = String(puri->password);
free_parsed_uri(puri);
_lwt_topic = String(lwt.lwtTopic);
_lwt_msg = String(lwt.lwtMsg);
_lwt_qos = lwt.lwtQos;
_lwt_retain = lwt.lwtRetain;
_keepalive = keepalive;
_connect_info.client_id = _client_id.c_str();
_connect_info.username = _username.c_str();
_connect_info.password = _password.c_str();
_connect_info.will_topic = _lwt_topic.c_str();
_connect_info.will_message = _lwt_msg.c_str();
_connect_info.will_qos = _lwt_qos;
_connect_info.will_retain = _lwt_retain;
_connect_info.keepalive = _keepalive;
_connect_info.clean_session = clean_session;
_state.in_buffer = (uint8_t *)malloc(DEFAULT_MQTT_BUFFER_SIZE_BYTES);
if(_state.in_buffer == NULL) {
LOG("Not enought memory\r\n");
return false;
}
_state.in_buffer_length = DEFAULT_MQTT_BUFFER_SIZE_BYTES;
_state.out_buffer = (uint8_t *)malloc(DEFAULT_MQTT_BUFFER_SIZE_BYTES);
if(_state.in_buffer == NULL) {
free(_state.in_buffer);
LOG("Not enought memory\r\n");
return false;
}
_state.out_buffer_length = DEFAULT_MQTT_BUFFER_SIZE_BYTES;
_state.connect_info = &_connect_info;
mqtt_msg_init(&_state.connection,
_state.out_buffer,
_state.out_buffer_length);
_transportTraits.reset(nullptr);
if(_scheme == "mqtt") {
_transportTraits = MQTTTransportTraitsPtr(new MQTTTransportTraits());
} else if(_scheme == "mqtts") {
_transportTraits = MQTTTransportTraitsPtr(new MQTTTransportTraits(true));
} else if(_scheme == "ws") {
_transportTraits = MQTTTransportTraitsPtr(new MQTTWSTraits());
} else if(_scheme == "wss") {
_transportTraits = MQTTTransportTraitsPtr(new MQTTWSTraits(true));
}
else {
free(_state.out_buffer);
free(_state.in_buffer);
LOG("ERROR: currently only support mqtt over tcp\r\n");
return false;
}
_tcp = _transportTraits->create();
_initialized = true;
return true;
}
bool MQTTClient::connected(void)
{
if(_tcp) {
return (_tcp->connected() || (_tcp->available() > 0));
}
return false;
}
bool MQTTClient::connect(void)
{
int write_len, read_len, connect_rsp_code;
int connect_tick = millis();
if(connected()) {
LOG("[MQTT-Client] connect. already connected, try reuse!\n");
while(_tcp->available() > 0) {
_tcp->read();
}
return true;
}
if(!_transportTraits) {
LOG("[MQTT-Client] connect: MQTTClient::begin was not called or returned error\n");
return false;
}
_tcp->setNoDelay(true);
if(!_transportTraits->connect(_tcp.get(), _host.c_str(), _port, _path.c_str())) {
LOG("[MQTT-Client] failed connect to %s:%u\n", _host.c_str(), _port);
return false;
}
LOG("[MQTT-Client] connected to %s:%u\n", _host.c_str(), _port);
if(_secure_cb && (_scheme == "wss" || _scheme == "mqtts")) {
LOG("[MQTT-Client] begin verifying %s:%u\n", _host.c_str(), _port);
// auto wcs = reinterpret_cast<WiFiClientSecure&>(*_tcp);
WiFiClientSecure *wcs = (WiFiClientSecure*) _tcp.get();
if(!_secure_cb(wcs, _host)) {
_tcp->stop();
LOG("[MQTT-Client] failed verify to %s:%u\n", _host.c_str(), _port);
return false;
}
}
if(!_tcp->connected())
return false;
_state.outbound_message = mqtt_msg_connect(&_state.connection,
_state.connect_info);
_state.pending_msg_type = mqtt_get_type(_state.outbound_message->data);
_state.pending_msg_id = mqtt_get_id(_state.outbound_message->data,
_state.outbound_message->length);
LOG("Sending MQTT CONNECT message, type: %d, id: %04X, len: %d\r\n",
_state.pending_msg_type,
_state.pending_msg_id,
_state.outbound_message->length);
write_len = _transportTraits->write(_tcp.get(), _state.outbound_message->data,
_state.outbound_message->length);
connect_tick = millis();
while(!_tcp->available()) {
if(!_tcp->connected())
return false;
yield();
if(millis() - connect_tick > MQTT_CONNECT_TIMEOUT) {
_tcp->stop();
return false;
}
}
LOG("Reading MQTT CONNECT response message\r\n");
read_len = _transportTraits->read(_tcp.get(), _state.in_buffer, DEFAULT_MQTT_BUFFER_SIZE_BYTES);
if(read_len < 0) {
LOG("Error network response\r\n");
return false;
}
if(mqtt_get_type(_state.in_buffer) != MQTT_MSG_TYPE_CONNACK) {
LOG("Invalid MSG_TYPE response: %d, read_len: %d\r\n", mqtt_get_type(_state.in_buffer), read_len);
_tcp->stop();
return false;
}
connect_rsp_code = mqtt_get_connect_return_code(_state.in_buffer);
switch(connect_rsp_code) {
case CONNECTION_ACCEPTED:
LOG("Connected\r\n");
return connected();
case CONNECTION_REFUSE_PROTOCOL:
case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
case CONNECTION_REFUSE_BAD_USERNAME:
case CONNECTION_REFUSE_NOT_AUTHORIZED:
LOG("Connection refuse, reason code: %d\r\n", connect_rsp_code);
return false;
default:
LOG("Connection refuse, Unknow reason\r\n");
return false;
}
return false;
}
void MQTTClient::onConnect(THandlerFunction fn)
{
_connected_cb = fn;
}
void MQTTClient::onDisconnect(THandlerFunction fn)
{
_disconnected_cb = fn;
}
void MQTTClient::onSubscribe(THandlerFunction_PubSub fn)
{
_subscribe_cb = fn;
}
void MQTTClient::onPublish(THandlerFunction_PubSub fn)
{
_publish_cb = fn;
}
void MQTTClient::onData(THandlerFunction_Data fn)
{
_data_cb = fn;
}
void MQTTClient::onSecure(THandlerFunction_Secure fn)
{
_secure_cb = fn;
}
void MQTTClient::handle(void)
{
mqtt_outbox *ob;
if(!_initialized)
return;
if(!connected()) {
if (!_disconnect_cb_called) {
_disconnected_cb();
_disconnect_cb_called = true;
}
if(_reconnect_tick != 0 && millis() - _reconnect_tick < MQTT_RECONNECT_TIMEOUT)
return;
_reconnect_tick = millis();
if(connect()) {
if(_connected_cb)
_connected_cb();
_disconnect_cb_called = false;
_keepalive_tick = millis();
} else {
return;
}
}
processRead();
if(millis() - _keepalive_tick > _keepalive / 2) {
_keepalive_tick = millis();
sendPing();
}
ob = ob_get_oldest_no_pending(_outbox);
if(ob != NULL) {
_transportTraits->write(_tcp.get(), (unsigned char*)ob->buffer,
ob->len);
ob->pending = 1;
if(ob->remove_on_sent) {
_tcp->flush();
ob_del_ob(ob);
}
LOG("Sent - Outbox size: %d\r\n", ob_get_size(_outbox));
}
ob_del_expired(_outbox, millis(), 60 * 60 * 1000); //remove all package not sent in 60 minutes
ob_cleanup(_outbox, DEFAULT_MQTT_MAX_QUEUE); //keep outbox maximum is DEFAULT_MQTT_MAX_QUEUE(8*1024) bytes
}
bool MQTTClient::deliverPublish(uint8_t *message)
{
mqtt_event_data_t event_data;
int more_data = 0, len_read_more = 0;
String topic, data;
char temp;
_state.message_length = mqtt_get_total_length(_state.in_buffer, _state.message_length_read);
event_data.topic_length = _state.message_length_read;
event_data.topic = mqtt_get_publish_topic(message, &event_data.topic_length);
event_data.data_length = _state.message_length_read;
event_data.data = mqtt_get_publish_data(message, &event_data.data_length);
LOG("Data received, total package len: %d, publish package len: %d, data len: %d\r\n", _state.message_length_read, _state.message_length, event_data.data_length);
if(_data_cb) {
temp = event_data.topic[event_data.topic_length];
event_data.topic[event_data.topic_length] = 0;
topic = String(event_data.topic);
event_data.topic[event_data.topic_length] = temp;
event_data.data[event_data.data_length] = 0;
data = String(event_data.data);
_data_cb(topic, data, false);
}
while(_state.message_length_read < _state.message_length) {
len_read_more = _transportTraits->read(_tcp.get(), _state.in_buffer, DEFAULT_MQTT_BUFFER_SIZE_BYTES);
LOG("Get more data: %d\r\n", len_read_more);
if(len_read_more <= 0)
break;
if(_data_cb) {
_state.in_buffer[len_read_more] = 0;
_data_cb(String((char*)event_data.topic), String((char*)_state.in_buffer), true);
}
_state.message_length_read += len_read_more;
}
return true;
}
int MQTTClient::processRead()
{
int read_len;
uint8_t msg_type;
uint8_t msg_qos;
uint16_t msg_id;
mqtt_outbox *valid_msg;
if(!connected())
return 0;
_tcp->setTimeout(DEFAULT_MQTT_READ_TIMEOUT);
read_len = _transportTraits->read(_tcp.get(), _state.in_buffer, DEFAULT_MQTT_BUFFER_SIZE_BYTES);
if(read_len <= 0)
return 0;
_state.message_length_read = read_len;
PROCESS_READ_AGAIN:
msg_type = mqtt_get_type(_state.in_buffer);
msg_qos = mqtt_get_qos(_state.in_buffer);
msg_id = mqtt_get_id(_state.in_buffer, _state.in_buffer_length);
LOG("Read len %d, id: %d, type: %d\r\n", read_len, msg_id, msg_type);
switch(msg_type)
{
case MQTT_MSG_TYPE_SUBACK:
valid_msg = ob_get(_outbox, msg_id);
if(valid_msg->msg_type == MQTT_MSG_TYPE_SUBSCRIBE && valid_msg->msg_id == msg_id) {
if(_subscribe_cb)
_subscribe_cb(msg_id);
ob_del_id(_outbox, msg_id);
_state.message_length = mqtt_get_total_length(_state.in_buffer, _state.message_length_read);
LOG("Subscribe successful, msgid: %d, outbox size: %d\r\n", msg_id, ob_get_size(_outbox));
}
break;
case MQTT_MSG_TYPE_UNSUBACK:
valid_msg = ob_get(_outbox, msg_id);
if(valid_msg && valid_msg->msg_type == MQTT_MSG_TYPE_UNSUBSCRIBE && valid_msg->msg_id == msg_id) {
LOG("UnSubscribe successful\r\n");
ob_del_id(_outbox, msg_id);
}
break;
case MQTT_MSG_TYPE_PUBLISH:
if(msg_qos == 1)
_state.outbound_message = mqtt_msg_puback(&_state.connection, msg_id);
else if(msg_qos == 2)
_state.outbound_message = mqtt_msg_pubrec(&_state.connection, msg_id);
deliverPublish(_state.in_buffer);
if(msg_qos == 0)
ob_del_id(_outbox, msg_id);
LOG("Outbox size: %d, msgid: %d\r\n", ob_get_size(_outbox), msg_id);
if(msg_qos == 1 || msg_qos == 2) {
LOG("Queue MQTT_MSG_TYPE_PUBACK/MQTT_MSG_TYPE_PUBREC: %d, delete on send\r\n", msg_qos);
queue(msg_qos == 1); //delete after send
}
break;
case MQTT_MSG_TYPE_PUBACK:
valid_msg = ob_get(_outbox, msg_id);
if(valid_msg && valid_msg->msg_type == MQTT_MSG_TYPE_PUBLISH && valid_msg->msg_id == msg_id) {
LOG("Received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish, msgid: %d, remove data outbox\r\n", msg_id);
ob_del_id(_outbox, msg_id);
LOG("Outbox size: %d\r\n", ob_get_size(_outbox));
}
break;
case MQTT_MSG_TYPE_PUBREC:
LOG("received MQTT_MSG_TYPE_PUBREC, msgid: %d\r\n", msg_id);
valid_msg = ob_get(_outbox, msg_id);
if(valid_msg && valid_msg->msg_type == MQTT_MSG_TYPE_PUBLISH && valid_msg->msg_id == msg_id) {
LOG("Reply with MQTT_MSG_TYPE_PUBREL msg_id: %d, %d\r\n", msg_id);
ob_del_id(_outbox, msg_id);
_state.outbound_message = mqtt_msg_pubrel(&_state.connection, msg_id);
queue(0);
LOG("Outbox size: %d\r\n", ob_get_size(_outbox));
}
break;
case MQTT_MSG_TYPE_PUBREL:
valid_msg = ob_get(_outbox, msg_id);
LOG("Received MQTT_MSG_TYPE_PUBREL, msg_id: %d, %d\r\n", msg_id);
if(valid_msg && valid_msg->msg_type == MQTT_MSG_TYPE_PUBREC && valid_msg->msg_id == msg_id) {
LOG("Reply with MQTT_MSG_TYPE_PUBCOMP, remove on sent, msg_id: %d, %d\r\n", msg_id);
ob_del_id(_outbox, msg_id);
_state.outbound_message = mqtt_msg_pubcomp(&_state.connection, msg_id);
queue(1);
LOG("Outbox size: %d\r\n", ob_get_size(_outbox));
}
break;
case MQTT_MSG_TYPE_PUBCOMP:
LOG("Received MQTT_MSG_TYPE_PUBCOMP, msg_id: %d\r\n", msg_id);
valid_msg = ob_get(_outbox, msg_id);
if(valid_msg && (valid_msg->msg_type == MQTT_MSG_TYPE_PUBCOMP || valid_msg->msg_type == MQTT_MSG_TYPE_PUBREL) && valid_msg->msg_id == msg_id) {
ob_del_id(_outbox, msg_id);
LOG("Outbox size: %d\r\n", ob_get_size(_outbox));
}
break;
case MQTT_MSG_TYPE_PINGREQ:
LOG("received MQTT_MSG_TYPE_PINGREQ\r\n");
_state.outbound_message = mqtt_msg_pingresp(&_state.connection);
queue(1);
break;
case MQTT_MSG_TYPE_PINGRESP:
LOG("MQTT_MSG_TYPE_PINGRESP\r\n");
// Ignore
break;
}
if(_state.message_length < _state.message_length_read) {
_state.message_length_read -= _state.message_length;
_state.in_buffer += _state.message_length;
goto PROCESS_READ_AGAIN;
}
return 1;
}
void MQTTClient::queue(int remove_on_sent)
{
_state.pending_msg_type = mqtt_get_type(_state.outbound_message->data);
_state.pending_msg_id = mqtt_get_id(_state.outbound_message->data, _state.outbound_message->length);
LOG("Queue: msgid: %d, msgtype: %d\r\n", _state.pending_msg_id, _state.pending_msg_type);
ob_put(_outbox,
_state.outbound_message->data,
_state.outbound_message->length,
_state.pending_msg_id,
_state.pending_msg_type,
millis(),
remove_on_sent);
LOG("Outbox size: %d\r\n", ob_get_size(_outbox));
}
void MQTTClient::sendPing()
{
_state.outbound_message = mqtt_msg_pingreq(&_state.connection);
_state.pending_msg_type = mqtt_get_type(_state.outbound_message->data);
_state.pending_msg_id = mqtt_get_id(_state.outbound_message->data,
_state.outbound_message->length);
LOG("Sending pingreq");
_transportTraits->write(_tcp.get(), _state.outbound_message->data,
_state.outbound_message->length);
}
int MQTTClient::subscribe(String topic)
{
return subscribe(topic, 0);
}
int MQTTClient::subscribe(String topic, uint8_t qos)
{
_state.outbound_message = mqtt_msg_subscribe(&_state.connection,
topic.c_str(), qos,
&_state.pending_msg_id);
LOG("Queue subscribe, topic\"%s\", id: %d\r\n", topic.c_str(), _state.pending_msg_id);
queue(0);
return _state.pending_msg_id;
}
int MQTTClient::unSubscribe(String topic)
{
_state.outbound_message = mqtt_msg_unsubscribe(&_state.connection,
topic.c_str(),
&_state.pending_msg_id);
LOG("Queue unsubscribe, topic\"%s\", id: %d\r\n", topic.c_str(), _state.pending_msg_id);
queue(0);
return _state.pending_msg_id;
}
int MQTTClient::publish(String topic, String data)
{
return publish(topic, data, 0, 0);
}
int MQTTClient::publish(String topic, String data, int qos, int retain)
{
int remove_on_sent = 0;
_state.outbound_message = mqtt_msg_publish(&_state.connection,
topic.c_str(), data.c_str(), data.length(),
qos, retain,
&_state.pending_msg_id);
LOG("Queue publish, topic\"%s\", id: %d\r\n", topic.c_str(), _state.pending_msg_id);
if(qos == 0)
remove_on_sent = 1;
queue(remove_on_sent);
return _state.pending_msg_id;
}

162
libraries/ESP8266MQTTClient/src/ESP8266MQTTClient.h Normal file
View File

@@ -0,0 +1,162 @@
/*
ESP8266 MQTT Client library for ESP8266 Arduino
Version 1.1
Copyright (c) 2016 Tuan PM (tuanpm@live.com)
ESP8266 port (c) 2015 Ivan Grokhotkov (ivan@esp8266.com)
License (MIT license):
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.
*/
#ifndef ESP8266MQTTClient_H_
#define ESP8266MQTTClient_H_
#include <memory>
#include <Arduino.h>
#include <WiFiClient.h>
#include <WiFiClientSecure.h>
#include <functional>
#include "uri_parser.h"
#include "mqtt_msg.h"
#include "mqtt_outbox.h"
// #ifdef DEBUG_ESP_MQTT_CLIENT
#ifdef DEBUG_ESP_PORT
#define LOG(...) DEBUG_ESP_PORT.printf( __VA_ARGS__ )
#endif
// #endif
#ifndef LOG
#define LOG(...)
#endif
#define MQTT_CHECK(cond, msg, ret) if(cond){LOG(msg); return ret;}
#define DEFAULT_MQTT_PORT 1883
#define DEFAULT_MQTT_KEEPALIVE 120000
#define DEFAULT_MQTT_BUFFER_SIZE_BYTES 1024
#define DEFAULT_MQTT_CLEAN_SESSION 1
#define MQTT_RECONNECT_TIMEOUT 5000
#define MQTT_CONNECT_TIMEOUT 10000
#define DEFAULT_MQTT_READ_TIMEOUT 200
#define DEFAULT_MQTT_MAX_QUEUE (1024*8)
typedef struct {
String lwtTopic;
String lwtMsg;
int lwtQos;
int lwtRetain;
} LwtOptions;
typedef struct mqtt_state_t
{
uint16_t port;
int auto_reconnect;
mqtt_connect_info_t* connect_info;
uint8_t* in_buffer;
uint8_t* out_buffer;
int in_buffer_length;
int out_buffer_length;
uint16_t message_length;
uint16_t message_length_read;
mqtt_message_t* outbound_message;
mqtt_connection_t connection;
uint16_t pending_msg_id;
int pending_msg_type;
int pending_publish_qos;
} mqtt_state_t;
typedef struct mqtt_event_data_t
{
uint8_t type;
char* topic;
char* data;
uint16_t topic_length;
uint16_t data_length;
uint16_t data_offset;
uint16_t data_total_length;
} mqtt_event_data_t;
class MQTTTransportTraits;
typedef std::unique_ptr<MQTTTransportTraits> MQTTTransportTraitsPtr;
class MQTTClient
{
public:
typedef std::function<void(void)> THandlerFunction;
typedef std::function<void(int)> THandlerFunction_Error;
typedef std::function<void(int)> THandlerFunction_PubSub;
typedef std::function<void(String, String, bool)> THandlerFunction_Data;
typedef std::function<bool(WiFiClientSecure *client, String host)> THandlerFunction_Secure;
MQTTClient();
~MQTTClient();
void onConnect(THandlerFunction fn);
void onDisconnect(THandlerFunction fn);
void onSubscribe(THandlerFunction_PubSub fn);
void onPublish(THandlerFunction_PubSub fn);
void onData(THandlerFunction_Data fn);
void onSecure(THandlerFunction_Secure fn);
bool begin(String uri);
bool begin(String uri, int keepalive, bool clean_session);
bool begin(String uri, LwtOptions lwt);
bool begin(String uri, LwtOptions lwt, int keepalive, bool clean_session);
void handle();
bool connect();
int subscribe(String topic);
int unSubscribe(String topic);
int subscribe(String topic, uint8_t qos);
int publish(String topic, String data);
int publish(String topic, String data, int qos, int retain);
protected:
std::unique_ptr<WiFiClient> _tcp;
MQTTTransportTraitsPtr _transportTraits;
bool connected();
int processRead();
void queue(int remove_on_sent);
void sendPing();
bool deliverPublish(uint8_t *message);
mqtt_state_t _state;
mqtt_connect_info_t _connect_info;
mqtt_outbox *_outbox;
uint32_t _keepalive_tick;
uint32_t _reconnect_tick;
String _scheme;
String _host;
String _path; //for websocket
int _port;
String _client_id;
String _username;
String _password;
String _lwt_topic;
String _lwt_msg;
int _lwt_qos;
int _lwt_retain;
int _clean_session;
int _keepalive;
bool _initialized;
bool _disconnect_cb_called;
THandlerFunction _connected_cb;
THandlerFunction _disconnected_cb;
THandlerFunction_PubSub _subscribe_cb;
THandlerFunction_PubSub _publish_cb;
THandlerFunction_Data _data_cb;
THandlerFunction_Secure _secure_cb;
};
#endif /* ESP8266MQTTClient_H_ */

223
libraries/ESP8266MQTTClient/src/MQTTTransport.cpp Normal file
View File

@@ -0,0 +1,223 @@
/*
ESP8266 MQTT Client library for ESP8266 Arduino
Version 0.1
Copyright (c) 2016 Tuan PM (tuanpm@live.com)
ESP8266 port (c) 2015 Ivan Grokhotkov (ivan@esp8266.com)
License (MIT license):
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.
*/
#include <Arduino.h>
#include <ESP8266WiFi.h>
#include <WiFiClientSecure.h>
#include <StreamString.h>
#include <base64.h>
#include <Hash.h>
#include "ESP8266MQTTClient.h"
#include "MQTTTransport.h"
MQTTTransportTraits::MQTTTransportTraits(): _isSecure(false)
{
}
MQTTTransportTraits::MQTTTransportTraits(bool secure): _isSecure(secure)
{
}
std::unique_ptr<WiFiClient> MQTTTransportTraits::create()
{
if(_isSecure)
return std::unique_ptr<WiFiClient>(new WiFiClientSecure());
return std::unique_ptr<WiFiClient>(new WiFiClient());
}
bool MQTTTransportTraits::connect(WiFiClient* client, const char* host, int port, const char *path)
{
if(_isSecure) {
WiFiClientSecure *client = (WiFiClientSecure*) client;
}
return client->connect(host, port);
}
int MQTTTransportTraits::write(WiFiClient* client, unsigned char *data, int size)
{
if(_isSecure) {
WiFiClientSecure *client = (WiFiClientSecure*) client;
}
return client->write(reinterpret_cast<const char*>(data), size);
}
int MQTTTransportTraits::read(WiFiClient* client, unsigned char *data, int size)
{
if(_isSecure) {
WiFiClientSecure *client = (WiFiClientSecure*) client;
}
return client->read(data, size);
}
/**
* MQTT Over WS
*/
MQTTWSTraits::MQTTWSTraits(): _isSecure(false)
{
randomSeed(RANDOM_REG32);
}
MQTTWSTraits::MQTTWSTraits(bool secure): _isSecure(secure)
{
randomSeed(RANDOM_REG32);
}
bool MQTTWSTraits::connect(WiFiClient* client, const char* host, int port, const char *path)
{
uint8_t randomKey[16] = { 0 }, timeout = 0;
int bite;
bool foundupgrade = false;
String serverKey, temp, acceptKey;
for(uint8_t i = 0; i < sizeof(randomKey); i++) {
randomKey[i] = random(0xFF);
}
_key = base64::encode(randomKey, 16);
LOG("Key: %s\r\n", _key.c_str());
String handshake = "GET "+ String(path) +" HTTP/1.1\r\n"
"Connection: Upgrade\r\n"
"Upgrade: websocket\r\n"
"Host: " + String(host) + ":" + String(port) + "\r\n"
"Sec-WebSocket-Version: 13\r\n"
"Origin: file://\r\n"
"Sec-WebSocket-Protocol: mqttv3.1\r\n"
"User-Agent: ESP8266MQTTClient\r\n"
"Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n"
"Sec-WebSocket-Key: " + _key + "\r\n\r\n";
if(!client->connect(host, port)) {
LOG("ERROR: Can't connect \r\n");
return false;
}
client->write(handshake.c_str(), handshake.length());
while(client->connected() && !client->available()) {
delay(100);
if(timeout++ > 10) {
LOG("ERROR Read timeout\r\n");
return false;
}
}
while((bite = client->read()) != -1) {
temp += (char)bite;
if((char)bite == '\n') {
if(!foundupgrade && (temp.startsWith("Upgrade: websocket") || temp.startsWith("upgrade: websocket"))) {
foundupgrade = true;
} else if(temp.startsWith("Sec-WebSocket-Accept: ") || temp.startsWith("sec-websocket-accept: ")) {
serverKey = temp.substring(22, temp.length() - 2); // Don't save last CR+LF
}
LOG("Data=%s", temp.c_str());
temp = "";
}
if(!client->available()) {
delay(100);
}
}
_key += "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
uint8_t sha1HashBin[20] = { 0 };
sha1(_key, &sha1HashBin[0]);
acceptKey = base64::encode(sha1HashBin, 20);
acceptKey.trim();
LOG("AcceptKey: %s\r\n", acceptKey.c_str());
LOG("ServerKey: %s\r\n", serverKey.c_str());
timeout = 0;
return acceptKey == serverKey;
}
int MQTTWSTraits::write(WiFiClient* client, unsigned char *data, int size)
{
char header_len = 0, *mask, *data_buffer;
int written = 0;
data_buffer = (char *) malloc(MAX_WEBSOCKET_HEADER_SIZE + size);
if(data_buffer == NULL)
return -1;
// Opcode; final fragment
data_buffer[header_len++] = WS_OPCODE_BINARY | WS_FIN;
// NOTE: no support for > 16-bit sized messages
if(size > 125) {
data_buffer[header_len++] = WS_SIZE16 | WS_MASK;
data_buffer[header_len++] = (uint8_t)(size >> 8);
data_buffer[header_len++] = (uint8_t)(size & 0xFF);
} else {
data_buffer[header_len++] = (uint8_t)(size | WS_MASK);
}
mask = &data_buffer[header_len];
data_buffer[header_len++] = random(0, 256);
data_buffer[header_len++] = random(0, 256);
data_buffer[header_len++] = random(0, 256);
data_buffer[header_len++] = random(0, 256);
for(int i = 0; i < size; ++i) {
data_buffer[header_len++] = (data[i] ^ mask[i % 4]);
}
client->write(reinterpret_cast<const char*>(data_buffer), header_len);
client->flush();
free(data_buffer);
return size;
}
int MQTTWSTraits::read(WiFiClient* client, unsigned char *data, int size)
{
unsigned char *data_buffer = (unsigned char*) malloc(size + MAX_WEBSOCKET_HEADER_SIZE), *data_ptr, opcode, mask, *maskKey = NULL;
int tcp_read_size, payloadLen;
data_ptr = data_buffer;
if(data_buffer == NULL)
return -1;
tcp_read_size = client->read(data_buffer, size + MAX_WEBSOCKET_HEADER_SIZE);
if(tcp_read_size <= 0)
{
free(data_buffer);
return -1;
}
opcode = (*data_ptr & 0x0F);
data_ptr ++;
mask = ((*data_ptr >> 7) & 0x01);
payloadLen = (*data_ptr & 0x7F);
data_ptr++;
LOG("Opcode: %d, mask: %d, len: %d\r\n", opcode, mask, payloadLen);
if(payloadLen == 126) {
// headerLen += 2;
payloadLen = data_ptr[0] << 8 | data_ptr[1];
data_ptr += 2;
} else if(payloadLen == 127) {
// headerLen += 8;
if(data_ptr[0] != 0 || data_ptr[1] != 0 || data_ptr[2] != 0 || data_ptr[3] != 0) {
// really too big!
payloadLen = 0xFFFFFFFF;
} else {
payloadLen = data_ptr[4] << 24 | data_ptr[5] << 16 | data_ptr[6] << 8 | data_ptr[7];
}
data_ptr += 8;
}
if(mask) {
maskKey = data_ptr;
data_ptr += 4;
for(size_t i = 0; i < payloadLen; i++) {
data[i] = (data_ptr[i] ^ maskKey[i % 4]);
}
} else {
memcpy(data, data_ptr, payloadLen);
}
return payloadLen;
}

43
libraries/ESP8266MQTTClient/src/MQTTTransport.h Normal file
View File

@@ -0,0 +1,43 @@
#ifndef _MQTT_TRANSPORT_H_
#define _MQTT_TRANSPORT_H_
// WebSocket protocol constants
// First byte
#define WS_FIN 0x80
#define WS_OPCODE_TEXT 0x01
#define WS_OPCODE_BINARY 0x02
#define WS_OPCODE_CLOSE 0x08
#define WS_OPCODE_PING 0x09
#define WS_OPCODE_PONG 0x0a
// Second byte
#define WS_MASK 0x80
#define WS_SIZE16 126
#define WS_SIZE64 127
#define MAX_WEBSOCKET_HEADER_SIZE 10
class MQTTTransportTraits
{
public:
MQTTTransportTraits();
MQTTTransportTraits(bool secure);
virtual std::unique_ptr<WiFiClient> create();
virtual bool connect(WiFiClient* client, const char* host, int port, const char *path);
virtual int write(WiFiClient* client, unsigned char *data, int size);
virtual int read(WiFiClient* client, unsigned char *data, int size);
protected:
std::unique_ptr<WiFiClient> _tcp;
bool _isSecure;
};
class MQTTWSTraits : public MQTTTransportTraits
{
public:
MQTTWSTraits();
MQTTWSTraits(bool secure);
bool connect(WiFiClient* client, const char* host, int port, const char *path) override;
int write(WiFiClient* client, unsigned char *data, int size) override;
int read(WiFiClient* client, unsigned char *data, int size) override;
protected:
String _key;
bool _isSecure;
};
#endif

488
libraries/ESP8266MQTTClient/src/mqtt_msg.c Normal file
View File

@@ -0,0 +1,488 @@
/*
* Copyright (c) 2014, Stephen Robinson
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* 3. Neither the name of the copyright holder nor the names of its
* contributors may be used to endorse or promote products derived
* from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*
*/
#include <stdlib.h>
#include <string.h>
#include "mqtt_msg.h"
#define MQTT_MAX_FIXED_HEADER_SIZE 3
#define PROTOCOL_NAMEv311
enum mqtt_connect_flag
{
MQTT_CONNECT_FLAG_USERNAME = 1 << 7,
MQTT_CONNECT_FLAG_PASSWORD = 1 << 6,
MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5,
MQTT_CONNECT_FLAG_WILL = 1 << 2,
MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1
};
struct __attribute((__packed__)) mqtt_connect_variable_header
{
uint8_t lengthMsb;
uint8_t lengthLsb;
#if defined(PROTOCOL_NAMEv31)
uint8_t magic[6];
#elif defined(PROTOCOL_NAMEv311)
uint8_t magic[4];
#else
#error "Please define protocol name"
#endif
uint8_t version;
uint8_t flags;
uint8_t keepaliveMsb;
uint8_t keepaliveLsb;
};
static int append_string(mqtt_connection_t* connection, const char* string, int len)
{
if (connection->message.length + len + 2 > connection->buffer_length)
return -1;
connection->buffer[connection->message.length++] = len >> 8;
connection->buffer[connection->message.length++] = len & 0xff;
memcpy(connection->buffer + connection->message.length, string, len);
connection->message.length += len;
return len + 2;
}
static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t message_id)
{
// If message_id is zero then we should assign one, otherwise
// we'll use the one supplied by the caller
while (message_id == 0)
message_id = (os_random() % 65535); //++connection->message_id;
if (connection->message.length + 2 > connection->buffer_length)
return 0;
connection->buffer[connection->message.length++] = message_id >> 8;
connection->buffer[connection->message.length++] = message_id & 0xff;
return message_id;
}
static int init_message(mqtt_connection_t* connection)
{
connection->message.length = MQTT_MAX_FIXED_HEADER_SIZE;
return MQTT_MAX_FIXED_HEADER_SIZE;
}
static mqtt_message_t* fail_message(mqtt_connection_t* connection)
{
connection->message.data = connection->buffer;
connection->message.length = 0;
return &connection->message;
}
static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int dup, int qos, int retain)
{
int remaining_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE;
if (remaining_length > 127)
{
connection->buffer[0] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
connection->buffer[1] = 0x80 | (remaining_length % 128);
connection->buffer[2] = remaining_length / 128;
connection->message.length = remaining_length + 3;
connection->message.data = connection->buffer;
}
else
{
connection->buffer[1] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
connection->buffer[2] = remaining_length;
connection->message.length = remaining_length + 2;
connection->message.data = connection->buffer + 1;
}
return &connection->message;
}
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length)
{
memset(connection, 0, sizeof(mqtt_connection_t));
connection->buffer = buffer;
connection->buffer_length = buffer_length;
}
int mqtt_get_total_length(uint8_t* buffer, uint16_t length)
{
int i;
int totlen = 0;
for (i = 1; i < length; ++i)
{
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
++i;
break;
}
}
totlen += i;
return totlen;
}
char* mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length)
{
int i;
int totlen = 0;
int topiclen;
for (i = 1; i < *length; ++i)
{
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
++i;
break;
}
}
totlen += i;
if (i + 2 >= *length)
return NULL;
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen > *length)
return NULL;
*length = topiclen;
return (char*)(buffer + i);
}
char* mqtt_get_publish_data(uint8_t* buffer, uint16_t* length)
{
int i;
int totlen = 0;
int topiclen;
int blength = *length;
*length = 0;
for (i = 1; i < blength; ++i)
{
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
++i;
break;
}
}
totlen += i;
if (i + 2 >= blength)
return NULL;
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen >= blength)
return NULL;
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (i + 2 >= blength)
return NULL;
i += 2;
}
if (totlen < i)
return NULL;
if (totlen <= blength)
*length = totlen - i;
else
*length = blength - i;
return (char*)(buffer + i);
}
uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length)
{
if (length < 1)
return 0;
switch (mqtt_get_type(buffer))
{
case MQTT_MSG_TYPE_PUBLISH:
{
int i;
int topiclen;
for (i = 1; i < length; ++i)
{
if ((buffer[i] & 0x80) == 0)
{
++i;
break;
}
}
if (i + 2 >= length)
return 0;
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen >= length)
return 0;
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (i + 2 >= length)
return 0;
//i += 2;
} else {
return 0;
}
return (buffer[i] << 8) | buffer[i + 1];
}
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_SUBSCRIBE:
case MQTT_MSG_TYPE_UNSUBSCRIBE:
{
// This requires the remaining length to be encoded in 1 byte,
// which it should be.
if (length >= 4 && (buffer[1] & 0x80) == 0)
return (buffer[2] << 8) | buffer[3];
else
return 0;
}
default:
return 0;
}
}
mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info)
{
struct mqtt_connect_variable_header* variable_header;
init_message(connection);
if (connection->message.length + sizeof(*variable_header) > connection->buffer_length)
return fail_message(connection);
variable_header = (void*)(connection->buffer + connection->message.length);
connection->message.length += sizeof(*variable_header);
variable_header->lengthMsb = 0;
#if defined(PROTOCOL_NAMEv31)
variable_header->lengthLsb = 6;
memcpy(variable_header->magic, "MQIsdp", 6);
variable_header->version = 3;
#elif defined(PROTOCOL_NAMEv311)
variable_header->lengthLsb = 4;
memcpy(variable_header->magic, "MQTT", 4);
variable_header->version = 4;
#else
#error "Please define protocol name"
#endif
variable_header->flags = 0;
variable_header->keepaliveMsb = info->keepalive >> 8;
variable_header->keepaliveLsb = info->keepalive & 0xff;
if (info->clean_session)
variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION;
if (info->client_id == NULL)
{
/* Never allowed */
return fail_message(connection);
}
else if (info->client_id[0] == '\0')
{
#ifdef PROTOCOL_NAMEv311
/* Allowed. Format 0 Length ID */
append_string(connection, info->client_id, 2) ;
#else
/* 0 Length not allowed */
return fail_message(connection);
#endif
}
else
{
/* No 0 data and at least 1 long. Good to go. */
if(append_string(connection, info->client_id, strlen(info->client_id)) < 0)
return fail_message(connection);
}
if (info->will_topic != NULL && info->will_topic[0] != '\0')
{
if (append_string(connection, info->will_topic, strlen(info->will_topic)) < 0)
return fail_message(connection);
if (append_string(connection, info->will_message, strlen(info->will_message)) < 0)
return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_WILL;
if (info->will_retain)
variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN;
variable_header->flags |= (info->will_qos & 3) << 3;
}
if (info->username != NULL && info->username[0] != '\0')
{
if (append_string(connection, info->username, strlen(info->username)) < 0)
return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME;
}
if (info->password != NULL && info->password[0] != '\0')
{
if (append_string(connection, info->password, strlen(info->password)) < 0)
return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD;
}
return fini_message(connection, MQTT_MSG_TYPE_CONNECT, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
return fail_message(connection);
if (append_string(connection, topic, strlen(topic)) < 0)
return fail_message(connection);
if (qos > 0)
{
if ((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection);
}
else
*message_id = 0;
if (connection->message.length + data_length > connection->buffer_length)
return fail_message(connection);
memcpy(connection->buffer + connection->message.length, data, data_length);
connection->message.length += data_length;
return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain);
}
mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PUBACK, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PUBREC, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PUBREL, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
return fail_message(connection);
if ((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection);
if (append_string(connection, topic, strlen(topic)) < 0)
return fail_message(connection);
if (connection->message.length + 1 > connection->buffer_length)
return fail_message(connection);
connection->buffer[connection->message.length++] = qos;
return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
return fail_message(connection);
if ((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection);
if (append_string(connection, topic, strlen(topic)) < 0)
return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PINGRESP, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0);
}

132
libraries/ESP8266MQTTClient/src/mqtt_msg.h Normal file
View File

@@ -0,0 +1,132 @@
#ifndef MQTT_MSG_H
#define MQTT_MSG_H
#include "c_types.h"
#ifdef __cplusplus
extern "C" {
#endif
/*
* Copyright (c) 2014, Stephen Robinson
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* 3. Neither the name of the copyright holder nor the names of its
* contributors may be used to endorse or promote products derived
* from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*
*/
/* 7 6 5 4 3 2 1 0*/
/*| --- Message Type---- | DUP Flag | QoS Level | Retain |
/* Remaining Length */
enum mqtt_message_type
{
MQTT_MSG_TYPE_CONNECT = 1,
MQTT_MSG_TYPE_CONNACK = 2,
MQTT_MSG_TYPE_PUBLISH = 3,
MQTT_MSG_TYPE_PUBACK = 4,
MQTT_MSG_TYPE_PUBREC = 5,
MQTT_MSG_TYPE_PUBREL = 6,
MQTT_MSG_TYPE_PUBCOMP = 7,
MQTT_MSG_TYPE_SUBSCRIBE = 8,
MQTT_MSG_TYPE_SUBACK = 9,
MQTT_MSG_TYPE_UNSUBSCRIBE = 10,
MQTT_MSG_TYPE_UNSUBACK = 11,
MQTT_MSG_TYPE_PINGREQ = 12,
MQTT_MSG_TYPE_PINGRESP = 13,
MQTT_MSG_TYPE_DISCONNECT = 14
};
enum mqtt_connect_return_code
{
CONNECTION_ACCEPTED = 0,
CONNECTION_REFUSE_PROTOCOL,
CONNECTION_REFUSE_ID_REJECTED,
CONNECTION_REFUSE_SERVER_UNAVAILABLE,
CONNECTION_REFUSE_BAD_USERNAME,
CONNECTION_REFUSE_NOT_AUTHORIZED
};
typedef struct mqtt_message
{
uint8_t* data;
uint16_t length;
} mqtt_message_t;
typedef struct mqtt_connection
{
mqtt_message_t message;
uint16_t message_id;
uint8_t* buffer;
uint16_t buffer_length;
} mqtt_connection_t;
typedef struct mqtt_connect_info
{
const char* client_id;
const char* username;
const char* password;
const char* will_topic;
const char* will_message;
uint32_t keepalive;
int will_qos;
int will_retain;
int clean_session;
} mqtt_connect_info_t;
static inline int mqtt_get_type(uint8_t* buffer) { return (buffer[0] & 0xf0) >> 4; }
static inline int mqtt_get_connect_return_code(uint8_t* buffer) { return buffer[3]; }
static inline int mqtt_get_dup(uint8_t* buffer) { return (buffer[0] & 0x08) >> 3; }
static inline int mqtt_get_qos(uint8_t* buffer) { return (buffer[0] & 0x06) >> 1; }
static inline int mqtt_get_retain(uint8_t* buffer) { return (buffer[0] & 0x01); }
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length);
int mqtt_get_total_length(uint8_t* buffer, uint16_t length);
char* mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length);
char* mqtt_get_publish_data(uint8_t* buffer, uint16_t* length);
uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length);
mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info);
mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id);
mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id);
mqtt_message_t* mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id);
mqtt_message_t* mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id);
mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id);
mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id);
mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id);
mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection);
mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection);
mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection);
#ifdef __cplusplus
}
#endif
#endif /* MQTT_MSG_H */

166
libraries/ESP8266MQTTClient/src/mqtt_outbox.c Normal file
View File

@@ -0,0 +1,166 @@
/*
* @Author: Tuan PM
* @Date: 2016-10-02 09:45:51
* @Last Modified by: TuanPM
* @Last Modified time: 2016-11-27 11:41:57
*/
#include "mqtt_outbox.h"
#include <stdlib.h>
#include <string.h>
static mqtt_outbox *ob_set_data(mqtt_outbox *ob, char *data, int len, int msg_id, int msg_type, int tick, int remove_on_sent)
{
ob->buffer = malloc(len);
if(ob->buffer == NULL)
return NULL;
ob->len = len;
ob->msg_id = msg_id;
ob->msg_type = msg_type;
ob->retry_count = 0;
ob->tick_created = tick;
ob->next = NULL;
ob->prev = NULL;
ob->pending = 0;
ob->remove_on_sent = remove_on_sent;
memcpy(ob->buffer, data, len);
return ob;
}
mqtt_outbox *ob_create()
{
mqtt_outbox *ob = (mqtt_outbox *) malloc(sizeof(mqtt_outbox));
if(ob)
memset(ob, 0, sizeof(mqtt_outbox));
return ob;
}
mqtt_outbox *ob_get_oldest_no_pending(mqtt_outbox *ob)
{
mqtt_outbox *oldest = ob->next;
while(oldest != NULL && oldest->pending == 1) {
oldest = oldest->next;
}
if(oldest != NULL && oldest->pending == 0)
return oldest;
return NULL;
}
mqtt_outbox *ob_get_top(mqtt_outbox *ob)
{
mqtt_outbox *top = ob;
while(top->next != NULL) {
top = top->next;
}
return top;
}
mqtt_outbox *ob_put(mqtt_outbox *ob, uint8_t *data, int len, int msg_id, int msg_type, int tick, int remove_on_sent)
{
mqtt_outbox *top = ob_get_top(ob);
top->next = ob_create();
if(top->next) {
ob_set_data(top->next, data, len, msg_id, msg_type, tick, remove_on_sent);
top->next->prev = top;
}
return top->next;
}
mqtt_outbox *ob_get(mqtt_outbox *ob, int msg_id)
{
mqtt_outbox *found = ob->next;
while(found != NULL) {
if(found->msg_id == msg_id) {
// found->retry_count ++;
return found;
} else {
found = found->next; //ignore root
}
}
return NULL;
}
mqtt_outbox *ob_del_ob(mqtt_outbox *del)
{
mqtt_outbox *next = NULL;
if(del->next) {
next = del->next;
del->prev->next = del->next;
del->next->prev = del->prev;
} else {
del->prev->next = NULL;
}
free(del->buffer);
free(del);
return next;
}
mqtt_outbox *ob_del_id_type(mqtt_outbox *ob, int msg_id, int msg_type)
{
mqtt_outbox *found = ob->next;
while(found != NULL) {
if(found->msg_id == msg_id && found->msg_type == msg_type) {
return ob_del_ob(found);
} else {
found = found->next;
}
}
return NULL;
}
int ob_del_id(mqtt_outbox *ob, int msg_id)
{
int deleted = 0;
mqtt_outbox *found = ob->next;
while(found != NULL) {
if(found->msg_id == msg_id) {
deleted ++;
found = ob_del_ob(found);
} else {
found = found->next;
}
}
return deleted;
}
mqtt_outbox *ob_del_oldest(mqtt_outbox *ob)
{
mqtt_outbox *oldest = ob->next;
if(oldest) {
return ob_del_ob(oldest);
}
return NULL;
}
int ob_del_expired(mqtt_outbox *ob, int current_tick, int timeout)
{
int del_count = 0;
mqtt_outbox *found = ob->next;
while(found != NULL) {
if(current_tick - found->tick_created > timeout) {
found = ob_del_ob(found);
del_count ++;
} else {
found = found->next;
}
}
return del_count;
}
int ob_get_size(mqtt_outbox *ob)
{
mqtt_outbox *found = ob->next;
int sz = 0;
while(found != NULL) {
sz += found->len;
found = found->next;
}
return sz;
}
int ob_cleanup(mqtt_outbox *ob, int max_size)
{
while(ob_get_size(ob) > max_size) {
ob_del_oldest(ob);
}
return ob_get_size(ob);
}
void ob_destroy(mqtt_outbox *ob)
{
while(ob_del_oldest(ob));
free(ob);
}

36
libraries/ESP8266MQTTClient/src/mqtt_outbox.h Normal file
View File

@@ -0,0 +1,36 @@
#ifndef _OUTBOX_H_
#define _OUTBOX_H_
#include "c_types.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct mqtt_outbox {
char *buffer;
int len;
int msg_id;
int msg_type;
int tick_created;
int retry_count;
int pending;
int remove_on_sent;
struct mqtt_outbox *next;
struct mqtt_outbox *prev;
} mqtt_outbox;
mqtt_outbox *ob_create();
mqtt_outbox *ob_put(mqtt_outbox *ob, uint8_t *data, int len, int msg_id, int msg_type, int tick, int remove_on_sent);
mqtt_outbox *ob_get(mqtt_outbox *ob, int msg_id);
int ob_del_id(mqtt_outbox *ob, int msg_id);
mqtt_outbox *ob_del_id_type(mqtt_outbox *ob, int msg_id, int msg_type);
mqtt_outbox *ob_del_ob(mqtt_outbox *del) ;
mqtt_outbox *ob_del_oldest(mqtt_outbox *ob);
mqtt_outbox *ob_get_oldest_no_pending(mqtt_outbox *ob);
int ob_del_expired(mqtt_outbox *ob, int current_tick, int timeout);
int ob_get_size(mqtt_outbox *ob);
int ob_cleanup(mqtt_outbox *ob, int max_size);
void ob_destroy(mqtt_outbox *ob);
mqtt_outbox *ob_get_top(mqtt_outbox *ob);
#ifdef __cplusplus
}
#endif
#endif

251
libraries/ESP8266MQTTClient/src/uri_parser.c Normal file
View File

@@ -0,0 +1,251 @@
/*
URI Parser
Copyright (c) 2016 Tuan PM (tuanpm@live.com)
Inspired by Hirochika Asai, http://draft.scyphus.co.jp/lang/c/url_parser.html
License (MIT license):
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.
*/
#include "uri_parser.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
static __inline__ int
_is_scheme_char(int c)
{
return (!isalpha(c) && '+' != c && '-' != c && '.' != c) ? 0 : 1;
}
#define JUMP_NEXT_STATE(var, state) { *curr_ptr = 0; curr_ptr ++; var = curr_ptr; parse_state = state; break;}
parsed_uri_t *parse_uri(const char *url)
{
parsed_uri_t *puri;
char *curr_ptr;
int bracket_flag;
enum parse_state_t {
PARSE_SCHEME = 0,
PARSE_USERNAME_OR_HOST,
PARSE_PASSWORD_OR_PORT,
PARSE_HOST,
PARSE_PORT,
PARSE_PATH,
PARSE_QUERY,
PARSE_FRAGMENT
} parse_state = 0;
puri = (parsed_uri_t *)malloc(sizeof(parsed_uri_t));
memset(puri, 0, sizeof(parsed_uri_t));
if(NULL == puri) {
return NULL;
}
puri->_uri_len = strlen(url);
puri->_uri = (char*) malloc(puri->_uri_len + 1);
memset(puri->_uri, 0, puri->_uri_len + 1);
if(puri->_uri == NULL) {
free_parsed_uri(puri);
return NULL;
}
strcpy(puri->_uri, url);
puri->_uri[puri->_uri_len] = 0;
puri->scheme = NULL;
puri->host = NULL;
puri->port = NULL;
puri->path = NULL;
puri->query = NULL;
puri->fragment = NULL;
puri->username = NULL;
puri->password = NULL;
curr_ptr = puri->_uri;
puri->scheme = curr_ptr;
parse_state = PARSE_SCHEME;
bracket_flag = 0;
while(*curr_ptr) {
// *curr_ptr = tolower((unsigned char)*curr_ptr);
switch(parse_state) {
case PARSE_SCHEME: /* parse scheme */
if(curr_ptr + 3 < (puri->_uri + puri->_uri_len) && memcmp(curr_ptr, "://", 3) == 0) {
*curr_ptr++ = 0;
*curr_ptr++ = 0;
*curr_ptr++ = 0;
puri->host = curr_ptr;
puri->username = curr_ptr;
parse_state = PARSE_USERNAME_OR_HOST; //next is username or host
break;
}
// if(!_is_scheme_char(*curr_ptr)) {
// free_parsed_uri(puri);
// return NULL;
// }
curr_ptr ++;
break;
case PARSE_USERNAME_OR_HOST: /* username or host*/
if('[' == *curr_ptr && bracket_flag == 0) {
bracket_flag = 1;
} else if(']' == *curr_ptr && bracket_flag == 1) {
bracket_flag = 0;
}
if(bracket_flag == 0 && *curr_ptr == ':') {
JUMP_NEXT_STATE(puri->port = puri->password, PARSE_PASSWORD_OR_PORT);
} else if(bracket_flag == 0 && *curr_ptr == '#') {
puri->username = NULL;
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
} else if(bracket_flag == 0 && *curr_ptr == '/') {
puri->username = NULL;
JUMP_NEXT_STATE(puri->path, PARSE_PATH);
}
curr_ptr ++;
break;
case PARSE_PASSWORD_OR_PORT: /* password or port */
if(*curr_ptr == '@') {
puri->port = NULL;
JUMP_NEXT_STATE(puri->host, PARSE_HOST);
break;
} else if(*curr_ptr == '/') {
puri->username = NULL;
puri->password = NULL;
JUMP_NEXT_STATE(puri->path, PARSE_PATH);
break;
} else if(*curr_ptr == '#') {
puri->username = NULL;
puri->password = NULL;
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
break;
}
curr_ptr ++;
break;
case PARSE_HOST: /* host */
if('[' == *curr_ptr && bracket_flag == 0) {
bracket_flag = 1;
} else if(']' == *curr_ptr && bracket_flag == 1) {
bracket_flag = 0;
}
if(bracket_flag == 0 && *curr_ptr == ':') {
JUMP_NEXT_STATE(puri->port, PARSE_PORT);
} else if(bracket_flag == 0 && *curr_ptr == '/') {
puri->port = NULL;
JUMP_NEXT_STATE(puri->path, PARSE_PATH);
} else if(bracket_flag == 0 && *curr_ptr == '#') {
puri->port = NULL;
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
}
curr_ptr ++;
break;
case PARSE_PORT: /* port */
if(*curr_ptr == '/') {
JUMP_NEXT_STATE(puri->path, PARSE_PATH);
} else if(*curr_ptr == '?') {
JUMP_NEXT_STATE(puri->query, PARSE_QUERY);
} else if(*curr_ptr == '#') {
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
}
curr_ptr ++;
break;
case PARSE_PATH: /* path */
if(*curr_ptr == '?') {
// JUMP_NEXT_STATE(puri->query, PARSE_QUERY);
} else if(*curr_ptr == '#') {
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
}
curr_ptr ++;
case PARSE_QUERY: /* query */
if(*curr_ptr == '#') {
JUMP_NEXT_STATE(puri->fragment, PARSE_FRAGMENT);
}
case PARSE_FRAGMENT: /* fragment*/
curr_ptr ++;
break;
}
}
if(parse_state < PARSE_HOST) {
puri->host = puri->username;
puri->port = puri->password;
puri->username = NULL;
puri->password = NULL;
}
if (puri->path && puri->path[0]!= 0){
char *temp = malloc(strlen(puri->path) + 2);
sprintf(temp, "/%s", puri->path);
puri->path = temp;
} else {
puri->path = malloc(2);
puri->path[0] = '/';
puri->path[1] = 0;
}
return puri;
}
void parse_uri_info(parsed_uri_t *puri)
{
printf( "scheme addr: %x\n"
"Username addr: %x\n"
"password addr: %x\n"
"host addr: %x\n"
"port addr: %x\n"
"path addr: %x\n"
"fragment addr: %x\n"
"extension addr: %x\n"
"host_ext addr: %x\r\n",
(int)puri->scheme,
(int)puri->username,
(int)puri->password,
(int)puri->host,
(int)puri->port,
(int)puri->path,
(int)puri->fragment,
(int)puri->extension,
(int)puri->host_ext);
if(puri->scheme && puri->scheme[0] != 0) {
printf("scheme: %s\n", puri->scheme);
}
if(puri->host && puri->host[0] != 0) {
printf("Host: %s\n", puri->host);
}
if(puri->path && puri->path[0] != 0) {
printf("path: %s\n", puri->path);
}
if(puri->port && puri->port[0] != 0) {
printf("port: %s\n", puri->port);
}
if(puri->username && puri->username[0] != 0) {
printf("username: %s\n", puri->username);
}
if(puri->password && puri->password[0] != 0) {
printf("password: %s\n", puri->password);
}
if(puri->fragment && puri->fragment[0] != 0) {
printf("fragment: %s\n", puri->fragment);
}
if(puri->extension && puri->extension[0] != 0) {
printf("extension: %s\n", puri->extension);
}
if(puri->host_ext && puri->host_ext[0] != 0) {
printf("host_ext: %s\n", puri->host_ext);
}
}
void free_parsed_uri(parsed_uri_t *puri)
{
if(NULL != puri) {
if(puri->path && puri->path[0] != 0) {
free(puri->path);
}
if(NULL != puri->_uri) {
free(puri->_uri);
}
free(puri);
}
}

28
libraries/ESP8266MQTTClient/src/uri_parser.h Normal file
View File

@@ -0,0 +1,28 @@
#ifndef _uri_parser_
#define _uri_parser_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
char *scheme; /* mandatory */
char *host; /* mandatory */
char *port; /* optional */
char *path; /* optional */
char *query; /* optional */
char *fragment; /* optional */
char *username; /* optional */
char *password; /* optional */
char *extension;
char *host_ext;
char *_uri; /* private */
int _uri_len; /* private */
} parsed_uri_t;
parsed_uri_t *parse_uri(const char *);
void free_parsed_uri(parsed_uri_t *);
void parse_uri_info(parsed_uri_t *puri);
#ifdef __cplusplus
}
#endif
#endif /* _uri_parser_ */