espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2025-07-29 18:28:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'idf' to 'master'

Browse Source 
Integration of ESP-MQTT to IDF

See merge request idf/esp-mqttmake
This commit is contained in:
David Cermak
2019-06-10 14:51:27 +02:00
parent 27780e3c45 9b003d3504
commit 11f884623b
52 changed files with 1747 additions and 3060 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
.gitlab-ci.yml Normal file
View File

@ -0,0 +1,140 @@
stages:
- build
- static_analysis
- deploy_report
- deploy
variables:
IDF_REPO: ${GITLAB_SSH_SERVER}/idf/esp-idf.git
OLDER_IDF: release/v3.1
RECENT_IDF: release/v3.3
.add_gh_key_remote: &add_gh_key_remote |
cit_add_ssh_key "${GH_PUSH_KEY}"
git remote remove github || true
git remote add github ${GH_PUSH_REPO}
before_script:
# Use CI Tools
- curl -sSL ${CIT_LOADER_URL} | sh
- source citools/import_functions
- PATH=$CI_PROJECT_DIR/esp-idf/tools:$PATH
build_with_older_idf:
stage: build
image: ${CI_DOCKER_REGISTRY}/esp32-ci-env
tags:
- build
dependencies: []
script:
- cit_add_ssh_key "${GITLAB_KEY}"
- git clone "${IDF_REPO}" && cd esp-idf && git checkout ${OLDER_IDF}
- ./tools/ci/mirror-submodule-update.sh
- export IDF_PATH=$(pwd)
- cd $CI_PROJECT_DIR
- ./modify_for_legacy_idf.sh ${RECENT_IDF} || true
- cd $CI_PROJECT_DIR/examples/tcp
- make defconfig
- make
build_with_idf:
stage: build
image: ${CI_DOCKER_REGISTRY}/esp32-ci-env
tags:
- build
dependencies: []
artifacts:
when: always
paths:
- tidybuild/*
expire_in: 1 day
script:
- cit_add_ssh_key "${GITLAB_KEY}"
- git clone "${IDF_REPO}"
- cd esp-idf
- ./tools/ci/mirror-submodule-update.sh
- export IDF_PATH=$(pwd)
- cd $IDF_PATH/components/mqtt/esp-mqtt
- rm -rf .git
- cp -r $CI_PROJECT_DIR/.git .
- git reset --hard $CI_COMMIT_SHA
# capture compile commands/flags for static analysis
- cd $IDF_PATH/examples/protocols/mqtt/tcp
- mkdir -p tidybuild && cd tidybuild
- cmake -DCMAKE_EXPORT_COMPILE_COMMANDS=ON ..
- mv $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild $CI_PROJECT_DIR/tidybuild
# enable pedantic flags for compilation
- export PEDANTIC_CFLAGS="-Werror -Werror=deprecated-declarations -Werror=unused-variable -Werror=unused-but-set-variable -Werror=unused-function"
- export EXTRA_CFLAGS=${PEDANTIC_CFLAGS} && export EXTRA_CXXFLAGS=${EXTRA_CFLAGS}
# build other examples
- 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/ssl_mutual_auth && idf.py build
- cd $IDF_PATH/examples/protocols/mqtt/ws && idf.py build
- cd $IDF_PATH/examples/protocols/mqtt/wss && idf.py build
clang_tidy_check:
stage: static_analysis
image: ${CI_DOCKER_REGISTRY}/clang-static-analysis
tags:
- host_test
dependencies:
- build_with_idf
artifacts:
reports:
junit: esp-idf/examples/protocols/mqtt/tcp/tidybuild/output.xml
when: always
paths:
- esp-idf/examples/protocols/mqtt/tcp/tidybuild/report/*
expire_in: 1 day
script:
- cit_add_ssh_key "${GITLAB_KEY}"
- git clone "${IDF_REPO}"
- cd esp-idf
- ./tools/ci/mirror-submodule-update.sh
- export IDF_PATH=$(pwd)
- cd $IDF_PATH/components/mqtt/esp-mqtt
- rm -rf .git
- cp -r $CI_PROJECT_DIR/.git .
- git reset --hard $CI_COMMIT_SHA
- mv $CI_PROJECT_DIR/tidybuild $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild
- cd $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild
- git clone $IDF_ANALYSIS_UTILS static_analysis_utils && cd static_analysis_utils
- ./generate_report.sh $CI_PROJECT_DIR/static-analysis-rules.yml $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild/report $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild/output.xml
deploy_report:
stage: deploy_report
image: $CI_DOCKER_REGISTRY/esp32-ci-env
tags:
- deploy
dependencies:
- clang_tidy_check
script:
- cit_add_ssh_key "${DOCS_DEPLOY_KEY}"
- echo -e "Host $DOCS_SERVER\n\tStrictHostKeyChecking no\n\tUser $DOCS_SERVER_USER\n" >> ~/.ssh/config
- export GIT_VER=$(git describe --always)
- cd esp-idf/examples/protocols/mqtt/tcp/tidybuild
- mv report $GIT_VER
- tar czvf $GIT_VER.tar.gz $GIT_VER
- ssh $DOCS_SERVER -x "mkdir -p $DOCS_PATH/clang-tidy"
- scp $GIT_VER.tar.gz $DOCS_SERVER:$DOCS_PATH/clang-tidy
- ssh $DOCS_SERVER -x "cd $DOCS_PATH/clang-tidy && tar xzvf $GIT_VER.tar.gz && rm -f latest && ln -s $GIT_VER latest"
# add link to view the report
- echo "[static analysis][clang tidy] $CI_DOCKER_REGISTRY/static_analysis/esp-idf/clang-tidy/${GIT_VER}/index.html"
- test ! -e ${GIT_VER}/FAILED_RULES || { echo 'Failed static analysis rules!'; cat ${GIT_VER}/FAILED_RULES; exit 1; }
push_master_to_github:
stage: deploy
image: ${CI_DOCKER_REGISTRY}/esp32-ci-env
tags:
- build
only:
- master
- idf
when: on_success
variables:
GIT_STRATEGY: clone
script:
- *add_gh_key_remote
- git push github HEAD:${CI_COMMIT_REF_NAME}

56
.travis.yml
View File

@ -13,6 +13,11 @@ addons:
before_install:
# Save path to the git respository
- PROJECT_PATH=$(pwd)
# Have to checkout a temp branch for later in tree reference
- git checkout -b temporary_ref_branch
- CI_COMMIT_SHA=$(git rev-parse HEAD)
# Test building with latest (stable == v3.3 for now) IDF
- LATEST_IDF=release/v3.3
install:
# Install ESP32 toochain following steps as desribed
@ -24,32 +29,37 @@ install:
- mkdir -p ~/esp
- cd ~/esp
# Download binary toolchain for the ESP32
- wget https://dl.espressif.com/dl/xtensa-esp32-elf-linux64-1.22.0-61-gab8375a-5.2.0.tar.gz
- tar -xzf xtensa-esp32-elf-linux64-1.22.0-61-gab8375a-5.2.0.tar.gz
# Make xtensa-esp32-elf available for all terminal sessions
- export PATH=$PATH:$HOME/esp/xtensa-esp32-elf/bin
# Get ESP-IDF from github
- git clone --recursive https://github.com/espressif/esp-idf.git
- cd esp-idf && git checkout --recurse-submodules v3.1
- 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 (non-recursive to save time, later we update submodules for different versions)
- git clone 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:
# Go back to the git repository
- cd $PROJECT_PATH/examples/mqtt_tcp
# Update configuration so that kconfig doesn't start interactive mode
- make defconfig
# Build project from the git repository
- make -j4
- cd $PROJECT_PATH/examples/mqtt_ssl
- make defconfig
- make -j4
- cd $PROJECT_PATH/examples/mqtt_ws
- make defconfig
- make -j4
- cd $PROJECT_PATH/examples/mqtt_wss
- make defconfig
- make -j4
- cd $PROJECT_PATH/examples/emitter-client
# Legacy build with IDF < 3.2
- cd $IDF_PATH
- git checkout v3.1 && git submodule update --init --recursive
- cd $PROJECT_PATH
- ./modify_for_legacy_idf.sh ${LATEST_IDF} || true
- cd $PROJECT_PATH/examples/tcp
- make defconfig
- make -j4
# Build with latest IDF
- cd $IDF_PATH
- git checkout ${LATEST_IDF} && git submodule update --init --recursive
- cd $IDF_PATH/components/mqtt/esp-mqtt
- git remote add local $PROJECT_PATH/.git
- git fetch local
- git reset --hard $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

8
CMakeLists.txt
View File

@ -1,8 +0,0 @@
set(COMPONENT_ADD_INCLUDEDIRS "include")
set(COMPONENT_PRIV_INCLUDEDIRS "lib/include")
set(COMPONENT_SRCDIRS ". lib")
set(COMPONENT_REQUIRES lwip nghttp mbedtls)
register_component()

96
Kconfig
View File

@ -1,96 +0,0 @@
menu "ESPMQTT Configurations"
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
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
endmenu

167
README.md
View File

@ -15,166 +15,25 @@
## How to use
From IDFv3.2 [ESP-MQTT]((https://github.com/espressif/esp-mqtt)) is integrated in [ESP-IDF](https://github.com/espressif/esp-idf) as a submodule. Please do not use separately.
For [ESP-IDF](https://github.com/espressif/esp-idf) versions prior to IDFv3.2, please checkout the [ESP-MQTT_FOR_IDF_3.1](https://github.com/espressif/esp-mqtt/tree/ESP-MQTT_FOR_IDF_3.1) tag and follow the instructions below:
Clone this component to [ESP-IDF](https://github.com/espressif/esp-idf) project (as submodule):
```
git submodule add https://github.com/tuanpmt/espmqtt.git components/espmqtt
```
Or run a sample (make sure you have installed the [toolchain](http://esp-idf.readthedocs.io/en/latest/get-started/index.html#setup-toolchain)):
```
git clone https://github.com/tuanpmt/espmqtt.git
cd espmqtt/examples/mqtt_tcp
make menuconfig
make flash monitor
```
[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
### URI
- Curently support `mqtt`, `mqtts`, `ws`, `wss` schemes
- MQTT over TCP samples:
+ `mqtt://iot.eclipse.org`: MQTT over TCP, default port 1883:
+ `mqtt://iot.eclipse.org:1884` MQTT over TCP, port 1884:
+ `mqtt://username:password@iot.eclipse.org:1884` MQTT over TCP, port 1884, with username and password
- MQTT over SSL samples:
+ `mqtts://iot.eclipse.org`: MQTT over SSL, port 8883
+ `mqtts://iot.eclipse.org:8884`: MQTT over SSL, port 8884
- MQTT over Websocket samples:
+ `ws://iot.eclipse.org:80/ws`
- MQTT over Websocket Secure samples:
+ `wss://iot.eclipse.org:443/ws`
- Minimal configurations:
* Please refer to the standard [ESP-IDF](https://github.com/espressif/esp-idf), documentation for the latest version: https://docs.espressif.com/projects/esp-idf/
```c
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtt://iot.eclipse.org",
.event_handle = mqtt_event_handler,
// .user_context = (void *)your_context
};
```
- If there are any options related to the URI in `esp_mqtt_client_config_t`, the option defined by the URI will be overridden. Sample:
```c
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtt://iot.eclipse.org:1234",
.event_handle = mqtt_event_handler,
.port = 4567,
};
//MQTT client will connect to iot.eclipse.org using port 4567
```
### SSL
- Get Certification from server, example: `iot.eclipse.org` `openssl s_client -showcerts -connect iot.eclipse.org:8883 </dev/null 2>/dev/null|openssl x509 -outform PEM >iot_eclipse_org.pem`
- Check the sample application: `examples/mqtt_ssl`
- Configuration:
```cpp
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtts://iot.eclipse.org:8883",
.event_handle = mqtt_event_handler,
.cert_pem = (const char *)iot_eclipse_org_pem_start,
};
```
### More options for `esp_mqtt_client_config_t`
- `event_handle` for MQTT events
- `host`: MQTT server domain (ipv4 as string)
- `port`: MQTT server port
- `client_id`: default client id is `ESP32_%CHIPID%`
- `username`: MQTT username
- `password`: MQTT password
- `lwt_topic, lwt_msg, lwt_qos, lwt_retain, lwt_msg_len`: are mqtt lwt options, default NULL
- `disable_clean_session`: mqtt clean session, default clean_session is true
- `keepalive`: (value in seconds) mqtt keepalive, default is 120 seconds
- `disable_auto_reconnect`: this mqtt client will reconnect to server (when errors/disconnect). Set `disable_auto_reconnect=true` to disable
- `user_context` pass user context to this option, then can receive that context in `event->user_context`
- `task_prio, task_stack` for MQTT task, default priority is 5, and task_stack = 6144 bytes (or default task stack can be set via `make menucofig`).
- `buffer_size` for MQTT send/receive buffer, default is 1024
- `cert_pem` pointer to CERT file for server verify (with SSL), default is NULL, not required to verify the server
- `client_cert_pem` pointer to CERT file 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.
- `client_key_pem` pointer to PEM private key file 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.
- `transport`: override URI transport
+ `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`
### Change settings in `menuconfig`
```
make menuconfig
-> Component config -> ESPMQTT Configuration
```
## Example
Check `examples/mqtt_tcp` and `examples/mqtt_ssl` project. In Short:
```cpp
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, "/topic/qos0", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
ESP_LOGI(TAG, "sent unsubscribe 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, "/topic/qos0", "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;
}
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtt://iot.eclipse.org",
.event_handle = mqtt_event_handler,
// .user_context = (void *)your_context
};
esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);
esp_mqtt_client_start(client);
```
* Documentation of ESP-MQTT API: https://docs.espressif.com/projects/esp-idf/en/latest/api-reference/protocols/mqtt.html
## License
- MQTT Package - [Stephen Robinson - contiki-mqtt](https://github.com/esar/contiki-mqtt)
- Others [@tuanpmt](https://twitter.com/tuanpmt)
Apache License
## Older IDF verisons
For [ESP-IDF](https://github.com/espressif/esp-idf) versions prior to IDFv3.2, please clone as a component of [ESP-IDF](https://github.com/espressif/esp-idf):
```
git submodule add https://github.com/espressif/esp-mqtt.git components/espmqtt
```
and checkout the [ESP-MQTT_FOR_IDF_3.1](https://github.com/espressif/esp-mqtt/tree/ESP-MQTT_FOR_IDF_3.1) tag

10
component.mk
View File

@ -1,10 +0,0 @@
#
# Component Makefile
#
# This Makefile should, at the very least, just include $(SDK_PATH)/make/component.mk. By default,
# this will take the sources in this directory, compile them and link them into
# lib(subdirectory_name).a in the build directory. This behaviour is entirely configurable,
# please read the SDK documents if you need to do this.
#
COMPONENT_SRCDIRS := . lib
COMPONENT_PRIV_INCLUDEDIRS := lib/include

19
examples/mqtt_ssl/CMakeLists.txt
View File

@ -1,19 +0,0 @@
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
examples/mqtt_ssl/Makefile
View File

@ -1,13 +0,0 @@
#
# 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 := mqtt_ssl
EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../
include $(IDF_PATH)/make/project.mk

5
examples/mqtt_ssl/README.md
View File

@ -1,5 +0,0 @@
# ESPMQTT SSL Sample application
Get iot.eclipse.org Certification
`openssl s_client -showcerts -connect iot.eclipse.org:8883 </dev/null 2>/dev/null|openssl x509 -outform PEM >iot_eclipse_org.pem`

15
examples/mqtt_ssl/main/Kconfig.projbuild
View File

@ -1,15 +0,0 @@
menu "MQTT Application sample"
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.
endmenu

149
examples/mqtt_ssl/main/app_main.c
View File

@ -1,149 +0,0 @@
#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);
}
extern const uint8_t iot_eclipse_org_pem_start[] asm("_binary_iot_eclipse_org_pem_start");
extern const uint8_t iot_eclipse_org_pem_end[] asm("_binary_iot_eclipse_org_pem_end");
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, "/topic/qos0", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
ESP_LOGI(TAG, "sent unsubscribe 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, "/topic/qos0", "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://iot.eclipse.org:8883",
.event_handle = mqtt_event_handler,
.cert_pem = (const char *)iot_eclipse_org_pem_start,
};
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();
}

1
examples/mqtt_ssl/main/component.mk
View File

@ -1 +0,0 @@
COMPONENT_EMBED_TXTFILES := iot_eclipse_org.pem

27
examples/mqtt_ssl/main/iot_eclipse_org.pem
View File

@ -1,27 +0,0 @@
-----BEGIN CERTIFICATE-----
MIIEkjCCA3qgAwIBAgIQCgFBQgAAAVOFc2oLheynCDANBgkqhkiG9w0BAQsFADA/
MSQwIgYDVQQKExtEaWdpdGFsIFNpZ25hdHVyZSBUcnVzdCBDby4xFzAVBgNVBAMT
DkRTVCBSb290IENBIFgzMB4XDTE2MDMxNzE2NDA0NloXDTIxMDMxNzE2NDA0Nlow
SjELMAkGA1UEBhMCVVMxFjAUBgNVBAoTDUxldCdzIEVuY3J5cHQxIzAhBgNVBAMT
GkxldCdzIEVuY3J5cHQgQXV0aG9yaXR5IFgzMIIBIjANBgkqhkiG9w0BAQEFAAOC
AQ8AMIIBCgKCAQEAnNMM8FrlLke3cl03g7NoYzDq1zUmGSXhvb418XCSL7e4S0EF
q6meNQhY7LEqxGiHC6PjdeTm86dicbp5gWAf15Gan/PQeGdxyGkOlZHP/uaZ6WA8
SMx+yk13EiSdRxta67nsHjcAHJyse6cF6s5K671B5TaYucv9bTyWaN8jKkKQDIZ0
Z8h/pZq4UmEUEz9l6YKHy9v6Dlb2honzhT+Xhq+w3Brvaw2VFn3EK6BlspkENnWA
a6xK8xuQSXgvopZPKiAlKQTGdMDQMc2PMTiVFrqoM7hD8bEfwzB/onkxEz0tNvjj
/PIzark5McWvxI0NHWQWM6r6hCm21AvA2H3DkwIDAQABo4IBfTCCAXkwEgYDVR0T
AQH/BAgwBgEB/wIBADAOBgNVHQ8BAf8EBAMCAYYwfwYIKwYBBQUHAQEEczBxMDIG
CCsGAQUFBzABhiZodHRwOi8vaXNyZy50cnVzdGlkLm9jc3AuaWRlbnRydXN0LmNv
bTA7BggrBgEFBQcwAoYvaHR0cDovL2FwcHMuaWRlbnRydXN0LmNvbS9yb290cy9k
c3Ryb290Y2F4My5wN2MwHwYDVR0jBBgwFoAUxKexpHsscfrb4UuQdf/EFWCFiRAw
VAYDVR0gBE0wSzAIBgZngQwBAgEwPwYLKwYBBAGC3xMBAQEwMDAuBggrBgEFBQcC
ARYiaHR0cDovL2Nwcy5yb290LXgxLmxldHNlbmNyeXB0Lm9yZzA8BgNVHR8ENTAz
MDGgL6AthitodHRwOi8vY3JsLmlkZW50cnVzdC5jb20vRFNUUk9PVENBWDNDUkwu
Y3JsMB0GA1UdDgQWBBSoSmpjBH3duubRObemRWXv86jsoTANBgkqhkiG9w0BAQsF
AAOCAQEA3TPXEfNjWDjdGBX7CVW+dla5cEilaUcne8IkCJLxWh9KEik3JHRRHGJo
uM2VcGfl96S8TihRzZvoroed6ti6WqEBmtzw3Wodatg+VyOeph4EYpr/1wXKtx8/
wApIvJSwtmVi4MFU5aMqrSDE6ea73Mj2tcMyo5jMd6jmeWUHK8so/joWUoHOUgwu
X4Po1QYz+3dszkDqMp4fklxBwXRsW10KXzPMTZ+sOPAveyxindmjkW8lGy+QsRlG
PfZ+G6Z6h7mjem0Y+iWlkYcV4PIWL1iwBi8saCbGS5jN2p8M+X+Q7UNKEkROb3N6
KOqkqm57TH2H3eDJAkSnh6/DNFu0Qg==
-----END CERTIFICATE-----

19
examples/mqtt_tcp/CMakeLists.txt
View File

@ -1,19 +0,0 @@
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_tcp")
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
examples/mqtt_tcp/Makefile
View File

@ -1,13 +0,0 @@
#
# 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 := mqtt_tcp
EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../
include $(IDF_PATH)/make/project.mk

1
examples/mqtt_tcp/README.md
View File

@ -1 +0,0 @@
# ESPMQTT Sample application

15
examples/mqtt_tcp/main/Kconfig.projbuild
View File

@ -1,15 +0,0 @@
menu "MQTT Application sample"
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.
endmenu

143
examples/mqtt_tcp/main/app_main.c
View File

@ -1,143 +0,0 @@
#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 = "MQTT_SAMPLE";
static EventGroupHandle_t wifi_event_group;
const static int CONNECTED_BIT = BIT0;
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, "/topic/qos0", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
ESP_LOGI(TAG, "sent unsubscribe 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, "/topic/qos0", "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 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 void mqtt_app_start(void)
{
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "mqtt://iot.eclipse.org",
.event_handle = mqtt_event_handler,
// .user_context = (void *)your_context
};
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
examples/mqtt_tcp/main/component.mk
View File

19
examples/mqtt_ws/CMakeLists.txt
View File

@ -1,19 +0,0 @@
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_ws")
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
examples/mqtt_ws/Makefile
View File

@ -1,13 +0,0 @@
#
# 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 := mqtt_ws
EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../
include $(IDF_PATH)/make/project.mk

1
examples/mqtt_ws/README.md
View File

@ -1 +0,0 @@
# ESPMQTT MQTT over Websocket

15
examples/mqtt_ws/main/Kconfig.projbuild
View File

@ -1,15 +0,0 @@
menu "MQTT Application sample"
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.
endmenu

144
examples/mqtt_ws/main/app_main.c
View File

@ -1,144 +0,0 @@
#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 = "MQTTWS_SAMPLE";
static EventGroupHandle_t wifi_event_group;
const static int CONNECTED_BIT = BIT0;
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, "/topic/qos0", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
ESP_LOGI(TAG, "sent unsubscribe 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, "/topic/qos0", "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 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 void mqtt_app_start(void)
{
const esp_mqtt_client_config_t mqtt_cfg = {
.uri = "ws://iot.eclipse.org:80/ws",
.event_handle = mqtt_event_handler,
// .user_context = (void *)your_context
};
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_WS", 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
examples/mqtt_ws/main/component.mk
View File

19
examples/mqtt_wss/CMakeLists.txt
View File

@ -1,19 +0,0 @@
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_wss")
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
examples/mqtt_wss/Makefile
View File

@ -1,13 +0,0 @@
#
# 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 := mqtt_wss
EXTRA_COMPONENT_DIRS += $(PROJECT_PATH)/../../../
include $(IDF_PATH)/make/project.mk

5
examples/mqtt_wss/README.md
View File

@ -1,5 +0,0 @@
# ESPMQTT MQTT over WSS Sample application
Get iot.eclipse.org Certification
`openssl s_client -showcerts -connect iot.eclipse.org:8883 </dev/null 2>/dev/null|openssl x509 -outform PEM >iot_eclipse_org.pem`

15
examples/mqtt_wss/main/Kconfig.projbuild
View File

@ -1,15 +0,0 @@
menu "MQTT Application sample"
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.
endmenu

148
examples/mqtt_wss/main/app_main.c
View File

@ -1,148 +0,0 @@
#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 = "MQTTWSS_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);
}
extern const uint8_t iot_eclipse_org_pem_start[] asm("_binary_iot_eclipse_org_pem_start");
extern const uint8_t iot_eclipse_org_pem_end[] asm("_binary_iot_eclipse_org_pem_end");
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, "/topic/qos0", 0);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
ESP_LOGI(TAG, "sent unsubscribe 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, "/topic/qos0", "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 = "wss://iot.eclipse.org:443/ws",
.event_handle = mqtt_event_handler,
.cert_pem = (const char *)iot_eclipse_org_pem_start,
};
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();
}

1
examples/mqtt_wss/main/component.mk
View File

@ -1 +0,0 @@
COMPONENT_EMBED_TXTFILES := iot_eclipse_org.pem

27
examples/mqtt_wss/main/iot_eclipse_org.pem
View File

@ -1,27 +0,0 @@
-----BEGIN CERTIFICATE-----
MIIEkjCCA3qgAwIBAgIQCgFBQgAAAVOFc2oLheynCDANBgkqhkiG9w0BAQsFADA/
MSQwIgYDVQQKExtEaWdpdGFsIFNpZ25hdHVyZSBUcnVzdCBDby4xFzAVBgNVBAMT
DkRTVCBSb290IENBIFgzMB4XDTE2MDMxNzE2NDA0NloXDTIxMDMxNzE2NDA0Nlow
SjELMAkGA1UEBhMCVVMxFjAUBgNVBAoTDUxldCdzIEVuY3J5cHQxIzAhBgNVBAMT
GkxldCdzIEVuY3J5cHQgQXV0aG9yaXR5IFgzMIIBIjANBgkqhkiG9w0BAQEFAAOC
AQ8AMIIBCgKCAQEAnNMM8FrlLke3cl03g7NoYzDq1zUmGSXhvb418XCSL7e4S0EF
q6meNQhY7LEqxGiHC6PjdeTm86dicbp5gWAf15Gan/PQeGdxyGkOlZHP/uaZ6WA8
SMx+yk13EiSdRxta67nsHjcAHJyse6cF6s5K671B5TaYucv9bTyWaN8jKkKQDIZ0
Z8h/pZq4UmEUEz9l6YKHy9v6Dlb2honzhT+Xhq+w3Brvaw2VFn3EK6BlspkENnWA
a6xK8xuQSXgvopZPKiAlKQTGdMDQMc2PMTiVFrqoM7hD8bEfwzB/onkxEz0tNvjj
/PIzark5McWvxI0NHWQWM6r6hCm21AvA2H3DkwIDAQABo4IBfTCCAXkwEgYDVR0T
AQH/BAgwBgEB/wIBADAOBgNVHQ8BAf8EBAMCAYYwfwYIKwYBBQUHAQEEczBxMDIG
CCsGAQUFBzABhiZodHRwOi8vaXNyZy50cnVzdGlkLm9jc3AuaWRlbnRydXN0LmNv
bTA7BggrBgEFBQcwAoYvaHR0cDovL2FwcHMuaWRlbnRydXN0LmNvbS9yb290cy9k
c3Ryb290Y2F4My5wN2MwHwYDVR0jBBgwFoAUxKexpHsscfrb4UuQdf/EFWCFiRAw
VAYDVR0gBE0wSzAIBgZngQwBAgEwPwYLKwYBBAGC3xMBAQEwMDAuBggrBgEFBQcC
ARYiaHR0cDovL2Nwcy5yb290LXgxLmxldHNlbmNyeXB0Lm9yZzA8BgNVHR8ENTAz
MDGgL6AthitodHRwOi8vY3JsLmlkZW50cnVzdC5jb20vRFNUUk9PVENBWDNDUkwu
Y3JsMB0GA1UdDgQWBBSoSmpjBH3duubRObemRWXv86jsoTANBgkqhkiG9w0BAQsF
AAOCAQEA3TPXEfNjWDjdGBX7CVW+dla5cEilaUcne8IkCJLxWh9KEik3JHRRHGJo
uM2VcGfl96S8TihRzZvoroed6ti6WqEBmtzw3Wodatg+VyOeph4EYpr/1wXKtx8/
wApIvJSwtmVi4MFU5aMqrSDE6ea73Mj2tcMyo5jMd6jmeWUHK8so/joWUoHOUgwu
X4Po1QYz+3dszkDqMp4fklxBwXRsW10KXzPMTZ+sOPAveyxindmjkW8lGy+QsRlG
PfZ+G6Z6h7mjem0Y+iWlkYcV4PIWL1iwBi8saCbGS5jN2p8M+X+Q7UNKEkROb3N6
KOqkqm57TH2H3eDJAkSnh6/DNFu0Qg==
-----END CERTIFICATE-----

263
include/mqtt_client.h Executable file → Normal file
View File

@ -13,84 +13,251 @@
#include "esp_err.h"
#include "mqtt_config.h"
#include "esp_event.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct esp_mqtt_client* esp_mqtt_client_handle_t;
#ifndef ESP_EVENT_DECLARE_BASE
// Define event loop types if macros not available
typedef void * esp_event_loop_handle_t;
typedef void * esp_event_handler_t;
#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,
MQTT_EVENT_DISCONNECTED,
MQTT_EVENT_SUBSCRIBED,
MQTT_EVENT_UNSUBSCRIBED,
MQTT_EVENT_PUBLISHED,
MQTT_EVENT_DATA,
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
Note: Multiple MQTT_EVENT_DATA could be fired for one message, if it is
longer than internal buffer. In that case only first event contains topic
pointer and length, other contain data only with current data length
and current data offset updating.
*/
MQTT_EVENT_BEFORE_CONNECT, /*!< The event occurs before connecting */
} esp_mqtt_event_id_t;
typedef enum {
MQTT_TRANSPORT_UNKNOWN = 0x0,
MQTT_TRANSPORT_OVER_TCP,
MQTT_TRANSPORT_OVER_SSL,
MQTT_TRANSPORT_OVER_WS,
MQTT_TRANSPORT_OVER_WSS
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;
esp_mqtt_client_handle_t client;
void *user_context;
char *data;
int data_len;
int total_data_len;
int current_data_offset;
char *topic;
int topic_len;
int msg_id;
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_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;
const char *host;
const char *uri;
uint32_t port;
const char *client_id;
const char *username;
const char *password;
const char *lwt_topic;
const char *lwt_msg;
int lwt_qos;
int lwt_retain;
int lwt_msg_len;
int disable_clean_session;
int keepalive;
bool disable_auto_reconnect;
void *user_context;
int task_prio;
int task_stack;
int buffer_size;
const char *cert_pem;
const char *client_cert_pem;
const char *client_key_pem;
esp_mqtt_transport_t transport;
mqtt_event_callback_t event_handle; /*!< handle for MQTT events as a callback in legacy mode */
esp_event_loop_handle_t event_loop_handle; /*!< handle for MQTT event loop library */
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 */
int refresh_connection_after_ms; /*!< Refresh connection after this value (in milliseconds) */
} esp_mqtt_client_config_t;
/**
* @brief Creates mqtt client handle based on the configuration
*
* @param config mqtt configuration structure
*
* @return mqtt_client_handle if successfully created, NULL on error
*/
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config);
/**
* @brief Sets mqtt connection URI. This API is usually used to overrides the URI
* configured in esp_mqtt_client_init
*
* @param client mqtt client hanlde
* @param uri
*
* @return ESP_FAIL if URI parse error, ESP_OK on success
*/
esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *uri);
/**
* @brief Starts mqtt client with already created client handle
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on other error
*/
esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client);
/**
* @brief This api is typically used to force reconnection upon a specific event
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_FAIL if client is in invalid state
*/
esp_err_t esp_mqtt_client_reconnect(esp_mqtt_client_handle_t client);
/**
* @brief Stops mqtt client tasks
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_FAIL if client is in invalid state
*/
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);
/**
* @brief Subscribe the client to defined topic with defined qos
*
* Notes:
* - Client must be connected to send subscribe message
* - This API is could be executed from a user task or
* from a mqtt event callback i.e. internal mqtt task
* (API is protected by internal mutex, so it might block
* if a longer data receive operation is in progress.
*
* @param client mqtt client handle
* @param topic
* @param qos
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos);
/**
* @brief Unsubscribe the client from defined topic
*
* Notes:
* - Client must be connected to send unsubscribe message
* - It is thread safe, please refer to `esp_mqtt_client_subscribe` for details
*
* @param client mqtt client handle
* @param topic
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic);
/**
* @brief Client to send a publish message to the broker
*
* Notes:
* - Client doesn't have to be connected to send publish message
* (although it would drop all qos=0 messages, qos>1 messages would be enqueued)
* - It is thread safe, please refer to `esp_mqtt_client_subscribe` for details
*
* @param client mqtt client handle
* @param topic topic string
* @param data payload string (set to NULL, sending empty payload message)
* @param len data length, if set to 0, length is calculated from payload string
* @param qos qos of publish message
* @param retain ratain flag
*
* @return message_id of the subscribe message on success
* 0 if cannot publish
*/
int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain);
/**
* @brief Destroys the client handle
*
* @param client mqtt client handle
*
* @return ESP_OK
*/
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client);
/**
* @brief Set configuration structure, typically used when updating the config (i.e. on "before_connect" event
*
* @param client mqtt client handle
*
* @param config mqtt configuration structure
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_OK on success
*/
esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config);
/**
* @brief Registers mqtt event
*
* @param client mqtt client handle
* @param event event type
* @param event_handler hanlder callback
* @param event_handler_arg handlers context
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_OK on success
*/
esp_err_t esp_mqtt_client_register_event(esp_mqtt_client_handle_t client, esp_mqtt_event_id_t event, esp_event_handler_t event_handler, void* event_handler_arg);
#ifdef __cplusplus
}
#endif //__cplusplus

