[nextion] Add tft_upload_http_buffer_size_tx option for TFT upload redirects (#21435)

CODEOWNERS

@@ -88,6 +88,7 @@ esphome/components/bl0940/* @dan-s-github @tobias-
 esphome/components/bl0942/* @dbuezas @dwmw2
 esphome/components/ble_client/* @buxtronix @clydebarrow
 esphome/components/ble_device_base/* @Bl00d-B0b
+esphome/components/ble_midi/* @bogdanr
 esphome/components/ble_nus/* @tomaszduda23
 esphome/components/bluetooth_connection/* @bdraco @jesserockz
 esphome/components/bluetooth_proxy/* @bdraco @jesserockz

esphome/components/ble_midi/__init__.py

@@ -0,0 +1,213 @@
+from esphome import automation
+import esphome.codegen as cg
+from esphome.components import ble_client
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_CHANNEL,
+    CONF_DATA,
+    CONF_ON_CONNECT,
+    CONF_ON_DISCONNECT,
+    CONF_ON_MESSAGE,
+    CONF_PAYLOAD,
+    CONF_VALUE,
+)
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@bogdanr"]
+DEPENDENCIES = ["ble_client"]
+DOMAIN = "ble_midi"
+MULTI_CONF = True
+
+CONF_CONTROL_NUMBER = "control_number"
+CONF_MAX_SYSEX_SIZE = "max_sysex_size"
+CONF_NOTE = "note"
+CONF_ON_CONTROL_CHANGE = "on_control_change"
+CONF_ON_NOTE_OFF = "on_note_off"
+CONF_ON_NOTE_ON = "on_note_on"
+CONF_ON_PITCH_BEND = "on_pitch_bend"
+CONF_ON_PROGRAM_CHANGE = "on_program_change"
+CONF_ON_SYSEX = "on_sysex"
+CONF_PAIR = "pair"
+CONF_PROGRAM = "program"
+CONF_SYSEX_BUFFER_ID = "sysex_buffer_id"
+CONF_VELOCITY = "velocity"
+
+ble_midi_ns = cg.esphome_ns.namespace("ble_midi")
+BLEMidi = ble_midi_ns.class_("BLEMidi", cg.Component, ble_client.BLEClientNode)
+# The message bytes as a view: data.size() and data[i] work, and nothing is copied
+ByteSpan = cg.std_span.template(cg.uint8.operator("const"))
+
+# MIDI data bytes are 7-bit and channels are 4-bit, matching what goes on the wire and what the triggers report.
+# Channel 0 is the channel that MIDI hardware usually labels "1".
+midi_data_byte = cv.int_range(min=0, max=127)
+midi_channel = cv.int_range(min=0, max=15)
+
+_NOTE_ARGS = [
+    (cg.uint8, CONF_NOTE),
+    (cg.uint8, CONF_VELOCITY),
+    (cg.uint8, CONF_CHANNEL),
+]
+_BYTES_ARGS = [(ByteSpan, CONF_DATA)]
+# data points into a notification or a reused buffer, so these automations cannot wait
+_BYTES_TRIGGERS = (CONF_ON_SYSEX, CONF_ON_MESSAGE)
+_CALLBACK_AUTOMATIONS = (
+    automation.CallbackAutomation(CONF_ON_CONNECT, "add_on_connect_callback"),
+    automation.CallbackAutomation(CONF_ON_DISCONNECT, "add_on_disconnect_callback"),
+    automation.CallbackAutomation(
+        CONF_ON_NOTE_ON, "add_on_note_on_callback", _NOTE_ARGS
+    ),
+    automation.CallbackAutomation(
+        CONF_ON_NOTE_OFF, "add_on_note_off_callback", _NOTE_ARGS
+    ),
+    automation.CallbackAutomation(
+        CONF_ON_CONTROL_CHANGE,
+        "add_on_control_change_callback",
+        [
+            (cg.uint8, CONF_CONTROL_NUMBER),
+            (cg.uint8, CONF_VALUE),
+            (cg.uint8, CONF_CHANNEL),
+        ],
+    ),
+    automation.CallbackAutomation(
+        CONF_ON_PROGRAM_CHANGE,
+        "add_on_program_change_callback",
+        [(cg.uint8, CONF_PROGRAM), (cg.uint8, CONF_CHANNEL)],
+    ),
+    automation.CallbackAutomation(
+        CONF_ON_PITCH_BEND,
+        "add_on_pitch_bend_callback",
+        [(cg.int16, CONF_VALUE), (cg.uint8, CONF_CHANNEL)],
+    ),
+    automation.CallbackAutomation(CONF_ON_SYSEX, "add_on_sysex_callback", _BYTES_ARGS),
+    automation.CallbackAutomation(
+        CONF_ON_MESSAGE, "add_on_message_callback", _BYTES_ARGS
+    ),
+)
+
+CONFIG_SCHEMA = (
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(BLEMidi),
+            cv.GenerateID(CONF_SYSEX_BUFFER_ID): cv.declare_id(cg.uint8),
+            cv.Optional(CONF_MAX_SYSEX_SIZE, default=256): cv.int_range(
+                min=8, max=2048
+            ),
+            cv.Optional(CONF_PAIR, default=False): cv.boolean,
+            **{
+                cv.Optional(entry.conf_key): (
+                    automation.validate_synchronous_automation(
+                        f"ble_midi {entry.conf_key}", "data"
+                    )
+                    if entry.conf_key in _BYTES_TRIGGERS
+                    else automation.validate_automation({})
+                )
+                for entry in _CALLBACK_AUTOMATIONS
+            },
+        }
+    )
+    .extend(cv.COMPONENT_SCHEMA)
+    .extend(ble_client.BLE_CLIENT_SCHEMA)
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    size = config[CONF_MAX_SYSEX_SIZE]
+    # SysEx messages are reassembled in a static array, so nothing is allocated at run time
+    buf = config[CONF_SYSEX_BUFFER_ID].id
+    cg.add_global(cg.RawStatement(f"static uint8_t {buf}[{size}];"))
+    var = await ble_client.new_ble_node_component(
+        config, cg.RawExpression(buf), size, config[CONF_PAIR]
+    )
+    await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
+
+
+def _action_schema(keys: dict) -> cv.Schema:
+    return cv.Schema(
+        {
+            cv.GenerateID(): cv.use_id(BLEMidi),
+            **keys,
+            cv.Optional(CONF_CHANNEL, default=0): cv.templatable(midi_channel),
+        }
+    )
+
+
+_NOTE_ACTION_SCHEMA = _action_schema(
+    {
+        cv.Required(CONF_NOTE): cv.templatable(midi_data_byte),
+        cv.Optional(CONF_VELOCITY, default=64): cv.templatable(midi_data_byte),
+    }
+)
+
+# Wider than the wire values so the C++ range check sees a lambda's result before it is narrowed
+_CHANNEL = (CONF_CHANNEL, cg.int32)
+_NOTE_CALL_ARGS = ((CONF_NOTE, cg.int32), (CONF_VELOCITY, cg.int32), _CHANNEL)
+
+automation.register_apply_action(
+    "ble_midi.send_note_on",
+    _NOTE_ACTION_SCHEMA,
+    automation.ApplyCall("send_note_on({}, {}, {})", _NOTE_CALL_ARGS),
+)
+
+automation.register_apply_action(
+    "ble_midi.send_note_off",
+    _NOTE_ACTION_SCHEMA,
+    automation.ApplyCall("send_note_off({}, {}, {})", _NOTE_CALL_ARGS),
+)
+
+automation.register_apply_action(
+    "ble_midi.send_control_change",
+    _action_schema(
+        {
+            cv.Required(CONF_CONTROL_NUMBER): cv.templatable(midi_data_byte),
+            cv.Required(CONF_VALUE): cv.templatable(midi_data_byte),
+        }
+    ),
+    automation.ApplyCall(
+        "send_control_change({}, {}, {})",
+        ((CONF_CONTROL_NUMBER, cg.int32), (CONF_VALUE, cg.int32), _CHANNEL),
+    ),
+)
+
+automation.register_apply_action(
+    "ble_midi.send_program_change",
+    _action_schema({cv.Required(CONF_PROGRAM): cv.templatable(midi_data_byte)}),
+    automation.ApplyCall(
+        "send_program_change({}, {})", ((CONF_PROGRAM, cg.int32), _CHANNEL)
+    ),
+)
+
+automation.register_apply_action(
+    "ble_midi.send_pitch_bend",
+    _action_schema(
+        {
+            cv.Required(CONF_VALUE): cv.templatable(cv.int_range(min=-8192, max=8191)),
+        }
+    ),
+    automation.ApplyCall("send_pitch_bend({}, {})", ((CONF_VALUE, cg.int32), _CHANNEL)),
+)
+
+
+def _bytes_action_schema(key: str) -> cv.Schema:
+    return cv.Schema(
+        {
+            cv.GenerateID(): cv.use_id(BLEMidi),
+            cv.Required(key): cv.templatable(cv.ensure_list(cv.hex_uint8_t)),
+        }
+    )
+
+
+automation.register_apply_action(
+    "ble_midi.send_sysex",
+    _bytes_action_schema(CONF_PAYLOAD),
+    automation.ApplyBytes(CONF_PAYLOAD, "send_sysex({})"),
+)
+
+automation.register_apply_action(
+    "ble_midi.send_raw",
+    _bytes_action_schema(CONF_DATA),
+    automation.ApplyBytes(CONF_DATA, "send_raw({})"),
+)
+
+automation.register_apply_condition(
+    "ble_midi.connected", cv.Schema({cv.GenerateID(): cv.use_id(BLEMidi)}), "is_ready()"
+)

esphome/components/ble_midi/ble_midi.cpp

@@ -0,0 +1,310 @@
+#include "ble_midi.h"
+
+#ifdef USE_ESP32
+
+#include "esphome/core/log.h"
+
+#include <algorithm>
+#include <cstring>
+
+namespace esphome::ble_midi {
+
+ESPHOME_LOG_TAG(TAG, "ble_midi");
+
+// Standard BLE-MIDI service and characteristic, as defined by the BLE-MIDI
+// specification.
+static constexpr const char *SERVICE_UUID = "03B80E5A-EDE8-4B33-A751-6CE34EC4C700";
+static constexpr const char *CHARACTERISTIC_UUID = "7772E5DB-3868-4112-A1A9-F2669D106BF3";
+
+// MIDI status bytes for the messages this component can transmit.
+static constexpr uint8_t STATUS_NOTE_OFF = 0x80;
+static constexpr uint8_t STATUS_NOTE_ON = 0x90;
+static constexpr uint8_t STATUS_CONTROL_CHANGE = 0xB0;
+static constexpr uint8_t STATUS_PROGRAM_CHANGE = 0xC0;
+static constexpr uint8_t STATUS_PITCH_BEND = 0xE0;
+
+// Number of received bytes included in verbose packet logs.
+static constexpr size_t LOG_MAX_BYTES = 32;
+
+void BLEMidi::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "BLE MIDI:\n"
+                "  MAC address: %s\n"
+                "  Maximum SysEx size: %u\n"
+                "  Pair on connect: %s",
+                this->parent()->address_str(), this->parser_.sysex_capacity(), YESNO(this->pair_));
+}
+
+void BLEMidi::gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if, esp_ble_gattc_cb_param_t *param) {
+  switch (event) {
+    case ESP_GATTC_OPEN_EVT: {
+      if (param->open.status != ESP_GATT_OK)
+        break;
+      // A warning left by an earlier session no longer applies
+      this->status_clear_warning();
+      this->parser_.reset();
+      if (this->pair_) {
+        esp_err_t err = this->parent()->pair();
+        if (err != ESP_OK) {
+          ESP_LOGW(TAG, "Pairing failed, error=%d", err);
+        }
+      }
+      break;
+    }
+    case ESP_GATTC_CFG_MTU_EVT: {
+      if (param->cfg_mtu.status == ESP_GATT_OK)
+        this->mtu_ = param->cfg_mtu.mtu;
+      break;
+    }
+    case ESP_GATTC_DISCONNECT_EVT:
+    case ESP_GATTC_CLOSE_EVT: {
+      this->parser_.reset();
+      this->characteristic_handle_ = 0;
+      this->cccd_handle_ = 0;
+      this->mtu_ = 23;
+      // Both events can arrive for the same disconnect, so the automation runs
+      // only for the one that follows an established MIDI connection.
+      const bool was_ready = this->notify_registered_;
+      this->notify_registered_ = false;
+      this->node_state = espbt::ClientState::IDLE;
+      if (was_ready)
+        this->disconnect_callback_.call();
+      break;
+    }
+    case ESP_GATTC_SEARCH_CMPL_EVT: {
+      if (param->search_cmpl.status != ESP_GATT_OK) {
+        ESP_LOGW(TAG, "Service discovery failed, status=%d", param->search_cmpl.status);
+        this->fail_session_();
+        break;
+      }
+      auto *characteristic = this->parent()->get_characteristic(espbt::ESPBTUUID::from_raw(SERVICE_UUID),
+                                                                espbt::ESPBTUUID::from_raw(CHARACTERISTIC_UUID));
+      // Reconnecting would not help a device without them, so the link is left as it is
+      if (characteristic == nullptr) {
+        this->status_set_warning(LOG_STR("No BLE MIDI characteristic found on this device"));
+        break;
+      }
+      auto *cccd = characteristic->get_descriptor(ESP_GATT_UUID_CHAR_CLIENT_CONFIG);
+      if (cccd == nullptr) {
+        this->status_set_warning(LOG_STR("BLE MIDI characteristic has no notification descriptor"));
+        break;
+      }
+      this->characteristic_handle_ = characteristic->handle;
+      this->cccd_handle_ = cccd->handle;
+      this->write_without_response_ = (characteristic->properties & ESP_GATT_CHAR_PROP_BIT_WRITE_NR) != 0;
+      esp_err_t err = this->parent()->register_for_notify(characteristic->handle);
+      if (err != ESP_OK) {
+        ESP_LOGW(TAG, "register_for_notify failed, error=%d", err);
+        this->fail_session_();
+      }
+      break;
+    }
+    case ESP_GATTC_REG_FOR_NOTIFY_EVT: {
+      if (param->reg_for_notify.handle != this->characteristic_handle_)
+        break;
+      if (param->reg_for_notify.status != ESP_GATT_OK) {
+        ESP_LOGW(TAG, "Registering for notifications failed, status=%d", param->reg_for_notify.status);
+        this->fail_session_();
+      }
+      // On success the parent writes the descriptor; the device has accepted it once WRITE_DESCR_EVT reports OK
+      break;
+    }
+    case ESP_GATTC_WRITE_DESCR_EVT: {
+      if (this->cccd_handle_ == 0 || param->write.handle != this->cccd_handle_ || this->notify_registered_)
+        break;
+      if (param->write.status != ESP_GATT_OK) {
+        ESP_LOGW(TAG, "Device refused to enable notifications, status=%d", param->write.status);
+        this->fail_session_();
+        break;
+      }
+      this->notify_registered_ = true;
+      this->node_state = espbt::ClientState::ESTABLISHED;
+      ESP_LOGI(TAG, "MIDI notifications enabled");
+      // The device can only be sent to once notifications are enabled, so this
+      // is the point where on_connect is useful.
+      this->connect_callback_.call();
+      break;
+    }
+    case ESP_GATTC_WRITE_CHAR_EVT: {
+      // A write with response reports its outcome here; one without response reports nothing
+      if (param->write.handle == this->characteristic_handle_ && param->write.status != ESP_GATT_OK) {
+        ESP_LOGW(TAG, "Write rejected, status=%d", param->write.status);
+      }
+      break;
+    }
+    case ESP_GATTC_NOTIFY_EVT: {
+      if (param->notify.handle != this->characteristic_handle_ || param->notify.value_len == 0)
+        break;
+      // format_hex_to() truncates to the buffer size, so long packets are logged
+      // only in part.
+      char hex[format_hex_size(LOG_MAX_BYTES)];
+      ESP_LOGV(TAG, "Received %u bytes: %s", param->notify.value_len,
+               format_hex_to(hex, param->notify.value, param->notify.value_len));
+      this->parser_.feed(param->notify.value, param->notify.value_len, BLEMidi::dispatch_message, this);
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+void BLEMidi::fail_session_() {
+  this->status_set_warning();
+  // ble_client reconnects, which runs the setup again
+  this->parent()->disconnect();
+}
+
+void BLEMidi::handle_message_(const MidiMessage &message) {
+  ESP_LOGV(
+      TAG, "Decoded %s",
+      LOG_STR_ARG(MessageTypeNames::get_log_str(static_cast<uint8_t>(message.type), MessageTypeNames::LAST_INDEX)));
+  // Only decoded messages reach on_message
+  if (message.type == MessageType::SYSEX_INCOMPLETE) {
+    ESP_LOGW(TAG, "SysEx of %u bytes dropped: cut short, or longer than max_sysex_size (%u)", message.length,
+             this->parser_.sysex_capacity());
+    return;
+  }
+  if (message.type == MessageType::INCOMPLETE) {
+    ESP_LOGW(TAG, "MIDI message 0x%02X dropped: cut short by another status byte", message.status());
+    return;
+  }
+  if (message.type == MessageType::UNKNOWN) {
+    ESP_LOGW(TAG, "Undecodable MIDI byte 0x%02X", message.status());
+    return;
+  }
+  this->message_callback_.call(ByteSpan(message.data, message.length));
+
+  switch (message.type) {
+    case MessageType::NOTE_ON:
+      this->note_on_callback_.call(message.data1(), message.data2(), message.channel());
+      break;
+    case MessageType::NOTE_OFF:
+      this->note_off_callback_.call(message.data1(), message.data2(), message.channel());
+      break;
+    case MessageType::CONTROL_CHANGE:
+      this->control_change_callback_.call(message.data1(), message.data2(), message.channel());
+      break;
+    case MessageType::PROGRAM_CHANGE:
+      this->program_change_callback_.call(message.data1(), message.channel());
+      break;
+    case MessageType::PITCH_BEND:
+      this->pitch_bend_callback_.call(message.pitch_bend(), message.channel());
+      break;
+    case MessageType::SYSEX:
+      // The payload without the 0xF0/0xF7 wrappers; a complete SysEx always has both
+      this->sysex_callback_.call(ByteSpan(message.data + 1, message.length - 2));
+      break;
+    default:
+      // Polyphonic and channel aftertouch, system common and real-time messages
+      // are only reported through on_message.
+      break;
+  }
+}
+
+bool BLEMidi::write_packet_(uint8_t *packet, size_t midi_length) {
+  if (!this->is_ready()) {
+    ESP_LOGW(TAG, "Cannot send: not connected");
+    return false;
+  }
+  // Every BLE-MIDI packet has to fit one ATT packet
+  const size_t packet_length = midi_length + 2;
+  const size_t limit = std::min<size_t>(MAX_TX_MIDI_BYTES + 2, this->mtu_ - 3);
+  if (packet_length > limit) {
+    ESP_LOGW(TAG, "Cannot send: message of %u bytes exceeds the %u byte limit", static_cast<unsigned>(midi_length),
+             static_cast<unsigned>(limit - 2));
+    return false;
+  }
+  packet[0] = PACKET_HEADER;
+  packet[1] = PACKET_TIMESTAMP;
+  esp_err_t err = esp_ble_gattc_write_char(
+      this->parent()->get_gattc_if(), this->parent()->get_conn_id(), this->characteristic_handle_, packet_length,
+      packet, this->write_without_response_ ? ESP_GATT_WRITE_TYPE_NO_RSP : ESP_GATT_WRITE_TYPE_RSP,
+      ESP_GATT_AUTH_REQ_NONE);
+  if (err != ESP_OK) {
+    ESP_LOGW(TAG, "Write failed, error=%d", err);
+    return false;
+  }
+  char hex[format_hex_size(LOG_MAX_BYTES)];
+  ESP_LOGV(TAG, "Sent %u bytes: %s", static_cast<unsigned>(packet_length), format_hex_to(hex, packet, packet_length));
+  return true;
+}
+
+bool BLEMidi::write_midi_(const uint8_t *midi_bytes, size_t length) {
+  if (length == 0 || length > MAX_TX_MIDI_BYTES) {
+    ESP_LOGW(TAG, "Cannot send: message of %u bytes, the limit is %u", static_cast<unsigned>(length),
+             static_cast<unsigned>(MAX_TX_MIDI_BYTES));
+    return false;
+  }
+  uint8_t packet[MAX_TX_MIDI_BYTES + 2];
+  std::memcpy(packet + 2, midi_bytes, length);
+  return this->write_packet_(packet, length);
+}
+
+void BLEMidi::send_channel_(uint8_t status, int32_t channel, int32_t data1, int32_t data2, size_t length) {
+  // Validation covers constants only; a lambda can return any value, and masking it would send a different message.
+  // The values arrive unnarrowed so a negative or wrapped one is caught here (the unsigned compare covers negatives).
+  if (static_cast<uint32_t>(channel) > 0x0F || static_cast<uint32_t>(data1) > 0x7F ||
+      static_cast<uint32_t>(data2) > 0x7F) {
+    ESP_LOGW(TAG, "Cannot send: channel %" PRId32 " or value %" PRId32 "/%" PRId32 " out of range", channel, data1,
+             data2);
+    return;
+  }
+  const uint8_t bytes[3] = {static_cast<uint8_t>(status | channel), static_cast<uint8_t>(data1),
+                            static_cast<uint8_t>(data2)};
+  this->write_midi_(bytes, length);
+}
+
+void BLEMidi::send_note_on(int32_t note, int32_t velocity, int32_t channel) {
+  this->send_channel_(STATUS_NOTE_ON, channel, note, velocity, 3);
+}
+
+void BLEMidi::send_note_off(int32_t note, int32_t velocity, int32_t channel) {
+  this->send_channel_(STATUS_NOTE_OFF, channel, note, velocity, 3);
+}
+
+void BLEMidi::send_control_change(int32_t control_number, int32_t value, int32_t channel) {
+  this->send_channel_(STATUS_CONTROL_CHANGE, channel, control_number, value, 3);
+}
+
+void BLEMidi::send_program_change(int32_t program, int32_t channel) {
+  this->send_channel_(STATUS_PROGRAM_CHANGE, channel, program, 0, 2);
+}
+
+void BLEMidi::send_pitch_bend(int32_t value, int32_t channel) {
+  if (value < -8192 || value > 8191) {
+    ESP_LOGW(TAG, "Cannot send: pitch bend %" PRId32 " out of range", value);
+    return;
+  }
+  const uint16_t unsigned_value = static_cast<uint16_t>(value + 8192);
+  this->send_channel_(STATUS_PITCH_BEND, channel, unsigned_value & 0x7F, unsigned_value >> 7, 3);
+}
+
+void BLEMidi::send_sysex(const uint8_t *payload, size_t length) {
+  uint8_t packet[MAX_TX_MIDI_BYTES + 2];
+  const size_t framed = frame_sysex(payload, length, packet + 2, MAX_TX_MIDI_BYTES);
+  if (framed == 0) {
+    ESP_LOGW(TAG, "Cannot send: SysEx of %u bytes is empty or exceeds the %u byte limit", static_cast<unsigned>(length),
+             static_cast<unsigned>(MAX_TX_MIDI_BYTES));
+    return;
+  }
+  this->write_packet_(packet, framed);
+}
+
+void BLEMidi::send_raw(const uint8_t *bytes, size_t length) {
+  // The packet carries one timestamp, in front of the first byte, so the payload has to be a single complete
+  // message with its status byte first; a SysEx goes through send_sysex
+  bool one_message = length > 0 && (bytes[0] & 0x80) != 0 && bytes[0] != STATUS_SYSEX_START &&
+                     bytes[0] != STATUS_SYSEX_END && length == 1u + MidiParser::expected_data_bytes(bytes[0]);
+  for (size_t i = 1; one_message && i < length; i++) {
+    one_message = (bytes[i] & 0x80) == 0;
+  }
+  if (!one_message) {
+    ESP_LOGW(TAG, "Cannot send: raw payload must be one message, a status byte followed by data bytes");
+    return;
+  }
+  this->write_midi_(bytes, length);
+}
+
+}  // namespace esphome::ble_midi
+
+#endif  // USE_ESP32

esphome/components/ble_midi/ble_midi.h

@@ -0,0 +1,138 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include "esphome/core/defines.h"
+#include "esphome/core/helpers.h"
+
+#ifdef USE_ESP32
+
+#include "esphome/components/ble_client/ble_client.h"
+#include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
+
+#include "midi_parser.h"
+
+#include <esp_gattc_api.h>
+
+#include <cstdint>
+#include <span>
+
+namespace esphome::ble_midi {
+
+namespace espbt = esphome::esp32_ble_tracker;
+
+/// The bytes of a received message, valid for the duration of the automation.
+using ByteSpan = std::span<const uint8_t>;
+
+/// Longest MIDI message that can be transmitted in a single BLE-MIDI packet, header and timestamp excluded.
+static constexpr size_t MAX_TX_MIDI_BYTES = 128;
+
+/// Client for devices implementing the standard BLE-MIDI service, e.g. wireless
+/// MIDI keyboards and pad controllers. Incoming messages are exposed as
+/// automation triggers; outgoing messages are sent with the ble_midi.send_*
+/// actions.
+///
+/// Specification: "Specification for MIDI over Bluetooth Low Energy
+/// (BLE-MIDI)", MIDI Association, document M1-2017-11-01.
+class BLEMidi final : public Component, public ble_client::BLEClientNode {
+ public:
+  /// sysex_buffer holds sysex_capacity bytes; longer SysEx messages are truncated.
+  BLEMidi(uint8_t *sysex_buffer, uint16_t sysex_capacity, bool pair)
+      : parser_(sysex_buffer, sysex_capacity), pair_(pair) {}
+
+  void dump_config() override;
+  float get_setup_priority() const override { return setup_priority::AFTER_BLUETOOTH; }
+
+  void gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if,
+                           esp_ble_gattc_cb_param_t *param) override;
+
+  /// True once the MIDI characteristic has been discovered and notifications
+  /// are enabled.
+  bool is_ready() const { return this->notify_registered_; }
+
+  // Channel is 0-15, note/velocity/controller/value/program are 0-127; anything else is refused with a warning.
+  void send_note_on(int32_t note, int32_t velocity, int32_t channel);
+  void send_note_off(int32_t note, int32_t velocity, int32_t channel);
+  void send_control_change(int32_t control_number, int32_t value, int32_t channel);
+  void send_program_change(int32_t program, int32_t channel);
+  /// Pitch bend value is -8192 to 8191, centered on zero.
+  void send_pitch_bend(int32_t value, int32_t channel);
+  /// Send a System Exclusive message. The 0xF0/0xF7 wrappers are added if the
+  /// payload does not already carry them.
+  void send_sysex(const uint8_t *payload, size_t length);
+  /// Send raw MIDI bytes (status byte followed by its data bytes).
+  void send_raw(const uint8_t *bytes, size_t length);
+
+  template<typename F> void add_on_connect_callback(F &&callback) {
+    this->connect_callback_.add(std::forward<F>(callback));
+  }
+  template<typename F> void add_on_disconnect_callback(F &&callback) {
+    this->disconnect_callback_.add(std::forward<F>(callback));
+  }
+  /// (note, velocity, channel)
+  template<typename F> void add_on_note_on_callback(F &&callback) {
+    this->note_on_callback_.add(std::forward<F>(callback));
+  }
+  /// (note, velocity, channel); a Note On with velocity zero is reported here as well
+  template<typename F> void add_on_note_off_callback(F &&callback) {
+    this->note_off_callback_.add(std::forward<F>(callback));
+  }
+  /// (control_number, value, channel)
+  template<typename F> void add_on_control_change_callback(F &&callback) {
+    this->control_change_callback_.add(std::forward<F>(callback));
+  }
+  /// (program, channel)
+  template<typename F> void add_on_program_change_callback(F &&callback) {
+    this->program_change_callback_.add(std::forward<F>(callback));
+  }
+  /// (value, channel), value being -8192 to 8191
+  template<typename F> void add_on_pitch_bend_callback(F &&callback) {
+    this->pitch_bend_callback_.add(std::forward<F>(callback));
+  }
+  /// The SysEx payload without the 0xF0/0xF7 wrappers
+  template<typename F> void add_on_sysex_callback(F &&callback) {
+    this->sysex_callback_.add(std::forward<F>(callback));
+  }
+  /// The raw bytes of every decoded message
+  template<typename F> void add_on_message_callback(F &&callback) {
+    this->message_callback_.add(std::forward<F>(callback));
+  }
+
+ protected:
+  static void dispatch_message(void *ctx, const MidiMessage &message) {
+    static_cast<BLEMidi *>(ctx)->handle_message_(message);
+  }
+  void handle_message_(const MidiMessage &message);
+  /// Wrap MIDI bytes in a BLE-MIDI packet and write them to the MIDI characteristic.
+  bool write_midi_(const uint8_t *midi_bytes, size_t length);
+  /// Write a packet whose MIDI bytes already sit at packet + 2; the header and timestamp are filled in.
+  bool write_packet_(uint8_t *packet, size_t midi_length);
+  /// Channel-voice message: the status byte takes the channel; out-of-range values are refused.
+  void send_channel_(uint8_t status, int32_t channel, int32_t data1, int32_t data2, size_t length);
+  /// Set a warning and drop the link so that ble_client reconnects and the setup runs again.
+  void fail_session_();
+
+  MidiParser parser_;
+  LazyCallbackManager<void()> connect_callback_;
+  LazyCallbackManager<void()> disconnect_callback_;
+  LazyCallbackManager<void(uint8_t, uint8_t, uint8_t)> note_on_callback_;
+  LazyCallbackManager<void(uint8_t, uint8_t, uint8_t)> note_off_callback_;
+  LazyCallbackManager<void(uint8_t, uint8_t, uint8_t)> control_change_callback_;
+  LazyCallbackManager<void(uint8_t, uint8_t)> program_change_callback_;
+  LazyCallbackManager<void(int16_t, uint8_t)> pitch_bend_callback_;
+  LazyCallbackManager<void(ByteSpan)> sysex_callback_;
+  LazyCallbackManager<void(ByteSpan)> message_callback_;
+  uint16_t characteristic_handle_{0};
+  // Notification descriptor of the MIDI characteristic; the link is ready once the device accepts the write to it
+  uint16_t cccd_handle_{0};
+  // The ATT MTU the link negotiated; every packet has to fit it
+  uint16_t mtu_{23};
+  bool notify_registered_{false};
+  // Write without response has lower latency and is preferred when the
+  // characteristic supports it.
+  bool write_without_response_{false};
+  const bool pair_;
+};
+
+}  // namespace esphome::ble_midi
+
+#endif  // USE_ESP32

esphome/components/ble_midi/midi_parser.cpp

@@ -0,0 +1,198 @@
+#include "midi_parser.h"
+
+#include <cstring>
+
+namespace esphome::ble_midi {
+
+static MessageType type_for_status(uint8_t status) {
+  if (status >= 0xF8)
+    return MessageType::SYSTEM_REAL_TIME;
+  if (status >= 0xF1 && status <= 0xF6)
+    return MessageType::SYSTEM_COMMON;
+  switch (status & 0xF0) {
+    case 0x80:
+      return MessageType::NOTE_OFF;
+    case 0x90:
+      return MessageType::NOTE_ON;
+    case 0xA0:
+      return MessageType::POLY_AFTERTOUCH;
+    case 0xB0:
+      return MessageType::CONTROL_CHANGE;
+    case 0xC0:
+      return MessageType::PROGRAM_CHANGE;
+    case 0xD0:
+      return MessageType::CHANNEL_AFTERTOUCH;
+    case 0xE0:
+      return MessageType::PITCH_BEND;
+    default:
+      return MessageType::UNKNOWN;
+  }
+}
+
+// Data bytes that follow a channel-voice or system-common status byte
+uint8_t MidiParser::expected_data_bytes(uint8_t status) {
+  if (status >= 0xF0) {
+    switch (status) {
+      case 0xF1:  // MIDI Time Code Quarter Frame
+      case 0xF3:  // Song Select
+        return 1;
+      case 0xF2:  // Song Position Pointer
+        return 2;
+      default:  // 0xF4, 0xF5 undefined, 0xF6 Tune Request
+        return 0;
+    }
+  }
+  switch (status & 0xF0) {
+    case 0xC0:  // Program Change
+    case 0xD0:  // Channel Aftertouch
+      return 1;
+    default:
+      return 2;
+  }
+}
+
+void MidiParser::reset() {
+  this->running_status_ = 0;
+  this->sysex_active_ = false;
+  this->sysex_length_ = 0;
+  this->current_length_ = 0;
+}
+
+void MidiParser::sysex_push_(uint8_t byte) {
+  if (this->sysex_length_ < this->sysex_capacity_) {
+    this->sysex_buffer_[this->sysex_length_++] = byte;
+  } else {
+    this->sysex_overflow_ = true;
+  }
+}
+
+// Ends the SysEx being assembled; it is complete when 0xF7 closed it and every byte fit the buffer
+void MidiParser::end_sysex_(bool complete, MidiMessageHandler handler, void *ctx) {
+  this->sysex_active_ = false;
+  this->running_status_ = 0;
+  this->emit_(complete && !this->sysex_overflow_ ? MessageType::SYSEX : MessageType::SYSEX_INCOMPLETE,
+              this->sysex_buffer_, this->sysex_length_, handler, ctx);
+}
+
+void MidiParser::emit_current_(MidiMessageHandler handler, void *ctx) {
+  const uint8_t status = this->current_message_[0];
+  MessageType type = type_for_status(status);
+  // A Note On with velocity zero is canonically a Note Off.
+  if (type == MessageType::NOTE_ON && this->current_message_[2] == 0)
+    type = MessageType::NOTE_OFF;
+  // System Common clears running status; channel-voice messages set it.
+  this->running_status_ = status >= 0xF0 ? 0 : status;
+  const uint16_t length = this->current_length_;
+  this->current_length_ = 0;
+  this->emit_(type, this->current_message_, length, handler, ctx);
+}
+
+// Takes a data byte for the message being assembled, or starts one from the running status
+void MidiParser::take_data_(const uint8_t *byte, MidiMessageHandler handler, void *ctx) {
+  if (this->current_length_ == 0) {
+    if (this->running_status_ == 0) {
+      // Data byte without any status context.
+      this->emit_(MessageType::UNKNOWN, byte, 1, handler, ctx);
+      return;
+    }
+    this->current_message_[0] = this->running_status_;
+    this->current_length_ = 1;
+    this->current_expected_ = expected_data_bytes(this->running_status_);
+  }
+  this->current_message_[this->current_length_++] = *byte;
+  if (this->current_length_ > this->current_expected_)
+    this->emit_current_(handler, ctx);
+}
+
+void MidiParser::feed(const uint8_t *data, size_t length, MidiMessageHandler handler, void *ctx) {
+  // A BLE-MIDI packet is at least a header byte plus one timestamp or SysEx
+  // continuation byte, and the header byte always has bit 7 set. Anything else
+  // is malformed.
+  if (length < 2 || (data[0] & 0x80) == 0) {
+    for (size_t i = 0; i < length; i++)
+      this->emit_(MessageType::UNKNOWN, data + i, 1, handler, ctx);
+    return;
+  }
+
+  // A timestamp byte precedes every status byte, even a real-time byte inside another message, so a byte with
+  // bit 7 set is a status byte only right after a timestamp and a timestamp everywhere else
+  bool expect_timestamp = true;
+  for (size_t i = 1; i < length; i++) {
+    const uint8_t byte = data[i];
+    if ((byte & 0x80) == 0) {
+      // A data byte continues the SysEx, the message being assembled, or the running status
+      if (this->sysex_active_) {
+        this->sysex_push_(byte);
+      } else {
+        this->take_data_(data + i, handler, ctx);
+      }
+      expect_timestamp = true;
+      continue;
+    }
+    if (expect_timestamp) {
+      expect_timestamp = false;
+      continue;
+    }
+    expect_timestamp = true;
+    if (byte >= 0xF8) {
+      // Real-time messages are single bytes that interrupt anything, a SysEx included, without affecting it
+      this->emit_(MessageType::SYSTEM_REAL_TIME, data + i, 1, handler, ctx);
+      continue;
+    }
+    // Any other status byte cuts a message still being assembled short
+    if (this->current_length_ > 0) {
+      const uint16_t length = this->current_length_;
+      this->current_length_ = 0;
+      this->emit_(MessageType::INCOMPLETE, this->current_message_, length, handler, ctx);
+    }
+    if (byte == STATUS_SYSEX_END) {
+      if (this->sysex_active_) {
+        this->sysex_push_(STATUS_SYSEX_END);
+        this->end_sysex_(true, handler, ctx);
+      } else {
+        this->emit_(MessageType::UNKNOWN, data + i, 1, handler, ctx);
+      }
+      continue;
+    }
+    // Any status byte but 0xF7 ends a SysEx still open
+    if (this->sysex_active_)
+      this->end_sysex_(false, handler, ctx);
+    if (byte == STATUS_SYSEX_START) {
+      this->sysex_active_ = true;
+      this->sysex_length_ = 0;
+      this->sysex_overflow_ = false;
+      this->sysex_push_(STATUS_SYSEX_START);
+      continue;
+    }
+    this->current_message_[0] = byte;
+    this->current_length_ = 1;
+    this->current_expected_ = expected_data_bytes(byte);
+    // A status byte without data bytes, e.g. Tune Request, is a message on its own
+    if (this->current_expected_ == 0)
+      this->emit_current_(handler, ctx);
+  }
+}
+
+size_t frame_sysex(const uint8_t *payload, size_t length, uint8_t *out, size_t out_size) {
+  if (length == 0)
+    return 0;
+  const size_t skip_start = payload[0] == STATUS_SYSEX_START ? 1 : 0;
+  const size_t skip_end = payload[length - 1] == STATUS_SYSEX_END ? 1 : 0;
+  const size_t body = length - skip_start - skip_end;
+  // 0xF0, the body, then the timestamp before 0xF7
+  const size_t framed = body + 3;
+  if (framed > out_size)
+    return 0;
+  // A body byte with bit 7 set would be read as a timestamp or status byte
+  for (size_t i = 0; i < body; i++) {
+    if (payload[skip_start + i] & 0x80)
+      return 0;
+  }
+  out[0] = STATUS_SYSEX_START;
+  std::memcpy(out + 1, payload + skip_start, body);
+  out[body + 1] = PACKET_TIMESTAMP;
+  out[body + 2] = STATUS_SYSEX_END;
+  return framed;
+}
+
+}  // namespace esphome::ble_midi

esphome/components/ble_midi/midi_parser.h

@@ -0,0 +1,131 @@
+#pragma once
+
+#include "esphome/core/progmem.h"
+
+#include <cstddef>
+#include <cstdint>
+
+namespace esphome::ble_midi {
+
+// Decoded MIDI message types. Status bytes have the high bit set; the low nibble of a channel-voice status byte
+// (0x80-0xEF) is the channel, system messages (0xF0-0xFF) have none.
+enum class MessageType : uint8_t {
+  NOTE_OFF,            // 0x80
+  NOTE_ON,             // 0x90
+  POLY_AFTERTOUCH,     // 0xA0
+  CONTROL_CHANGE,      // 0xB0
+  PROGRAM_CHANGE,      // 0xC0
+  CHANNEL_AFTERTOUCH,  // 0xD0
+  PITCH_BEND,          // 0xE0
+  SYSEX,               // 0xF0 ... 0xF7, possibly spread over several BLE packets
+  SYSEX_INCOMPLETE,    // a SysEx cut short by another status byte or longer than the buffer
+  SYSTEM_COMMON,       // 0xF1-0xF6, e.g. quarter frame, song position
+  SYSTEM_REAL_TIME,    // 0xF8-0xFF, e.g. clock, start, stop
+  INCOMPLETE,          // a channel or system common message cut short by another status byte
+  UNKNOWN,             // malformed/undecodable byte
+};
+
+PROGMEM_STRING_TABLE(MessageTypeNames, "NOTE_OFF", "NOTE_ON", "POLY_AFTERTOUCH", "CONTROL_CHANGE", "PROGRAM_CHANGE",
+                     "CHANNEL_AFTERTOUCH", "PITCH_BEND", "SYSEX", "SYSEX_INCOMPLETE", "SYSTEM_COMMON",
+                     "SYSTEM_REAL_TIME", "INCOMPLETE", "UNKNOWN");
+
+/// A decoded MIDI message. `data` points into the parser's storage or the packet being fed, so it is only valid
+/// for the duration of the handler.
+struct MidiMessage {
+  // Status byte followed by its data bytes; a SysEx includes the 0xF0/0xF7 wrappers
+  const uint8_t *data{nullptr};
+  uint16_t length{0};
+  MessageType type{MessageType::UNKNOWN};
+
+  uint8_t status() const { return this->data[0]; }
+  uint8_t channel() const { return this->data[0] & 0x0F; }
+  // The data bytes of a complete channel-voice message: note and velocity, controller and value, or program
+  uint8_t data1() const { return this->data[1]; }
+  uint8_t data2() const { return this->data[2]; }
+  // PITCH_BEND: 14-bit value re-centered on zero (-8192..8191).
+  int16_t pitch_bend() const {
+    return static_cast<int16_t>((static_cast<uint16_t>(this->data[2]) << 7) | this->data[1]) - 8192;
+  }
+};
+
+/// Called once per decoded message with the context passed to feed().
+using MidiMessageHandler = void (*)(void *ctx, const MidiMessage &message);
+
+/// Decoder for the BLE-MIDI packet format (Apple/MMA "Bluetooth Low Energy
+/// MIDI" specification, section 3).
+///
+/// Packet layout:
+///   [header] [timestamp] [MIDI 0] ... ([timestamp] [MIDI n])*
+///
+/// - header    : 0b10hhhhhh, bit 7 set, bits 5-0 hold the high bits of the
+/// timestamp
+/// - timestamp : 0b1lllllll, bit 7 set, bits 6-0 hold the low bits of the
+/// timestamp
+///
+/// A timestamp byte precedes every status byte, except where running status
+/// applies. SysEx messages may span several BLE packets and may be interrupted
+/// by real-time messages (0xF8-0xFF).
+///
+/// Timestamps are used for jitter reconstruction by MIDI hosts and are ignored
+/// here.
+///
+/// This class holds no BLE or ESPHome state so that it can be exercised in
+/// isolation.
+class MidiParser final {
+ public:
+  /// `sysex_buffer` reassembles SysEx messages; one longer than `sysex_capacity` bytes is reported incomplete.
+  MidiParser(uint8_t *sysex_buffer, uint16_t sysex_capacity)
+      : sysex_buffer_(sysex_buffer), sysex_capacity_(sysex_capacity) {}
+
+  uint16_t sysex_capacity() const { return this->sysex_capacity_; }
+
+  /// Drop all parser state. Call on disconnect so running status and partial
+  /// messages do not leak into the next session.
+  void reset();
+
+  /// Feed one BLE notification payload. Undecodable bytes are reported as MessageType::UNKNOWN so the
+  /// caller can log them.
+  void feed(const uint8_t *data, size_t length, MidiMessageHandler handler, void *ctx);
+
+  /// Number of data bytes that follow a status byte other than SysEx.
+  static uint8_t expected_data_bytes(uint8_t status);
+
+ protected:
+  void emit_(MessageType type, const uint8_t *data, uint16_t length, MidiMessageHandler handler, void *ctx) {
+    handler(ctx, MidiMessage{data, length, type});
+  }
+  void emit_current_(MidiMessageHandler handler, void *ctx);
+  void take_data_(const uint8_t *byte, MidiMessageHandler handler, void *ctx);
+  void end_sysex_(bool complete, MidiMessageHandler handler, void *ctx);
+  void sysex_push_(uint8_t byte);
+
+  // SysEx reassembly across packets.
+  uint8_t *const sysex_buffer_;
+  const uint16_t sysex_capacity_;
+  uint16_t sysex_length_{0};
+  bool sysex_active_{false};
+  bool sysex_overflow_{false};
+
+  // Last channel-voice status byte, sticky across bytes and packets until a new
+  // status byte or a system-common message clears it (MIDI running status).
+  uint8_t running_status_{0};
+
+  // Message being assembled: status byte plus up to two data bytes.
+  uint8_t current_message_[3]{};
+  uint8_t current_length_{0};
+  uint8_t current_expected_{0};
+};
+
+// A BLE-MIDI packet starts with a header byte and a timestamp byte, both with bit 7 set. The remaining
+// timestamp bits serve jitter reconstruction only and are left at zero when transmitting.
+static constexpr uint8_t PACKET_HEADER = 0x80;
+static constexpr uint8_t PACKET_TIMESTAMP = 0x80;
+static constexpr uint8_t STATUS_SYSEX_START = 0xF0;
+static constexpr uint8_t STATUS_SYSEX_END = 0xF7;
+
+/// Frame a SysEx payload for one BLE-MIDI packet: 0xF0 and 0xF7 are added when the payload lacks them, with
+/// the timestamp byte the specification requires before 0xF7. Returns the bytes written to `out`, or 0 when
+/// the payload is empty, holds a byte with bit 7 set, or does not fit `out_size` once framed.
+size_t frame_sysex(const uint8_t *payload, size_t length, uint8_t *out, size_t out_size);
+
+}  // namespace esphome::ble_midi

esphome/components/mqtt/__init__.py

@@ -1,6 +1,6 @@
 from collections.abc import Iterator
 
-from esphome import automation
+from esphome import automation, preferences
 import esphome.codegen as cg
 from esphome.components import logger, network, socket
 from esphome.components.const import CONF_VERIFY_SSL
@@ -107,6 +107,7 @@ def AUTO_LOAD():
 CONF_TLS = "tls"
 CONF_IDF_SEND_ASYNC = "idf_send_async"
 CONF_WAIT_FOR_CONNECTION = "wait_for_connection"
+CONF_PERSIST_SUBSCRIPTIONS = "persist_subscriptions"
 CONF_TRANSPORT = "transport"
 CONF_WS_PATH = "ws_path"
 
@@ -304,6 +305,10 @@ def validate_config(value):
         raise cv.Invalid(
             f"'{CONF_WS_PATH}' is only valid when 'transport' is 'ws' or 'wss'"
         )
+    if value.get(CONF_PERSIST_SUBSCRIPTIONS) and value.get(CONF_CLEAN_SESSION):
+        raise cv.Invalid(
+            f"'{CONF_PERSIST_SUBSCRIPTIONS}' requires '{CONF_CLEAN_SESSION}: false'"
+        )
 
     # Populate default fields
     out = value.copy()
@@ -394,6 +399,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_USERNAME, default=""): cv.string,
             cv.Optional(CONF_PASSWORD, default=""): cv.sensitive(),
             cv.Optional(CONF_CLEAN_SESSION, default=False): cv.boolean,
+            cv.Optional(CONF_PERSIST_SUBSCRIPTIONS, default=False): cv.boolean,
             cv.Optional(CONF_CLIENT_ID): cv.string,
             cv.Optional(
                 CONF_TRANSPORT, default=MQTT_TRANSPORT_TCP
@@ -666,6 +672,10 @@ async def to_code(config):
     cg.add(var.set_username(config[CONF_USERNAME]))
     cg.add(var.set_password(config[CONF_PASSWORD]))
     cg.add(var.set_clean_session(config[CONF_CLEAN_SESSION]))
+    if config[CONF_PERSIST_SUBSCRIPTIONS]:
+        cg.add_define("USE_MQTT_SESSION_PERSISTENCE")
+        # The session fingerprint, a uint32_t kept in RTC memory
+        preferences.reserve_preferences(1, 4, in_flash=False)
     cg.add_optional(var.set_client_id, config.get(CONF_CLIENT_ID))
 
     discovery = config[CONF_DISCOVERY]

esphome/components/mqtt/mqtt_backend.h

@@ -39,7 +39,7 @@ class MQTTBackend {
  public:
   using on_connect_callback_t = void(bool session_present);
   using on_disconnect_callback_t = void(MQTTClientDisconnectReason reason);
-  using on_subscribe_callback_t = void(uint16_t packet_id, uint8_t qos);
+  using on_subscribe_callback_t = void(uint16_t packet_id, uint8_t return_code);
   using on_unsubscribe_callback_t = void(uint16_t packet_id);
   using on_message_callback_t = void(const char *topic, const char *payload, size_t len, size_t index, size_t total);
   using on_publish_user_callback_t = void(uint16_t packet_id);

esphome/components/mqtt/mqtt_backend_esp32.cpp

@@ -170,8 +170,9 @@ void MQTTBackendESP32::mqtt_event_handler_(const Event &event) {
 
     case MQTT_EVENT_SUBSCRIBED:
       ESP_LOGV(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event.msg_id);
-      // hardcode QoS to 0. QoS is not used in this context but required to mirror the AsyncMqtt interface
-      this->on_subscribe_.call((int) event.msg_id, 0);
+      // One topic per SUBSCRIBE, so data[0] is its return code
+      this->on_subscribe_.call((int) event.msg_id,
+                               event.data.empty() ? SUBACK_FAILURE : static_cast<uint8_t>(event.data[0]));
       break;
     case MQTT_EVENT_UNSUBSCRIBED:
       ESP_LOGV(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event.msg_id);
@@ -224,6 +225,16 @@ void MQTTBackendESP32::mqtt_event_handler(void *handler_args, esp_event_base_t b
     if (msg_id == instance->tracked_msg_id_.load())
       instance->tracked_acked_.store(1);
   }
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  // Also counted here: the queue holds 31 events, too few for the SUBACKs of a device with many command topics
+  if (event_id == MQTT_EVENT_CONNECTED) {
+    instance->subacks_.reset();
+  } else if (event_id == MQTT_EVENT_SUBSCRIBED) {
+    const auto *mqtt_event = static_cast<esp_mqtt_event_t *>(event_data);
+    // One topic per SUBSCRIBE, so data[0] is its return code
+    instance->subacks_.add(mqtt_event->data_len > 0 ? static_cast<uint8_t>(mqtt_event->data[0]) : SUBACK_FAILURE);
+  }
+#endif
   // queue event to decouple processing from ESP-IDF MQTT task to main loop
   auto *event = instance->mqtt_event_pool_.allocate();
   if (event == nullptr) {

esphome/components/mqtt/mqtt_backend_esp32.h

@@ -14,6 +14,7 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/lock_free_queue.h"
 #include "esphome/core/event_pool.h"
+#include "mqtt_subscription_fingerprint.h"
 
 namespace esphome::mqtt {
 
@@ -214,6 +215,9 @@ class MQTTBackendESP32 final : public MQTTBackend {
   using MQTTBackend::publish;
   bool publish_tracked(const MQTTMessage &message) final;
   bool publish_acknowledged() const final { return this->tracked_acked_.load(std::memory_order_acquire) != 0; }
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  const SubackCounter &get_subacks() const { return this->subacks_; }
+#endif
 
   void loop() final;
 
@@ -294,6 +298,10 @@ class MQTTBackendESP32 final : public MQTTBackend {
   std::atomic<uint16_t> acked_msg_id_{0};
   // uint8_t, not bool: GCC on Xtensa emits an out-of-line call for atomic<bool>
   std::atomic<uint8_t> tracked_acked_{0};
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  // Counted in the ESP-IDF MQTT task, see mqtt_event_handler()
+  SubackCounter subacks_;
+#endif
   // Pool sized to queue capacity (SIZE-1) because LockFreeQueue<T,N> is a ring
   // buffer that holds N-1 elements (one slot distinguishes full from empty).
   // This guarantees allocate() returns nullptr before push() can fail, which:

esphome/components/mqtt/mqtt_client.cpp

@@ -21,6 +21,16 @@
 #ifdef USE_API
 #include "esphome/components/api/api_server.h"
 #endif
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+#include "mqtt_subscription_fingerprint.h"
+#ifdef USE_ESP8266
+#include "esphome/core/preferences.h"
+#elif defined(USE_ESP32)
+#include <esp_attr.h>
+#include <soc/soc_caps.h>
+#include "esphome/core/preferences_rtc.h"
+#endif
+#endif
 #ifdef USE_DASHBOARD_IMPORT
 #include "esphome/components/dashboard_import/dashboard_import.h"
 #endif
@@ -32,6 +42,39 @@ ESPHOME_LOG_TAG(TAG, "mqtt");
 // Maximum number of MQTT component resends per loop iteration.
 // Limits work to avoid triggering the task watchdog on reconnect.
 static constexpr uint8_t MAX_RESENDS_PER_LOOP = 8;
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+static constexpr uint32_t SESSION_FINGERPRINT_HASH = fnv1a_hash("mqtt_session_fingerprint");
+#ifdef USE_ESP8266
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+static ESPPreferenceObject s_session_fingerprint_pref;
+#elif defined(USE_ESP32) && SOC_RTC_MEM_SUPPORTED
+#define USE_MQTT_SESSION_FINGERPRINT_RTC_NOINIT
+// Own 8 bytes rather than core's RTC preference pool, which would add 248 bytes of RTC RAM (usable memory on
+// variants with RTC fast memory only). The fingerprint and its checksum; power-on garbage fails the checksum.
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+static RTC_NOINIT_ATTR uint32_t s_session_fingerprint[2];
+#endif
+
+// RTC memory where available; RAM only otherwise, so a reset resubscribes
+static void load_session_fingerprint([[maybe_unused]] uint32_t *fingerprint) {
+#ifdef USE_ESP8266
+  s_session_fingerprint_pref = global_preferences->make_preference<uint32_t>(SESSION_FINGERPRINT_HASH, false);
+  s_session_fingerprint_pref.load(fingerprint);
+#elif defined(USE_MQTT_SESSION_FINGERPRINT_RTC_NOINIT)
+  rtc_pref_decode(s_session_fingerprint, SESSION_FINGERPRINT_HASH, 1, reinterpret_cast<uint8_t *>(fingerprint),
+                  sizeof(*fingerprint));
+#endif
+}
+
+static void store_session_fingerprint([[maybe_unused]] const uint32_t *fingerprint) {
+#ifdef USE_ESP8266
+  s_session_fingerprint_pref.save(fingerprint);
+#elif defined(USE_MQTT_SESSION_FINGERPRINT_RTC_NOINIT)
+  rtc_pref_encode(s_session_fingerprint, SESSION_FINGERPRINT_HASH, 1, reinterpret_cast<const uint8_t *>(fingerprint),
+                  sizeof(*fingerprint));
+#endif
+}
+#endif
 
 // Disconnect reason strings indexed by MQTTClientDisconnectReason enum (0-8)
 PROGMEM_STRING_TABLE(MQTTDisconnectReasonStrings, "TCP disconnected", "Unacceptable Protocol Version",
@@ -80,6 +123,19 @@ void MQTTClientComponent::setup() {
     this->state_ = MQTT_CLIENT_DISCONNECTED;
     this->disconnect_reason_ = reason;
   });
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  this->mqtt_backend_.set_on_connect([this](bool session_present) {
+    this->session_present_ = session_present;
+#ifndef USE_ESP32
+    // AsyncMqttClient delivers CONNACK and SUBACK in the same context, so the counter has one writer
+    this->subacks_.reset();
+#endif
+  });
+#ifndef USE_ESP32
+  this->mqtt_backend_.set_on_subscribe([this](uint16_t, uint8_t return_code) { this->subacks_.add(return_code); });
+#endif
+  load_session_fingerprint(&this->session_fingerprint_);
+#endif
 #ifdef USE_LOGGER
   if (this->is_log_message_enabled() && logger::global_logger != nullptr) {
     logger::global_logger->add_log_callback(
@@ -361,6 +417,9 @@ void MQTTClientComponent::start_connect_() {
                                  this->last_will_.payload.c_str());
   }
 
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  this->session_present_ = false;
+#endif
   this->mqtt_backend_.connect();
   this->state_ = MQTT_CLIENT_CONNECTING;
   this->connect_begin_ = millis();
@@ -385,14 +444,30 @@ void MQTTClientComponent::check_connected() {
   // MQTT Client needs some time to be fully set up.
   delay(100);  // NOLINT
 
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  this->subscribes_sent_ = 0;
+  const bool resumed =
+      session_has_subscriptions(this->session_present_, this->session_fingerprint_, this->subscriptions_fingerprint_());
+  this->subacks_settled_ = resumed;
+  if (resumed) {
+    for (auto &subscription : this->subscriptions_)
+      subscription.subscribed = true;
+  } else {
+    // Untrusted until all are acknowledged again, so a reset part way through does not trust a partial session;
+    // topics only an older firmware used stay subscribed in the session
+    this->set_session_fingerprint_(UNTRUSTED_SESSION_FINGERPRINT);
+  }
+#endif
   this->resubscribe_subscriptions_();
 #ifdef USE_MQTT_CRASH_LOG
-  // Best effort: a refused subscription is tried again at the next connection
-  if (this->crash_log_offered_) {
-    char topic[ESPHOME_TOPIC_BUFFER_SIZE];
-    format_esphome_topic(topic, "logs");
-    this->subscribe_(topic, 0);
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  // Otherwise sent once the SUBACKs are in, so its own SUBACK is never counted
+  if (resumed) {
+    this->subscribe_crash_log_();
   }
+#else
+  this->subscribe_crash_log_();
+#endif
 #endif
   this->send_device_info_();
 
@@ -441,7 +516,12 @@ void MQTTClientComponent::loop() {
         }
 
         this->last_connected_ = now;
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+        if (this->resubscribe_subscriptions_() && !this->subacks_settled_)
+          this->check_subacks_();
+#else
         this->resubscribe_subscriptions_();
+#endif
 
         // Process pending resends for all MQTT components centrally
         // Limit work per loop iteration to avoid triggering task WDT on reconnect
@@ -499,15 +579,69 @@ void MQTTClientComponent::resubscribe_subscription_(MQTTSubscription *sub) {
   if (do_resub) {
     sub->subscribed = this->subscribe_topic_(sub->topic, sub->qos);
     sub->resubscribe_timeout = now;
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+    // Only the subscriptions the fingerprint covers are counted
+    if (sub->subscribed)
+      this->subscribes_sent_++;
+#endif
   }
 }
-void MQTTClientComponent::resubscribe_subscriptions_() {
+bool MQTTClientComponent::resubscribe_subscriptions_() {
+  bool all_subscribed = true;
   for (auto &subscription : this->subscriptions_) {
     this->resubscribe_subscription_(&subscription);
+    all_subscribed &= subscription.subscribed;
   }
+  return all_subscribed;
 }
 
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+// Out of line so the topic buffer is not part of loop()'s stack frame
+__attribute__((noinline)) uint32_t MQTTClientComponent::subscriptions_fingerprint_() {
+  // Seeded with the config hash, so a new firmware config sends every subscription once
+  uint32_t fingerprint = App.get_config_hash();
+  char buf[MQTT_DEFAULT_TOPIC_MAX_LEN];
+  for (const auto &subscription : this->subscriptions_)
+    fingerprint = add_subscription_fingerprint(fingerprint, subscription.topic.build(buf).c_str(), subscription.qos);
+  return fingerprint;
+}
+
+void MQTTClientComponent::check_subacks_() {
+#ifdef USE_ESP32
+  const SubackTally tally = this->mqtt_backend_.get_subacks().tally(this->subscribes_sent_);
+#else
+  const SubackTally tally = this->subacks_.tally(this->subscribes_sent_);
+#endif
+  if (tally == SubackTally::SUBACK_TALLY_PENDING)
+    return;
+  this->subacks_settled_ = true;
+  // Trusted only once the broker granted every SUBSCRIBE, so a reset or a lost link before that sends them again,
+  // and a refused one is tried again at the next connect
+  if (tally == SubackTally::SUBACK_TALLY_ALL_GRANTED)
+    this->set_session_fingerprint_(this->subscriptions_fingerprint_());
 #ifdef USE_MQTT_CRASH_LOG
+  this->subscribe_crash_log_();
+#endif
+}
+
+void MQTTClientComponent::set_session_fingerprint_(uint32_t fingerprint) {
+  if (this->session_fingerprint_ == fingerprint)
+    return;
+  this->session_fingerprint_ = fingerprint;
+  store_session_fingerprint(&this->session_fingerprint_);
+}
+#endif
+
+#ifdef USE_MQTT_CRASH_LOG
+void MQTTClientComponent::subscribe_crash_log_() {
+  // Best effort: a refused subscription is tried again at the next connection
+  if (!this->crash_log_offered_)
+    return;
+  char topic[ESPHOME_TOPIC_BUFFER_SIZE];
+  format_esphome_topic(topic, "logs");
+  this->subscribe_(topic, 0);
+}
+
 void MQTTClientComponent::handle_crash_log_request_(const std::string &topic, const std::string &payload) {
   // Only the CLI's empty request counts: a log_topic pointing at this topic would otherwise replay the crash
   // log for every line it logs
@@ -572,6 +706,10 @@ __attribute__((noinline)) static bool topic_equals(const MQTTTopicSource &source
 }
 
 void MQTTClientComponent::unsubscribe(const std::string &topic) {
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  // Before the broker drops the topic, so a reset in between cannot leave a matching fingerprint
+  this->set_session_fingerprint_(UNTRUSTED_SESSION_FINGERPRINT);
+#endif
   bool ret = this->mqtt_backend_.unsubscribe(topic.c_str());
   yield();
   if (ret) {

esphome/components/mqtt/mqtt_client.h

@@ -23,6 +23,9 @@
 #elif defined(USE_LIBRETINY)
 #include "mqtt_backend_libretiny.h"
 #endif
+#if defined(USE_MQTT_SESSION_PERSISTENCE) && !defined(USE_ESP32)
+#include "mqtt_subscription_fingerprint.h"
+#endif
 #include "lwip/ip_addr.h"
 
 #include <span>
@@ -380,10 +383,18 @@ class MQTTClientComponent final : public Component {
   }
   void erase_removed_subscriptions_();
 #ifdef USE_MQTT_CRASH_LOG
+  void subscribe_crash_log_();
   void handle_crash_log_request_(const std::string &topic, const std::string &payload);
 #endif
   void add_subscription_(MQTTTopicSource topic, mqtt_subscription_callback_t callback, bool boxed, uint8_t qos);
-  void resubscribe_subscriptions_();
+  /// Returns true when every subscription is subscribed
+  bool resubscribe_subscriptions_();
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  uint32_t subscriptions_fingerprint_();
+  /// Once every SUBACK of this connection is in, saves the fingerprint if all were granted
+  void check_subacks_();
+  void set_session_fingerprint_(uint32_t fingerprint);
+#endif
 
   MQTTCredentials credentials_;
   /// The last will message. Disabled optional denotes it being default and
@@ -440,6 +451,18 @@ class MQTTClientComponent final : public Component {
   bool wait_for_connection_{false};
   bool dispatching_{false};
   bool removed_while_dispatching_{false};
+#ifdef USE_MQTT_SESSION_PERSISTENCE
+  bool session_present_{false};
+  /// Set once every SUBACK of this connection is in (or the session was resumed without subscribing)
+  bool subacks_settled_{false};
+  /// SUBSCRIBEs sent on this connection for the subscriptions the fingerprint covers
+  uint16_t subscribes_sent_{0};
+#ifndef USE_ESP32
+  SubackCounter subacks_;
+#endif
+  /// Fingerprint of the set the broker session is known to hold in full; 0 while a rebuild is in progress
+  uint32_t session_fingerprint_{0};
+#endif
 };
 
 extern MQTTClientComponent *global_mqtt_client;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)

esphome/components/mqtt/mqtt_subscription_fingerprint.h

@@ -0,0 +1,65 @@
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+
+#include "esphome/core/helpers.h"
+
+namespace esphome::mqtt {
+
+/// Adds one subscription to a fingerprint of a set of subscriptions; the order they are added in does not matter.
+inline uint32_t add_subscription_fingerprint(uint32_t fingerprint, const char *topic, uint8_t qos) {
+  return fingerprint + fnv1a_hash_extend(fnv1a_hash(topic), qos);
+}
+
+/// Session fingerprint while a rebuild is in progress, and after a power loss or a cold boot.
+static constexpr uint32_t UNTRUSTED_SESSION_FINGERPRINT = 0;
+
+/// True when a resumed session holds in full the subscriptions it is trusted with, and they are the current ones.
+constexpr bool session_has_subscriptions(bool session_present, uint32_t trusted, uint32_t current) {
+  return session_present && trusted != UNTRUSTED_SESSION_FINGERPRINT && trusted == current;
+}
+
+/// First SUBACK return code that refuses a subscription; also stands in for a SUBACK with no return code.
+static constexpr uint8_t SUBACK_FAILURE = 0x80;
+
+/// True when a SUBACK return code grants the subscription.
+constexpr bool subscribe_granted(uint8_t return_code) { return return_code < SUBACK_FAILURE; }
+
+enum class SubackTally : uint8_t {
+  SUBACK_TALLY_PENDING,
+  SUBACK_TALLY_ALL_GRANTED,
+  SUBACK_TALLY_NOT_ALL_GRANTED,
+};
+
+/// SUBACKs received on the current connection. Written only by the context that delivers broker events (so plain
+/// loads and stores, no read-modify-write), read by the main loop.
+class SubackCounter {
+ public:
+  void reset() {
+    this->received_.store(0, std::memory_order_relaxed);
+    this->granted_.store(0, std::memory_order_relaxed);
+  }
+  void add(uint8_t return_code) {
+    if (subscribe_granted(return_code))
+      this->granted_.store(this->granted_.load(std::memory_order_relaxed) + 1, std::memory_order_relaxed);
+    // Released after granted_, so a reader that sees this SUBACK in received_ also sees it in granted_
+    this->received_.store(this->received_.load(std::memory_order_relaxed) + 1, std::memory_order_release);
+  }
+  /// Compares the SUBACKs received with the number of SUBSCRIBEs sent on this connection.
+  SubackTally tally(uint16_t sent) const {
+    const uint16_t received = this->received_.load(std::memory_order_acquire);
+    if (received < sent)
+      return SubackTally::SUBACK_TALLY_PENDING;
+    // More than sent means a SUBACK from an earlier connection was counted, so none can be trusted
+    return received == sent && this->granted_.load(std::memory_order_relaxed) == sent
+               ? SubackTally::SUBACK_TALLY_ALL_GRANTED
+               : SubackTally::SUBACK_TALLY_NOT_ALL_GRANTED;
+  }
+
+ protected:
+  std::atomic<uint16_t> received_{0};
+  std::atomic<uint16_t> granted_{0};
+};
+
+}  // namespace esphome::mqtt

esphome/components/nextion/base_component.py

@@ -31,6 +31,7 @@ CONF_PRECISION = "precision"
 CONF_SKIP_CONNECTION_HANDSHAKE = "skip_connection_handshake"
 CONF_START_UP_PAGE = "start_up_page"
 CONF_STARTUP_OVERRIDE_MS = "startup_override_ms"
+CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX = "tft_upload_http_buffer_size_tx"
 CONF_TFT_UPLOAD_HTTP_RETRIES = "tft_upload_http_retries"
 CONF_TFT_UPLOAD_HTTP_TIMEOUT = "tft_upload_http_timeout"
 CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT = "tft_upload_watchdog_timeout"

esphome/components/nextion/display.py

@@ -33,6 +33,7 @@ from .base_component import (
     CONF_SKIP_CONNECTION_HANDSHAKE,
     CONF_START_UP_PAGE,
     CONF_STARTUP_OVERRIDE_MS,
+    CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX,
     CONF_TFT_UPLOAD_HTTP_RETRIES,
     CONF_TFT_UPLOAD_HTTP_TIMEOUT,
     CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT,
@@ -53,10 +54,9 @@ def AUTO_LOAD() -> list[str]:
 
 
 def _validate_tft_upload(config):
-    if CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT in config and not CORE.is_esp32:
-        raise cv.Invalid(
-            f"{CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT} is only available on ESP32"
-        )
+    for key in (CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT, CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX):
+        if key in config and not CORE.is_esp32:
+            raise cv.Invalid(f"{key} is only available on ESP32")
     return config
 
 
@@ -105,6 +105,9 @@ CONFIG_SCHEMA = cv.All(
                 ),
             ),
             cv.Optional(CONF_START_UP_PAGE): cv.uint8_t,
+            cv.Optional(CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX): cv.int_range(
+                min=512, max=8192
+            ),
             cv.Optional(CONF_TFT_UPLOAD_HTTP_RETRIES): cv.int_range(min=1, max=255),
             cv.Optional(CONF_TFT_UPLOAD_HTTP_TIMEOUT): cv.All(
                 cv.positive_time_period_milliseconds,
@@ -136,6 +139,7 @@ CONFIG_SCHEMA = cv.All(
         (
             CONF_TFT_UPLOAD_HTTP_TIMEOUT,
             CONF_TFT_UPLOAD_HTTP_RETRIES,
+            CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX,
             CONF_TFT_UPLOAD_WATCHDOG_TIMEOUT,
         ),
         CONF_TFT_URL,
@@ -244,6 +248,14 @@ async def to_code(config):
                 )
             )
 
+        # TFT upload HTTP TX buffer size (ESP32 only; unset = ESP-IDF default)
+        if CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX in config:
+            cg.add(
+                var.set_tft_upload_http_buffer_size_tx(
+                    config[CONF_TFT_UPLOAD_HTTP_BUFFER_SIZE_TX]
+                )
+            )
+
         # TFT upload HTTP retries (default: 5)
         if CONF_TFT_UPLOAD_HTTP_RETRIES in config:
             cg.add(

esphome/components/nextion/nextion.h

@@ -1089,6 +1089,19 @@ class Nextion final : public NextionBase, public PollingComponent, public uart::
   void set_tft_upload_http_timeout(uint16_t timeout_ms) { this->tft_upload_http_timeout_ = timeout_ms; }
 
 #ifdef USE_ESP32
+  /**
+   * @brief Set the HTTP client TX buffer size for TFT upload requests.
+   *
+   * The TX buffer must hold the request line, which carries the full URL. After a redirect
+   * to a long signed URL (e.g. a GitHub release asset, ~600+ characters) the ESP-IDF default
+   * is too small and the request fails with "HTTP_CLIENT: Out of buffer".
+   *
+   * A value of 0 keeps the ESP-IDF default (default).
+   *
+   * @param size Buffer size in bytes.
+   */
+  void set_tft_upload_http_buffer_size_tx(uint16_t size) { this->tft_upload_http_buffer_size_tx_ = size; }
+
   /**
    * @brief Set the watchdog timeout during TFT upload.
    *
@@ -1573,7 +1586,8 @@ class Nextion final : public NextionBase, public PollingComponent, public uart::
   uint8_t tft_upload_http_retries_{5};      ///< HTTP retry count (default: 5)
 
 #ifdef USE_ESP32
-  uint32_t tft_upload_watchdog_timeout_{0};  ///< WDT timeout in ms (0 = no adjustment)
+  uint32_t tft_upload_watchdog_timeout_{0};     ///< WDT timeout in ms (0 = no adjustment)
+  uint16_t tft_upload_http_buffer_size_tx_{0};  ///< HTTP TX buffer in bytes (0 = ESP-IDF default)
 
   /**
    * will request 4096 bytes chunks from the web server

esphome/components/nextion/nextion_upload_esp32.cpp

@@ -283,6 +283,7 @@ bool Nextion::upload_tft(uint32_t baud_rate, bool exit_reparse) {
       .timeout_ms = static_cast<int>(this->tft_upload_http_timeout_),
       .disable_auto_redirect = false,
       .max_redirection_count = 10,
+      .buffer_size_tx = this->tft_upload_http_buffer_size_tx_,  // 0 = ESP-IDF default
   };
   // Initialize the HTTP client with the configuration
   esp_http_client_handle_t http_client = esp_http_client_init(&config);

esphome/core/defines.h

@@ -381,6 +381,7 @@
 #if !defined(USE_RP2) && !defined(USE_HOST)  // no MQTT backend or esp_wireguard library on RP2 and host
 #define USE_MQTT
 #define USE_MQTT_ABBREVIATIONS
+#define USE_MQTT_SESSION_PERSISTENCE
 #define USE_MQTT_COVER_JSON
 #define USE_WIREGUARD
 #endif

tests/component_tests/mqtt/config/persist_subscriptions.yaml

@@ -0,0 +1,15 @@
+---
+esphome:
+  name: test-device
+
+esp8266:
+  board: d1_mini_lite
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+mqtt:
+  broker: 192.168.1.2
+  clean_session: false
+  persist_subscriptions: true

tests/component_tests/mqtt/test_persist_subscriptions.py

@@ -0,0 +1,51 @@
+"""Tests for the mqtt persist_subscriptions option."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.esp8266.const import KEY_ESP8266, KEY_PREFERENCES
+from esphome.components.mqtt import CONF_PERSIST_SUBSCRIPTIONS, validate_config
+from esphome.const import CONF_CLEAN_SESSION, CONF_TOPIC_PREFIX
+from esphome.core import CORE
+
+
+def test_rejected_with_clean_session() -> None:
+    with pytest.raises(cv.Invalid, match="requires 'clean_session: false'"):
+        validate_config(
+            {
+                CONF_TOPIC_PREFIX: "node",
+                CONF_CLEAN_SESSION: True,
+                CONF_PERSIST_SUBSCRIPTIONS: True,
+            }
+        )
+
+
+def test_accepted_without_clean_session() -> None:
+    out = validate_config(
+        {
+            CONF_TOPIC_PREFIX: "node",
+            CONF_CLEAN_SESSION: False,
+            CONF_PERSIST_SUBSCRIPTIONS: True,
+        }
+    )
+    assert out[CONF_PERSIST_SUBSCRIPTIONS] is True
+
+
+@pytest.mark.parametrize(
+    ("config_file", "enabled"),
+    [("persist_subscriptions.yaml", True), ("no_mac_suffix.yaml", False)],
+)
+def test_define_and_rtc_reservation(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+    enabled: bool,
+) -> None:
+    generate_main(component_config_path(config_file))
+    assert ("USE_MQTT_SESSION_PERSISTENCE" in {d.name for d in CORE.defines}) is enabled
+    # One uint32_t in ESP8266 RTC memory: a data word plus the checksum word, not in flash
+    reservations = CORE.data[KEY_ESP8266].get(KEY_PREFERENCES, [])
+    assert reservations.count((2, False)) == int(enabled)

tests/components/ble_midi/common.yaml

@@ -0,0 +1,83 @@
+esp32_ble_tracker:
+
+ble_client:
+  - mac_address: 01:02:03:04:05:06
+    id: test_midi_client
+
+ble_midi:
+  id: test_midi
+  ble_client_id: test_midi_client
+  max_sysex_size: 64
+  pair: false
+  on_connect:
+    - logger.log: "MIDI device connected"
+    - ble_midi.send_program_change:
+        id: test_midi
+        program: 0
+  on_disconnect:
+    - logger.log: "MIDI device disconnected"
+  on_note_on:
+    - logger.log:
+        format: "Note on %u velocity %u channel %u"
+        args: [note, velocity, channel]
+  on_note_off:
+    - logger.log:
+        format: "Note off %u channel %u"
+        args: [note, channel]
+  on_control_change:
+    - logger.log:
+        format: "CC %u = %u on channel %u"
+        args: [control_number, value, channel]
+  on_program_change:
+    - logger.log:
+        format: "Program %u on channel %u"
+        args: [program, channel]
+  on_pitch_bend:
+    - logger.log:
+        format: "Pitch bend %d on channel %u"
+        args: [value, channel]
+  on_sysex:
+    - logger.log:
+        format: "SysEx of %u bytes"
+        args: ["data.size()"]
+  on_message:
+    - logger.log:
+        format: "Raw message of %u bytes"
+        args: ["data.size()"]
+
+button:
+  - platform: template
+    name: "MIDI send tests"
+    on_press:
+      - ble_midi.send_note_on:
+          id: test_midi
+          note: 60
+      - ble_midi.send_note_on:
+          id: test_midi
+          note: !lambda "return 60;"
+          velocity: !lambda "return 100;"
+          channel: !lambda "return 1;"
+      - ble_midi.send_note_off:
+          id: test_midi
+          note: 60
+          velocity: 0
+          channel: 1
+      - ble_midi.send_control_change:
+          id: test_midi
+          control_number: 7
+          value: 64
+      - ble_midi.send_pitch_bend:
+          id: test_midi
+          value: -8192
+      - ble_midi.send_sysex:
+          id: test_midi
+          payload: [0x7E, 0x7F, 0x06, 0x01]
+      - ble_midi.send_raw:
+          id: test_midi
+          data: !lambda "return std::vector<uint8_t>{0x90, 0x3C, 0x40};"
+      - if:
+          condition:
+            ble_midi.connected:
+              id: test_midi
+          then:
+            - logger.log: "Still connected"

tests/components/ble_midi/midi_parser_test.cpp

@@ -0,0 +1,283 @@
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <vector>
+
+#include "esphome/components/ble_midi/midi_parser.h"
+
+namespace esphome::ble_midi::testing {
+
+// A BLE-MIDI packet is a header byte, then timestamp/MIDI byte pairs. Both
+// header and timestamp have bit 7 set; their remaining bits carry a timestamp
+// that this component ignores.
+static constexpr uint8_t HDR = 0x80;
+static constexpr uint8_t TS = 0x80;
+
+struct Decoded {
+  MessageType type;
+  std::vector<uint8_t> bytes;
+};
+
+class MidiParserTest : public ::testing::Test {
+ protected:
+  static void collect(void *ctx, const MidiMessage &message) {
+    static_cast<MidiParserTest *>(ctx)->messages_.push_back(
+        {message.type, std::vector<uint8_t>(message.data, message.data + message.length)});
+  }
+  void feed_(const std::vector<uint8_t> &packet) { this->parser_.feed(packet.data(), packet.size(), collect, this); }
+
+  uint8_t sysex_buffer_[64]{};
+  MidiParser parser_{sysex_buffer_, sizeof(sysex_buffer_)};
+  std::vector<Decoded> messages_;
+};
+
+TEST_F(MidiParserTest, NoteOn) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::NOTE_ON);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0x90, 0x3C, 0x40}));
+}
+
+TEST_F(MidiParserTest, ChannelIsTakenFromTheStatusByte) {
+  this->feed_({HDR, TS, 0x95, 0x3C, 0x40});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  MidiMessage message{this->messages_[0].bytes.data(), 3, this->messages_[0].type};
+  EXPECT_EQ(message.channel(), 5);
+  EXPECT_EQ(message.data1(), 60);
+  EXPECT_EQ(message.data2(), 64);
+}
+
+TEST_F(MidiParserTest, RunningStatusReusesThePreviousStatusByte) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40, 0x3E, 0x41});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[1].type, MessageType::NOTE_ON);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0x90, 0x3E, 0x41}));
+}
+
+TEST_F(MidiParserTest, RunningStatusSurvivesAcrossPackets) {
+  this->feed_({HDR, TS, 0xB0, 0x07, 0x10});
+  this->feed_({HDR, 0x07, 0x20});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[1].type, MessageType::CONTROL_CHANGE);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0xB0, 0x07, 0x20}));
+}
+
+TEST_F(MidiParserTest, ResetClearsRunningStatus) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40});
+  this->parser_.reset();
+  this->messages_.clear();
+  this->feed_({HDR, TS, 0x3E, 0x41});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::UNKNOWN);
+  EXPECT_EQ(this->messages_[1].type, MessageType::UNKNOWN);
+}
+
+TEST_F(MidiParserTest, NoteOnWithZeroVelocityIsANoteOff) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x00});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::NOTE_OFF);
+}
+
+TEST_F(MidiParserTest, ControlChange) {
+  this->feed_({HDR, TS, 0xB2, 0x07, 0x64});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  MidiMessage message{this->messages_[0].bytes.data(), 3, this->messages_[0].type};
+  EXPECT_EQ(message.type, MessageType::CONTROL_CHANGE);
+  EXPECT_EQ(message.channel(), 2);
+  EXPECT_EQ(message.data1(), 7);
+  EXPECT_EQ(message.data2(), 100);
+}
+
+TEST_F(MidiParserTest, ProgramChangeHasASingleDataByte) {
+  this->feed_({HDR, TS, 0xC0, 0x05, TS, 0xC0, 0x06});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::PROGRAM_CHANGE);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xC0, 0x05}));
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0xC0, 0x06}));
+}
+
+TEST_F(MidiParserTest, PitchBendIsCenteredOnZero) {
+  this->feed_({HDR, TS, 0xE0, 0x00, 0x40, TS, 0xE0, 0x00, 0x00, TS, 0xE0, 0x7F, 0x7F});
+  ASSERT_EQ(this->messages_.size(), 3u);
+  auto bend = [this](size_t index) {
+    MidiMessage message{this->messages_[index].bytes.data(), 3, this->messages_[index].type};
+    return message.pitch_bend();
+  };
+  EXPECT_EQ(this->messages_[0].type, MessageType::PITCH_BEND);
+  EXPECT_EQ(bend(0), 0);
+  EXPECT_EQ(bend(1), -8192);
+  EXPECT_EQ(bend(2), 8191);
+}
+
+TEST_F(MidiParserTest, RealTimeMessageIsASingleByte) {
+  this->feed_({HDR, TS, 0xF8});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSTEM_REAL_TIME);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF8}));
+}
+
+TEST_F(MidiParserTest, RealTimeMessageInsideAChannelMessageLeavesItIntact) {
+  this->feed_({HDR, TS, 0x90, 0x3C, TS, 0xF8, 0x40});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSTEM_REAL_TIME);
+  EXPECT_EQ(this->messages_[1].type, MessageType::NOTE_ON);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0x90, 0x3C, 0x40}));
+}
+
+TEST_F(MidiParserTest, SysexCutsAPartialChannelMessageShort) {
+  this->feed_({HDR, TS, 0x90, 0x3C, TS, 0xF0, 0x01, TS, 0xF7, 0x40});
+  ASSERT_EQ(this->messages_.size(), 3u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::INCOMPLETE);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0x90, 0x3C}));
+  EXPECT_EQ(this->messages_[1].type, MessageType::SYSEX);
+  // No status is in effect after a SysEx, so the stray data byte does not finish the Note On
+  EXPECT_EQ(this->messages_[2].type, MessageType::UNKNOWN);
+  EXPECT_EQ(this->messages_[2].bytes, std::vector<uint8_t>({0x40}));
+}
+
+TEST_F(MidiParserTest, SystemCommonMessageWithTwoDataBytes) {
+  this->feed_({HDR, TS, 0xF2, 0x01, 0x02});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSTEM_COMMON);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF2, 0x01, 0x02}));
+}
+
+TEST_F(MidiParserTest, SystemCommonClearsRunningStatus) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40, TS, 0xF6});
+  this->messages_.clear();
+  this->feed_({HDR, 0x3E, 0x41});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::UNKNOWN);
+}
+
+TEST_F(MidiParserTest, SysexInASinglePacket) {
+  this->feed_({HDR, TS, 0xF0, 0x7E, 0x7F, 0x06, 0x01, TS, 0xF7});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSEX);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7}));
+}
+
+TEST_F(MidiParserTest, SysexReassembledAcrossPackets) {
+  this->feed_({HDR, TS, 0xF0, 0x7E, 0x7F});
+  EXPECT_TRUE(this->messages_.empty());
+  this->feed_({HDR, 0x06, 0x01, TS, 0xF7});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7}));
+}
+
+TEST_F(MidiParserTest, RealTimeMessageInterruptsSysex) {
+  this->feed_({HDR, TS, 0xF0, 0x7E, TS, 0xF8, 0x7F, TS, 0xF7});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSTEM_REAL_TIME);
+  EXPECT_EQ(this->messages_[1].type, MessageType::SYSEX);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0xF0, 0x7E, 0x7F, 0xF7}));
+}
+
+TEST_F(MidiParserTest, SysexLongerThanTheBufferIsReportedIncomplete) {
+  uint8_t small_buffer[8]{};
+  MidiParser small_parser{small_buffer, sizeof(small_buffer)};
+  std::vector<uint8_t> packet{HDR, TS, 0xF0};
+  for (uint8_t i = 0; i < 64; i++)
+    packet.push_back(i & 0x7F);
+  packet.push_back(TS);
+  packet.push_back(0xF7);
+
+  small_parser.feed(packet.data(), packet.size(), collect, this);
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSEX_INCOMPLETE);
+  EXPECT_EQ(this->messages_[0].bytes.size(), 8u);
+
+  // The next SysEx that fits is reported whole
+  const std::vector<uint8_t> start{HDR, TS, 0xF0, 0x00, 0x01};
+  const std::vector<uint8_t> end{HDR, TS, 0xF7};
+  small_parser.feed(start.data(), start.size(), collect, this);
+  small_parser.feed(end.data(), end.size(), collect, this);
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[1].type, MessageType::SYSEX);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0xF0, 0x00, 0x01, 0xF7}));
+}
+
+TEST_F(MidiParserTest, StatusByteCuttingASysexShortDropsItAndIsDecoded) {
+  this->feed_({HDR, TS, 0xF0, 0x7E, TS, 0x90, 0x3C, 0x40});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSEX_INCOMPLETE);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF0, 0x7E}));
+  EXPECT_EQ(this->messages_[1].type, MessageType::NOTE_ON);
+  EXPECT_EQ(this->messages_[1].bytes, std::vector<uint8_t>({0x90, 0x3C, 0x40}));
+}
+
+TEST_F(MidiParserTest, SysexCarriedOverEndsWhenTheNextPacketStartsAMessage) {
+  this->feed_({HDR, TS, 0xF0, 0x7E});
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40, 0x3E, 0x41});
+  ASSERT_EQ(this->messages_.size(), 3u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::SYSEX_INCOMPLETE);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0xF0, 0x7E}));
+  EXPECT_EQ(this->messages_[1].type, MessageType::NOTE_ON);
+  // Running status works again after the SysEx was closed
+  EXPECT_EQ(this->messages_[2].bytes, std::vector<uint8_t>({0x90, 0x3E, 0x41}));
+}
+
+TEST(FrameSysex, AddsTheWrappersAndTheTimestampBeforeTheEnd) {
+  uint8_t out[16];
+  const uint8_t payload[] = {0x7E, 0x7F, 0x06, 0x01};
+  ASSERT_EQ(frame_sysex(payload, sizeof(payload), out, sizeof(out)), 7u);
+  EXPECT_EQ(std::vector<uint8_t>(out, out + 7), std::vector<uint8_t>({0xF0, 0x7E, 0x7F, 0x06, 0x01, TS, 0xF7}));
+}
+
+TEST(FrameSysex, KeepsWrappersThatArePresent) {
+  uint8_t out[16];
+  const uint8_t payload[] = {0xF0, 0x7E, 0xF7};
+  ASSERT_EQ(frame_sysex(payload, sizeof(payload), out, sizeof(out)), 4u);
+  EXPECT_EQ(std::vector<uint8_t>(out, out + 4), std::vector<uint8_t>({0xF0, 0x7E, TS, 0xF7}));
+}
+
+TEST(FrameSysex, RejectsEmptyAndOversizedPayloads) {
+  uint8_t out[4];
+  const uint8_t payload[] = {0x7E, 0x7F};
+  EXPECT_EQ(frame_sysex(payload, 0, out, sizeof(out)), 0u);
+  EXPECT_EQ(frame_sysex(payload, sizeof(payload), out, sizeof(out)), 0u);
+}
+
+TEST(FrameSysex, RejectsABodyByteWithBit7Set) {
+  uint8_t out[16];
+  const uint8_t payload[] = {0x7E, 0x80, 0x01};
+  EXPECT_EQ(frame_sysex(payload, sizeof(payload), out, sizeof(out)), 0u);
+}
+
+TEST_F(MidiParserTest, DataByteWithoutStatusContextIsReportedAsUnknown) {
+  this->feed_({HDR, TS, 0x3C});
+  ASSERT_EQ(this->messages_.size(), 1u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::UNKNOWN);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0x3C}));
+}
+
+TEST_F(MidiParserTest, PacketWithoutAHeaderByteIsRejected) {
+  this->feed_({0x3C, 0x40});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::UNKNOWN);
+  EXPECT_EQ(this->messages_[1].type, MessageType::UNKNOWN);
+}
+
+TEST_F(MidiParserTest, EmptyAndSingleByteInputAreHandled) {
+  this->feed_({});
+  this->feed_({HDR});
+  EXPECT_EQ(this->messages_.size(), 1u);
+}
+
+TEST_F(MidiParserTest, IncompleteMessageIsReportedWhenANewStatusByteArrives) {
+  this->feed_({HDR, TS, 0x90, 0x3C, TS, 0xB0, 0x07, 0x10});
+  ASSERT_EQ(this->messages_.size(), 2u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::INCOMPLETE);
+  EXPECT_EQ(this->messages_[0].bytes, std::vector<uint8_t>({0x90, 0x3C}));
+  EXPECT_EQ(this->messages_[1].type, MessageType::CONTROL_CHANGE);
+}
+
+TEST_F(MidiParserTest, MultipleMessagesInOnePacket) {
+  this->feed_({HDR, TS, 0x90, 0x3C, 0x40, TS, 0x80, 0x3C, 0x00, TS, 0xB0, 0x07, 0x7F});
+  ASSERT_EQ(this->messages_.size(), 3u);
+  EXPECT_EQ(this->messages_[0].type, MessageType::NOTE_ON);
+  EXPECT_EQ(this->messages_[1].type, MessageType::NOTE_OFF);
+  EXPECT_EQ(this->messages_[2].type, MessageType::CONTROL_CHANGE);
+}
+
+}  // namespace esphome::ble_midi::testing

tests/components/ble_midi/test.esp32-idf.yaml

@@ -0,0 +1,3 @@
+packages:
+  ble: !include ../../test_build_components/common/ble/esp32-idf.yaml
+  ble_midi: !include common.yaml

tests/components/mqtt/common.yaml

@@ -12,7 +12,8 @@ mqtt:
   username: debug
   password: debug
   enable_on_boot: false
-  clean_session: True
+  clean_session: false
+  persist_subscriptions: true
   client_id: someclient
   use_abbreviations: false
   discovery: true

tests/components/mqtt/test_subscription_fingerprint.cpp

@@ -0,0 +1,94 @@
+#include <gtest/gtest.h>
+
+#include <initializer_list>
+#include <utility>
+
+#include "esphome/components/mqtt/mqtt_subscription_fingerprint.h"
+
+namespace esphome::mqtt::testing {
+
+static constexpr uint32_t SEED = 0x12345678;
+
+static uint32_t fingerprint(std::initializer_list<std::pair<const char *, uint8_t>> subscriptions) {
+  uint32_t value = SEED;
+  for (const auto &[topic, qos] : subscriptions)
+    value = add_subscription_fingerprint(value, topic, qos);
+  return value;
+}
+
+TEST(SubscriptionFingerprint, OrderDoesNotMatter) {
+  EXPECT_EQ(fingerprint({{"a/set", 0}, {"b/set", 1}, {"c/set", 0}}),
+            fingerprint({{"c/set", 0}, {"a/set", 0}, {"b/set", 1}}));
+}
+
+TEST(SubscriptionFingerprint, TopicQosAndCountChangeIt) {
+  const uint32_t base = fingerprint({{"a/set", 0}, {"b/set", 0}});
+  EXPECT_NE(base, fingerprint({{"a/set", 0}, {"b/cmd", 0}}));
+  EXPECT_NE(base, fingerprint({{"a/set", 0}, {"b/set", 1}}));
+  EXPECT_NE(base, fingerprint({{"a/set", 0}}));
+  // Duplicates do not cancel each other out
+  EXPECT_NE(base, fingerprint({{"a/set", 0}, {"b/set", 0}, {"b/set", 0}, {"b/set", 0}}));
+}
+
+TEST(SubscriptionFingerprint, ChangedSetSubscribesAgain) {
+  const uint32_t trusted = fingerprint({{"a/set", 0}, {"b/set", 1}});
+  EXPECT_TRUE(session_has_subscriptions(true, trusted, fingerprint({{"b/set", 1}, {"a/set", 0}})));
+  EXPECT_FALSE(session_has_subscriptions(true, trusted, fingerprint({{"a/set", 0}, {"c/set", 1}})));
+}
+
+TEST(SubscriptionFingerprint, NewSessionSubscribesAgain) {
+  const uint32_t trusted = fingerprint({{"a/set", 0}});
+  EXPECT_FALSE(session_has_subscriptions(false, trusted, trusted));
+}
+
+TEST(SubscriptionFingerprint, UntrustedNeverMatches) {
+  EXPECT_FALSE(session_has_subscriptions(true, UNTRUSTED_SESSION_FINGERPRINT, UNTRUSTED_SESSION_FINGERPRINT));
+}
+
+TEST(SubscriptionFingerprint, RefusedSubackIsNotGranted) {
+  EXPECT_TRUE(subscribe_granted(0));
+  EXPECT_TRUE(subscribe_granted(2));
+  EXPECT_FALSE(subscribe_granted(0x80));
+  // MQTT 5 reason codes
+  EXPECT_FALSE(subscribe_granted(0x87));
+}
+
+TEST(SubackCounter, PendingUntilEverySubackIsIn) {
+  SubackCounter subacks;
+  EXPECT_EQ(subacks.tally(2), SubackTally::SUBACK_TALLY_PENDING);
+  subacks.add(0);
+  EXPECT_EQ(subacks.tally(2), SubackTally::SUBACK_TALLY_PENDING);
+  subacks.add(1);
+  EXPECT_EQ(subacks.tally(2), SubackTally::SUBACK_TALLY_ALL_GRANTED);
+}
+
+TEST(SubackCounter, NothingSentIsSettled) {
+  SubackCounter subacks;
+  EXPECT_EQ(subacks.tally(0), SubackTally::SUBACK_TALLY_ALL_GRANTED);
+}
+
+TEST(SubackCounter, RefusalSettlesWithoutTrust) {
+  SubackCounter subacks;
+  subacks.add(0x80);
+  EXPECT_EQ(subacks.tally(2), SubackTally::SUBACK_TALLY_PENDING);
+  subacks.add(0);
+  EXPECT_EQ(subacks.tally(2), SubackTally::SUBACK_TALLY_NOT_ALL_GRANTED);
+}
+
+TEST(SubackCounter, ExtraSubackIsNotTrusted) {
+  // A SUBACK for a SUBSCRIBE resent from an earlier connection
+  SubackCounter subacks;
+  subacks.add(0);
+  subacks.add(0);
+  EXPECT_EQ(subacks.tally(1), SubackTally::SUBACK_TALLY_NOT_ALL_GRANTED);
+}
+
+TEST(SubackCounter, ResetStartsANewConnection) {
+  SubackCounter subacks;
+  subacks.add(0x80);
+  subacks.reset();
+  subacks.add(0);
+  EXPECT_EQ(subacks.tally(1), SubackTally::SUBACK_TALLY_ALL_GRANTED);
+}
+
+}  // namespace esphome::mqtt::testing

tests/components/nextion/common_tft_upload_watchdog.yaml

@@ -1,3 +1,4 @@
 display:
   - id: !extend main_lcd
     tft_upload_watchdog_timeout: 30s
+    tft_upload_http_buffer_size_tx: 1024