From dc719b300b20b07fecebb295327cdbd24cec9036 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 12 Oct 2026 07:16:29 -1000 Subject: [PATCH] [udp] Use the socket UDP API on ESP8266 and RP2040 (#20558) --- esphome/components/socket/socket.cpp | 9 +- esphome/components/socket/socket.h | 6 +- esphome/components/udp/automation.h | 2 +- esphome/components/udp/udp_component.cpp | 90 +++++++------------ esphome/components/udp/udp_component.h | 25 ++---- .../fixtures/udp_multicast_join.yaml | 22 +++++ tests/integration/test_udp_multicast.py | 75 ++++++++++++++++ 7 files changed, 149 insertions(+), 80 deletions(-) create mode 100644 tests/integration/fixtures/udp_multicast_join.yaml create mode 100644 tests/integration/test_udp_multicast.py diff --git a/esphome/components/socket/socket.cpp b/esphome/components/socket/socket.cpp index a91c455e0fd..36cd6d6ab7f 100644 --- a/esphome/components/socket/socket.cpp +++ b/esphome/components/socket/socket.cpp @@ -218,8 +218,9 @@ template static bool foreach_eligible_ipv6_if(F &&callback) { return found_any; #endif } +#endif // USE_SOCKET_IMPL_BSD_SOCKETS || USE_SOCKET_IMPL_LWIP_SOCKETS -bool join_multicast_group(Socket *sock, const char *ip_address, uint32_t *if_index_out) { +bool join_multicast_group(UDPSocket *sock, const char *ip_address, uint32_t *if_index_out) { if (strchr(ip_address, ':') == nullptr) { struct in_addr ipv4mc {}; #ifdef USE_ZEPHYR @@ -249,7 +250,8 @@ bool join_multicast_group(Socket *sock, const char *ip_address, uint32_t *if_ind } return true; } -#if USE_NETWORK_IPV6 + // The raw lwIP implementation has no IPv6 membership support +#if USE_NETWORK_IPV6 && (defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)) struct ipv6_mreq imreq6 {}; #ifdef USE_SOCKET_IMPL_BSD_SOCKETS #ifdef USE_ZEPHYR @@ -300,9 +302,10 @@ bool join_multicast_group(Socket *sock, const char *ip_address, uint32_t *if_ind #else errno = EINVAL; return false; -#endif // USE_NETWORK_IPV6 +#endif // USE_NETWORK_IPV6 && fd based sockets } +#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) #if USE_NETWORK_IPV6 bool set_ipv6_multicast_if(Socket *sock, uint32_t if_index_in) { #if defined(USE_ZEPHYR) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) diff --git a/esphome/components/socket/socket.h b/esphome/components/socket/socket.h index 8f5860e26e1..7310e32afb6 100644 --- a/esphome/components/socket/socket.h +++ b/esphome/components/socket/socket.h @@ -191,16 +191,16 @@ inline socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const st /// Set a sockaddr to the any address and specified port for the IP version used by socket_ip(). socklen_t set_sockaddr_any(struct sockaddr *addr, socklen_t addrlen, uint16_t port); -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) /// Join a multicast group on the given socket. /// @param sock The socket to join on -/// @param ip_address Null-terminated multicast address string (IPv4 or IPv6) +/// @param ip_address Null-terminated multicast address string (IPv4, or IPv6 except on LWIP_TCP) /// @param if_index_out If non-null, receives the interface index used (IPv4 always writes 0). For /// link-local IPv6 groups, the caller must also copy this into sin6_scope_id /// before bind() -- POSIX requires it, LwIP ignores it. /// @return true on success, false on failure (errno set) -bool join_multicast_group(Socket *sock, const char *ip_address, uint32_t *if_index_out = nullptr); +bool join_multicast_group(UDPSocket *sock, const char *ip_address, uint32_t *if_index_out = nullptr); +#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) #if USE_NETWORK_IPV6 /// Configure the outgoing IPv6 multicast interface on the given socket. /// @param sock The socket to configure diff --git a/esphome/components/udp/automation.h b/esphome/components/udp/automation.h index a6a40829b11..0936aa3ab50 100644 --- a/esphome/components/udp/automation.h +++ b/esphome/components/udp/automation.h @@ -13,7 +13,7 @@ template class UDPWriteAction final : public Action, publ TEMPLATABLE_BYTES(data) void play(const Ts &...x) override { - // One write per packet: WiFiUDP appends all or nothing, so a failed allocation never sends a partial packet + // One sendto per packet, so a failed allocation never sends a partial packet this->data_.template visit( [this](const uint8_t *data, size_t len) { this->parent_->send_packet(data, len); }, x...); } diff --git a/esphome/components/udp/udp_component.cpp b/esphome/components/udp/udp_component.cpp index 3918ddc1fa8..8e824038341 100644 --- a/esphome/components/udp/udp_component.cpp +++ b/esphome/components/udp/udp_component.cpp @@ -8,9 +8,10 @@ namespace esphome::udp { ESPHOME_LOG_TAG(TAG, "udp"); +static constexpr uint32_t JOIN_POLL_ID = 0; +static constexpr uint32_t JOIN_POLL_INTERVAL_MS = 1000; void UDPComponent::setup() { -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) for (const char *const *it = this->addresses_; it != nullptr && *it != nullptr; it++) { const char *address = *it; struct sockaddr saddr {}; @@ -24,7 +25,7 @@ void UDPComponent::setup() { } // set up broadcast socket if (this->should_broadcast_) { - this->broadcast_socket_ = socket::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP); + this->broadcast_socket_ = socket::socket_udp_send(AF_INET, IPPROTO_IP); if (this->broadcast_socket_ == nullptr) { this->status_set_error(LOG_STR("Could not create socket")); this->mark_failed(); @@ -44,7 +45,7 @@ void UDPComponent::setup() { // create listening socket if we either want to subscribe to providers, or need to listen // for ping key broadcasts. if (this->should_listen_) { - this->listen_socket_ = socket::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP); + this->listen_socket_ = socket::socket_udp(AF_INET, IPPROTO_IP); if (this->listen_socket_ == nullptr) { this->status_set_error(LOG_STR("Could not create socket")); this->mark_failed(); @@ -69,22 +70,13 @@ void UDPComponent::setup() { server.sin_addr.s_addr = ESPHOME_INADDR_ANY; server.sin_port = htons(this->listen_port_); - if (this->listen_address_.has_value()) { - // Only 16 bytes needed for IPv4, but use standard size for consistency - char addr_buf[network::IP_ADDRESS_BUFFER_SIZE]; - this->listen_address_.value().str_to(addr_buf); - struct ip_mreq imreq = {}; - imreq.imr_interface.s_addr = ESPHOME_INADDR_ANY; - inet_aton(addr_buf, &imreq.imr_multiaddr); - server.sin_addr.s_addr = imreq.imr_multiaddr.s_addr; - ESP_LOGD(TAG, "Join multicast %s", addr_buf); - err = this->listen_socket_->setsockopt(IPPROTO_IP, IP_ADD_MEMBERSHIP, &imreq, sizeof(imreq)); - if (err < 0) { - ESP_LOGE(TAG, "Failed to set IP_ADD_MEMBERSHIP. Error %d", errno); - this->status_set_error(LOG_STR("Failed to set IP_ADD_MEMBERSHIP")); - this->mark_failed(); - return; - } + if (this->listen_address_ != nullptr) { + inet_aton(this->listen_address_, &server.sin_addr); + // Joined once the network is up; RP2040 has no interface to join before then + this->set_interval(JOIN_POLL_ID, JOIN_POLL_INTERVAL_MS, [this]() { + if (network::is_connected() && this->join_multicast_()) + this->cancel_interval(JOIN_POLL_ID); + }); } err = this->listen_socket_->bind((struct sockaddr *) &server, sizeof(server)); @@ -95,36 +87,35 @@ void UDPComponent::setup() { return; } } -#endif -#ifdef USE_SOCKET_IMPL_LWIP_TCP - // 8266 and RP2040 `Duino - for (const char *const *it = this->addresses_; it != nullptr && *it != nullptr; it++) { - const char *address = *it; - auto ipaddr = IPAddress(); - if (!ipaddr.fromString(address)) { - ESP_LOGW(TAG, "Invalid address %s", address); - this->status_set_warning(LOG_STR("invalid address")); - continue; +} + +bool UDPComponent::join_multicast_() { + if (socket::join_multicast_group(this->listen_socket_.get(), this->listen_address_)) { + ESP_LOGD(TAG, "Joined multicast %s", this->listen_address_); + if (this->join_warning_) { + this->join_warning_ = false; + this->status_clear_warning(); } - this->ipaddrs_.push_back(ipaddr); + return true; } - if (this->should_listen_) - this->udp_client_.begin(this->listen_port_); -#endif + // Log the first failure only; the poll keeps retrying every second + if (!this->join_failed_) { + this->join_failed_ = true; + ESP_LOGW(TAG, "Failed to set IP_ADD_MEMBERSHIP. Error %d", errno); + // One warning bit: only a warning raised here is cleared on success, one from setup() stays + if (!this->status_has_warning()) { + this->status_set_warning(LOG_STR("Failed to set IP_ADD_MEMBERSHIP")); + this->join_warning_ = true; + } + } + return false; } void UDPComponent::loop() { if (this->should_listen_) { std::array buf; for (;;) { -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) auto len = this->listen_socket_->read(buf.data(), buf.size()); -#endif -#ifdef USE_SOCKET_IMPL_LWIP_TCP - auto len = this->udp_client_.parsePacket(); - if (len > 0) - len = this->udp_client_.read(buf.data(), buf.size()); -#endif if (len <= 0) break; size_t packet_len = static_cast(len); @@ -144,9 +135,8 @@ void UDPComponent::dump_config() { const char *address = *it; ESP_LOGCONFIG(TAG, " Address: %s", address); } - if (this->listen_address_.has_value()) { - char addr_buf[network::IP_ADDRESS_BUFFER_SIZE]; - ESP_LOGCONFIG(TAG, " Listen address: %s", this->listen_address_.value().str_to(addr_buf)); + if (this->listen_address_ != nullptr) { + ESP_LOGCONFIG(TAG, " Listen address: %s", this->listen_address_); } ESP_LOGCONFIG(TAG, " Broadcasting: %s\n" @@ -155,26 +145,12 @@ void UDPComponent::dump_config() { } void UDPComponent::send_packet(const uint8_t *data, size_t size) { -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) for (const auto &saddr : this->sockaddrs_) { auto result = this->broadcast_socket_->sendto(data, size, 0, &saddr, sizeof(saddr)); if (result < 0) { ESP_LOGW(TAG, "sendto() error %d", errno); } } -#endif -#ifdef USE_SOCKET_IMPL_LWIP_TCP - auto iface = IPAddress(0, 0, 0, 0); - for (const auto &saddr : this->ipaddrs_) { - if (this->udp_client_.beginPacketMulticast(saddr, this->broadcast_port_, iface, 128) != 0) { - this->udp_client_.write(data, size); - auto result = this->udp_client_.endPacket(); - if (result == 0) { - ESP_LOGW(TAG, "udp.write() error"); - } - } - } -#endif } } // namespace esphome::udp diff --git a/esphome/components/udp/udp_component.h b/esphome/components/udp/udp_component.h index 17e27fb7704..276fd5adf17 100644 --- a/esphome/components/udp/udp_component.h +++ b/esphome/components/udp/udp_component.h @@ -3,13 +3,7 @@ #include "esphome/core/defines.h" #ifdef USE_NETWORK #include "esphome/core/helpers.h" -#include "esphome/components/network/ip_address.h" -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) #include "esphome/components/socket/socket.h" -#endif -#ifdef USE_SOCKET_IMPL_LWIP_TCP -#include -#endif #include #include #include @@ -22,7 +16,8 @@ class UDPComponent final : public Component { public: /// nullptr-terminated table of addresses; must outlive the component. void set_addresses(const char *const *addresses) { this->addresses_ = addresses; } - void set_listen_address(const char *listen_addr) { this->listen_address_ = network::IPAddress(listen_addr); } + /// Multicast group literal; must outlive the component. + void set_listen_address(const char *listen_addr) { this->listen_address_ = listen_addr; } void set_listen_port(uint16_t port) { this->listen_port_ = port; } void set_broadcast_port(uint16_t port) { this->broadcast_port_ = port; } void set_should_broadcast() { this->should_broadcast_ = true; } @@ -36,24 +31,22 @@ class UDPComponent final : public Component { float get_setup_priority() const override { return setup_priority::AFTER_WIFI; }; protected: + bool join_multicast_(); + uint16_t listen_port_{}; uint16_t broadcast_port_{}; bool should_broadcast_{}; bool should_listen_{}; + bool join_failed_{}; + bool join_warning_{}; CallbackManager)> packet_listeners_{}; -#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) - std::unique_ptr broadcast_socket_ = nullptr; - std::unique_ptr listen_socket_ = nullptr; + std::unique_ptr broadcast_socket_ = nullptr; + std::unique_ptr listen_socket_ = nullptr; std::vector sockaddrs_{}; -#endif -#ifdef USE_SOCKET_IMPL_LWIP_TCP - std::vector ipaddrs_{}; - WiFiUDP udp_client_{}; -#endif const char *const *addresses_{nullptr}; - optional listen_address_{}; + const char *listen_address_{nullptr}; }; } // namespace esphome::udp diff --git a/tests/integration/fixtures/udp_multicast_join.yaml b/tests/integration/fixtures/udp_multicast_join.yaml new file mode 100644 index 00000000000..e2c6c350bfc --- /dev/null +++ b/tests/integration/fixtures/udp_multicast_join.yaml @@ -0,0 +1,22 @@ +esphome: + name: udp-multicast-test + +host: + +api: + +logger: + level: DEBUG + +udp: + - id: test_udp + listen_address: 239.1.2.3 + addresses: + - "239.1.2.3" + port: + listen_port: UDP_LISTEN_PORT_PLACEHOLDER + broadcast_port: UDP_LISTEN_PORT_PLACEHOLDER + on_receive: + - logger.log: + format: "Received UDP: %d bytes" + args: [data.size()] diff --git a/tests/integration/test_udp_multicast.py b/tests/integration/test_udp_multicast.py new file mode 100644 index 00000000000..98a90591421 --- /dev/null +++ b/tests/integration/test_udp_multicast.py @@ -0,0 +1,75 @@ +"""Integration test for the UDP component's deferred multicast join.""" + +from __future__ import annotations + +import asyncio +import socket + +import pytest + +from .types import APIClientConnectedFactory, RunCompiledFunction + +MULTICAST_GROUP = "239.1.2.3" + + +def _get_free_udp_port() -> int: + """Get a free UDP port by binding to port 0 and releasing.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +@pytest.mark.asyncio +async def test_udp_multicast_join( + yaml_config: str, + run_compiled: RunCompiledFunction, + api_client_connected: APIClientConnectedFactory, +) -> None: + """The group is joined from the poll after setup, once, and then receives.""" + log_lines: list[str] = [] + joined_event = asyncio.Event() + receive_event = asyncio.Event() + + def on_log_line(line: str) -> None: + log_lines.append(line) + if f"Joined multicast {MULTICAST_GROUP}" in line: + joined_event.set() + if "Received UDP:" in line: + receive_event.set() + + listen_port = _get_free_udp_port() + config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port)) + + async with ( + run_compiled(config, line_callback=on_log_line), + api_client_connected() as client, + ): + device_info = await client.device_info() + assert device_info is not None + assert device_info.name == "udp-multicast-test" + + try: + await asyncio.wait_for(joined_event.wait(), timeout=5.0) + except TimeoutError: + pytest.fail(f"Multicast group was not joined. Log: {log_lines[-20:]}") + + # The poll cancels itself after the join, so a second poll period must not join again + await asyncio.sleep(2.5) + join_lines = [line for line in log_lines if "Joined multicast" in line] + assert len(join_lines) == 1, f"Expected one join, got: {join_lines}" + assert not any("IP_ADD_MEMBERSHIP" in line for line in log_lines), ( + f"Join reported a failure. Log: {log_lines[-20:]}" + ) + + send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + send_sock.sendto(b"MULTICAST_PAYLOAD", (MULTICAST_GROUP, listen_port)) + finally: + send_sock.close() + + try: + await asyncio.wait_for(receive_event.wait(), timeout=5.0) + except TimeoutError: + pytest.fail(f"Multicast datagram was not received. Log: {log_lines[-20:]}")