17
include/mqtt_config.h
View File

@ -10,6 +10,7 @@
#define MQTT_PROTOCOL_311 CONFIG_MQTT_PROTOCOL_311
#define MQTT_RECONNECT_TIMEOUT_MS (10*1000)
#define MQTT_POLL_READ_TIMEOUT_MS (1000)
#if CONFIG_MQTT_BUFFER_SIZE
#define MQTT_BUFFER_SIZE_BYTE CONFIG_MQTT_BUFFER_SIZE
@ -61,14 +62,18 @@
#define MQTT_CORE_SELECTION_ENABLED CONFIG_MQTT_TASK_CORE_SELECTION_ENABLED
#ifdef CONFIG_MQTT_DISABLE_API_LOCKS
#define MQTT_DISABLE_API_LOCKS CONFIG_MQTT_DISABLE_API_LOCKS
#endif
#ifdef CONFIG_MQTT_USE_CORE_0
#define MQTT_TASK_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
#ifdef CONFIG_MQTT_USE_CORE_1
#define MQTT_TASK_CORE 1
#else
#define MQTT_TASK_CORE 0
#endif
#endif

43
include/mqtt_supported_features.h Normal file
View File

@ -0,0 +1,43 @@
// Copyright 2015-2019 Espressif Systems (Shanghai) PTE LTD
//
// 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.
#ifndef _MQTT_SUPPORTED_FEATURES_H_
#define _MQTT_SUPPORTED_FEATURES_H_
#if __has_include("esp_idf_version.h")
#include "esp_idf_version.h"
#endif
/**
* @brief This header defines supported features of IDF which mqtt module
* could use depending on specific version of ESP-IDF.
* In case "esp_idf_version.h" were not found, all additional
* features would be disabled
*/
#ifdef ESP_IDF_VERSION
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(3, 3, 0)
// Features supported from 3.3
#define MQTT_SUPPORTED_FEATURE_EVENT_LOOP
#endif
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(4, 0, 0)
// Features supported in 4.0
#define MQTT_SUPPORTED_FEATURE_WS_SUBPROTOCOL
#endif
#endif
#endif // _MQTT_SUPPORTED_FEATURES_H_

