espressif/ESP8266_RTOS_SDK
1
0
Fork 0
mirror of https://github.com/espressif/ESP8266_RTOS_SDK.git synced 2025-05-20 16:46:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): Add esp-mqtt

Browse Source 
Commit ID: 13018449

Add global "mqtt" to manage both "ESP-MQTT" and "IBM-MQTT". Using select the target MQTT to using like selecting SSL.
This commit is contained in:
dongheng
2019-03-18 14:43:28 +08:00
parent 2ce8246b8f
commit 04ee11ee53
49 changed files with 2745 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
components/mqtt/CMakeLists.txt
View File

@ -1,3 +1,4 @@
if(CONFIG_MQTT_USING_IBM)
set(COMPONENT_ADD_INCLUDEDIRS
paho/MQTTClient-C/src
paho/MQTTClient-C/src/FreeRTOS
@ -9,7 +10,18 @@ set(COMPONENT_SRCDIRS
paho/MQTTPacket/src)
set(COMPONENT_REQUIRES freertos lwip ssl)
elif (CONFIG_MQTT_USING_ESP)
set(COMPONENT_ADD_INCLUDEDIRS esp-mqtt/include)
set(COMPONENT_PRIV_INCLUDEDIRS "esp-mqtt/lib/include")
set(COMPONENT_SRCS "esp-mqtt/mqtt_client.c"
"esp-mqtt/lib/mqtt_msg.c"
"esp-mqtt/lib/mqtt_outbox.c"
"esp-mqtt/lib/platform_esp32_idf.c")
set(COMPONENT_REQUIRES lwip http_parser mbedtls tcp_transport)
endif()
register_component()
target_compile_options(${COMPONENT_NAME} PUBLIC -DMQTT_TASK)

127
components/mqtt/Kconfig
View File

@ -1,4 +1,21 @@
menu "MQTT(Paho)"
menu "MQTT"
choice MQTT_LIBRARY_CHOOSE
prompt "Choose MQTT library"
default MQTT_USING_IBM
help
Choose the MQTT library which you want to use.
Currently we support ESP-MQTT and IBM-MQTT(paho).
config MQTT_USING_ESP
bool "ESP-MQTT"
config MQTT_USING_IBM
bool "IBM-MQTT(paho)"
endchoice
menu "IBM-MQTT(paho)"
depends on MQTT_USING_IBM
choice MQTT_VERSION
prompt "MQTT version"
@ -133,4 +150,110 @@ config MQTT_PING_TIMEOUT
If the ESP device does not receive any ping response within MQTT_PING_TIMEOUT,
it will terminate the MQTT connection.
endmenu
endmenu # IBM-MQTT(paho)
menu "ESP-MQTT"
depends on MQTT_USING_ESP
config MQTT_PROTOCOL_311
bool "Enable MQTT protocol 3.1.1"
default y
help
If not, this library will use MQTT protocol 3.1
config MQTT_TRANSPORT_SSL
bool "Enable MQTT over SSL"
default y
help
Enable MQTT transport over SSL with mbedtls
config MQTT_TRANSPORT_WEBSOCKET
bool "Enable MQTT over Websocket"
default y
help
Enable MQTT transport over Websocket.
config MQTT_TRANSPORT_WEBSOCKET_SECURE
bool "Enable MQTT over Websocket Secure"
default y
depends on MQTT_TRANSPORT_WEBSOCKET
depends on MQTT_TRANSPORT_SSL
help
Enable MQTT transport over Websocket Secure.
config MQTT_USE_CUSTOM_CONFIG
bool "MQTT Using custom configurations"
default n
help
Custom MQTT configurations.
config MQTT_TCP_DEFAULT_PORT
int "Default MQTT over TCP port"
default 1883
depends on MQTT_USE_CUSTOM_CONFIG
help
Default MQTT over TCP port
config MQTT_SSL_DEFAULT_PORT
int "Default MQTT over SSL port"
default 8883
depends on MQTT_USE_CUSTOM_CONFIG
depends on MQTT_TRANSPORT_SSL
help
Default MQTT over SSL port
config MQTT_WS_DEFAULT_PORT
int "Default MQTT over Websocket port"
default 80
depends on MQTT_USE_CUSTOM_CONFIG
depends on MQTT_TRANSPORT_WEBSOCKET
help
Default MQTT over Websocket port
config MQTT_WSS_DEFAULT_PORT
int "Default MQTT over Websocket Secure port"
default 443
depends on MQTT_USE_CUSTOM_CONFIG
depends on MQTT_TRANSPORT_WEBSOCKET
depends on MQTT_TRANSPORT_WEBSOCKET_SECURE
help
Default MQTT over Websocket Secure port
config MQTT_BUFFER_SIZE
int "Default MQTT Buffer Size"
default 1024
depends on MQTT_USE_CUSTOM_CONFIG
help
This buffer size using for both transmit and receive
config MQTT_TASK_STACK_SIZE
int "MQTT task stack size"
default 6144
depends on MQTT_USE_CUSTOM_CONFIG
help
MQTT task stack size
config MQTT_TASK_CORE_SELECTION_ENABLED
bool "Enable MQTT task core selection"
default false
help
This will enable core selection
choice MQTT_TASK_CORE_SELECTION
depends on MQTT_TASK_CORE_SELECTION_ENABLED
prompt "Core to use ?"
config MQTT_USE_CORE_0
bool "Core 0"
config MQTT_USE_CORE_1
bool "Core 1"
endchoice
config MQTT_CUSTOM_OUTBOX
bool "Enable custom outbox implementation"
default n
help
Set to true if a specific implementation of message outbox is needed (e.g. persistant outbox in NVM or similar).
endmenu # ESP-MQTT
endmenu

2
components/mqtt/Makefile.projbuild
View File

@ -1 +1,3 @@
ifdef CONFIG_MQTT_USING_IBM
CPPFLAGS += -DMQTT_TASK -DMQTTCLIENT_PLATFORM_HEADER=MQTTFreeRTOS.h
endif

26
components/mqtt/component.mk
View File

@ -1,12 +1,20 @@
#
# Component Makefile
#
COMPONENT_ADD_INCLUDEDIRS += paho/MQTTClient-C/src \
paho/MQTTClient-C/src/FreeRTOS \
paho/MQTTPacket/src
COMPONENT_SRCDIRS += paho/MQTTClient-C/src \
paho/MQTTClient-C/src/FreeRTOS \
paho/MQTTPacket/src
ifdef CONFIG_MQTT_USING_IBM
COMPONENT_ADD_INCLUDEDIRS += ibm-mqtt/MQTTClient-C/src \
ibm-mqtt/MQTTClient-C/src/FreeRTOS \
ibm-mqtt/MQTTPacket/src
COMPONENT_SRCDIRS += ibm-mqtt/MQTTClient-C/src \
ibm-mqtt/MQTTClient-C/src/FreeRTOS \
ibm-mqtt/MQTTPacket/src
CFLAGS += -DMQTT_TASK
else
ifdef CONFIG_MQTT_USING_ESP
COMPONENT_SUBMODULES += esp-mqtt
COMPONENT_ADD_INCLUDEDIRS := esp-mqtt/include
COMPONENT_SRCDIRS := esp-mqtt esp-mqtt/lib
COMPONENT_PRIV_INCLUDEDIRS := esp-mqtt/lib/include
endif
endif

34
components/mqtt/esp-mqtt/.editorconfig Normal file
View File

@ -0,0 +1,34 @@
# EditorConfig helps developers define and maintain consistent
# coding styles between different editors and IDEs
# http://editorconfig.org
root = true
[*]
indent_style = space
indent_size = 4
end_of_line = lf
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
[{*.md,*.rst}]
trim_trailing_whitespace = false
[{Makefile,*.mk,*.bat}]
indent_style = tab
indent_size = 2
[*/freertos/**]
indent_style = tab
indent_size = 4
[{*/freertos/**.S,**/FreeRTOSConfig.h}]
indent_style = space
indent_size = 4
[*.pem]
insert_final_newline = false
[*.py]
max_line_length = 119

36
components/mqtt/esp-mqtt/.gitignore vendored Normal file
View File

@ -0,0 +1,36 @@
# Object files
*.o
*.ko
*.obj
*.elf
# Precompiled Headers
*.gch
*.pch
# Libraries
*.lib
*.a
*.la
*.lo
# Shared objects (inc. Windows DLLs)
*.dll
*.so
*.so.*
*.dylib
# Executables
*.exe
*.out
*.app
*.i*86
*.x86_64
*.hex
# Debug files
*.dSYM/
*.su
build
examples/**/build
examples/**/sdkconfig*

