espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2026-08-03 20:14:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'feat/mqtt-conformance-app-cbor' into 'master'

Browse Source 
MR: Add RPC like based on json to conformance test app
See merge request espressif/esp-mqtt!284
This commit is contained in:
Euripedes Rocha
2026-07-14 12:26:19 +02:00
parent ccd6cf6cb6 de09b7c0c0
commit e08670365a
11 changed files with 1141 additions and 472 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/apps/mqtt_conformance/README.md
+73 -2
View File
@@ -4,14 +4,85 @@ This app exposes a console API for pytest-embedded HIL tests that target MQTT co
## Console commands
- `init`: Create and configure MQTT client
- `set_uri <uri>`: Override broker URI before `start`
- `init <base64_json>`: Create MQTT client from a base64-encoded JSON config (must include `uri`)
- `config <base64_json>`: Apply base64-encoded JSON config to initialized client
- `start`: Start MQTT client
- `stop`: Stop MQTT client
- `disconnect`: Request disconnect
- `reconnect`: Request reconnect
- `destroy`: Destroy MQTT client
- `subscribe <topic> <qos>`: Subscribe to topic
- `unsubscribe <topic>`: Unsubscribe topic
- `publish <topic> <pattern> <pattern_repetitions> <qos> <retain> <enqueue>`: Publish payload
## JSON config keys
All configuration is passed as a base64-encoded JSON object with exactly one top-level key, naming which config category the blob targets. The JSON shape mirrors the real esp_mqtt C struct layout, so field names/paths match `mqtt_client.h` / `mqtt5_client.h` directly.
### `mqtt_config` (used with `init`)
Mirrors `esp_mqtt_client_config_t`'s nesting:
```json
{
"mqtt_config": {
"broker": { "address": { "uri": "mqtt://192.168.1.1:1883" } },
"credentials": { "client_id": "my-client" },
"session": { "keepalive": 30, "disable_clean_session": false, "protocol_ver": 3 },
"network": { "disable_auto_reconnect": true }
}
}
```
| Path | Type | Description |
|------|------|-------------|
| `broker.address.uri` | string | Broker URI (e.g. `mqtt://192.168.1.1:1883`) |
| `credentials.client_id` | string | Client identifier |
| `session.keepalive` | int | Keepalive interval (seconds) |
| `session.disable_clean_session` | bool | `true` = persistent session (clean start = false) |
| `session.protocol_ver` | int | Raw `esp_mqtt_protocol_ver_t` ordinal: `0`=UNDEFINED, `1`=MQTT 3.1, `2`=MQTT 3.1.1, `3`=MQTT 5.0 |
| `network.disable_auto_reconnect` | bool | Disable MQTT client automatic reconnect |
### `connect_property` (MQTT5 connect properties)
Already flat in C, so the JSON object is flat too:
| Key | Type | Description |
|-----|------|-------------|
| `session_expiry_interval` | int | Session expiry (seconds) |
| `receive_maximum` | int | Receive maximum |
| `topic_alias_maximum` | int | Topic alias maximum |
| `maximum_packet_size` | int | Maximum packet size |
| `will_delay_interval` | int | Will delay interval (seconds) |
### `publish_property` (MQTT5 publish properties)
| Key | Type | Description |
|-----|------|-------------|
| `message_expiry_interval` | int | Message expiry (seconds) |
| `payload_format_indicator` | bool | `true` = UTF-8 encoded payload |
| `topic_alias` | int | Topic alias |
| `content_type` | string | Content type |
| `response_topic` | string | Response topic |
### `subscribe_property` (MQTT5 subscribe properties)
| Key | Type | Description |
|-----|------|-------------|
| `subscribe_id` | int | Subscription identifier |
| `no_local_flag` | bool | No local flag |
| `retain_as_published_flag` | bool | Retain as published flag |
| `retain_handle` | int | Retain handling option (0/1/2) |
| `is_share_subscribe` | bool | Shared subscription flag |
| `share_name` | string | Shared subscription group name |
### `disconnect_property` (MQTT5 disconnect properties)
| Key | Type | Description |
|-----|------|-------------|
| `session_expiry_interval` | int | Session expiry override on disconnect |
| `disconnect_reason` | int | Disconnect reason code |
## Conformance mapping
Each pytest case should document the MQTT specification section it validates where practical.
test/apps/mqtt_conformance/main/CMakeLists.txt
+7 -2
View File
@@ -1,3 +1,8 @@
idf_component_register(SRCS "mqtt_conformance.c" "mqtt_conformance_console.c"
idf_component_register(SRCS "mqtt_conformance.cpp" "mqtt_conformance_console.cpp"
INCLUDE_DIRS "."
REQUIRES mqtt nvs_flash console esp_netif)
REQUIRES mqtt nvs_flash console esp_netif nlohmann-json)
set_source_files_properties(mqtt_conformance.cpp PROPERTIES
COMPILE_OPTIONS "-Wno-deprecated-declarations")
target_compile_definitions(${COMPONENT_LIB} PRIVATE JSON_NO_IO)
test/apps/mqtt_conformance/main/idf_component.yml
+3 -2
View File
@@ -2,5 +2,6 @@ dependencies:
protocol_examples_common:
path: ${IDF_PATH}/examples/common_components/protocol_examples_common
mqtt:
version: "*"
override_path: "../../../.."
version: '*'
override_path: ../../../..
mittelab/nlohmann-json: ^3.11.3
test/apps/mqtt_conformance/main/mqtt_conformance.c
-125
View File
@@ -1,125 +0,0 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include "esp_event.h"
#include "esp_random.h"
#include "esp_system.h"
#include "esp_log.h"
#include "mqtt_client.h"
#if CONFIG_MQTT_PROTOCOL_5
#include "mqtt5_client.h"
#endif
#include "mqtt_conformance.h"
static const char *TAG = "mqtt_conformance";
#define CLIENT_ID_SIZE 20
static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
{
esp_mqtt_event_handle_t event = event_data;
switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
break;
case MQTT_EVENT_DISCONNECTED:
#if CONFIG_MQTT_PROTOCOL_5
if (event->error_handle) {
ESP_LOGW(TAG, "DISCONNECT_REASON=%d", event->reason_code);
}
#endif
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
break;
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
if (event->data_len > 0 && event->data) {
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED data_len=%d return_code=0x%02x",
event->data_len, (unsigned int)(uint8_t)event->data[0]);
}
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
if (event->data_len > 0 && event->data) {
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED data_len=%d reason_code=0x%02x",
event->data_len, (unsigned int)(uint8_t)event->data[0]);
}
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 topic=%.*s qos=%d len=%d offset=%d total=%d", event->topic_len, event->topic,
event->qos, event->data_len, event->current_data_offset, event->total_data_len);
ESP_LOGI(TAG, "MQTT_EVENT_DATA_PAYLOAD %.*s", event->data_len, event->data ? event->data : "");
if (event->current_data_offset + event->data_len == event->total_data_len) {
ESP_LOGI(TAG, "MQTT_EVENT_DATA_COMPLETE msg_id=%d total=%d", event->msg_id, event->total_data_len);
}
break;
case MQTT_EVENT_ERROR:
ESP_LOGE(TAG, "MQTT_EVENT_ERROR");
if (event->error_handle) {
ESP_LOGE(TAG, "error_type=%" PRId32 " connect_return_code=%" PRId32,
(int32_t)event->error_handle->error_type,
(int32_t)event->error_handle->connect_return_code);
}
if (event->data_len > 0 && event->data) {
ESP_LOGE(TAG, "MQTT_EVENT_ERROR data_len=%d data=%.*s",
event->data_len, event->data_len, event->data);
}
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
}
void conformance_register_event_handlers(command_context_t *ctx)
{
esp_mqtt_client_register_event(ctx->mqtt_client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL);
}
void conformance_unregister_event_handlers(command_context_t *ctx)
{
esp_mqtt_client_unregister_event(ctx->mqtt_client, ESP_EVENT_ANY_ID, mqtt_event_handler);
}
void conformance_set_broker_uri(command_context_t *ctx, const char *uri)
{
esp_mqtt_client_config_t config = {0};
config.broker.address.uri = uri;
esp_mqtt_set_config(ctx->mqtt_client, &config);
}
void conformance_configure_client(command_context_t *ctx)
{
static char client_id[CLIENT_ID_SIZE];
snprintf(client_id, sizeof(client_id), "esp-%08" PRIx32, esp_random());
esp_mqtt_client_config_t config = {0};
config.credentials.client_id = client_id;
ESP_LOGI(TAG, "Client configured, client_id=%s (broker URI set via set_uri command)", client_id);
esp_mqtt_set_config(ctx->mqtt_client, &config);
}
test/apps/mqtt_conformance/main/mqtt_conformance.cpp
+496
View File
@@ -0,0 +1,496 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#include <concepts>
#include <cstdint>
#include <deque>
#include <expected>
#include <functional>
#include <memory>
#include <ranges>
#include <string>
#include <string_view>
#include <type_traits>
#include <vector>
#include <nlohmann/json_impl.hpp>
#include "esp_log.h"
#include "mbedtls/base64.h"
#include "mqtt_client.h"
#include "mqtt5_client.h"
#include "mqtt_conformance.hpp"
static constexpr auto TAG = "mqtt_conformance";
namespace
{
template <typename... Ts> struct overloaded : Ts... {
using Ts::operator()...;
};
template <typename... Ts> overloaded(Ts...) -> overloaded<Ts...>;
void log_user_properties(const char *prefix,
mqtt5_user_property_handle_t user_property)
{
uint8_t count = esp_mqtt5_client_get_user_property_count(user_property);
if (count == 0) {
return;
}
auto items = std::make_unique<esp_mqtt5_user_property_item_t[]>(count);
if (esp_mqtt5_client_get_user_property(user_property, items.get(), &count) !=
ESP_OK) {
return;
}
for (uint8_t i = 0; i < count; i++) {
ESP_LOGI(TAG, "%s key=%s val=%s", prefix, items[i].key ? items[i].key : "",
items[i].value ? items[i].value : "");
free(const_cast<char *>(items[i].key));
free(const_cast<char *>(items[i].value));
}
esp_mqtt5_client_delete_user_property(user_property);
}
} // namespace
void conformance_mqtt_event_handler(void *, esp_event_base_t, int32_t event_id,
void *event_data)
{
auto *event = static_cast<esp_mqtt_event_handle_t>(event_data);
switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED session_present=%d",
event->session_present);
if (event->property) {
log_user_properties("CONNACK_USER_PROPERTY",
event->property->user_property);
}
break;
case MQTT_EVENT_DISCONNECTED:
if (event->property) {
ESP_LOGW(TAG, "DISCONNECT_REASON=%d", event->reason_code);
}
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
break;
case MQTT_EVENT_SUBSCRIBED:
if (event->error_handle &&
event->error_handle->error_type == MQTT_ERROR_TYPE_SUBSCRIBE_FAILED) {
ESP_LOGW(TAG, "MQTT_EVENT_SUBSCRIBE_FAILED msg_id=%d", event->msg_id);
} else {
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
}
if (event->data_len > 0 && event->data) {
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED data_len=%d return_code=0x%02x",
event->data_len,
static_cast<unsigned>(static_cast<uint8_t>(event->data[0])));
}
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
if (event->data_len > 0 && event->data) {
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED data_len=%d reason_code=0x%02x",
event->data_len,
static_cast<unsigned>(static_cast<uint8_t>(event->data[0])));
}
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA:
if (event->topic && event->topic_len > 0) {
ESP_LOGI(TAG, "MQTT_EVENT_DATA topic=%.*s qos=%d len=%d offset=%d total=%d",
event->topic_len, event->topic, event->qos, event->data_len,
event->current_data_offset, event->total_data_len);
} else {
ESP_LOGI(TAG, "MQTT_EVENT_DATA qos=%d len=%d offset=%d total=%d",
event->qos, event->data_len,
event->current_data_offset, event->total_data_len);
}
if (event->data && event->data_len > 0) {
ESP_LOGI(TAG, "MQTT_EVENT_DATA_PAYLOAD %.*s", event->data_len, event->data);
}
if (event->property) {
if (event->property->payload_format_indicator) {
ESP_LOGI(TAG, "DATA_PROP payload_format_indicator=1");
}
if (event->property->content_type &&
event->property->content_type_len > 0) {
ESP_LOGI(TAG, "DATA_PROP content_type=%.*s",
event->property->content_type_len,
event->property->content_type);
}
if (event->property->response_topic &&
event->property->response_topic_len > 0) {
ESP_LOGI(TAG, "DATA_PROP response_topic=%.*s",
event->property->response_topic_len,
event->property->response_topic);
}
if (event->property->correlation_data &&
event->property->correlation_data_len > 0) {
ESP_LOGI(TAG, "DATA_PROP correlation_data=%.*s",
event->property->correlation_data_len,
event->property->correlation_data);
}
if (event->property->subscribe_id > 0) {
ESP_LOGI(TAG, "DATA_PROP subscribe_id=%d",
event->property->subscribe_id);
}
log_user_properties("DATA_PROP user_property",
event->property->user_property);
}
if (event->current_data_offset + event->data_len == event->total_data_len) {
ESP_LOGI(TAG, "MQTT_EVENT_DATA_COMPLETE msg_id=%d total=%d",
event->msg_id, event->total_data_len);
}
break;
case MQTT_EVENT_ERROR:
if (event->error_handle) {
if (event->error_handle->error_type ==
MQTT_ERROR_TYPE_CONNECTION_REFUSED) {
ESP_LOGW(TAG, "MQTT_EVENT_ERROR CONNECTION_REFUSED code=%d",
event->error_handle->connect_return_code);
} else if (event->error_handle->error_type ==
MQTT_ERROR_TYPE_TCP_TRANSPORT) {
ESP_LOGE(TAG, "MQTT_EVENT_ERROR TCP_TRANSPORT");
} else {
ESP_LOGE(TAG, "MQTT_EVENT_ERROR type=%d",
event->error_handle->error_type);
}
} else {
ESP_LOGE(TAG, "MQTT_EVENT_ERROR (no error_handle)");
}
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
}
namespace
{
using json = nlohmann::json;
using namespace nlohmann::literals;
template <typename T>
struct field_table_entry {
using Config = T;
std::string_view field_name;
void (*set)(owned_config<T> &, const json &);
};
template <typename E>
concept field_table = std::same_as<E, field_table_entry<typename E::Config>>;
template <typename P>
concept member_object_pointer = std::is_member_object_pointer_v<P>;
template <member_object_pointer auto member>
void set_scalar(auto &config, const json &value)
{
using member_type = std::remove_reference_t<decltype((*config.data).*member)>;
if constexpr(std::is_same_v<member_type, const char *>) {
const std::string &stored = config.data_storage.emplace_back(value.get<std::string>());
(*config.data).*member = stored.c_str();
} else {
(*config.data).*member = value.get<member_type>();
}
}
constexpr field_table_entry<esp_mqtt5_connection_property_config_t> connection_property_fields_table[] = {
{"session_expiry_interval", &set_scalar<&esp_mqtt5_connection_property_config_t::session_expiry_interval>},
{"receive_maximum", &set_scalar<&esp_mqtt5_connection_property_config_t::receive_maximum>},
{"topic_alias_maximum", &set_scalar<&esp_mqtt5_connection_property_config_t::topic_alias_maximum>},
{"maximum_packet_size", &set_scalar<&esp_mqtt5_connection_property_config_t::maximum_packet_size>},
{"will_delay_interval", &set_scalar<&esp_mqtt5_connection_property_config_t::will_delay_interval>},
};
constexpr field_table_entry<esp_mqtt5_publish_property_config_t> publish_property_fields_table[] = {
{"message_expiry_interval", &set_scalar<&esp_mqtt5_publish_property_config_t::message_expiry_interval>},
{"payload_format_indicator", &set_scalar<&esp_mqtt5_publish_property_config_t::payload_format_indicator>},
{"topic_alias", &set_scalar<&esp_mqtt5_publish_property_config_t::topic_alias>},
{"content_type", &set_scalar<&esp_mqtt5_publish_property_config_t::content_type>},
{"response_topic", &set_scalar<&esp_mqtt5_publish_property_config_t::response_topic>},
};
constexpr field_table_entry<esp_mqtt5_subscribe_property_config_t> subscribe_property_fields_table[] = {
{"subscribe_id", &set_scalar<&esp_mqtt5_subscribe_property_config_t::subscribe_id>},
{"no_local_flag", &set_scalar<&esp_mqtt5_subscribe_property_config_t::no_local_flag>},
{"retain_as_published_flag", &set_scalar<&esp_mqtt5_subscribe_property_config_t::retain_as_published_flag>},
{"retain_handle", &set_scalar<&esp_mqtt5_subscribe_property_config_t::retain_handle>},
{"is_share_subscribe", &set_scalar<&esp_mqtt5_subscribe_property_config_t::is_share_subscribe>},
{"share_name", &set_scalar<&esp_mqtt5_subscribe_property_config_t::share_name>},
};
constexpr field_table_entry<esp_mqtt5_disconnect_property_config_t> disconnect_property_fields_table[] = {
{"session_expiry_interval", &set_scalar<&esp_mqtt5_disconnect_property_config_t::session_expiry_interval>},
{"disconnect_reason", &set_scalar<&esp_mqtt5_disconnect_property_config_t::disconnect_reason>},
};
[[nodiscard]] auto find_field(std::ranges::forward_range auto const &table, std::string_view name)
-> const std::ranges::range_value_t<decltype(table)> *
requires field_table<std::ranges::range_value_t<decltype(table)>>
{
for (const auto &entry : table) {
if (entry.field_name == name) {
return &entry;
}
}
return nullptr;
}
void build_from_table(auto &cfg, std::ranges::forward_range auto const &table, const json &node)
requires field_table<std::ranges::range_value_t<decltype(table)>>
{
for (const auto &[field_name, value] : node.items()) {
const auto *field = find_field(table, field_name);
if (!field) {
ESP_LOGW(TAG, "json config: unknown field name '%s', skipping", field_name.c_str());
continue;
}
try {
field->set(cfg, value);
} catch (const json::exception &e) {
ESP_LOGE(TAG, "json config: field '%s' has an unexpected JSON type (%s)", field_name.c_str(), e.what());
throw;
}
}
}
void apply_if_present(const json &node, const json::json_pointer &ptr, const std::function<void(const json &)> &assign)
{
if (!node.contains(ptr)) {
return;
}
try {
assign(node.at(ptr));
} catch (const json::exception &e) {
ESP_LOGE(TAG, "json config: '%s' has an unexpected JSON type (%s)", ptr.to_string().c_str(), e.what());
throw;
}
}
[[nodiscard]] parsed_config build_mqtt_config(const json &node)
{
unique_mqtt_config cfg{std::make_unique<esp_mqtt_client_config_t>(), {}};
apply_if_present(node, "/broker/address/uri"_json_pointer, [&](const json & v) {
const std::string &uri = cfg.data_storage.emplace_back(v.get<std::string>());
cfg.data->broker.address.uri = uri.c_str();
});
apply_if_present(node, "/credentials/client_id"_json_pointer, [&](const json & v) {
const std::string &client_id = cfg.data_storage.emplace_back(v.get<std::string>());
cfg.data->credentials.client_id = client_id.c_str();
});
apply_if_present(node, "/session/keepalive"_json_pointer, [&](const json & v) {
cfg.data->session.keepalive = v.get<int>();
});
apply_if_present(node, "/session/disable_clean_session"_json_pointer, [&](const json & v) {
cfg.data->session.disable_clean_session = v.get<bool>();
});
apply_if_present(node, "/session/protocol_ver"_json_pointer, [&](const json & v) {
cfg.data->session.protocol_ver = v.get<esp_mqtt_protocol_ver_t>();
});
apply_if_present(node, "/network/disable_auto_reconnect"_json_pointer, [&](const json & v) {
cfg.data->network.disable_auto_reconnect = v.get<bool>();
});
return cfg;
}
[[nodiscard]] parsed_config build_connect_property(const json &node)
{
unique_connection_property_config cfg{std::make_unique<esp_mqtt5_connection_property_config_t>(), {}};
build_from_table(cfg, connection_property_fields_table, node);
return cfg;
}
[[nodiscard]] parsed_config build_publish_property(const json &node)
{
unique_publish_property_config cfg{std::make_unique<esp_mqtt5_publish_property_config_t>(), {}};
build_from_table(cfg, publish_property_fields_table, node);
return cfg;
}
[[nodiscard]] parsed_config build_subscribe_property(const json &node)
{
unique_subscribe_property_config cfg{std::make_unique<esp_mqtt5_subscribe_property_config_t>(), {}};
build_from_table(cfg, subscribe_property_fields_table, node);
return cfg;
}
[[nodiscard]] parsed_config build_disconnect_property(const json &node)
{
unique_disconnect_property_config cfg{std::make_unique<esp_mqtt5_disconnect_property_config_t>(), {}};
build_from_table(cfg, disconnect_property_fields_table, node);
return cfg;
}
struct config_builder {
std::string_view config_name;
parsed_config(*build)(const json &);
};
// The blob's single top-level key names the category directly, so this
// table doubles as both the category detector and the dispatch table.
constexpr config_builder config_entries[] = {
{"mqtt_config", &build_mqtt_config},
{"connect_property", &build_connect_property},
{"publish_property", &build_publish_property},
{"subscribe_property", &build_subscribe_property},
{"disconnect_property", &build_disconnect_property},
};
[[nodiscard]] auto decode_base64(std::string_view b64)
-> std::expected<std::vector<uint8_t>, esp_err_t>
{
size_t out_len = 0;
auto *base64_data = reinterpret_cast<const unsigned char *>(b64.data());
mbedtls_base64_decode(nullptr, 0, &out_len, base64_data, b64.size());
if (out_len == 0) {
ESP_LOGE(TAG, "json config: base64 length probe failed");
return std::unexpected(ESP_ERR_INVALID_ARG);
}
std::vector<uint8_t> buf(out_len);
size_t written = 0;
int rc = mbedtls_base64_decode(buf.data(), buf.size(), &written, base64_data, b64.size());
if (rc != 0) {
ESP_LOGE(TAG, "json config: base64 decode failed (%d)", rc);
return std::unexpected(ESP_ERR_INVALID_ARG);
}
buf.resize(written);
return buf;
}
[[nodiscard]] auto parse_json(std::vector<uint8_t> decoded) -> std::expected<json, esp_err_t>
{
try {
json doc = json::parse(decoded.begin(), decoded.end());
if (!doc.is_object()) {
ESP_LOGE(TAG, "json config: expected a JSON object");
return std::unexpected(ESP_ERR_INVALID_ARG);
}
return doc;
} catch (const json::parse_error &e) {
ESP_LOGE(TAG, "json config: %s", e.what());
return std::unexpected(ESP_ERR_INVALID_ARG);
}
}
[[nodiscard]] std::expected<parsed_config, esp_err_t> build_command_config(const json &command)
{
const auto builder = std::ranges::find_if(config_entries, [&](const config_builder & category) {
return command.contains(category.config_name);
});
if (builder == std::ranges::end(config_entries)) {
ESP_LOGE(TAG, "json config: no recognized category present");
return std::unexpected(ESP_ERR_INVALID_ARG);
}
const json &node = command[builder->config_name];
if (!node.is_object()) {
ESP_LOGE(TAG, "json config: category '%.*s' must be a JSON object",
static_cast<int>(builder->config_name.size()), builder->config_name.data());
return std::unexpected(ESP_ERR_INVALID_ARG);
}
try {
return builder->build(node);
} catch (const json::exception &) {
return std::unexpected(ESP_ERR_INVALID_ARG);
}
}
} // namespace
[[nodiscard]] std::expected<parsed_config, esp_err_t>
conformance_parse_json_config(std::string_view command_b64)
{
return decode_base64(command_b64)
.and_then(parse_json)
.and_then(build_command_config);
}
void conformance_apply_config(command_context_t *ctx, parsed_config config)
{
std::visit(
overloaded{
[&](unique_mqtt_config & client_config)
{
if (client_config.data->credentials.client_id) {
ESP_LOGI(TAG, "client_id=%s", client_config.data->credentials.client_id);
}
esp_mqtt_set_config(ctx->mqtt_client, client_config.data.get());
},
[&](unique_connection_property_config & property_config)
{
esp_mqtt5_client_set_connect_property(ctx->mqtt_client, property_config.data.get());
},
[&](unique_publish_property_config & property_config)
{
esp_mqtt5_client_set_publish_property(ctx->mqtt_client, property_config.data.get());
},
[&](unique_subscribe_property_config & property_config)
{
// esp-mqtt5 keeps a pointer to *subscribe_property* itself (not a
// copy) for use at the next subscribe, so share_name's backing
// storage has to live in ctx, not in this (about to be destroyed)
// parsed config.
ctx->subscribe_property = *property_config.data;
ctx->subscribe_share_name = property_config.data_storage.empty() ? std::string{} : std::move(property_config.data_storage.front());
ctx->subscribe_property.share_name = ctx->subscribe_share_name.c_str();
esp_mqtt5_client_set_subscribe_property(ctx->mqtt_client,
&ctx->subscribe_property);
},
[&](unique_disconnect_property_config & c)
{
esp_mqtt5_client_set_disconnect_property(ctx->mqtt_client, c.data.get());
},
},
config);
}
test/apps/mqtt_conformance/main/mqtt_conformance.h
-42
View File
@@ -1,42 +0,0 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#pragma once
#include "mqtt_client.h"
struct arg_int;
struct arg_str;
struct arg_end;
typedef struct {
esp_mqtt_client_handle_t mqtt_client;
} command_context_t;
typedef struct {
struct arg_str *uri;
struct arg_end *end;
} set_uri_args_t;
typedef struct {
struct arg_str *topic;
struct arg_int *qos;
struct arg_end *end;
} subscribe_args_t;
typedef struct {
struct arg_str *topic;
struct arg_str *pattern;
struct arg_int *pattern_repetitions;
struct arg_int *qos;
struct arg_int *retain;
struct arg_int *enqueue;
struct arg_end *end;
} publish_args_t;
void conformance_register_event_handlers(command_context_t *ctx);
void conformance_unregister_event_handlers(command_context_t *ctx);
void conformance_configure_client(command_context_t *ctx);
void conformance_set_broker_uri(command_context_t *ctx, const char *uri);
test/apps/mqtt_conformance/main/mqtt_conformance.hpp
+90
View File
@@ -0,0 +1,90 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#pragma once
#include <deque>
#include <expected>
#include <memory>
#include <string>
#include <string_view>
#include <variant>
#include "esp_err.h"
#include "mqtt5_client.h"
#include "mqtt_client.h"
struct arg_int;
struct arg_str;
struct arg_end;
struct command_context_t {
esp_mqtt_client_handle_t mqtt_client = nullptr;
esp_mqtt5_subscribe_property_config_t subscribe_property = {};
// Backing storage for subscribe_property.share_name: esp-mqtt5 retains a
// pointer to *subscribe_property* itself (not a copy) between `config`
// and the next subscribe, so this string must outlive the config call.
std::string subscribe_share_name;
};
struct subscribe_args_t {
struct arg_str *topic;
struct arg_int *qos;
struct arg_end *end;
};
struct unsubscribe_args_t {
struct arg_str *topic;
struct arg_end *end;
};
struct publish_args_t {
struct arg_str *topic;
struct arg_str *pattern;
struct arg_int *pattern_repetitions;
struct arg_int *qos;
struct arg_int *retain;
struct arg_int *enqueue;
struct arg_end *end;
};
struct json_config_args_t {
struct arg_str *b64;
struct arg_end *end;
};
/**
* Owns a heap-allocated C config struct plus the backing storage for any of
* its const char* fields that were populated from JSON. Strings live in a
* deque. pointers handed out via .c_str() stay valid for the lifetime of this
* object — no manual malloc/free bookkeeping needed. Plain aggregate: no
* invariant to protect, so callers use the members directly.
*/
template <typename T>
struct owned_config {
std::unique_ptr<T> data;
std::deque<std::string> data_storage;
};
using unique_mqtt_config = owned_config<esp_mqtt_client_config_t>;
using unique_connection_property_config = owned_config<esp_mqtt5_connection_property_config_t>;
using unique_publish_property_config = owned_config<esp_mqtt5_publish_property_config_t>;
using unique_subscribe_property_config = owned_config<esp_mqtt5_subscribe_property_config_t>;
using unique_disconnect_property_config = owned_config<esp_mqtt5_disconnect_property_config_t>;
/**
* Config parsed from a single JSON blob.
*/
using parsed_config = std::variant<unique_mqtt_config, unique_connection_property_config, unique_publish_property_config, unique_subscribe_property_config, unique_disconnect_property_config>;
/** MQTT event handler — registered directly by the console with esp_mqtt_client_register_event. */
void conformance_mqtt_event_handler(void *handler_args, esp_event_base_t base,
int32_t event_id, void *event_data);
/** JSON layer: decode base64 JSON blob → owned config. */
[[nodiscard]] std::expected<parsed_config, esp_err_t> conformance_parse_json_config(std::string_view b64);
/** Apply a parsed config to an initialised client. */
void conformance_apply_config(command_context_t *ctx, parsed_config cfg);
test/apps/mqtt_conformance/main/mqtt_conformance_console.c
-297
View File
@@ -1,297 +0,0 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#include <inttypes.h>
#include <stdio.h>
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include "esp_system.h"
#include "mqtt_client.h"
#include "nvs_flash.h"
#include "esp_event.h"
#include "esp_netif.h"
#include "protocol_examples_common.h"
#include "esp_console.h"
#include "argtable3/argtable3.h"
#include "esp_log.h"
#include "mqtt_conformance.h"
static const char *TAG = "mqtt_conformance";
static command_context_t command_context;
static set_uri_args_t set_uri_args;
static subscribe_args_t subscribe_args;
static publish_args_t publish_args;
#define RETURN_ON_PARSE_ERROR(args) do { \
int nerrors = arg_parse(argc, argv, (void **) &(args)); \
if (nerrors != 0) { \
arg_print_errors(stderr, (args).end, argv[0]); \
return 1; \
}} while(0)
static int require_client(void)
{
if (!command_context.mqtt_client) {
ESP_LOGE(TAG, "MQTT client not initialized, call init first");
return 1;
}
return 0;
}
static int do_init(int argc, char **argv)
{
if (command_context.mqtt_client) {
ESP_LOGW(TAG, "MQTT client already initialized");
return 0;
}
const esp_mqtt_client_config_t mqtt_cfg = {
.broker.address.uri = "mqtt://127.0.0.1:1234",
.network.disable_auto_reconnect = true,
#if CONFIG_MQTT_PROTOCOL_5
.session.protocol_ver = MQTT_PROTOCOL_V_5,
#endif
};
command_context.mqtt_client = esp_mqtt_client_init(&mqtt_cfg);
if (!command_context.mqtt_client) {
ESP_LOGE(TAG, "Failed to initialize client");
return 1;
}
conformance_configure_client(&command_context);
conformance_register_event_handlers(&command_context);
ESP_LOGI(TAG, "Mqtt client initialized");
return 0;
}
static int do_set_uri(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(set_uri_args);
if (require_client() != 0) {
return 1;
}
conformance_set_broker_uri(&command_context, set_uri_args.uri->sval[0]);
ESP_LOGI(TAG, "Broker URI updated to %s", set_uri_args.uri->sval[0]);
return 0;
}
static int do_start(int argc, char **argv)
{
if (require_client() != 0) {
return 1;
}
if (esp_mqtt_client_start(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to start mqtt client task");
return 1;
}
ESP_LOGI(TAG, "Mqtt client started");
return 0;
}
static int do_stop(int argc, char **argv)
{
if (require_client() != 0) {
return 1;
}
if (esp_mqtt_client_stop(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to stop mqtt client task");
return 1;
}
ESP_LOGI(TAG, "Mqtt client stopped");
return 0;
}
static int do_destroy(int argc, char **argv)
{
if (!command_context.mqtt_client) {
return 0;
}
conformance_unregister_event_handlers(&command_context);
esp_mqtt_client_destroy(command_context.mqtt_client);
command_context.mqtt_client = NULL;
ESP_LOGI(TAG, "mqtt client for tests destroyed");
return 0;
}
static int do_subscribe(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(subscribe_args);
if (require_client() != 0) {
return 1;
}
int msg_id = esp_mqtt_client_subscribe(command_context.mqtt_client, subscribe_args.topic->sval[0],
subscribe_args.qos->ival[0]);
if (msg_id < 0) {
ESP_LOGE(TAG, "Subscribe failed, msg_id=%d", msg_id);
return 1;
}
ESP_LOGI(TAG, "Subscribe requested, msg_id=%d", msg_id);
return 0;
}
static int do_publish(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(publish_args);
if (require_client() != 0) {
return 1;
}
const char *pattern = publish_args.pattern->sval[0];
int repetitions = publish_args.pattern_repetitions->ival[0];
size_t pattern_len = strlen(pattern);
size_t payload_len = pattern_len * (size_t)repetitions;
char *payload = NULL;
if (repetitions < 0) {
ESP_LOGE(TAG, "Invalid pattern repetitions");
return 1;
}
if (payload_len > 0) {
payload = malloc(payload_len);
if (!payload) {
ESP_LOGE(TAG, "Failed to allocate payload");
return 1;
}
for (int i = 0; i < repetitions; i++) {
memcpy(payload + (size_t)i * pattern_len, pattern, pattern_len);
}
}
int msg_id;
if (publish_args.enqueue->ival[0]) {
msg_id = esp_mqtt_client_enqueue(command_context.mqtt_client, publish_args.topic->sval[0], payload, payload_len,
publish_args.qos->ival[0], publish_args.retain->ival[0], true);
} else {
msg_id = esp_mqtt_client_publish(command_context.mqtt_client, publish_args.topic->sval[0], payload, payload_len,
publish_args.qos->ival[0], publish_args.retain->ival[0]);
}
free(payload);
if (msg_id < 0) {
ESP_LOGE(TAG, "Publish failed, msg_id=%d", msg_id);
return 1;
}
ESP_LOGI(TAG, "Publish requested, msg_id=%d", msg_id);
return 0;
}
static void register_common_commands(void)
{
const esp_console_cmd_t init = {
.command = "init",
.help = "Initialize mqtt client",
.hint = NULL,
.func = &do_init,
};
const esp_console_cmd_t set_uri = {
.command = "set_uri",
.help = "Set broker URI",
.hint = NULL,
.func = &do_set_uri,
.argtable = &set_uri_args,
};
const esp_console_cmd_t start = {
.command = "start",
.help = "Start mqtt client",
.hint = NULL,
.func = &do_start,
};
const esp_console_cmd_t stop = {
.command = "stop",
.help = "Stop mqtt client",
.hint = NULL,
.func = &do_stop,
};
const esp_console_cmd_t destroy = {
.command = "destroy",
.help = "Destroy mqtt client",
.hint = NULL,
.func = &do_destroy,
};
ESP_ERROR_CHECK(esp_console_cmd_register(&init));
ESP_ERROR_CHECK(esp_console_cmd_register(&set_uri));
ESP_ERROR_CHECK(esp_console_cmd_register(&start));
ESP_ERROR_CHECK(esp_console_cmd_register(&stop));
ESP_ERROR_CHECK(esp_console_cmd_register(&destroy));
}
static void register_pubsub_commands(void)
{
set_uri_args.uri = arg_str1(NULL, NULL, "<uri>", "Broker URI");
set_uri_args.end = arg_end(1);
subscribe_args.topic = arg_str1(NULL, NULL, "<topic>", "Subscribe topic");
subscribe_args.qos = arg_int1(NULL, NULL, "<qos>", "Subscribe qos");
subscribe_args.end = arg_end(1);
publish_args.topic = arg_str1(NULL, NULL, "<topic>", "Publish topic");
publish_args.pattern = arg_str1(NULL, NULL, "<pattern>", "Payload pattern");
publish_args.pattern_repetitions = arg_int1(NULL, NULL, "<pattern repetitions>", "Number of pattern repetitions");
publish_args.qos = arg_int1(NULL, NULL, "<qos>", "Publish qos");
publish_args.retain = arg_int1(NULL, NULL, "<retain>", "Publish retain flag");
publish_args.enqueue = arg_int1(NULL, NULL, "<enqueue>", "0=publish,1=enqueue");
publish_args.end = arg_end(1);
const esp_console_cmd_t subscribe = {
.command = "subscribe",
.help = "Subscribe to a topic",
.hint = NULL,
.func = &do_subscribe,
.argtable = &subscribe_args,
};
const esp_console_cmd_t publish = {
.command = "publish",
.help = "Publish a message",
.hint = NULL,
.func = &do_publish,
.argtable = &publish_args,
};
ESP_ERROR_CHECK(esp_console_cmd_register(&subscribe));
ESP_ERROR_CHECK(esp_console_cmd_register(&publish));
}
void app_main(void)
{
static const size_t max_line = 256;
ESP_LOGI(TAG, "[APP] Free memory: %" PRIu32 " 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("wifi", ESP_LOG_ERROR);
esp_log_level_set("mqtt_client", ESP_LOG_INFO);
esp_log_level_set("outbox", ESP_LOG_INFO);
ESP_ERROR_CHECK(nvs_flash_init());
ESP_ERROR_CHECK(esp_netif_init());
ESP_ERROR_CHECK(esp_event_loop_create_default());
ESP_ERROR_CHECK(example_connect());
esp_console_repl_t *repl = NULL;
esp_console_repl_config_t repl_config = ESP_CONSOLE_REPL_CONFIG_DEFAULT();
repl_config.prompt = "mqtt>";
repl_config.max_cmdline_length = max_line;
esp_console_register_help_command();
register_pubsub_commands();
register_common_commands();
esp_console_dev_uart_config_t hw_config = ESP_CONSOLE_DEV_UART_CONFIG_DEFAULT();
ESP_ERROR_CHECK(esp_console_new_repl_uart(&hw_config, &repl_config, &repl));
ESP_ERROR_CHECK(esp_console_start_repl(repl));
}
test/apps/mqtt_conformance/main/mqtt_conformance_console.cpp
+431
View File
@@ -0,0 +1,431 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#include <cinttypes>
#include <cstddef>
#include <cstdio>
#include <cstring>
#include <memory>
#include <variant>
#include "esp_system.h"
#include "mqtt_client.h"
#include "nvs_flash.h"
#include "esp_event.h"
#include "esp_netif.h"
#include "protocol_examples_common.h"
#include "esp_console.h"
#include "argtable3/argtable3.h"
#include "esp_log.h"
#include "mqtt_conformance.hpp"
namespace
{
constexpr auto TAG = "mqtt_conformance";
command_context_t command_context;
subscribe_args_t subscribe_args;
unsubscribe_args_t unsubscribe_args;
publish_args_t publish_args;
json_config_args_t init_args;
json_config_args_t config_args;
#define RETURN_ON_PARSE_ERROR(args) do { \
int nerrors = arg_parse(argc, argv, (void **) &(args)); \
if (nerrors != 0) { \
arg_print_errors(stderr, (args).end, argv[0]); \
return 1; \
}} while(0)
[[nodiscard]] bool client_available()
{
if (!command_context.mqtt_client) {
ESP_LOGE(TAG, "MQTT client not initialized, call init first");
return false;
}
return true;
}
int do_init(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(init_args);
if (command_context.mqtt_client) {
ESP_LOGW(TAG, "MQTT client already initialized");
return 0;
}
auto mqtt_config = []() -> unique_mqtt_config {
auto parsed_config = conformance_parse_json_config(init_args.b64->sval[0]);
if (!parsed_config) {
ESP_LOGE(TAG, "Invalid JSON config");
return {};
}
auto *mqtt_config = std::get_if<unique_mqtt_config>(&*parsed_config);
if (!mqtt_config)
{
return {};
}
return std::move(*mqtt_config);
}();
if (!mqtt_config.data) {
ESP_LOGE(TAG, "init requires mqtt client config");
return 1;
}
command_context.mqtt_client = esp_mqtt_client_init(mqtt_config.data.get());
if (!command_context.mqtt_client) {
ESP_LOGE(TAG, "Failed to initialize client");
return 1;
}
esp_mqtt_client_register_event(command_context.mqtt_client,
static_cast<esp_mqtt_event_id_t>(ESP_EVENT_ANY_ID),
conformance_mqtt_event_handler, nullptr);
ESP_LOGI(TAG, "Mqtt client initialized");
return 0;
}
int do_config(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(config_args);
if (!client_available()) {
return 1;
}
auto cfg = conformance_parse_json_config(config_args.b64->sval[0]);
if (!cfg) {
ESP_LOGE(TAG, "Failed to parse json config");
return 1;
}
conformance_apply_config(&command_context, std::move(*cfg));
return 0;
}
int do_start(int argc, char **argv)
{
if (!client_available()) {
return 1;
}
if (esp_mqtt_client_start(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to start mqtt client task");
return 1;
}
ESP_LOGI(TAG, "Mqtt client started");
return 0;
}
int do_stop(int argc, char **argv)
{
if (!client_available()) {
return 1;
}
if (esp_mqtt_client_stop(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to stop mqtt client task");
return 1;
}
ESP_LOGI(TAG, "Mqtt client stopped");
return 0;
}
int do_disconnect(int argc, char **argv)
{
(void)argc;
(void)argv;
if (!client_available()) {
return 1;
}
if (esp_mqtt_client_disconnect(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to request disconnection");
return 1;
}
ESP_LOGI(TAG, "Mqtt client disconnected");
return 0;
}
int do_reconnect(int argc, char **argv)
{
(void)argc;
(void)argv;
if (!client_available()) {
return 1;
}
if (esp_mqtt_client_reconnect(command_context.mqtt_client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to request reconnection");
return 1;
}
ESP_LOGI(TAG, "Mqtt client will reconnect");
return 0;
}
int do_destroy(int argc, char **argv)
{
if (!command_context.mqtt_client) {
return 0;
}
esp_mqtt_client_unregister_event(command_context.mqtt_client,
static_cast<esp_mqtt_event_id_t>(ESP_EVENT_ANY_ID),
conformance_mqtt_event_handler);
esp_mqtt_client_destroy(command_context.mqtt_client);
command_context.mqtt_client = nullptr;
command_context.subscribe_property = {};
command_context.subscribe_share_name.clear();
ESP_LOGI(TAG, "mqtt client for tests destroyed");
return 0;
}
int do_subscribe(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(subscribe_args);
if (!client_available()) {
return 1;
}
int msg_id = esp_mqtt_client_subscribe(command_context.mqtt_client, subscribe_args.topic->sval[0],
subscribe_args.qos->ival[0]);
if (msg_id < 0) {
ESP_LOGE(TAG, "Subscribe failed, msg_id=%d", msg_id);
return 1;
}
ESP_LOGI(TAG, "Subscribe requested, msg_id=%d", msg_id);
return 0;
}
int do_unsubscribe(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(unsubscribe_args);
if (!client_available()) {
return 1;
}
int msg_id = esp_mqtt_client_unsubscribe(command_context.mqtt_client, unsubscribe_args.topic->sval[0]);
if (msg_id < 0) {
ESP_LOGE(TAG, "Unsubscribe failed, msg_id=%d", msg_id);
return 1;
}
ESP_LOGI(TAG, "Unsubscribe requested, msg_id=%d", msg_id);
return 0;
}
int do_publish(int argc, char **argv)
{
RETURN_ON_PARSE_ERROR(publish_args);
if (!client_available()) {
return 1;
}
const char *pattern = publish_args.pattern->sval[0];
int repetitions = publish_args.pattern_repetitions->ival[0];
if (repetitions < 0) {
ESP_LOGE(TAG, "Invalid pattern repetitions");
return 1;
}
size_t pattern_len = std::strlen(pattern);
size_t payload_len = pattern_len * static_cast<size_t>(repetitions);
std::unique_ptr<char[]> payload;
if (payload_len > 0) {
payload = std::make_unique<char[]>(payload_len);
for (int i = 0; i < repetitions; i++) {
std::memcpy(payload.get() + static_cast<size_t>(i) * pattern_len, pattern, pattern_len);
}
}
int msg_id;
if (publish_args.enqueue->ival[0]) {
msg_id = esp_mqtt_client_enqueue(command_context.mqtt_client, publish_args.topic->sval[0],
payload.get(), payload_len,
publish_args.qos->ival[0], publish_args.retain->ival[0], true);
} else {
msg_id = esp_mqtt_client_publish(command_context.mqtt_client, publish_args.topic->sval[0],
payload.get(), payload_len,
publish_args.qos->ival[0], publish_args.retain->ival[0]);
}
if (msg_id < 0) {
ESP_LOGE(TAG, "Publish failed, msg_id=%d", msg_id);
return 1;
}
ESP_LOGI(TAG, "Publish requested, msg_id=%d", msg_id);
return 0;
}
void register_commands()
{
init_args.b64 = arg_str1(nullptr, nullptr, "<base64_json>", "JSON config blob (base64-encoded)");
init_args.end = arg_end(1);
const esp_console_cmd_t init = {
.command = "init",
.help = "Initialize mqtt client with base64-encoded JSON config",
.hint = nullptr,
.func = &do_init,
.argtable = &init_args,
.func_w_context = nullptr,
.context = nullptr,
};
config_args.b64 = arg_str1(nullptr, nullptr, "<base64_json>", "JSON config blob (base64-encoded)");
config_args.end = arg_end(1);
const esp_console_cmd_t config_cmd = {
.command = "config",
.help = "Apply base64-encoded JSON config to initialized client",
.hint = nullptr,
.func = &do_config,
.argtable = &config_args,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t start = {
.command = "start",
.help = "Start mqtt client",
.hint = nullptr,
.func = &do_start,
.argtable = nullptr,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t stop = {
.command = "stop",
.help = "Stop mqtt client",
.hint = nullptr,
.func = &do_stop,
.argtable = nullptr,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t disconnect = {
.command = "disconnect",
.help = "Disconnect mqtt client",
.hint = nullptr,
.func = &do_disconnect,
.argtable = nullptr,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t reconnect = {
.command = "reconnect",
.help = "Reconnect mqtt client",
.hint = nullptr,
.func = &do_reconnect,
.argtable = nullptr,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t destroy = {
.command = "destroy",
.help = "Destroy mqtt client",
.hint = nullptr,
.func = &do_destroy,
.argtable = nullptr,
.func_w_context = nullptr,
.context = nullptr,
};
subscribe_args.topic = arg_str1(nullptr, nullptr, "<topic>", "Subscribe topic");
subscribe_args.qos = arg_int1(nullptr, nullptr, "<qos>", "Subscribe qos");
subscribe_args.end = arg_end(1);
unsubscribe_args.topic = arg_str1(nullptr, nullptr, "<topic>", "Unsubscribe topic");
unsubscribe_args.end = arg_end(1);
publish_args.topic = arg_str1(nullptr, nullptr, "<topic>", "Publish topic");
publish_args.pattern = arg_str1(nullptr, nullptr, "<pattern>", "Payload pattern");
publish_args.pattern_repetitions = arg_int1(nullptr, nullptr, "<pattern repetitions>", "Number of pattern repetitions");
publish_args.qos = arg_int1(nullptr, nullptr, "<qos>", "Publish qos");
publish_args.retain = arg_int1(nullptr, nullptr, "<retain>", "Publish retain flag");
publish_args.enqueue = arg_int1(nullptr, nullptr, "<enqueue>", "0=publish,1=enqueue");
publish_args.end = arg_end(1);
const esp_console_cmd_t subscribe = {
.command = "subscribe",
.help = "Subscribe to a topic",
.hint = nullptr,
.func = &do_subscribe,
.argtable = &subscribe_args,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t unsubscribe = {
.command = "unsubscribe",
.help = "Unsubscribe from a topic",
.hint = nullptr,
.func = &do_unsubscribe,
.argtable = &unsubscribe_args,
.func_w_context = nullptr,
.context = nullptr,
};
const esp_console_cmd_t publish = {
.command = "publish",
.help = "Publish a message",
.hint = nullptr,
.func = &do_publish,
.argtable = &publish_args,
.func_w_context = nullptr,
.context = nullptr,
};
ESP_ERROR_CHECK(esp_console_cmd_register(&init));
ESP_ERROR_CHECK(esp_console_cmd_register(&config_cmd));
ESP_ERROR_CHECK(esp_console_cmd_register(&start));
ESP_ERROR_CHECK(esp_console_cmd_register(&stop));
ESP_ERROR_CHECK(esp_console_cmd_register(&disconnect));
ESP_ERROR_CHECK(esp_console_cmd_register(&reconnect));
ESP_ERROR_CHECK(esp_console_cmd_register(&destroy));
ESP_ERROR_CHECK(esp_console_cmd_register(&subscribe));
ESP_ERROR_CHECK(esp_console_cmd_register(&unsubscribe));
ESP_ERROR_CHECK(esp_console_cmd_register(&publish));
}
} // namespace
extern "C" void app_main(void)
{
constexpr size_t max_line = 512;
ESP_LOGI(TAG, "[APP] Free memory: %" PRIu32 " 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_ERROR_CHECK(nvs_flash_init());
ESP_ERROR_CHECK(esp_netif_init());
ESP_ERROR_CHECK(esp_event_loop_create_default());
ESP_ERROR_CHECK(example_connect());
register_commands();
esp_console_register_help_command();
esp_console_repl_t *repl = nullptr;
esp_console_repl_config_t repl_config = ESP_CONSOLE_REPL_CONFIG_DEFAULT();
repl_config.prompt = "mqtt>";
repl_config.max_cmdline_length = max_line;
repl_config.task_stack_size = 12288;
esp_console_dev_uart_config_t hw_config = ESP_CONSOLE_DEV_UART_CONFIG_DEFAULT();
ESP_ERROR_CHECK(esp_console_new_repl_uart(&hw_config, &repl_config, &repl));
ESP_ERROR_CHECK(esp_console_start_repl(repl));
}
test/apps/mqtt_conformance/pytest_mqtt_conformance.py
+40 -2
View File
@@ -2,7 +2,9 @@
# SPDX-License-Identifier: Unlicense OR CC0-1.0
from __future__ import annotations
import base64
import contextlib
import json
import os
import random
import re
@@ -49,6 +51,43 @@ def build_topic() -> str:
return f"test/conformance/{suffix}"
# Raw esp_mqtt_protocol_ver_t ordinals (see mqtt_client.h): the wire format
# mirrors the real enum value rather than a human-friendly "3/4/5" mapping.
MQTT_PROTOCOL_V_3_1_1 = 2
MQTT_PROTOCOL_V_5 = 3
def esp_mqtt_config(
*,
uri: str,
client_id: str | None = None,
protocol_ver: int | None = None,
disable_auto_reconnect: bool = True,
) -> str:
"""Encode an ``mqtt_config`` blob (base64 JSON) for the DUT `init <b64>` command.
The JSON shape mirrors esp_mqtt_client_config_t's real field nesting
(broker.address.uri, credentials.client_id, session.*, network.*).
A random ``client_id`` is injected unless the caller supplies one.
"""
client_id = client_id or "esp-" + "".join(random.choices(string.digits + "abcdef", k=8))
session: dict[str, object] = {}
if protocol_ver is not None:
session["protocol_ver"] = protocol_ver
mqtt_config: dict[str, object] = {
"broker": {"address": {"uri": uri}},
"credentials": {"client_id": client_id},
"network": {"disable_auto_reconnect": disable_auto_reconnect},
}
if session:
mqtt_config["session"] = session
return base64.b64encode(json.dumps({"mqtt_config": mqtt_config}).encode()).decode()
def require_paho_testing_checked_out() -> None:
"""Hard requirement: fail the test if the paho.mqtt.testing submodule is not available."""
if not PAHO_SPEC_FILE.exists():
@@ -140,8 +179,7 @@ def broker_uri(broker: _BrokerHandle) -> str:
def mqtt_client(dut: Dut, broker_uri: str):
require_paho_testing_checked_out()
dut.expect(re.compile(rb"mqtt>"), timeout=DUT_READY_TIMEOUT)
dut.write("init")
dut.write(f"set_uri {broker_uri}")
dut.write(f"init {esp_mqtt_config(protocol_ver=MQTT_PROTOCOL_V_3_1_1, uri=broker_uri)}")
yield dut
dut.write("destroy")
test/apps/mqtt_conformance/sdkconfig.ci.default
+1
View File
@@ -1,4 +1,5 @@
CONFIG_MQTT_PROTOCOL_5=y
CONFIG_COMPILER_CXX_EXCEPTIONS=y
CONFIG_EXAMPLE_CONNECT_ETHERNET=y
CONFIG_EXAMPLE_CONNECT_WIFI=n
CONFIG_ESP_NETIF_RECEIVE_REPORT_ERRORS=y