99
lib/include/mqtt_msg.h
View File

@ -40,8 +40,7 @@ extern "C" {
/* Remaining Length */
enum mqtt_message_type
{
enum mqtt_message_type {
MQTT_MSG_TYPE_CONNECT = 1,
MQTT_MSG_TYPE_CONNACK = 2,
MQTT_MSG_TYPE_PUBLISH = 3,
@ -58,8 +57,7 @@ enum mqtt_message_type
MQTT_MSG_TYPE_DISCONNECT = 14
};
enum mqtt_connect_return_code
{
enum mqtt_connect_return_code {
CONNECTION_ACCEPTED = 0,
CONNECTION_REFUSE_PROTOCOL,
CONNECTION_REFUSE_ID_REJECTED,
@ -68,30 +66,28 @@ enum mqtt_connect_return_code
CONNECTION_REFUSE_NOT_AUTHORIZED
};
typedef struct mqtt_message
{
uint8_t* data;
typedef struct mqtt_message {
uint8_t *data;
uint32_t length;
uint32_t fragmented_msg_total_length; /*!< total len of fragmented messages (zero for all other messages) */
uint32_t fragmented_msg_data_offset; /*!< data offset of fragmented messages (zero for all other messages) */
} mqtt_message_t;
typedef struct mqtt_connection
{
typedef struct mqtt_connection {
mqtt_message_t message;
uint16_t message_id;
uint8_t* buffer;
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;
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;
@ -101,29 +97,54 @@ typedef struct mqtt_connect_info
} 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); }
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 void mqtt_set_dup(uint8_t *buffer)
{
buffer[0] |= 0x08;
}
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);
void mqtt_msg_init(mqtt_connection_t *connection, uint8_t *buffer, uint16_t buffer_length);
bool mqtt_header_complete(uint8_t *buffer, uint16_t buffer_length);
uint32_t mqtt_get_total_length(uint8_t *buffer, uint16_t length, int *fixed_size_len);
char *mqtt_get_publish_topic(uint8_t *buffer, uint32_t *length);
char *mqtt_get_publish_data(uint8_t *buffer, uint32_t *length);
uint16_t mqtt_get_id(uint8_t *buffer, uint16_t length);
int mqtt_has_valid_msg_hdr(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);
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

39
lib/include/mqtt_outbox.h
View File

@ -11,32 +11,39 @@
extern "C" {
#endif
typedef struct outbox_item {
char *buffer;
struct outbox_item;
typedef struct outbox_list_t *outbox_handle_t;
typedef struct outbox_item *outbox_item_handle_t;
typedef struct outbox_message *outbox_message_handle_t;
typedef struct outbox_message {
uint8_t *data;
int len;
int msg_id;
int msg_qos;
int msg_type;
int tick;
int retry_count;
bool pending;
STAILQ_ENTRY(outbox_item) next;
} outbox_item_t;
uint8_t *remaining_data;
int remaining_len;
} outbox_message_t;
STAILQ_HEAD(outbox_list_t, outbox_item);
typedef enum pending_state {
QUEUED,
TRANSMITTED,
CONFIRMED
} pending_state_t;
typedef struct outbox_list_t * outbox_handle_t;
typedef outbox_item_t *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_handle_t outbox_init(void);
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, int tick);
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, int *tick);
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id);
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos);
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);
int 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);
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id, pending_state_t pending);
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);