59
components/mqtt/esp-mqtt/.travis.yml Normal file
View File

@ -0,0 +1,59 @@
sudo: false
language: bash
os:
- linux
addons:
apt:
packages:
- gperf
- python
- python-serial
before_install:
# Save path to the git respository
- PROJECT_PATH=$(pwd)
- CI_COMMIT_SHA=$(git rev-parse HEAD)
install:
# Install ESP32 toochain following steps as desribed
# in http://esp-idf.readthedocs.io/en/latest/linux-setup.html
#
# Get required packages - already done above, see addons: apt: packages:
# - sudo apt-get install git wget make libncurses-dev flex bison gperf python python-serial
# Prepare directory for the toolchain
- mkdir -p ~/esp
- cd ~/esp
# Download binary toolchain for the ESP32
- wget https://dl.espressif.com/dl/xtensa-esp32-elf-linux64-1.22.0-80-g6c4433a-5.2.0.tar.gz
- tar -xzf xtensa-esp32-elf-linux64-1.22.0-80-g6c4433a-5.2.0.tar.gz
# Get ESP-IDF from github
- git clone --recursive https://github.com/espressif/esp-idf.git
# Set the path to ESP-IDF directory
- export IDF_PATH=~/esp/esp-idf
- python -m pip install --user -r $IDF_PATH/requirements.txt
# Setup build tool: xtensa-esp32-elf and idf.py
- export PATH=$PATH:$HOME/esp/xtensa-esp32-elf/bin:$IDF_PATH/tools
script:
# Legacy build with IDF < 3.2
- cd $IDF_PATH && git checkout --recurse-submodules v3.1
- cd $PROJECT_PATH
- ./modify_for_legacy_idf.sh || true
- cd $PROJECT_PATH/examples/tcp
- make defconfig
- make -j4
# Master build with latest IDF
- cd $IDF_PATH && git checkout --recurse-submodules master
- cd $IDF_PATH/components/mqtt/esp-mqtt
- git remote add local $PROJECT_PATH/.git
- git fetch local
- git checkout $CI_COMMIT_SHA
- cd $IDF_PATH/examples/protocols/mqtt/tcp
- idf.py build
- cd $IDF_PATH/examples/protocols/mqtt/ssl
- idf.py build
- cd $IDF_PATH/examples/protocols/mqtt/ws
- idf.py build
- cd $IDF_PATH/examples/protocols/mqtt/wss
- idf.py build

202
components/mqtt/esp-mqtt/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 2016 Tuan PM
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.

27
components/mqtt/esp-mqtt/README.md Normal file
View File