6
lib/include/platform_esp32_idf.h
View File

@ -18,14 +18,14 @@
#include "lwip/netdb.h"
#include "lwip/dns.h"
#include "rom/queue.h"
#include "sys/queue.h"
#include "esp_err.h"
#include "esp_log.h"
#include "esp_system.h"
char *platform_create_id_string();
char *platform_create_id_string(void);
int platform_random(int max);
long long platform_tick_get_ms();
long long platform_tick_get_ms(void);
void ms_to_timeval(int timeout_ms, struct timeval *tv);
#define ESP_MEM_CHECK(TAG, a, action) if (!(a)) { \

242
lib/include/transport.h
View File

@ -1,242 +0,0 @@
/*
* 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 _TRANSPORT_H_
#define _TRANSPORT_H_
#include <esp_err.h>
#ifdef __cplusplus
extern "C" {
#endif
typedef struct transport_list_t* transport_list_handle_t;
typedef struct transport_item_t* transport_handle_t;
typedef int (*connect_func)(transport_handle_t t, const char *host, int port, int timeout_ms);
typedef int (*io_func)(transport_handle_t t, const char *buffer, int len, int timeout_ms);
typedef int (*io_read_func)(transport_handle_t t, char *buffer, int len, int timeout_ms);
typedef int (*trans_func)(transport_handle_t t);
typedef int (*poll_func)(transport_handle_t t, int timeout_ms);
/**
* @brief Create transport list
*
* @return A handle can hold all transports
*/
transport_list_handle_t transport_list_init();
/**
* @brief Cleanup and free all transports, include itself,
* this function will invoke transport_destroy of every transport have added this the list
*
* @param[in] list The list
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_list_destroy(transport_list_handle_t list);
/**
* @brief Add a transport to the list, and define a scheme to indentify this transport in the list
*
* @param[in] list The list
* @param[in] t The Transport
* @param[in] scheme The scheme
*
* @return
* - ESP_OK
*/
esp_err_t transport_list_add(transport_list_handle_t list, transport_handle_t t, const char *scheme);
/**
* @brief This function will remove all transport from the list,
* invoke transport_destroy of every transport have added this the list
*
* @param[in] list The list
*
* @return
* - ESP_OK
* - ESP_ERR_INVALID_ARG
*/
esp_err_t transport_list_clean(transport_list_handle_t list);
/**
* @brief Get the transport by scheme, which has been defined when calling function `transport_list_add`
*
* @param[in] list The list
* @param[in] tag The tag
*
* @return The transport handle
*/
transport_handle_t transport_list_get_transport(transport_list_handle_t list, const char *scheme);
/**
* @brief Initialize a transport handle object
*
* @return The transport handle
*/
transport_handle_t transport_init();
/**
* @brief Cleanup and free memory the transport
*
* @param[in] t The transport handle
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_destroy(transport_handle_t t);
/**
* @brief Get default port number used by this transport
*
* @param[in] t The transport handle
*
* @return the port number
*/
int transport_get_default_port(transport_handle_t t);
/**
* @brief Set default port number that can be used by this transport
*
* @param[in] t The transport handle
* @param[in] port The port number
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_set_default_port(transport_handle_t t, int port);
/**
* @brief Transport connection function, to make a connection to server
*
* @param t The transport handle
* @param[in] host Hostname
* @param[in] port Port
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - socket for will use by this transport
* - (-1) if there are any errors, should check errno
*/
int transport_connect(transport_handle_t t, const char *host, int port, int timeout_ms);
/**
* @brief Transport read function
*
* @param t The transport handle
* @param buffer The buffer
* @param[in] len The length
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - Number of bytes was read
* - (-1) if there are any errors, should check errno
*/
int transport_read(transport_handle_t t, char *buffer, int len, int timeout_ms);
/**
* @brief Poll the transport until readable or timeout
*
* @param[in] t The transport handle
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - 0 Timeout
* - (-1) If there are any errors, should check errno
* - other The transport can read
*/
int transport_poll_read(transport_handle_t t, int timeout_ms);
/**
* @brief Transport write function
*
* @param t The transport handle
* @param buffer The buffer
* @param[in] len The length
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - Number of bytes was written
* - (-1) if there are any errors, should check errno
*/
int transport_write(transport_handle_t t, const char *buffer, int len, int timeout_ms);
/**
* @brief Poll the transport until writeable or timeout
*
* @param[in] t The transport handle
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - 0 Timeout
* - (-1) If there are any errors, should check errno
* - other The transport can write
*/
int transport_poll_write(transport_handle_t t, int timeout_ms);
/**
* @brief Transport close
*
* @param t The transport handle
*
* @return
* - 0 if ok
* - (-1) if there are any errors, should check errno
*/
int transport_close(transport_handle_t t);
/**
* @brief Get user data context of this transport
*
* @param[in] t The transport handle
*
* @return The user data context
*/
void *transport_get_context_data(transport_handle_t t);
/**
* @brief Set the user context data for this transport
*
* @param[in] t The transport handle
* @param data The user data context
*
* @return
* - ESP_OK
*/
esp_err_t transport_set_context_data(transport_handle_t t, void *data);
/**
* @brief Set transport functions for the transport handle
*
* @param[in] t The transport handle
* @param[in] _connect The connect function pointer
* @param[in] _read The read function pointer
* @param[in] _write The write function pointer
* @param[in] _close The close function pointer
* @param[in] _poll_read The poll read function pointer
* @param[in] _poll_write The poll write function pointer
* @param[in] _destroy The destroy function pointer
*
* @return
* - ESP_OK
*/
esp_err_t transport_set_func(transport_handle_t t,
connect_func _connect,
io_read_func _read,
io_func _write,
trans_func _close,
poll_func _poll_read,
poll_func _poll_write,
trans_func _destroy);
#ifdef __cplusplus
}
#endif
#endif

41
lib/include/transport_ssl.h
View File

@ -1,41 +0,0 @@
/*
* 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 _TRANSPORT_SSL_H_
#define _TRANSPORT_SSL_H_
#include "transport.h"
#ifdef __cplusplus
extern "C" {
#endif
/**
* @brief Create new SSL transport, the transport handle must be release transport_destroy callback
*
* @return the allocated transport_handle_t, or NULL if the handle can not be allocated
*/
transport_handle_t transport_ssl_init();
/**
* @brief Set SSL certificate data (as PEM format).
* Note that, this function stores the pointer to data, rather than making a copy.
* So we need to make sure to keep the data lifetime before cleanup the connection
*
* @param t ssl transport
* @param[in] data The pem data
* @param[in] len The length
*/
void transport_ssl_set_cert_data(transport_handle_t t, const char *data, int len);
void transport_ssl_set_client_cert_data(transport_handle_t t, const char *data, int len);
void transport_ssl_set_client_key_data(transport_handle_t t, const char *data, int len);
#ifdef __cplusplus
}
#endif
#endif

27
lib/include/transport_tcp.h
View File

@ -1,27 +0,0 @@
/*
* 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 _TRANSPORT_TCP_H_
#define _TRANSPORT_TCP_H_
#include "transport.h"
#ifdef __cplusplus
extern "C" {
#endif
/**
* @brief Create TCP transport, the transport handle must be release transport_destroy callback
*
* @return the allocated transport_handle_t, or NULL if the handle can not be allocated
*/
transport_handle_t transport_tcp_init();
#ifdef __cplusplus
}
#endif
#endif

46
lib/include/transport_ws.h
View File

@ -1,46 +0,0 @@
/*
* 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 _TRANSPORT_WS_H_
#define _TRANSPORT_WS_H_
#include "transport.h"
#ifdef __cplusplus
extern "C" {
#endif
#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
#define WS_RESPONSE_OK 101
/**
* @brief Create TCP transport
*
* @return
* - transport
* - NULL
*/
transport_handle_t transport_ws_init(transport_handle_t parent_handle);
void transport_ws_set_path(transport_handle_t t, const char *path);
#ifdef __cplusplus
}
#endif
#endif

450
lib/mqtt_msg.c
View File

@ -29,15 +29,15 @@
*
*/
#include <stdint.h>
#include <stdbool.h>
#include <string.h>
#include "mqtt_msg.h"
#include "mqtt_config.h"
#include "platform.h"
#define MQTT_MAX_FIXED_HEADER_SIZE 3
#define MQTT_MAX_FIXED_HEADER_SIZE 5
enum mqtt_connect_flag
{
enum mqtt_connect_flag {
MQTT_CONNECT_FLAG_USERNAME = 1 << 7,
MQTT_CONNECT_FLAG_PASSWORD = 1 << 6,
MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5,
@ -45,8 +45,7 @@ enum mqtt_connect_flag
MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1
};
struct __attribute((__packed__)) mqtt_connect_variable_header
{
struct __attribute((__packed__)) mqtt_connect_variable_header {
uint8_t lengthMsb;
uint8_t lengthLsb;
#if defined(MQTT_PROTOCOL_311)
@ -60,10 +59,11 @@ struct __attribute((__packed__)) mqtt_connect_variable_header
uint8_t keepaliveLsb;
};
static int append_string(mqtt_connection_t* connection, const char* string, int len)
static int append_string(mqtt_connection_t *connection, const char *string, int len)
{
if (connection->message.length + len + 2 > connection->buffer_length)
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;
@ -73,7 +73,7 @@ static int append_string(mqtt_connection_t* connection, const char* string, int
return len + 2;
}
static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t message_id)
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
@ -81,8 +81,9 @@ static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t messag
message_id = platform_random(65535);
}
if (connection->message.length + 2 > connection->buffer_length)
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;
@ -90,98 +91,150 @@ static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t messag
return message_id;
}
static int init_message(mqtt_connection_t* connection)
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)
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)
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;
int message_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE;
int total_length = message_length;
int encoded_length = 0;
uint8_t encoded_lens[4] = {0};
// Check if we have fragmented message and update total_len
if (connection->message.fragmented_msg_total_length) {
total_length = connection->message.fragmented_msg_total_length - MQTT_MAX_FIXED_HEADER_SIZE;
}
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;
// Encode MQTT message length
int len_bytes = 0; // size of encoded message length
do {
encoded_length = total_length % 128;
total_length /= 128;
if (total_length > 0) {
encoded_length |= 0x80;
}
encoded_lens[len_bytes] = encoded_length;
len_bytes++;
} while (total_length > 0);
// Sanity check for MQTT header
if (len_bytes + 1 > MQTT_MAX_FIXED_HEADER_SIZE) {
return fail_message(connection);
}
// Save the header bytes
connection->message.length = message_length + len_bytes + 1; // msg len + encoded_size len + type (1 byte)
int offs = MQTT_MAX_FIXED_HEADER_SIZE - 1 - len_bytes;
connection->message.data = connection->buffer + offs;
connection->message.fragmented_msg_data_offset -= offs;
// type byte
connection->buffer[offs++] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
// length bytes
for (int j = 0; j < len_bytes; j++) {
connection->buffer[offs++] = encoded_lens[j];
}
return &connection->message;
}
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length)
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)
uint32_t mqtt_get_total_length(uint8_t *buffer, uint16_t length, int *fixed_size_len)
{
int i;
uint32_t totlen = 0;
for (i = 1; i < length; ++i)
{
for (i = 1; i < length; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (fixed_size_len) {
*fixed_size_len = i;
}
return totlen;
}
const char* mqtt_get_publish_topic(uint8_t* buffer, uint32_t* length)
bool mqtt_header_complete(uint8_t *buffer, uint16_t buffer_length)
{
uint16_t i;
uint16_t topiclen;
for (i = 1; i < MQTT_MAX_FIXED_HEADER_SIZE; ++i) {
if (i >= buffer_length) {
return false;
}
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
// i is now the length of the fixed header
if (i + 2 >= buffer_length) {
return false;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
i += topiclen;
if (mqtt_get_qos(buffer) > 0) {
i += 2;
}
// i is now the length of the fixed + variable header
return buffer_length >= i;
}
char *mqtt_get_publish_topic(uint8_t *buffer, uint32_t *length)
{
int i;
int totlen = 0;
int topiclen;
for (i = 1; i < *length; ++i)
{
for (i = 1; i < *length; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (i + 2 >= *length)
if (i + 2 >= *length) {
return NULL;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen > *length)
if (i + topiclen > *length) {
return NULL;
}
*length = topiclen;
return (const char*)(buffer + i);
}
return (char *)(buffer + i);
}
const char* mqtt_get_publish_data(uint8_t* buffer, uint32_t* length)
char *mqtt_get_publish_data(uint8_t *buffer, uint32_t *length)
{
int i;
int totlen = 0;
@ -189,115 +242,118 @@ const char* mqtt_get_publish_data(uint8_t* buffer, uint32_t* length)
int blength = *length;
*length = 0;
for (i = 1; i < blength; ++i)
{
for (i = 1; i < blength; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (i + 2 >= blength)
if (i + 2 >= blength) {
return NULL;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen >= blength)
if (i + topiclen >= blength) {
return NULL;
}
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (i + 2 >= blength)
if (mqtt_get_qos(buffer) > 0) {
if (i + 2 >= blength) {
return NULL;
}
i += 2;
}
if (totlen < i)
if (totlen < i) {
return NULL;
}
if (totlen <= blength)
if (totlen <= blength) {
*length = totlen - i;
else
} else {
*length = blength - i;
return (const char*)(buffer + i);
}
return (char *)(buffer + i);
}
uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length)
uint16_t mqtt_get_id(uint8_t *buffer, uint16_t length)
{
if (length < 1)
if (length < 1) {
return 0;
}
switch (mqtt_get_type(buffer))
{
case MQTT_MSG_TYPE_PUBLISH:
{
int i;
int topiclen;
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;
for (i = 1; i < length; ++i) {
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
default:
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)
mqtt_message_t *mqtt_msg_connect(mqtt_connection_t *connection, mqtt_connect_info_t *info)
{
struct mqtt_connect_variable_header* variable_header;
struct mqtt_connect_variable_header *variable_header;
init_message(connection);
if (connection->message.length + sizeof(*variable_header) > connection->buffer_length)
if (connection->message.length + sizeof(*variable_header) > connection->buffer_length) {
return fail_message(connection);
variable_header = (void*)(connection->buffer + connection->message.length);
}
variable_header = (void *)(connection->buffer + connection->message.length);
connection->message.length += sizeof(*variable_header);
variable_header->lengthMsb = 0;
@ -315,43 +371,46 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
variable_header->keepaliveMsb = info->keepalive >> 8;
variable_header->keepaliveLsb = info->keepalive & 0xff;
if (info->clean_session)
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
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)
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)
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)
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)
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)
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;
}
@ -359,114 +418,173 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
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)
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')
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)
if (append_string(connection, topic, strlen(topic)) < 0) {
return fail_message(connection);
memcpy(connection->buffer + connection->message.length, data, data_length);
connection->message.length += data_length;
}
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) {
// Not enough size in buffer -> fragment this message
connection->message.fragmented_msg_data_offset = connection->message.length;
memcpy(connection->buffer + connection->message.length, data, connection->buffer_length - connection->message.length);
connection->message.length = connection->buffer_length;
connection->message.fragmented_msg_total_length = data_length + connection->message.fragmented_msg_data_offset;
} else {
memcpy(connection->buffer + connection->message.length, data, data_length);
connection->message.length += data_length;
connection->message.fragmented_msg_total_length = 0;
}
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)
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)
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)
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)
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)
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)
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)
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)
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)
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')
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0)
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
if (append_string(connection, topic, strlen(topic)) < 0)
if (append_string(connection, topic, strlen(topic)) < 0) {
return fail_message(connection);
}
if (connection->message.length + 1 > connection->buffer_length)
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)
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')
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0)
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
if (append_string(connection, topic, strlen(topic)) < 0)
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)
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)
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)
mqtt_message_t *mqtt_msg_disconnect(mqtt_connection_t *connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0);
}
/*
* check flags: [MQTT-2.2.2-1], [MQTT-2.2.2-2]
* returns 0 if flags are invalid, otherwise returns 1
*/
int mqtt_has_valid_msg_hdr(uint8_t *buffer, uint16_t length)
{
int qos, dup;
if (length < 1) {
return 0;
}
switch (mqtt_get_type(buffer)) {
case MQTT_MSG_TYPE_CONNECT:
case MQTT_MSG_TYPE_CONNACK:
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_PINGREQ:
case MQTT_MSG_TYPE_PINGRESP:
case MQTT_MSG_TYPE_DISCONNECT:
return (buffer[0] & 0x0f) == 0; /* all flag bits are 0 */
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_SUBSCRIBE:
case MQTT_MSG_TYPE_UNSUBSCRIBE:
return (buffer[0] & 0x0f) == 0x02; /* only bit 1 is set */
case MQTT_MSG_TYPE_PUBLISH:
qos = mqtt_get_qos(buffer);
dup = mqtt_get_dup(buffer);
/*
* there is no qos=3 [MQTT-3.3.1-4]
* dup flag must be set to 0 for all qos=0 messages [MQTT-3.3.1-2]
*/
return (qos < 3) && ((qos > 0) || (dup == 0));
default:
return 0;
}
}

83
lib/mqtt_outbox.c
View File