@ -0,0 +1,27 @@
[![](https://travis-ci.org/tuanpmt/espmqtt.svg?branch=master)](https://travis-ci.org/tuanpmt/espmqtt)
[![](http://hits.dwyl.io/tuanpmt/espmqtt.svg)](http://hits.dwyl.io/tuanpmt/espmqtt)
[![Twitter Follow](https://img.shields.io/twitter/follow/tuanpmt.svg?style=social&label=Follow)](https://twitter.com/tuanpmt)
![GitHub contributors](https://img.shields.io/github/contributors/tuanpmt/espmqtt.svg)
# ESP32 MQTT Library
## Features
- Based on: https://github.com/tuanpmt/esp_mqtt
- Support MQTT over TCP, SSL with mbedtls, MQTT over Websocket, MQTT over Websocket Secure
- Easy to setup with URI
- Multiple instances (Multiple clients in one application)
- Support subscribing, publishing, authentication, will messages, keep alive pings and all 3 QoS levels (it should be a fully functional client).
## How to use
[ESP-MQTT](https://github.com/espressif/esp-mqtt) is a standard [ESP-IDF](https://github.com/espressif/esp-idf) component.
Please refer to instructions in [ESP-IDF](https://github.com/espressif/esp-idf)
## Documentation
Please refer to the standard [ESP-IDF](https://github.com/espressif/esp-idf) documentation here https://github.com/espressif/esp-idf/blob/master/docs/en/api-reference/protocols/mqtt.rst
## License
- MQTT Package - [Stephen Robinson - contiki-mqtt](https://github.com/esar/contiki-mqtt)
- Others [@tuanpmt](https://twitter.com/tuanpmt)
Apache License

19
components/mqtt/esp-mqtt/examples/emitter-client/CMakeLists.txt Normal file
View File

@ -0,0 +1,19 @@
cmake_minimum_required(VERSION 3.5)
get_filename_component(DEV_ROOT "${CMAKE_CURRENT_SOURCE_DIR}" ABSOLUTE)
set(PROJECT_ROOT "${DEV_ROOT}/")
set(SUBMODULE_ROOT "${DEV_ROOT}/../../../")
set(PROJECT_NAME "mqtt_ssl")
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
set(MAIN_SRCS ${PROJECT_ROOT}/main/app_main.c)
set(EXTRA_COMPONENT_DIRS "${EXTRA_COMPONENT_DIRS} ${SUBMODULE_ROOT}")
set(BUILD_COMPONENTS "${BUILD_COMPONENTS} espmqtt")
project(${PROJECT_NAME})

13
components/mqtt/esp-mqtt/examples/emitter-client/Makefile Normal file
View File

@ -0,0 +1,13 @@
#
# This is a project Makefile. It is assumed the directory this Makefile resides in is a
# project subdirectory.
#
#
# This is a project Makefile. It is assumed the directory this Makefile resides in is a
# project subdirectory.
#
PROJECT_NAME := emitter_client
EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../
include $(IDF_PATH)/make/project.mk

10
components/mqtt/esp-mqtt/examples/emitter-client/README.md Normal file
View File

@ -0,0 +1,10 @@
# ESPMQTT Emitter client
## Before you run this Example
- Register an account from https://emitter.io/
- Login and create channel key, grant access for the channel `/topic/` as the images bellow
- `make menuconfig` provide Wi-Fi information and CHANNEL_KEY to `MQTT Application example`
- `make flash monitor`
![](generate-key-0.png)
![](generate-key-1.png)

BIN
components/mqtt/esp-mqtt/examples/emitter-client/generate-key-0.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 234 KiB

BIN
components/mqtt/esp-mqtt/examples/emitter-client/generate-key-1.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 61 KiB

22
components/mqtt/esp-mqtt/examples/emitter-client/main/Kconfig.projbuild Normal file
View File

@ -0,0 +1,22 @@
menu "MQTT Application example"
config WIFI_SSID
string "WiFi SSID"
default "myssid"
help
SSID (network name) for the example to connect to.
config WIFI_PASSWORD
string "WiFi Password"
default "mypassword"
help
WiFi password (WPA or WPA2) for the example to use.
config EMITTER_CHANNEL_KEY
string "Emitter channel key"
default ""
help
The Emitter channel key using to pub/sub
endmenu

143
components/mqtt/esp-mqtt/examples/emitter-client/main/app_main.c Executable file
View File

@ -0,0 +1,143 @@
#include <stdio.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#include "esp_wifi.h"
#include "esp_system.h"
#include "nvs_flash.h"
#include "esp_event_loop.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/semphr.h"
#include "freertos/queue.h"
#include "freertos/event_groups.h"
#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"
#include "esp_log.h"
#include "mqtt_client.h"
static const char *TAG = "MQTTS_SAMPLE";
static EventGroupHandle_t wifi_event_group;
const static int CONNECTED_BIT = BIT0;
static esp_err_t wifi_event_handler(void *ctx, system_event_t *event)
{
switch (event->event_id) {
case SYSTEM_EVENT_STA_START:
esp_wifi_connect();
break;
case SYSTEM_EVENT_STA_GOT_IP:
xEventGroupSetBits(wifi_event_group, CONNECTED_BIT);
break;
case SYSTEM_EVENT_STA_DISCONNECTED:
esp_wifi_connect();
xEventGroupClearBits(wifi_event_group, CONNECTED_BIT);
break;
default:
break;
}
return ESP_OK;
}
static void wifi_init(void)
{
tcpip_adapter_init();
wifi_event_group = xEventGroupCreate();
ESP_ERROR_CHECK(esp_event_loop_init(wifi_event_handler, NULL));
wifi_init_config_t cfg = WIFI_INIT_CONFIG_DEFAULT();
ESP_ERROR_CHECK(esp_wifi_init(&cfg));
ESP_ERROR_CHECK(esp_wifi_set_storage(WIFI_STORAGE_RAM));
wifi_config_t wifi_config = {
.sta = {
.ssid = CONFIG_WIFI_SSID,
.password = CONFIG_WIFI_PASSWORD,
},
};
ESP_ERROR_CHECK(esp_wifi_set_mode(WIFI_MODE_STA));
ESP_ERROR_CHECK(esp_wifi_set_config(ESP_IF_WIFI_STA, &wifi_config));
ESP_LOGI(TAG, "start the WIFI SSID:[%s] password:[%s]", CONFIG_WIFI_SSID, "******");
ESP_ERROR_CHECK(esp_wifi_start());
ESP_LOGI(TAG, "Waiting for wifi");
xEventGroupWaitBits(wifi_event_group, CONNECTED_BIT, false, true, portMAX_DELAY);
}
static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
{
esp_mqtt_client_handle_t client = event->client;
int msg_id;
// your_context_t *context = event->context;
switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
msg_id = esp_mqtt_client_subscribe(client, CONFIG_EMITTER_CHANNEL_KEY"/topic/", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
break;
case MQTT_EVENT_DISCONNECTED:
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
break;
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
msg_id = esp_mqtt_client_publish(client, CONFIG_EMITTER_CHANNEL_KEY"/topic/", "data", 0, 0, 0);
ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA:
ESP_LOGI(TAG, "MQTT_EVENT_DATA");
printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
printf("DATA=%.*s\r\n", event->data_len, event->data);
break;
case MQTT_EVENT_ERROR:
ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
break;
}
return ESP_OK;
}
static void mqtt_app_start(void)
{
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtts://api.emitter.io:443", // for mqtt over ssl
// .uri = "mqtt://api.emitter.io:8080", //for mqtt over tcp
// .uri = "ws://api.emitter.io:8080", //for mqtt over websocket
// .uri = "wss://api.emitter.io:443", //for mqtt over websocket secure
.event_handle = mqtt_event_handler,
};
ESP_LOGI(TAG, "[APP] Free memory: %d bytes", esp_get_free_heap_size());
esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);
esp_mqtt_client_start(client);
}
void app_main()
{
ESP_LOGI(TAG, "[APP] Startup..");
ESP_LOGI(TAG, "[APP] Free memory: %d bytes", esp_get_free_heap_size());
ESP_LOGI(TAG, "[APP] IDF version: %s", esp_get_idf_version());
esp_log_level_set("*", ESP_LOG_INFO);
esp_log_level_set("MQTT_CLIENT", ESP_LOG_VERBOSE);
esp_log_level_set("TRANSPORT_TCP", ESP_LOG_VERBOSE);
esp_log_level_set("TRANSPORT_SSL", ESP_LOG_VERBOSE);
esp_log_level_set("TRANSPORT", ESP_LOG_VERBOSE);
esp_log_level_set("OUTBOX", ESP_LOG_VERBOSE);
nvs_flash_init();
wifi_init();
mqtt_app_start();
}

0
components/mqtt/esp-mqtt/examples/emitter-client/main/component.mk Normal file
View File

121
components/mqtt/esp-mqtt/include/mqtt_client.h Executable file
View File

@ -0,0 +1,121 @@
/*
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _MQTT_CLIENT_H_
#define _MQTT_CLIENT_H_
#include <stdint.h>
#include <stdbool.h>
#include <string.h>
#include "esp_err.h"
#include "mqtt_config.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct esp_mqtt_client* esp_mqtt_client_handle_t;
/**
* @brief MQTT event types.
*
* User event handler receives context data in `esp_mqtt_event_t` structure with
* - `user_context` - user data from `esp_mqtt_client_config_t`
* - `client` - mqtt client handle
* - various other data depending on event type
*
*/
typedef enum {
MQTT_EVENT_ERROR = 0,
MQTT_EVENT_CONNECTED, /*!< connected event, additional context: session_present flag */
MQTT_EVENT_DISCONNECTED, /*!< disconnected event */
MQTT_EVENT_SUBSCRIBED, /*!< subscribed event, additional context: msg_id */
MQTT_EVENT_UNSUBSCRIBED, /*!< unsubscribed event */
MQTT_EVENT_PUBLISHED, /*!< published event, additional context: msg_id */
MQTT_EVENT_DATA, /*!< data event, additional context:
- msg_id message id
- topic pointer to the received topic
- topic_len length of the topic
- data pointer to the received data
- data_len length of the data for this event
- current_data_offset offset of the current data for this event
- total_data_len total length of the data received
*/
} esp_mqtt_event_id_t;
typedef enum {
MQTT_TRANSPORT_UNKNOWN = 0x0,
MQTT_TRANSPORT_OVER_TCP, /*!< MQTT over TCP, using scheme: ``mqtt`` */
MQTT_TRANSPORT_OVER_SSL, /*!< MQTT over SSL, using scheme: ``mqtts`` */
MQTT_TRANSPORT_OVER_WS, /*!< MQTT over Websocket, using scheme:: ``ws`` */
MQTT_TRANSPORT_OVER_WSS /*!< MQTT over Websocket Secure, using scheme: ``wss`` */
} esp_mqtt_transport_t;
/**
* MQTT event configuration structure
*/
typedef struct {
esp_mqtt_event_id_t event_id; /*!< MQTT event type */
esp_mqtt_client_handle_t client; /*!< MQTT client handle for this event */
void *user_context; /*!< User context passed from MQTT client config */
char *data; /*!< Data asociated with this event */
int data_len; /*!< Lenght of the data for this event */
int total_data_len; /*!< Total length of the data (longer data are supplied with multiple events) */
int current_data_offset; /*!< Actual offset for the data asociated with this event */
char *topic; /*!< Topic asociated with this event */
int topic_len; /*!< Length of the topic for this event asociated with this event */
int msg_id; /*!< MQTT messaged id of message */
int session_present; /*!< MQTT session_present flag for connection event */
} esp_mqtt_event_t;
typedef esp_mqtt_event_t* esp_mqtt_event_handle_t;
typedef esp_err_t (* mqtt_event_callback_t)(esp_mqtt_event_handle_t event);
/**
* MQTT client configuration structure
*/
typedef struct {
mqtt_event_callback_t event_handle; /*!< handle for MQTT events */
const char *host; /*!< MQTT server domain (ipv4 as string) */
const char *uri; /*!< Complete MQTT broker URI */
uint32_t port; /*!< MQTT server port */
const char *client_id; /*!< default client id is ``ESP32_%CHIPID%`` where %CHIPID% are last 3 bytes of MAC address in hex format */
const char *username; /*!< MQTT username */
const char *password; /*!< MQTT password */
const char *lwt_topic; /*!< LWT (Last Will and Testament) message topic (NULL by default) */
const char *lwt_msg; /*!< LWT message (NULL by default) */
int lwt_qos; /*!< LWT message qos */
int lwt_retain; /*!< LWT retained message flag */
int lwt_msg_len; /*!< LWT message length */
int disable_clean_session; /*!< mqtt clean session, default clean_session is true */
int keepalive; /*!< mqtt keepalive, default is 120 seconds */
bool disable_auto_reconnect; /*!< this mqtt client will reconnect to server (when errors/disconnect). Set disable_auto_reconnect=true to disable */
void *user_context; /*!< pass user context to this option, then can receive that context in ``event->user_context`` */
int task_prio; /*!< MQTT task priority, default is 5, can be changed in ``make menuconfig`` */
int task_stack; /*!< MQTT task stack size, default is 6144 bytes, can be changed in ``make menuconfig`` */
int buffer_size; /*!< size of MQTT send/receive buffer, default is 1024 */
const char *cert_pem; /*!< Pointer to certificate data in PEM format for server verify (with SSL), default is NULL, not required to verify the server */
const char *client_cert_pem; /*!< Pointer to certificate data in PEM format for SSL mutual authentication, default is NULL, not required if mutual authentication is not needed. If it is not NULL, also `client_key_pem` has to be provided. */
const char *client_key_pem; /*!< Pointer to private key data in PEM format for SSL mutual authentication, default is NULL, not required if mutual authentication is not needed. If it is not NULL, also `client_cert_pem` has to be provided. */
esp_mqtt_transport_t transport; /*!< overrides URI transport */
} esp_mqtt_client_config_t;
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config);
esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *uri);
esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client);
esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client);
esp_err_t esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos);
esp_err_t esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic);
int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain);
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client);
#ifdef __cplusplus
}
#endif //__cplusplus
#endif

81
components/mqtt/esp-mqtt/include/mqtt_config.h Normal file
View File

@ -0,0 +1,81 @@
/*
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _MQTT_CONFIG_H_
#define _MQTT_CONFIG_H_
#include "sdkconfig.h"
#define MQTT_PROTOCOL_311 CONFIG_MQTT_PROTOCOL_311
#define MQTT_RECONNECT_TIMEOUT_MS (10*1000)
#if CONFIG_MQTT_BUFFER_SIZE
#define MQTT_BUFFER_SIZE_BYTE CONFIG_MQTT_BUFFER_SIZE
#else
#define MQTT_BUFFER_SIZE_BYTE 1024
#endif
#define MQTT_MAX_HOST_LEN 64
#define MQTT_MAX_CLIENT_LEN 32
#define MQTT_MAX_USERNAME_LEN 32
#define MQTT_MAX_PASSWORD_LEN 65
#define MQTT_MAX_LWT_TOPIC 32
#define MQTT_MAX_LWT_MSG 128
#define MQTT_TASK_PRIORITY 5
#if CONFIG_MQTT_TASK_STACK_SIZE
#define MQTT_TASK_STACK CONFIG_MQTT_TASK_STACK_SIZE
#else
#define MQTT_TASK_STACK (6*1024)
#endif
#define MQTT_KEEPALIVE_TICK (120)
#define MQTT_CMD_QUEUE_SIZE (10)
#define MQTT_NETWORK_TIMEOUT_MS (10000)
#ifdef CONFIG_MQTT_TCP_DEFAULT_PORT
#define MQTT_TCP_DEFAULT_PORT CONFIG_MQTT_TCP_DEFAULT_PORT
#else
#define MQTT_TCP_DEFAULT_PORT 1883
#endif
#ifdef CONFIG_MQTT_SSL_DEFAULT_PORT
#define MQTT_SSL_DEFAULT_PORT CONFIG_MQTT_SSL_DEFAULT_PORT
#else
#define MQTT_SSL_DEFAULT_PORT 8883
#endif
#ifdef CONFIG_MQTT_WS_DEFAULT_PORT
#define MQTT_WS_DEFAULT_PORT CONFIG_MQTT_WS_DEFAULT_PORT
#else
#define MQTT_WS_DEFAULT_PORT 80
#endif
#ifdef MQTT_WSS_DEFAULT_PORT
#define MQTT_WSS_DEFAULT_PORT CONFIG_MQTT_WSS_DEFAULT_PORT
#else
#define MQTT_WSS_DEFAULT_PORT 443
#endif
#define MQTT_CORE_SELECTION_ENABLED CONFIG_MQTT_TASK_CORE_SELECTION_ENABLED
#ifdef CONFIG_MQTT_USE_CORE_0
#define MQTT_TASK_CORE 0
#else
#ifdef CONFIG_MQTT_USE_CORE_1
#define MQTT_TASK_CORE 1
#else
#define MQTT_TASK_CORE 0
#endif
#endif
#define MQTT_ENABLE_SSL CONFIG_MQTT_TRANSPORT_SSL
#define MQTT_ENABLE_WS CONFIG_MQTT_TRANSPORT_WEBSOCKET
#define MQTT_ENABLE_WSS CONFIG_MQTT_TRANSPORT_WEBSOCKET_SECURE
#define OUTBOX_EXPIRED_TIMEOUT_MS (30*1000)
#define OUTBOX_MAX_SIZE (4*1024)
#endif

135
components/mqtt/esp-mqtt/lib/include/mqtt_msg.h Normal file
View File

@ -0,0 +1,135 @@
#ifndef MQTT_MSG_H
#define MQTT_MSG_H
#include "mqtt_config.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;
uint32_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
{
char* client_id;
char* username;
char* password;
char* will_topic;
char* will_message;
int keepalive;
int will_length;
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_session_present(uint8_t* buffer) { return buffer[2] & 0x01; }
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);
uint32_t mqtt_get_total_length(uint8_t* buffer, uint16_t length);
const char* mqtt_get_publish_topic(uint8_t* buffer, uint32_t* length);
const char* mqtt_get_publish_data(uint8_t* buffer, uint32_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 */

36
components/mqtt/esp-mqtt/lib/include/mqtt_outbox.h Normal file
View File

@ -0,0 +1,36 @@
/*
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _MQTT_OUTOBX_H_
#define _MQTT_OUTOBX_H_
#include "platform.h"
#ifdef __cplusplus
extern "C" {
#endif
struct outbox_item;
typedef struct outbox_list_t * outbox_handle_t;
typedef struct outbox_item * outbox_item_handle_t;
outbox_handle_t outbox_init();
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, uint8_t *data, int len, int msg_id, int msg_type, int tick);
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox);
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id);
esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type);
esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id);
esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type);
esp_err_t outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout);
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id);
int outbox_get_size(outbox_handle_t outbox);
esp_err_t outbox_cleanup(outbox_handle_t outbox, int max_size);
void outbox_destroy(outbox_handle_t outbox);
#ifdef __cplusplus
}
#endif
#endif

14
components/mqtt/esp-mqtt/lib/include/platform.h Normal file
View File

@ -0,0 +1,14 @@
/*
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _PLATFORM_H__
#define _PLATFORM_H__
//Support ESP32
#ifdef ESP_PLATFORM
#include "platform_esp32_idf.h"
#endif
#endif

35
components/mqtt/esp-mqtt/lib/include/platform_esp32_idf.h Normal file
View File

@ -0,0 +1,35 @@
/*
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _ESP_PLATFORM_H__
#define _ESP_PLATFORM_H__
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/semphr.h"
#include "freertos/queue.h"
#include "freertos/event_groups.h"
#include "lwip/err.h"
#include "lwip/sockets.h"
#include "lwip/sys.h"
#include "lwip/netdb.h"
#include "lwip/dns.h"
#include "rom/queue.h"
#include "esp_err.h"
#include "esp_log.h"
#include "esp_system.h"
char *platform_create_id_string();
int platform_random(int max);
long long platform_tick_get_ms();
void ms_to_timeval(int timeout_ms, struct timeval *tv);
#define ESP_MEM_CHECK(TAG, a, action) if (!(a)) { \
ESP_LOGE(TAG,"%s:%d (%s): %s", __FILE__, __LINE__, __FUNCTION__, "Memory exhausted"); \
action; \
}
#endif

472
components/mqtt/esp-mqtt/lib/mqtt_msg.c Normal file
View File

@ -0,0 +1,472 @@
/*
* 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 <stdint.h>
#include <string.h>
#include "mqtt_msg.h"
#include "mqtt_config.h"
#include "platform.h"
#define MQTT_MAX_FIXED_HEADER_SIZE 3
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(MQTT_PROTOCOL_311)
uint8_t magic[4];
#else
uint8_t magic[6];
#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 = platform_random(65535);
}
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;
}
uint32_t mqtt_get_total_length(uint8_t* buffer, uint16_t length)
{
int i;
uint32_t 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;
}
const char* mqtt_get_publish_topic(uint8_t* buffer, uint32_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 (const char*)(buffer + i);
}
const char* mqtt_get_publish_data(uint8_t* buffer, uint32_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 (const 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:
{
// 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(CONFIG_MQTT_PROTOCOL_311)
variable_header->lengthLsb = 4;
memcpy(variable_header->magic, "MQTT", 4);
variable_header->version = 4;
#else
variable_header->lengthLsb = 6;
memcpy(variable_header->magic, "MQIsdp", 6);
variable_header->version = 3;
#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 && info->client_id[0] != '\0')
{
if (append_string(connection, info->client_id, strlen(info->client_id)) < 0)
return fail_message(connection);
}
else
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, info->will_length) < 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);
}

170
components/mqtt/esp-mqtt/lib/mqtt_outbox.c Normal file
View File

@ -0,0 +1,170 @@
#include "mqtt_outbox.h"
#include <stdlib.h>
#include <string.h>
#include "rom/queue.h"
#include "esp_log.h"
#ifndef CONFIG_MQTT_CUSTOM_OUTBOX
static const char *TAG = "OUTBOX";
typedef struct outbox_item {
char *buffer;
int len;
int msg_id;
int msg_type;
int tick;
int retry_count;
bool pending;
STAILQ_ENTRY(outbox_item) next;
} outbox_item_t;
STAILQ_HEAD(outbox_list_t, outbox_item);
outbox_handle_t outbox_init()
{
outbox_handle_t outbox = calloc(1, sizeof(struct outbox_list_t));
ESP_MEM_CHECK(TAG, outbox, return NULL);
STAILQ_INIT(outbox);
return outbox;
}
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, uint8_t *data, int len, int msg_id, int msg_type, int tick)
{
outbox_item_handle_t item = calloc(1, sizeof(outbox_item_t));
ESP_MEM_CHECK(TAG, item, return NULL);
item->msg_id = msg_id;
item->msg_type = msg_type;
item->tick = tick;
item->len = len;
item->buffer = malloc(len);
ESP_MEM_CHECK(TAG, item->buffer, {
free(item);
return NULL;
});
memcpy(item->buffer, data, len);
STAILQ_INSERT_TAIL(outbox, item, next);
ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%d", msg_id, msg_type, len, outbox_get_size(outbox));
return item;
}
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (item->msg_id == msg_id) {
return item;
}
}
return NULL;
}
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (!item->pending) {
return item;
}
}
return NULL;
}
esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_id == msg_id && item->msg_type == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
ESP_LOGD(TAG, "DELETED msgid=%d, msg_type=%d, remain size=%d", msg_id, msg_type, outbox_get_size(outbox));
return ESP_OK;
}
}
return ESP_FAIL;
}
esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_id == msg_id) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
}
}
return ESP_OK;
}
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id)
{
outbox_item_handle_t item = outbox_get(outbox, msg_id);
if (item) {
item->pending = true;
return ESP_OK;
}
return ESP_FAIL;
}
esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_type == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
}
}
return ESP_OK;
}
esp_err_t outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (current_tick - item->tick > timeout) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
}
}
return ESP_OK;
}
int outbox_get_size(outbox_handle_t outbox)
{
int siz = 0;
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
siz += item->len;
}
return siz;
}
esp_err_t outbox_cleanup(outbox_handle_t outbox, int max_size)
{
while(outbox_get_size(outbox) > max_size) {
outbox_item_handle_t item = outbox_dequeue(outbox);
if (item == NULL) {
return ESP_FAIL;
}
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
}
return ESP_OK;
}
void outbox_destroy(outbox_handle_t outbox)
{
outbox_cleanup(outbox, 0);
free(outbox);
}
#endif /* CONFIG_MQTT_CUSTOM_OUTBOX */

41
components/mqtt/esp-mqtt/lib/platform_idf.c Normal file
View File

@ -0,0 +1,41 @@
#include "platform.h"
#ifdef ESP_PLATFORM
#include "esp_system.h"
#include "esp_log.h"
#include <sys/time.h>
static const char *TAG = "PLATFORM";
#define MAX_ID_STRING (32)
char *platform_create_id_string()
{
uint8_t mac[6];
char *id_string = calloc(1, MAX_ID_STRING);
ESP_MEM_CHECK(TAG, id_string, return NULL);
esp_read_mac(mac, ESP_MAC_WIFI_STA);
sprintf(id_string, "ESP32_%02x%02X%02X", mac[3], mac[4], mac[5]);
return id_string;
}
int platform_random(int max)
{
return esp_random()%max;
}
long long platform_tick_get_ms()
{
struct timeval te;
gettimeofday(&te, NULL); // get current time
long long milliseconds = te.tv_sec*1000LL + te.tv_usec/1000; // calculate milliseconds
// printf("milliseconds: %lld\n", milliseconds);
return milliseconds;
}
void ms_to_timeval(int timeout_ms, struct timeval *tv)
{
tv->tv_sec = timeout_ms / 1000;
tv->tv_usec = (timeout_ms - (tv->tv_sec * 1000)) * 1000;
}
#endif

27
components/mqtt/esp-mqtt/modify_for_legacy_idf.sh Executable file
View File

@ -0,0 +1,27 @@
#!/usr/bin/env bash
# This snipped prepares environment for using esp-mqtt repository separately from idf -- legacy use before IDFv3.2
#
esp_mqtt_path=`pwd`
mkdir -p ${esp_mqtt_path}/examples
pushd
cd $IDF_PATH
former_commit_id=`git rev-parse HEAD`
git checkout master
for example in tcp; do
cp -r $IDF_PATH/examples/protocols/mqtt/${example} ${esp_mqtt_path}/examples
echo 'EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../' > ${esp_mqtt_path}/examples/${example}/Makefile
cat $IDF_PATH/examples/protocols/mqtt/${example}/Makefile >> ${esp_mqtt_path}/examples/${example}/Makefile
echo "CONFIG_MQTT_TRANSPORT_SSL=" >> ${esp_mqtt_path}/examples/${example}/sdkconfig.defaults
echo "CONFIG_MQTT_TRANSPORT_WEBSOCKET=" >> ${esp_mqtt_path}/examples/${example}/sdkconfig.defaults
done
cp -r $IDF_PATH/components/tcp_transport ${esp_mqtt_path}/..
rm ${esp_mqtt_path}/../tcp_transport/transport_ssl.c
echo -e "#include \"esp_transport.h\"\nvoid esp_transport_ws_set_path(esp_transport_handle_t t, const char *path) {}" > ${esp_mqtt_path}/../tcp_transport/transport_ws.c
cp $IDF_PATH/components/mqtt/Kconfig ${esp_mqtt_path}
sed 's/esp-mqtt/\./g' $IDF_PATH/components/mqtt/component.mk > ${esp_mqtt_path}/component.mk
git checkout $former_commit_id
popd

892
components/mqtt/esp-mqtt/mqtt_client.c Normal file
View File

@ -0,0 +1,892 @@
#include <stdio.h>
#include "platform.h"
#include "mqtt_client.h"
#include "mqtt_msg.h"
#include "esp_transport.h"
#include "esp_transport_tcp.h"
#include "esp_transport_ssl.h"
#include "esp_transport_ws.h"
#include "platform.h"
#include "mqtt_outbox.h"
/* using uri parser */
#include "http_parser.h"
static const char *TAG = "MQTT_CLIENT";
typedef struct mqtt_state
{
mqtt_connect_info_t *connect_info;
uint8_t *in_buffer;
uint8_t *out_buffer;
int in_buffer_length;
int out_buffer_length;
uint32_t message_length;
uint32_t message_length_read;
mqtt_message_t *outbound_message;
mqtt_connection_t mqtt_connection;
uint16_t pending_msg_id;
int pending_msg_type;
int pending_publish_qos;
int pending_msg_count;
} mqtt_state_t;
typedef struct {
mqtt_event_callback_t event_handle;
int task_stack;
int task_prio;
char *uri;
char *host;
char *path;
char *scheme;
int port;
bool auto_reconnect;
void *user_context;
int network_timeout_ms;
} mqtt_config_storage_t;
typedef enum {
MQTT_STATE_ERROR = -1,
MQTT_STATE_UNKNOWN = 0,
MQTT_STATE_INIT,
MQTT_STATE_CONNECTED,
MQTT_STATE_WAIT_TIMEOUT,
} mqtt_client_state_t;
struct esp_mqtt_client {
esp_transport_list_handle_t transport_list;
esp_transport_handle_t transport;
mqtt_config_storage_t *config;
mqtt_state_t mqtt_state;
mqtt_connect_info_t connect_info;
mqtt_client_state_t state;
long long keepalive_tick;
long long reconnect_tick;
int wait_timeout_ms;
int auto_reconnect;
esp_mqtt_event_t event;
bool run;
bool wait_for_ping_resp;
outbox_handle_t outbox;
EventGroupHandle_t status_bits;
};
const static int STOPPED_BIT = BIT0;
static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config);
static esp_err_t esp_mqtt_destroy_config(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_ms);
static esp_err_t esp_mqtt_abort_connection(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_client_ping(esp_mqtt_client_handle_t client);
static char *create_string(const char *ptr, int len);
static esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config)
{
//Copy user configurations to client context
esp_err_t err = ESP_OK;
mqtt_config_storage_t *cfg = calloc(1, sizeof(mqtt_config_storage_t));
ESP_MEM_CHECK(TAG, cfg, return ESP_ERR_NO_MEM);
client->config = cfg;
cfg->task_prio = config->task_prio;
if (cfg->task_prio <= 0) {
cfg->task_prio = MQTT_TASK_PRIORITY;
}
cfg->task_stack = config->task_stack;
if (cfg->task_stack == 0) {
cfg->task_stack = MQTT_TASK_STACK;
}
err = ESP_ERR_NO_MEM;
if (config->host) {
cfg->host = strdup(config->host);
ESP_MEM_CHECK(TAG, cfg->host, goto _mqtt_set_config_failed);
}
cfg->port = config->port;
if (config->username) {
client->connect_info.username = strdup(config->username);
ESP_MEM_CHECK(TAG, client->connect_info.username, goto _mqtt_set_config_failed);
}
if (config->password) {
client->connect_info.password = strdup(config->password);
ESP_MEM_CHECK(TAG, client->connect_info.password, goto _mqtt_set_config_failed);
}
if (config->client_id) {
client->connect_info.client_id = strdup(config->client_id);
} else {
client->connect_info.client_id = platform_create_id_string();
}
ESP_MEM_CHECK(TAG, client->connect_info.client_id, goto _mqtt_set_config_failed);
ESP_LOGD(TAG, "MQTT client_id=%s", client->connect_info.client_id);
if (config->uri) {
cfg->uri = strdup(config->uri);
ESP_MEM_CHECK(TAG, cfg->uri, goto _mqtt_set_config_failed);
}
if (config->lwt_topic) {
client->connect_info.will_topic = strdup(config->lwt_topic);
ESP_MEM_CHECK(TAG, client->connect_info.will_topic, goto _mqtt_set_config_failed);
}
if (config->lwt_msg_len) {
client->connect_info.will_message = malloc(config->lwt_msg_len);
ESP_MEM_CHECK(TAG, client->connect_info.will_message, goto _mqtt_set_config_failed);
memcpy(client->connect_info.will_message, config->lwt_msg, config->lwt_msg_len);
client->connect_info.will_length = config->lwt_msg_len;
} else if (config->lwt_msg) {
client->connect_info.will_message = strdup(config->lwt_msg);
ESP_MEM_CHECK(TAG, client->connect_info.will_message, goto _mqtt_set_config_failed);
client->connect_info.will_length = strlen(config->lwt_msg);
}
client->connect_info.will_qos = config->lwt_qos;
client->connect_info.will_retain = config->lwt_retain;
client->connect_info.clean_session = 1;
if (config->disable_clean_session) {
client->connect_info.clean_session = false;
}
client->connect_info.keepalive = config->keepalive;
if (client->connect_info.keepalive == 0) {
client->connect_info.keepalive = MQTT_KEEPALIVE_TICK;
}
cfg->network_timeout_ms = MQTT_NETWORK_TIMEOUT_MS;
cfg->user_context = config->user_context;
cfg->event_handle = config->event_handle;
cfg->auto_reconnect = true;
if (config->disable_auto_reconnect) {
cfg->auto_reconnect = false;
}
return err;
_mqtt_set_config_failed:
esp_mqtt_destroy_config(client);
return err;
}
static esp_err_t esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
{
mqtt_config_storage_t *cfg = client->config;
free(cfg->host);
free(cfg->uri);
free(cfg->path);
free(cfg->scheme);
free(client->connect_info.will_topic);
free(client->connect_info.will_message);
free(client->connect_info.client_id);
free(client->connect_info.username);
free(client->connect_info.password);
free(client->config);
return ESP_OK;
}
static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_ms)
{
int write_len, read_len, connect_rsp_code;
client->wait_for_ping_resp = false;
mqtt_msg_init(&client->mqtt_state.mqtt_connection,
client->mqtt_state.out_buffer,
client->mqtt_state.out_buffer_length);
client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection,
client->mqtt_state.connect_info);
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
ESP_LOGI(TAG, "Sending MQTT CONNECT message, type: %d, id: %04X",
client->mqtt_state.pending_msg_type,
client->mqtt_state.pending_msg_id);
write_len = esp_transport_write(client->transport,
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
if (write_len < 0) {
ESP_LOGE(TAG, "Writing failed, errno= %d", errno);
return ESP_FAIL;
}
read_len = esp_transport_read(client->transport,
(char *)client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_length,
client->config->network_timeout_ms);
if (read_len < 0) {
ESP_LOGE(TAG, "Error network response");
return ESP_FAIL;
}
if (mqtt_get_type(client->mqtt_state.in_buffer) != MQTT_MSG_TYPE_CONNACK) {
ESP_LOGE(TAG, "Invalid MSG_TYPE response: %d, read_len: %d", mqtt_get_type(client->mqtt_state.in_buffer), read_len);
return ESP_FAIL;
}
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
switch (connect_rsp_code) {
case CONNECTION_ACCEPTED:
ESP_LOGD(TAG, "Connected");
return ESP_OK;
case CONNECTION_REFUSE_PROTOCOL:
ESP_LOGW(TAG, "Connection refused, bad protocol");
return ESP_FAIL;
case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Connection refused, server unavailable");
return ESP_FAIL;
case CONNECTION_REFUSE_BAD_USERNAME:
ESP_LOGW(TAG, "Connection refused, bad username or password");
return ESP_FAIL;
case CONNECTION_REFUSE_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Connection refused, not authorized");
return ESP_FAIL;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
return ESP_FAIL;
}
return ESP_OK;
}
static esp_err_t esp_mqtt_abort_connection(esp_mqtt_client_handle_t client)
{
esp_transport_close(client->transport);
client->wait_timeout_ms = MQTT_RECONNECT_TIMEOUT_MS;
client->reconnect_tick = platform_tick_get_ms();
client->state = MQTT_STATE_WAIT_TIMEOUT;
ESP_LOGI(TAG, "Reconnect after %d ms", client->wait_timeout_ms);
client->event.event_id = MQTT_EVENT_DISCONNECTED;
client->wait_for_ping_resp = false;
esp_mqtt_dispatch_event(client);
return ESP_OK;
}
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config)
{
esp_mqtt_client_handle_t client = calloc(1, sizeof(struct esp_mqtt_client));
ESP_MEM_CHECK(TAG, client, return NULL);
esp_mqtt_set_config(client, config);
client->transport_list = esp_transport_list_init();
ESP_MEM_CHECK(TAG, client->transport_list, goto _mqtt_init_failed);
esp_transport_handle_t tcp = esp_transport_tcp_init();
ESP_MEM_CHECK(TAG, tcp, goto _mqtt_init_failed);
esp_transport_set_default_port(tcp, MQTT_TCP_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, tcp, "mqtt");
if (config->transport == MQTT_TRANSPORT_OVER_TCP) {
client->config->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#if MQTT_ENABLE_WS
esp_transport_handle_t ws = esp_transport_ws_init(tcp);
ESP_MEM_CHECK(TAG, ws, goto _mqtt_init_failed);
esp_transport_set_default_port(ws, MQTT_WS_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, ws, "ws");
if (config->transport == MQTT_TRANSPORT_OVER_WS) {
client->config->scheme = create_string("ws", 2);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
#if MQTT_ENABLE_SSL
esp_transport_handle_t ssl = esp_transport_ssl_init();
ESP_MEM_CHECK(TAG, ssl, goto _mqtt_init_failed);
esp_transport_set_default_port(ssl, MQTT_SSL_DEFAULT_PORT);
if (config->cert_pem) {
esp_transport_ssl_set_cert_data(ssl, config->cert_pem, strlen(config->cert_pem));
}
if (config->client_cert_pem) {
esp_transport_ssl_set_client_cert_data(ssl, config->client_cert_pem, strlen(config->client_cert_pem));
}
if (config->client_key_pem) {
esp_transport_ssl_set_client_key_data(ssl, config->client_key_pem, strlen(config->client_key_pem));
}
esp_transport_list_add(client->transport_list, ssl, "mqtts");
if (config->transport == MQTT_TRANSPORT_OVER_SSL) {
client->config->scheme = create_string("mqtts", 5);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
#if MQTT_ENABLE_WSS
esp_transport_handle_t wss = esp_transport_ws_init(ssl);
ESP_MEM_CHECK(TAG, wss, goto _mqtt_init_failed);
esp_transport_set_default_port(wss, MQTT_WSS_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, wss, "wss");
if (config->transport == MQTT_TRANSPORT_OVER_WSS) {
client->config->scheme = create_string("wss", 3);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
if (client->config->uri) {
if (esp_mqtt_client_set_uri(client, client->config->uri) != ESP_OK) {
goto _mqtt_init_failed;
}
}
if (client->config->scheme == NULL) {
client->config->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
client->keepalive_tick = platform_tick_get_ms();
client->reconnect_tick = platform_tick_get_ms();
client->wait_for_ping_resp = false;
int buffer_size = config->buffer_size;
if (buffer_size <= 0) {
buffer_size = MQTT_BUFFER_SIZE_BYTE;
}
client->mqtt_state.in_buffer = (uint8_t *)malloc(buffer_size);
ESP_MEM_CHECK(TAG, client->mqtt_state.in_buffer, goto _mqtt_init_failed);
client->mqtt_state.in_buffer_length = buffer_size;
client->mqtt_state.out_buffer = (uint8_t *)malloc(buffer_size);
ESP_MEM_CHECK(TAG, client->mqtt_state.out_buffer, goto _mqtt_init_failed);
client->mqtt_state.out_buffer_length = buffer_size;
client->mqtt_state.connect_info = &client->connect_info;
client->outbox = outbox_init();
ESP_MEM_CHECK(TAG, client->outbox, goto _mqtt_init_failed);
client->status_bits = xEventGroupCreate();
ESP_MEM_CHECK(TAG, client->status_bits, goto _mqtt_init_failed);
return client;
_mqtt_init_failed:
esp_mqtt_client_destroy(client);
return NULL;
}
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client)
{
esp_mqtt_client_stop(client);
esp_mqtt_destroy_config(client);
esp_transport_list_destroy(client->transport_list);
outbox_destroy(client->outbox);
vEventGroupDelete(client->status_bits);
free(client->mqtt_state.in_buffer);
free(client->mqtt_state.out_buffer);
free(client);
return ESP_OK;
}
static char *create_string(const char *ptr, int len)
{
char *ret;
if (len <= 0) {
return NULL;
}
ret = calloc(1, len + 1);
ESP_MEM_CHECK(TAG, ret, return NULL);
memcpy(ret, ptr, len);
return ret;
}
esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *uri)
{
struct http_parser_url puri;
http_parser_url_init(&puri);
int parser_status = http_parser_parse_url(uri, strlen(uri), 0, &puri);
if (parser_status != 0) {
ESP_LOGE(TAG, "Error parse uri = %s", uri);
return ESP_FAIL;
}
if (client->config->scheme == NULL) {
client->config->scheme = create_string(uri + puri.field_data[UF_SCHEMA].off, puri.field_data[UF_SCHEMA].len);
}
if (client->config->host == NULL) {
client->config->host = create_string(uri + puri.field_data[UF_HOST].off, puri.field_data[UF_HOST].len);
}
if (client->config->path == NULL) {
client->config->path = create_string(uri + puri.field_data[UF_PATH].off, puri.field_data[UF_PATH].len);
}
if (client->config->path) {
esp_transport_handle_t trans = esp_transport_list_get_transport(client->transport_list, "ws");
if (trans) {
esp_transport_ws_set_path(trans, client->config->path);
}
trans = esp_transport_list_get_transport(client->transport_list, "wss");
if (trans) {
esp_transport_ws_set_path(trans, client->config->path);
}
}
if (puri.field_data[UF_PORT].len) {
client->config->port = strtol((const char*)(uri + puri.field_data[UF_PORT].off), NULL, 10);
}
char *user_info = create_string(uri + puri.field_data[UF_USERINFO].off, puri.field_data[UF_USERINFO].len);
if (user_info) {
char *pass = strchr(user_info, ':');
if (pass) {
pass[0] = 0; //terminal username
pass ++;
client->connect_info.password = strdup(pass);
}
client->connect_info.username = strdup(user_info);
free(user_info);
}
return ESP_OK;
}
static esp_err_t mqtt_write_data(esp_mqtt_client_handle_t client)
{
int write_len = esp_transport_write(client->transport,
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
// client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
if (write_len <= 0) {
ESP_LOGE(TAG, "Error write data or timeout, written len = %d", write_len);
return ESP_FAIL;
}
/* we've just sent a mqtt control packet, update keepalive counter
* [MQTT-3.1.2-23]
*/
client->keepalive_tick = platform_tick_get_ms();
return ESP_OK;
}
static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client)
{
client->event.msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
client->event.user_context = client->config->user_context;
client->event.client = client;
if (client->config->event_handle) {
return client->config->event_handle(&client->event);
}
return ESP_FAIL;
}
typedef struct {
char *path;
char *buffer;
esp_transport_handle_t parent;
} transport_ws_t;
static void deliver_publish(esp_mqtt_client_handle_t client, uint8_t *message, int length)
{
const char *mqtt_topic, *mqtt_data;
uint32_t mqtt_topic_length, mqtt_data_length;
uint32_t mqtt_len, mqtt_offset = 0, total_mqtt_len = 0;
int len_read;
esp_transport_handle_t transport = client->transport;
do
{
if (total_mqtt_len == 0) {
mqtt_topic_length = length;
mqtt_topic = mqtt_get_publish_topic(message, &mqtt_topic_length);
mqtt_data_length = length;
mqtt_data = mqtt_get_publish_data(message, &mqtt_data_length);
total_mqtt_len = client->mqtt_state.message_length - client->mqtt_state.message_length_read + mqtt_data_length;
mqtt_len = mqtt_data_length;
/* any further reading only the underlying payload */
transport = esp_transport_get_payload_transport_handle(transport);
} else {
mqtt_len = len_read;
mqtt_data = (const char*)client->mqtt_state.in_buffer;
mqtt_topic = NULL;
mqtt_topic_length = 0;
}
ESP_LOGD(TAG, "Get data len= %d, topic len=%d", mqtt_len, mqtt_topic_length);
client->event.event_id = MQTT_EVENT_DATA;
client->event.data = (char *)mqtt_data;
client->event.data_len = mqtt_len;
client->event.total_data_len = total_mqtt_len;
client->event.current_data_offset = mqtt_offset;
client->event.topic = (char *)mqtt_topic;
client->event.topic_len = mqtt_topic_length;
esp_mqtt_dispatch_event(client);
mqtt_offset += mqtt_len;
if (client->mqtt_state.message_length_read >= client->mqtt_state.message_length) {
break;
}
len_read = esp_transport_read(transport,
(char *)client->mqtt_state.in_buffer,
client->mqtt_state.message_length - client->mqtt_state.message_length_read > client->mqtt_state.in_buffer_length ?
client->mqtt_state.in_buffer_length : client->mqtt_state.message_length - client->mqtt_state.message_length_read,
client->config->network_timeout_ms);
if (len_read <= 0) {
ESP_LOGE(TAG, "Read error or timeout: %d", errno);
break;
}
client->mqtt_state.message_length_read += len_read;
} while (1);
}
static bool is_valid_mqtt_msg(esp_mqtt_client_handle_t client, int msg_type, int msg_id)
{
ESP_LOGD(TAG, "pending_id=%d, pending_msg_count = %d", client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_count);
if (client->mqtt_state.pending_msg_count == 0) {
return false;
}
if (outbox_delete(client->outbox, msg_id, msg_type) == ESP_OK) {
client->mqtt_state.pending_msg_count --;
return true;
}
if (client->mqtt_state.pending_msg_type == msg_type && client->mqtt_state.pending_msg_id == msg_id) {
client->mqtt_state.pending_msg_count --;
return true;
}
return false;
}
static void mqtt_enqueue(esp_mqtt_client_handle_t client)
{
ESP_LOGD(TAG, "mqtt_enqueue id: %d, type=%d successful",
client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_type);
//lock mutex
if (client->mqtt_state.pending_msg_count > 0) {
//Copy to queue buffer
outbox_enqueue(client->outbox,
client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->mqtt_state.pending_msg_id,
client->mqtt_state.pending_msg_type,
platform_tick_get_ms());
}
//unlock
}
static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
{
int read_len;
uint8_t msg_type;
uint8_t msg_qos;
uint16_t msg_id;
read_len = esp_transport_read(client->transport, (char *)client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length, 1000);
if (read_len < 0) {
ESP_LOGE(TAG, "Read error or end of stream");
return ESP_FAIL;
}
if (read_len == 0) {
return ESP_OK;
}
msg_type = mqtt_get_type(client->mqtt_state.in_buffer);
msg_qos = mqtt_get_qos(client->mqtt_state.in_buffer);
msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
ESP_LOGD(TAG, "msg_type=%d, msg_id=%d", msg_type, msg_id);
switch (msg_type)
{
case MQTT_MSG_TYPE_SUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "Subscribe successful");
client->event.event_id = MQTT_EVENT_SUBSCRIBED;
esp_mqtt_dispatch_event(client);
}
break;
case MQTT_MSG_TYPE_UNSUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "UnSubscribe successful");
client->event.event_id = MQTT_EVENT_UNSUBSCRIBED;
esp_mqtt_dispatch_event(client);
}
break;
case MQTT_MSG_TYPE_PUBLISH:
if (msg_qos == 1) {
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
}
else if (msg_qos == 2) {
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
}
if (msg_qos == 1 || msg_qos == 2) {
ESP_LOGD(TAG, "Queue response QoS: %d", msg_qos);
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error write qos msg repsonse, qos = %d", msg_qos);
// TODO: Shoule reconnect?
// return ESP_FAIL;
}
}
client->mqtt_state.message_length_read = read_len;
client->mqtt_state.message_length = mqtt_get_total_length(client->mqtt_state.in_buffer, client->mqtt_state.message_length_read);
ESP_LOGI(TAG, "deliver_publish, message_length_read=%d, message_length=%d", read_len, client->mqtt_state.message_length);
deliver_publish(client, client->mqtt_state.in_buffer, client->mqtt_state.message_length_read);
break;
case MQTT_MSG_TYPE_PUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
}
break;
case MQTT_MSG_TYPE_PUBREC:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREC");
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBREL:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREL");
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBCOMP:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
}
break;
case MQTT_MSG_TYPE_PINGRESP:
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PINGRESP");
client->wait_for_ping_resp = false;
break;
}
return ESP_OK;
}
static void esp_mqtt_task(void *pv)
{
esp_mqtt_client_handle_t client = (esp_mqtt_client_handle_t) pv;
client->run = true;
//get transport by scheme
client->transport = esp_transport_list_get_transport(client->transport_list, client->config->scheme);
if (client->transport == NULL) {
ESP_LOGE(TAG, "There are no transports valid, stop mqtt client, config scheme = %s", client->config->scheme);
client->run = false;
}
//default port
if (client->config->port == 0) {
client->config->port = esp_transport_get_default_port(client->transport);
}
client->state = MQTT_STATE_INIT;
xEventGroupClearBits(client->status_bits, STOPPED_BIT);
while (client->run) {
switch ((int)client->state) {
case MQTT_STATE_INIT:
if (client->transport == NULL) {
ESP_LOGE(TAG, "There are no transport");
client->run = false;
}
if (esp_transport_connect(client->transport,
client->config->host,
client->config->port,
client->config->network_timeout_ms) < 0) {
ESP_LOGE(TAG, "Error transport connect");
esp_mqtt_abort_connection(client);
break;
}
ESP_LOGD(TAG, "Transport connected to %s://%s:%d", client->config->scheme, client->config->host, client->config->port);
if (esp_mqtt_connect(client, client->config->network_timeout_ms) != ESP_OK) {
ESP_LOGI(TAG, "Error MQTT Connected");
esp_mqtt_abort_connection(client);
break;
}
client->event.event_id = MQTT_EVENT_CONNECTED;
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
client->state = MQTT_STATE_CONNECTED;
esp_mqtt_dispatch_event(client);
break;
case MQTT_STATE_CONNECTED:
// receive and process data
if (mqtt_process_receive(client) == ESP_FAIL) {
esp_mqtt_abort_connection(client);
break;
}
if (platform_tick_get_ms() - client->keepalive_tick > client->connect_info.keepalive * 1000 / 2) {
//No ping resp from last ping => Disconnected
if(client->wait_for_ping_resp){
ESP_LOGE(TAG, "No PING_RESP, disconnected");
esp_mqtt_abort_connection(client);
client->wait_for_ping_resp = false;
break;
}
if (esp_mqtt_client_ping(client) == ESP_FAIL) {
ESP_LOGE(TAG, "Can't send ping, disconnected");
esp_mqtt_abort_connection(client);
break;
} else {
client->wait_for_ping_resp = true;
}
ESP_LOGD(TAG, "PING sent");
}
//Delete mesaage after 30 senconds
outbox_delete_expired(client->outbox, platform_tick_get_ms(), OUTBOX_EXPIRED_TIMEOUT_MS);
//
outbox_cleanup(client->outbox, OUTBOX_MAX_SIZE);
break;
case MQTT_STATE_WAIT_TIMEOUT:
if (!client->config->auto_reconnect) {
client->run = false;
break;
}
if (platform_tick_get_ms() - client->reconnect_tick > client->wait_timeout_ms) {
client->state = MQTT_STATE_INIT;
client->reconnect_tick = platform_tick_get_ms();
ESP_LOGD(TAG, "Reconnecting...");
}
vTaskDelay(client->wait_timeout_ms / 2 / portTICK_RATE_MS);
break;
}
}
esp_transport_close(client->transport);
xEventGroupSetBits(client->status_bits, STOPPED_BIT);
vTaskDelete(NULL);
}
esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client)
{
if (client->state >= MQTT_STATE_INIT) {
ESP_LOGE(TAG, "Client has started");
return ESP_FAIL;
}
#if MQTT_CORE_SELECTION_ENABLED
ESP_LOGD(TAG, "Core selection enabled on %u", MQTT_TASK_CORE);
if (xTaskCreatePinnedToCore(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, NULL, MQTT_TASK_CORE) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
#else
ESP_LOGD(TAG, "Core selection disabled");
if (xTaskCreate(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, NULL) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
#endif
return ESP_OK;
}
esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client)
{
if (client->run) {
client->run = false;
xEventGroupWaitBits(client->status_bits, STOPPED_BIT, false, true, portMAX_DELAY);
client->state = MQTT_STATE_UNKNOWN;
return ESP_OK;
} else {
ESP_LOGW(TAG, "Client asked to stop, but was not started");
return ESP_FAIL;
}
}
static esp_err_t esp_mqtt_client_ping(esp_mqtt_client_handle_t client)
{
client->mqtt_state.outbound_message = mqtt_msg_pingreq(&client->mqtt_state.mqtt_connection);
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error sending ping");
return ESP_FAIL;
}
ESP_LOGD(TAG, "Sent PING successful");
return ESP_OK;
}
int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos)
{
if (client->state != MQTT_STATE_CONNECTED) {
ESP_LOGE(TAG, "Client has not connected");
return -1;
}
mqtt_enqueue(client); //move pending msg to outbox (if have)
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
&client->mqtt_state.pending_msg_id);
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_count ++;
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error to subscribe topic=%s, qos=%d", topic, qos);
return -1;
}
ESP_LOGD(TAG, "Sent subscribe topic=%s, id: %d, type=%d successful", topic, client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_type);
return client->mqtt_state.pending_msg_id;
}
int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic)
{
if (client->state != MQTT_STATE_CONNECTED) {
ESP_LOGE(TAG, "Client has not connected");
return -1;
}
mqtt_enqueue(client);
client->mqtt_state.outbound_message = mqtt_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
topic,
&client->mqtt_state.pending_msg_id);
ESP_LOGD(TAG, "unsubscribe, topic\"%s\", id: %d", topic, client->mqtt_state.pending_msg_id);
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_count ++;
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error to unsubscribe topic=%s", topic);
return -1;
}
ESP_LOGD(TAG, "Sent Unsubscribe topic=%s, id: %d, successful", topic, client->mqtt_state.pending_msg_id);
return client->mqtt_state.pending_msg_id;
}
int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain)
{
uint16_t pending_msg_id = 0;
if (client->state != MQTT_STATE_CONNECTED) {
ESP_LOGE(TAG, "Client has not connected");
return -1;
}
if (len <= 0) {
len = strlen(data);
}
mqtt_message_t *publish_msg = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id);
/* We have to set as pending all the qos>0 messages) */
if (qos > 0) {
mqtt_enqueue(client);
client->mqtt_state.outbound_message = publish_msg;
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_id = pending_msg_id;
client->mqtt_state.pending_msg_count ++;
} else {
client->mqtt_state.outbound_message = publish_msg;
}
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error to public data to topic=%s, qos=%d", topic, qos);
return -1;
}
return pending_msg_id;
}