@ -1,12 +1,30 @@
#include "mqtt_outbox.h"
#include <stdlib.h>
#include <string.h>
#include "rom/queue.h"
#include "sys/queue.h"
#include "esp_log.h"
#ifndef CONFIG_MQTT_CUSTOM_OUTBOX
static const char *TAG = "OUTBOX";
outbox_handle_t outbox_init()
typedef struct outbox_item {
char *buffer;
int len;
int msg_id;
int msg_type;
int msg_qos;
int tick;
int retry_count;
pending_state_t pending;
STAILQ_ENTRY(outbox_item) next;
} outbox_item_t;
STAILQ_HEAD(outbox_list_t, outbox_item);
outbox_handle_t outbox_init(void)
{
outbox_handle_t outbox = calloc(1, sizeof(struct outbox_list_t));
ESP_MEM_CHECK(TAG, outbox, return NULL);
@ -14,22 +32,27 @@ outbox_handle_t outbox_init()
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 outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, 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->msg_id = message->msg_id;
item->msg_type = message->msg_type;
item->msg_qos = message->msg_qos;
item->tick = tick;
item->len = len;
item->buffer = malloc(len);
item->len = message->len + message->remaining_len;
item->pending = QUEUED;
item->buffer = malloc(message->len + message->remaining_len);
ESP_MEM_CHECK(TAG, item->buffer, {
free(item);
return NULL;
});
memcpy(item->buffer, data, len);
memcpy(item->buffer, message->data, message->len);
if (message->remaining_data) {
memcpy(item->buffer + message->len, message->remaining_data, message->remaining_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));
ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%d", message->msg_id, message->msg_type, message->len + message->remaining_len, outbox_get_size(outbox));
return item;
}
@ -44,21 +67,37 @@ outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id)
return NULL;
}
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox)
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, int *tick)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (!item->pending) {
if (item->pending == pending) {
if (tick) {
*tick = item->tick;
}
return item;
}
}
return NULL;
}
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos)
{
if (item) {
*len = item->len;
*msg_id = item->msg_id;
*msg_type = item->msg_type;
*qos = item->msg_qos;
return (uint8_t *)item->buffer;
}
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) {
if (item->msg_id == msg_id && (0xFF & (item->msg_type)) == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
@ -82,11 +121,11 @@ esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id)
}
return ESP_OK;
}
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id)
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id, pending_state_t pending)
{
outbox_item_handle_t item = outbox_get(outbox, msg_id);
if (item) {
item->pending = true;
item->pending = pending;
return ESP_OK;
}
return ESP_FAIL;
@ -106,18 +145,20 @@ esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type)
return ESP_OK;
}
esp_err_t outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout)
int outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout)
{
int deleted_items = 0;
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);
deleted_items ++;
}
}
return ESP_OK;
return deleted_items;
}
int outbox_get_size(outbox_handle_t outbox)
@ -125,15 +166,17 @@ int outbox_get_size(outbox_handle_t outbox)
int siz = 0;
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
siz += item->len;
// Suppressing "use after free" warning as this could happen only if queue is in inconsistent state
// which never happens if STAILQ interface used
siz += item->len; // NOLINT(clang-analyzer-unix.Malloc)
}
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);
while (outbox_get_size(outbox) > max_size) {
outbox_item_handle_t item = outbox_dequeue(outbox, CONFIRMED, NULL);
if (item == NULL) {
return ESP_FAIL;
}
@ -149,3 +192,5 @@ void outbox_destroy(outbox_handle_t outbox)
outbox_cleanup(outbox, 0);
free(outbox);
}
#endif /* CONFIG_MQTT_CUSTOM_OUTBOX */

8
lib/platform_esp32_idf.c
View File

@ -9,7 +9,7 @@ static const char *TAG = "PLATFORM";
#define MAX_ID_STRING (32)
char *platform_create_id_string()
char *platform_create_id_string(void)
{
uint8_t mac[6];
char *id_string = calloc(1, MAX_ID_STRING);
@ -21,14 +21,14 @@ char *platform_create_id_string()
int platform_random(int max)
{
return esp_random()%max;
return esp_random() % max;
}
long long platform_tick_get_ms()
long long platform_tick_get_ms(void)
{
struct timeval te;
gettimeofday(&te, NULL); // get current time
long long milliseconds = te.tv_sec*1000LL + te.tv_usec/1000; // calculate milliseconds
long long milliseconds = te.tv_sec * 1000LL + te.tv_usec / 1000; // calculate milliseconds
// printf("milliseconds: %lld\n", milliseconds);
return milliseconds;
}

218
lib/transport.c
View File

@ -1,218 +0,0 @@
#include <stdlib.h>
#include <string.h>
#include "rom/queue.h"
#include "esp_log.h"
#include "transport.h"
#include "platform.h"
static const char *TAG = "TRANSPORT";
/**
* Transport layer structure, which will provide functions, basic properties for transport types
*/
struct transport_item_t {
int port;
int socket; /*!< Socket to use in this transport */
char *scheme; /*!< Tag name */
void *context; /*!< Context data */
void *data; /*!< Additional transport data */
connect_func _connect; /*!< Connect function of this transport */
io_read_func _read; /*!< Read */
io_func _write; /*!< Write */
trans_func _close; /*!< Close */
poll_func _poll_read; /*!< Poll and read */
poll_func _poll_write; /*!< Poll and write */
trans_func _destroy; /*!< Destroy and free transport */
STAILQ_ENTRY(transport_item_t) next;
};
/**
* This list will hold all transport available
*/
STAILQ_HEAD(transport_list_t, transport_item_t);
transport_list_handle_t transport_list_init()
{
transport_list_handle_t list = calloc(1, sizeof(struct transport_list_t));
ESP_MEM_CHECK(TAG, list, return NULL);
STAILQ_INIT(list);
return list;
}
esp_err_t transport_list_add(transport_list_handle_t list, transport_handle_t t, const char *scheme)
{
if (list == NULL || t == NULL) {
return ESP_ERR_INVALID_ARG;
}
t->scheme = calloc(1, strlen(scheme) + 1);
ESP_MEM_CHECK(TAG, t->scheme, return ESP_ERR_NO_MEM);
strcpy(t->scheme, scheme);
STAILQ_INSERT_TAIL(list, t, next);
return ESP_OK;
}
transport_handle_t transport_list_get_transport(transport_list_handle_t list, const char *scheme)
{
if (!list) {
return NULL;
}
if (scheme == NULL) {
return STAILQ_FIRST(list);
}
transport_handle_t item;
STAILQ_FOREACH(item, list, next) {
if (strcasecmp(item->scheme, scheme) == 0) {
return item;
}
}
return NULL;
}
esp_err_t transport_list_destroy(transport_list_handle_t list)
{
transport_list_clean(list);
free(list);
return ESP_OK;
}
esp_err_t transport_list_clean(transport_list_handle_t list)
{
transport_handle_t item = STAILQ_FIRST(list);
transport_handle_t tmp;
while (item != NULL) {
tmp = STAILQ_NEXT(item, next);
if (item->_destroy) {
item->_destroy(item);
}
transport_destroy(item);
item = tmp;
}
STAILQ_INIT(list);
return ESP_OK;
}
transport_handle_t transport_init()
{
transport_handle_t t = calloc(1, sizeof(struct transport_item_t));
ESP_MEM_CHECK(TAG, t, return NULL);
return t;
}
esp_err_t transport_destroy(transport_handle_t t)
{
if (t->scheme) {
free(t->scheme);
}
free(t);
return ESP_OK;
}
int transport_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
int ret = -1;
if (t && t->_connect) {
return t->_connect(t, host, port, timeout_ms);
}
return ret;
}
int transport_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
if (t && t->_read) {
return t->_read(t, buffer, len, timeout_ms);
}
return -1;
}
int transport_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
if (t && t->_write) {
return t->_write(t, buffer, len, timeout_ms);
}
return -1;
}
int transport_poll_read(transport_handle_t t, int timeout_ms)
{
if (t && t->_poll_read) {
return t->_poll_read(t, timeout_ms);
}
return -1;
}
int transport_poll_write(transport_handle_t t, int timeout_ms)
{
if (t && t->_poll_write) {
return t->_poll_write(t, timeout_ms);
}
return -1;
}
int transport_close(transport_handle_t t)
{
if (t && t->_close) {
return t->_close(t);
}
return 0;
}
void *transport_get_context_data(transport_handle_t t)
{
if (t) {
return t->data;
}
return NULL;
}
esp_err_t transport_set_context_data(transport_handle_t t, void *data)
{
if (t) {
t->data = data;
return ESP_OK;
}
return ESP_FAIL;
}
esp_err_t transport_set_func(transport_handle_t t,
connect_func _connect,
io_read_func _read,
io_func _write,
trans_func _close,
poll_func _poll_read,
poll_func _poll_write,
trans_func _destroy)
{
if (t == NULL) {
return ESP_FAIL;
}
t->_connect = _connect;
t->_read = _read;
t->_write = _write;
t->_close = _close;
t->_poll_read = _poll_read;
t->_poll_write = _poll_write;
t->_destroy = _destroy;
return ESP_OK;
}
int transport_get_default_port(transport_handle_t t)
{
if (t == NULL) {
return -1;
}
return t->port;
}
esp_err_t transport_set_default_port(transport_handle_t t, int port)
{
if (t == NULL) {
return ESP_FAIL;
}
t->port = port;
return ESP_OK;
}

310
lib/transport_ssl.c
View File

@ -1,310 +0,0 @@
#include <string.h>
#include <stdlib.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "lwip/err.h"
#include "lwip/sockets.h"
#include "lwip/sys.h"
#include "lwip/netdb.h"
#include "lwip/dns.h"
#include "mbedtls/platform.h"
#include "mbedtls/net_sockets.h"
#include "mbedtls/esp_debug.h"
#include "mbedtls/ssl.h"
#include "mbedtls/entropy.h"
#include "mbedtls/ctr_drbg.h"
#include "mbedtls/error.h"
#include "mbedtls/certs.h"
#include "esp_log.h"
#include "esp_system.h"
#include "platform.h"
#include "transport.h"
#include "transport_ssl.h"
static const char *TAG = "TRANS_SSL";
/**
* mbedtls specific transport data
*/
typedef struct {
mbedtls_entropy_context entropy;
mbedtls_ctr_drbg_context ctr_drbg;
mbedtls_ssl_context ctx;
mbedtls_x509_crt cacert;
mbedtls_x509_crt client_cert;
mbedtls_pk_context client_key;
mbedtls_ssl_config conf;
mbedtls_net_context client_fd;
void *cert_pem_data;
int cert_pem_len;
void *client_cert_pem_data;
int client_cert_pem_len;
void *client_key_pem_data;
int client_key_pem_len;
bool mutual_authentication;
bool ssl_initialized;
bool verify_server;
} transport_ssl_t;
static int ssl_close(transport_handle_t t);
static int ssl_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
int ret = -1, flags;
struct timeval tv;
transport_ssl_t *ssl = transport_get_context_data(t);
if (!ssl) {
return -1;
}
ssl->ssl_initialized = true;
mbedtls_ssl_init(&ssl->ctx);
mbedtls_ctr_drbg_init(&ssl->ctr_drbg);
mbedtls_ssl_config_init(&ssl->conf);
mbedtls_entropy_init(&ssl->entropy);
if ((ret = mbedtls_ssl_config_defaults(&ssl->conf,
MBEDTLS_SSL_IS_CLIENT,
MBEDTLS_SSL_TRANSPORT_STREAM,
MBEDTLS_SSL_PRESET_DEFAULT)) != 0) {
ESP_LOGE(TAG, "mbedtls_ssl_config_defaults returned %d", ret);
goto exit;
}
if ((ret = mbedtls_ctr_drbg_seed(&ssl->ctr_drbg, mbedtls_entropy_func, &ssl->entropy, NULL, 0)) != 0) {
ESP_LOGE(TAG, "mbedtls_ctr_drbg_seed returned %d", ret);
goto exit;
}
if (ssl->cert_pem_data) {
mbedtls_x509_crt_init(&ssl->cacert);
ssl->verify_server = true;
if ((ret = mbedtls_x509_crt_parse(&ssl->cacert, ssl->cert_pem_data, ssl->cert_pem_len + 1)) < 0) {
ESP_LOGE(TAG, "mbedtls_x509_crt_parse returned -0x%x\n\nDATA=%s,len=%d", -ret, (char*)ssl->cert_pem_data, ssl->cert_pem_len);
goto exit;
}
mbedtls_ssl_conf_ca_chain(&ssl->conf, &ssl->cacert, NULL);
mbedtls_ssl_conf_authmode(&ssl->conf, MBEDTLS_SSL_VERIFY_REQUIRED);
if ((ret = mbedtls_ssl_set_hostname(&ssl->ctx, host)) != 0) {
ESP_LOGE(TAG, "mbedtls_ssl_set_hostname returned -0x%x", -ret);
goto exit;
}
} else {
mbedtls_ssl_conf_authmode(&ssl->conf, MBEDTLS_SSL_VERIFY_NONE);
}
if (ssl->client_cert_pem_data && ssl->client_key_pem_data) {
mbedtls_x509_crt_init(&ssl->client_cert);
mbedtls_pk_init(&ssl->client_key);
ssl->mutual_authentication = true;
if ((ret = mbedtls_x509_crt_parse(&ssl->client_cert, ssl->client_cert_pem_data, ssl->client_cert_pem_len + 1)) < 0) {
ESP_LOGE(TAG, "mbedtls_x509_crt_parse returned -0x%x\n\nDATA=%s,len=%d", -ret, (char*)ssl->client_cert_pem_data, ssl->client_cert_pem_len);
goto exit;
}
if ((ret = mbedtls_pk_parse_key(&ssl->client_key, ssl->client_key_pem_data, ssl->client_key_pem_len + 1, NULL, 0)) < 0) {
ESP_LOGE(TAG, "mbedtls_pk_parse_keyfile returned -0x%x\n\nDATA=%s,len=%d", -ret, (char*)ssl->client_key_pem_data, ssl->client_key_pem_len);
goto exit;
}
if ((ret = mbedtls_ssl_conf_own_cert(&ssl->conf, &ssl->client_cert, &ssl->client_key)) < 0) {
ESP_LOGE(TAG, "mbedtls_ssl_conf_own_cert returned -0x%x\n", -ret);
goto exit;
}
} else if (ssl->client_cert_pem_data || ssl->client_key_pem_data) {
ESP_LOGE(TAG, "You have to provide both client_cert_pem and client_key_pem for mutual authentication");
goto exit;
}
mbedtls_ssl_conf_rng(&ssl->conf, mbedtls_ctr_drbg_random, &ssl->ctr_drbg);
#ifdef CONFIG_MBEDTLS_DEBUG
mbedtls_esp_enable_debug_log(&ssl->conf, 4);
#endif
if ((ret = mbedtls_ssl_setup(&ssl->ctx, &ssl->conf)) != 0) {
ESP_LOGE(TAG, "mbedtls_ssl_setup returned -0x%x\n\n", -ret);
goto exit;
}
mbedtls_net_init(&ssl->client_fd);
ms_to_timeval(timeout_ms, &tv);
setsockopt(ssl->client_fd.fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
ESP_LOGD(TAG, "Connect to %s:%d", host, port);
char port_str[8] = {0};
sprintf(port_str, "%d", port);
if ((ret = mbedtls_net_connect(&ssl->client_fd, host, port_str, MBEDTLS_NET_PROTO_TCP)) != 0) {
ESP_LOGE(TAG, "mbedtls_net_connect returned -%x", -ret);
goto exit;
}
mbedtls_ssl_set_bio(&ssl->ctx, &ssl->client_fd, mbedtls_net_send, mbedtls_net_recv, NULL);
if((ret = mbedtls_ssl_set_hostname(&ssl->ctx, host)) != 0) {
ESP_LOGE(TAG, " failed\n ! mbedtls_ssl_set_hostname returned %d\n\n", ret);
goto exit;
}
ESP_LOGD(TAG, "Performing the SSL/TLS handshake...");
while ((ret = mbedtls_ssl_handshake(&ssl->ctx)) != 0) {
if (ret != MBEDTLS_ERR_SSL_WANT_READ && ret != MBEDTLS_ERR_SSL_WANT_WRITE) {
ESP_LOGE(TAG, "mbedtls_ssl_handshake returned -0x%x", -ret);
goto exit;
}
}
ESP_LOGD(TAG, "Verifying peer X.509 certificate...");
if ((flags = mbedtls_ssl_get_verify_result(&ssl->ctx)) != 0) {
/* In real life, we probably want to close connection if ret != 0 */
ESP_LOGW(TAG, "Failed to verify peer certificate!");
if (ssl->cert_pem_data) {
goto exit;
}
} else {
ESP_LOGD(TAG, "Certificate verified.");
}
ESP_LOGD(TAG, "Cipher suite is %s", mbedtls_ssl_get_ciphersuite(&ssl->ctx));
return ret;
exit:
ssl_close(t);
return ret;
}
static int ssl_poll_read(transport_handle_t t, int timeout_ms)
{
transport_ssl_t *ssl = transport_get_context_data(t);
fd_set readset;
FD_ZERO(&readset);
FD_SET(ssl->client_fd.fd, &readset);
struct timeval timeout;
ms_to_timeval(timeout_ms, &timeout);
return select(ssl->client_fd.fd + 1, &readset, NULL, NULL, &timeout);
}
static int ssl_poll_write(transport_handle_t t, int timeout_ms)
{
transport_ssl_t *ssl = transport_get_context_data(t);
fd_set writeset;
FD_ZERO(&writeset);
FD_SET(ssl->client_fd.fd, &writeset);
struct timeval timeout;
ms_to_timeval(timeout_ms, &timeout);
return select(ssl->client_fd.fd + 1, NULL, &writeset, NULL, &timeout);
}
static int ssl_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
int poll, ret;
transport_ssl_t *ssl = transport_get_context_data(t);
if ((poll = transport_poll_write(t, timeout_ms)) <= 0) {
ESP_LOGW(TAG, "Poll timeout or error, errno=%s, fd=%d, timeout_ms=%d", strerror(errno), ssl->client_fd.fd, timeout_ms);
return poll;
}
ret = mbedtls_ssl_write(&ssl->ctx, (const unsigned char *) buffer, len);
if (ret <= 0) {
ESP_LOGE(TAG, "mbedtls_ssl_write error, errno=%s", strerror(errno));
}
return ret;
}
static int ssl_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
int poll = -1, ret;
transport_ssl_t *ssl = transport_get_context_data(t);
if (mbedtls_ssl_get_bytes_avail(&ssl->ctx) <= 0) {
if ((poll = transport_poll_read(t, timeout_ms)) <= 0) {
return poll;
}
}
ret = mbedtls_ssl_read(&ssl->ctx, (unsigned char *)buffer, len);
if (ret == 0) {
return -1;
}
return ret;
}
static int ssl_close(transport_handle_t t)
{
int ret = -1;
transport_ssl_t *ssl = transport_get_context_data(t);
if (ssl->ssl_initialized) {
ESP_LOGD(TAG, "Cleanup mbedtls");
mbedtls_ssl_close_notify(&ssl->ctx);
mbedtls_ssl_session_reset(&ssl->ctx);
mbedtls_net_free(&ssl->client_fd);
mbedtls_ssl_config_free(&ssl->conf);
if (ssl->verify_server) {
mbedtls_x509_crt_free(&ssl->cacert);
}
if (ssl->mutual_authentication) {
mbedtls_x509_crt_free(&ssl->client_cert);
mbedtls_pk_free(&ssl->client_key);
}
mbedtls_ctr_drbg_free(&ssl->ctr_drbg);
mbedtls_entropy_free(&ssl->entropy);
mbedtls_ssl_free(&ssl->ctx);
ssl->mutual_authentication = false;
ssl->ssl_initialized = false;
ssl->verify_server = false;
}
return ret;
}
static int ssl_destroy(transport_handle_t t)
{
transport_ssl_t *ssl = transport_get_context_data(t);
transport_close(t);
free(ssl);
return 0;
}
void transport_ssl_set_cert_data(transport_handle_t t, const char *data, int len)
{
transport_ssl_t *ssl = transport_get_context_data(t);
if (t && ssl) {
ssl->cert_pem_data = (void *)data;
ssl->cert_pem_len = len;
}
}
void transport_ssl_set_client_cert_data(transport_handle_t t, const char *data, int len)
{
transport_ssl_t *ssl = transport_get_context_data(t);
if (t && ssl) {
ssl->client_cert_pem_data = (void *)data;
ssl->client_cert_pem_len = len;
}
}
void transport_ssl_set_client_key_data(transport_handle_t t, const char *data, int len)
{
transport_ssl_t *ssl = transport_get_context_data(t);
if (t && ssl) {
ssl->client_key_pem_data = (void *)data;
ssl->client_key_pem_len = len;
}
}
transport_handle_t transport_ssl_init()
{
transport_handle_t t = transport_init();
transport_ssl_t *ssl = calloc(1, sizeof(transport_ssl_t));
ESP_MEM_CHECK(TAG, ssl, return NULL);
mbedtls_net_init(&ssl->client_fd);
transport_set_context_data(t, ssl);
transport_set_func(t, ssl_connect, ssl_read, ssl_write, ssl_close, ssl_poll_read, ssl_poll_write, ssl_destroy);
return t;
}

152
lib/transport_tcp.c
View File