0
components/mqtt/paho/MQTTClient-C/src/FreeRTOS/MQTTFreeRTOS.c → components/mqtt/ibm-mqtt/MQTTClient-C/src/FreeRTOS/MQTTFreeRTOS.c
View File

0
components/mqtt/paho/MQTTClient-C/src/FreeRTOS/MQTTFreeRTOS.h → components/mqtt/ibm-mqtt/MQTTClient-C/src/FreeRTOS/MQTTFreeRTOS.h
View File

0
components/mqtt/paho/MQTTClient-C/src/MQTTClient.c → components/mqtt/ibm-mqtt/MQTTClient-C/src/MQTTClient.c
View File

0
components/mqtt/paho/MQTTClient-C/src/MQTTClient.h → components/mqtt/ibm-mqtt/MQTTClient-C/src/MQTTClient.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTConnect.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTConnect.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTConnectClient.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTConnectClient.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTConnectServer.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTConnectServer.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTDeserializePublish.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTDeserializePublish.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTFormat.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTFormat.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTFormat.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTFormat.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTPacket.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTPacket.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTPacket.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTPacket.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTPublish.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTPublish.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTSerializePublish.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTSerializePublish.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTSubscribe.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTSubscribe.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTSubscribeClient.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTSubscribeClient.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTSubscribeServer.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTSubscribeServer.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTUnsubscribe.h → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTUnsubscribe.h
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTUnsubscribeClient.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTUnsubscribeClient.c
View File

0
components/mqtt/paho/MQTTPacket/src/MQTTUnsubscribeServer.c → components/mqtt/ibm-mqtt/MQTTPacket/src/MQTTUnsubscribeServer.c
View File

0
components/mqtt/paho/MQTTPacket/src/StackTrace.h → components/mqtt/ibm-mqtt/MQTTPacket/src/StackTrace.h
View File