@ -1,152 +0,0 @@
#include <stdlib.h>
#include <string.h>
#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"
#include "esp_log.h"
#include "esp_system.h"
#include "esp_err.h"
#include "platform.h"
#include "transport.h"
static const char *TAG = "TRANS_TCP";
typedef struct {
int sock;
} transport_tcp_t;
static int resolve_dns(const char *host, struct sockaddr_in *ip) {
struct hostent *he;
struct in_addr **addr_list;
he = gethostbyname(host);
if (he == NULL) {
return ESP_FAIL;
}
addr_list = (struct in_addr **)he->h_addr_list;
if (addr_list[0] == NULL) {
return ESP_FAIL;
}
ip->sin_family = AF_INET;
memcpy(&ip->sin_addr, addr_list[0], sizeof(ip->sin_addr));
return ESP_OK;
}
static int tcp_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
struct sockaddr_in remote_ip;
struct timeval tv;
transport_tcp_t *tcp = transport_get_context_data(t);
bzero(&remote_ip, sizeof(struct sockaddr_in));
//if stream_host is not ip address, resolve it AF_INET,servername,&serveraddr.sin_addr
if (inet_pton(AF_INET, host, &remote_ip.sin_addr) != 1) {
if (resolve_dns(host, &remote_ip) < 0) {
return -1;
}
}
tcp->sock = socket(PF_INET, SOCK_STREAM, 0);
if (tcp->sock < 0) {
ESP_LOGE(TAG, "Error create socket");
return -1;
}
remote_ip.sin_family = AF_INET;
remote_ip.sin_port = htons(port);
ms_to_timeval(timeout_ms, &tv);
setsockopt(tcp->sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
ESP_LOGD(TAG, "[sock=%d],connecting to server IP:%s,Port:%d...",
tcp->sock, ipaddr_ntoa((const ip_addr_t*)&remote_ip.sin_addr.s_addr), port);
if (connect(tcp->sock, (struct sockaddr *)(&remote_ip), sizeof(struct sockaddr)) != 0) {
close(tcp->sock);
tcp->sock = -1;
return -1;
}
return tcp->sock;
}
static int tcp_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
int poll;
transport_tcp_t *tcp = transport_get_context_data(t);
if ((poll = transport_poll_write(t, timeout_ms)) <= 0) {
return poll;
}
return write(tcp->sock, buffer, len);
}
static int tcp_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_context_data(t);
int poll = -1;
if ((poll = transport_poll_read(t, timeout_ms)) <= 0) {
return poll;
}
int read_len = read(tcp->sock, buffer, len);
if (read_len == 0) {
return -1;
}
return read_len;
}
static int tcp_poll_read(transport_handle_t t, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_context_data(t);
fd_set readset;
FD_ZERO(&readset);
FD_SET(tcp->sock, &readset);
struct timeval timeout;
ms_to_timeval(timeout_ms, &timeout);
return select(tcp->sock + 1, &readset, NULL, NULL, &timeout);
}
static int tcp_poll_write(transport_handle_t t, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_context_data(t);
fd_set writeset;
FD_ZERO(&writeset);
FD_SET(tcp->sock, &writeset);
struct timeval timeout;
ms_to_timeval(timeout_ms, &timeout);
return select(tcp->sock + 1, NULL, &writeset, NULL, &timeout);
}
static int tcp_close(transport_handle_t t)
{
transport_tcp_t *tcp = transport_get_context_data(t);
int ret = -1;
if (tcp->sock >= 0) {
ret = close(tcp->sock);
tcp->sock = -1;
}
return ret;
}
static esp_err_t tcp_destroy(transport_handle_t t)
{
transport_tcp_t *tcp = transport_get_context_data(t);
transport_close(t);
free(tcp);
return 0;
}
transport_handle_t transport_tcp_init()
{
transport_handle_t t = transport_init();
transport_tcp_t *tcp = calloc(1, sizeof(transport_tcp_t));
ESP_MEM_CHECK(TAG, tcp, return NULL);
tcp->sock = -1;
transport_set_func(t, tcp_connect, tcp_read, tcp_write, tcp_close, tcp_poll_read, tcp_poll_write, tcp_destroy);
transport_set_context_data(t, tcp);
return t;
}

251
lib/transport_ws.c
View File

@ -1,251 +0,0 @@
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include "platform.h"
#include "transport.h"
#include "transport_tcp.h"
#include "transport_ws.h"
#include "mbedtls/base64.h"
#include "mbedtls/sha1.h"
static const char *TAG = "TRANSPORT_WS";
#define DEFAULT_WS_BUFFER (1024)
typedef struct {
char *path;
char *buffer;
transport_handle_t parent;
} transport_ws_t;
static char *trimwhitespace(const char *str)
{
char *end;
// Trim leading space
while (isspace((unsigned char)*str)) str++;
if (*str == 0) {
return (char *)str;
}
// Trim trailing space
end = (char *)(str + strlen(str) - 1);
while (end > str && isspace((unsigned char)*end)) end--;
// Write new null terminator
*(end + 1) = 0;
return (char *)str;
}
static char *get_http_header(const char *buffer, const char *key)
{
char *found = strstr(buffer, key);
if (found) {
found += strlen(key);
char *found_end = strstr(found, "\r\n");
if (found_end) {
found_end[0] = 0;//terminal string
return trimwhitespace(found);
}
}
return NULL;
}
static int ws_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
transport_ws_t *ws = transport_get_context_data(t);
if (transport_connect(ws->parent, host, port, timeout_ms) < 0) {
ESP_LOGE(TAG, "Error connect to ther server");
}
unsigned char random_key[16] = { 0 }, client_key[32] = {0};
int i;
for (i = 0; i < sizeof(random_key); i++) {
random_key[i] = rand() & 0xFF;
}
size_t outlen = 0;
mbedtls_base64_encode(client_key, 32, &outlen, random_key, 16);
int len = snprintf(ws->buffer, DEFAULT_WS_BUFFER,
"GET %s HTTP/1.1\r\n"
"Connection: Upgrade\r\n"
"Host: %s:%d\r\n"
"Upgrade: websocket\r\n"
"Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Protocol: mqtt\r\n"
"Sec-WebSocket-Key: %s\r\n"
"User-Agent: ESP32 MQTT Client\r\n\r\n",
ws->path,
host, port,
client_key);
ESP_LOGD(TAG, "Write upgrate request\r\n%s", ws->buffer);
if (transport_write(ws->parent, ws->buffer, len, timeout_ms) <= 0) {
ESP_LOGE(TAG, "Error write Upgrade header %s", ws->buffer);
return -1;
}
if ((len = transport_read(ws->parent, ws->buffer, DEFAULT_WS_BUFFER, timeout_ms)) <= 0) {
ESP_LOGE(TAG, "Error read response for Upgrade header %s", ws->buffer);
return -1;
}
char *server_key = get_http_header(ws->buffer, "Sec-WebSocket-Accept:");
if (server_key == NULL) {
ESP_LOGE(TAG, "Sec-WebSocket-Accept not found");
return -1;
}
unsigned char client_key_b64[64], valid_client_key[20], accept_key[32] = {0};
int key_len = sprintf((char*)client_key_b64, "%s258EAFA5-E914-47DA-95CA-C5AB0DC85B11", (char*)client_key);
mbedtls_sha1(client_key_b64, (size_t)key_len, valid_client_key);
mbedtls_base64_encode(accept_key, 32, &outlen, valid_client_key, 20);
accept_key[outlen] = 0;
ESP_LOGD(TAG, "server key=%s, send_key=%s, accept_key=%s", (char *)server_key, (char*)client_key, accept_key);
if (strcmp((char*)accept_key, (char*)server_key) != 0) {
ESP_LOGE(TAG, "Invalid websocket key");
return -1;
}
return 0;
}
static int ws_write(transport_handle_t t, const char *buff, int len, int timeout_ms)
{
transport_ws_t *ws = transport_get_context_data(t);
char ws_header[MAX_WEBSOCKET_HEADER_SIZE];
char *mask;
int header_len = 0, i;
char *buffer = (char *)buff;
int poll_write;
if ((poll_write = transport_poll_write(ws->parent, timeout_ms)) <= 0) {
return poll_write;
}
ws_header[header_len++] = WS_OPCODE_BINARY | WS_FIN;
// NOTE: no support for > 16-bit sized messages
if (len > 125) {
ws_header[header_len++] = WS_SIZE16 | WS_MASK;
ws_header[header_len++] = (uint8_t)(len >> 8);
ws_header[header_len++] = (uint8_t)(len & 0xFF);
} else {
ws_header[header_len++] = (uint8_t)(len | WS_MASK);
}
mask = &ws_header[header_len];
ws_header[header_len++] = rand() & 0xFF;
ws_header[header_len++] = rand() & 0xFF;
ws_header[header_len++] = rand() & 0xFF;
ws_header[header_len++] = rand() & 0xFF;
for (i = 0; i < len; ++i) {
buffer[i] = (buffer[i] ^ mask[i % 4]);
}
if (transport_write(ws->parent, ws_header, header_len, timeout_ms) != header_len) {
ESP_LOGE(TAG, "Error write header");
return -1;
}
return transport_write(ws->parent, buffer, len, timeout_ms);
}
static int ws_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
transport_ws_t *ws = transport_get_context_data(t);
int payload_len;
char *data_ptr = buffer, opcode, mask, *mask_key = NULL;
int rlen;
int poll_read;
if ((poll_read = transport_poll_read(ws->parent, timeout_ms)) <= 0) {
return poll_read;
}
if ((rlen = transport_read(ws->parent, buffer, len, timeout_ms)) <= 0) {
ESP_LOGE(TAG, "Error read data");
return rlen;
}
opcode = (*data_ptr & 0x0F);
data_ptr ++;
mask = ((*data_ptr >> 7) & 0x01);
payload_len = (*data_ptr & 0x7F);
data_ptr++;
ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", opcode, mask, payload_len);
if (payload_len == 126) {
// headerLen += 2;
payload_len = data_ptr[0] << 8 | data_ptr[1];
data_ptr += 2;
} else if (payload_len == 127) {
// headerLen += 8;
if (data_ptr[0] != 0 || data_ptr[1] != 0 || data_ptr[2] != 0 || data_ptr[3] != 0) {
// really too big!
payload_len = 0xFFFFFFFF;
} else {
payload_len = data_ptr[4] << 24 | data_ptr[5] << 16 | data_ptr[6] << 8 | data_ptr[7];
}
data_ptr += 8;
}
if (mask) {
mask_key = data_ptr;
data_ptr += 4;
for (int i = 0; i < payload_len; i++) {
buffer[i] = (data_ptr[i] ^ mask_key[i % 4]);
}
} else {
memmove(buffer, data_ptr, payload_len);
}
return payload_len;
}
static int ws_poll_read(transport_handle_t t, int timeout_ms)
{
transport_ws_t *ws = transport_get_context_data(t);
return transport_poll_read(ws->parent, timeout_ms);
}
static int ws_poll_write(transport_handle_t t, int timeout_ms)
{
transport_ws_t *ws = transport_get_context_data(t);
return transport_poll_write(ws->parent, timeout_ms);;
}
static int ws_close(transport_handle_t t)
{
transport_ws_t *ws = transport_get_context_data(t);
return transport_close(ws->parent);
}
static esp_err_t ws_destroy(transport_handle_t t)
{
transport_ws_t *ws = transport_get_context_data(t);
free(ws->buffer);
free(ws->path);
free(ws);
return 0;
}
void transport_ws_set_path(transport_handle_t t, const char *path)
{
transport_ws_t *ws = transport_get_context_data(t);
ws->path = realloc(ws->path, strlen(path) + 1);
strcpy(ws->path, path);
}
transport_handle_t transport_ws_init(transport_handle_t parent_handle)
{
transport_handle_t t = transport_init();
transport_ws_t *ws = calloc(1, sizeof(transport_ws_t));
ESP_MEM_CHECK(TAG, ws, return NULL);
ws->parent = parent_handle;
ws->path = strdup("/");
ESP_MEM_CHECK(TAG, ws->path, return NULL);
ws->buffer = malloc(DEFAULT_WS_BUFFER);
ESP_MEM_CHECK(TAG, ws->buffer, {
free(ws->path);
free(ws);
return NULL;
});
transport_set_func(t, ws_connect, ws_read, ws_write, ws_close, ws_poll_read, ws_poll_write, ws_destroy);
transport_set_context_data(t, ws);
return t;
}

33
modify_for_legacy_idf.sh Executable file
View File

@ -0,0 +1,33 @@
#!/usr/bin/env bash
if [[ -z $1 ]]; then
LATEST_IDF=master
else
LATEST_IDF=$1
fi
# 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 ${LATEST_IDF}
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

1153
mqtt_client.c
View File

File diff suppressed because it is too large Load Diff

9
static-analysis-rules.yml Normal file
View File

@ -0,0 +1,9 @@
limits:
"clang-analyzer-core.NullDereference" : 0
"clang-analyzer-unix.Malloc" : 0
ignore:
- "llvm-header-guard"
- "llvm-include-order"
skip: