[tcp] Add TCP packet interface link

Packets go over one TCP connection, as a client that connects to a host
or a server that accepts one client at a time. TCP resends what is lost
and joins exactly two devices, so the link provides the reliability,
addressing and session layers itself, and those cannot be configured on
it. A link platform names such layers with the new provides argument.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
clydebarrow
2026-10-20 09:35:39 +11:00
co-authored by Claude Opus 5.5
parent 8a07d4204b
commit 6c36da4e84
12 changed files with 600 additions and 2 deletions
+2
View File
@@ -635,6 +635,8 @@ esphome/components/tca8418/* @zebble
esphome/components/tca9548a/* @andreashergert1984
esphome/components/tca9555/* @mobrembski
esphome/components/tcl112/* @glmnet
esphome/components/tcp/* @clydebarrow
esphome/components/tcp/packet_interface/* @clydebarrow
esphome/components/tcp_uart/* @Bascht74
esphome/components/tee501/* @Stock-M
esphome/components/teleinfo/* @0hax @FredM67
@@ -136,8 +136,12 @@ def packet_interface_schema(
class_: MockObjClass,
default_max_packet_size: int,
max_packet_size_limit: int = 4096,
provides: tuple[str, ...] = (),
) -> cv.Schema:
"""Schema for a link platform, with the layers that can be stacked on it."""
"""Schema for a link platform; provides names the layer keys, such as reliability, that the link does itself."""
link_layers = {
key: schema for key, schema in LINK_LAYER_SCHEMAS.items() if key not in provides
}
return (
cv.Schema(
{
@@ -145,7 +149,7 @@ def packet_interface_schema(
cv.Optional(
CONF_MAX_PACKET_SIZE, default=default_max_packet_size
): cv.int_range(16, max_packet_size_limit),
**LINK_LAYER_SCHEMAS,
**link_layers,
}
)
.extend(cv.COMPONENT_SCHEMA)
+4
View File
@@ -0,0 +1,4 @@
"""TCP links. The packet_interface platform carries packets over a TCP connection."""
CODEOWNERS = ["@clydebarrow"]
DOMAIN = "tcp"
@@ -0,0 +1,98 @@
import esphome.codegen as cg
from esphome.components import binary_sensor, packet_interface, socket
from esphome.components.const import (
CONF_ALLOWED_IPS,
CONF_CONNECTED,
CONF_HOST,
CONF_RECONNECT_INTERVAL,
CONF_ROLE,
)
from esphome.components.packet_interface import (
new_packet_interface,
packet_interface_schema,
)
from esphome.components.packet_interface.layers import (
CONF_ADDRESSING,
CONF_RELIABILITY,
CONF_SESSION,
)
from esphome.components.packet_interface.layers.basic_session import CONF_KEEP_ALIVE
import esphome.config_validation as cv
from esphome.const import (
CONF_ID,
CONF_PORT,
DEVICE_CLASS_CONNECTIVITY,
ENTITY_CATEGORY_DIAGNOSTIC,
)
from esphome.types import ConfigType
CODEOWNERS = ["@clydebarrow"]
DEPENDENCIES = ["network"]
def AUTO_LOAD(config: ConfigType | None = None) -> list[str]:
"""The binary_sensor component only for the connected sensor; without a config, as tooling calls it, both."""
if config is None or CONF_CONNECTED in config:
return ["socket", "binary_sensor"]
return ["socket"]
# Each packet and its 2 byte length must fit the socket link's 1024 byte send buffer
MAX_PACKET_SIZE = 1022
tcp_ns = cg.esphome_ns.namespace("tcp")
TcpPacketInterface = tcp_ns.class_(
"TcpPacketInterface", packet_interface.PacketInterface
)
BASE_SCHEMA = packet_interface_schema(
TcpPacketInterface,
MAX_PACKET_SIZE,
MAX_PACKET_SIZE,
# TCP resends what is lost, joins exactly two devices, and its connection state is the session
provides=(CONF_ADDRESSING, CONF_RELIABILITY, CONF_SESSION),
).extend(
{
cv.Required(CONF_PORT): cv.port,
cv.Optional(
CONF_RECONNECT_INTERVAL, default="5s"
): cv.positive_time_period_milliseconds,
cv.Optional(CONF_KEEP_ALIVE, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(min=cv.TimePeriod(seconds=1)),
),
cv.Optional(CONF_CONNECTED): binary_sensor.binary_sensor_schema(
device_class=DEVICE_CLASS_CONNECTIVITY,
entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
),
}
)
CONFIG_SCHEMA = cv.typed_schema(
{
"client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
"server": BASE_SCHEMA.extend(
{cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
),
},
key=CONF_ROLE,
default_type="client",
)
async def to_code(config: ConfigType) -> None:
var = await new_packet_interface(config)
if config[CONF_ROLE] == "server":
socket.require_tcp_listener()
cg.add(var.set_server(True))
socket.add_ipv4_allow(
var.set_allow, config.get(CONF_ALLOWED_IPS), config[CONF_ID]
)
else:
socket.require_tcp_client_link()
cg.add(var.set_host(config[CONF_HOST]))
cg.add(var.set_port(config[CONF_PORT]))
cg.add(var.set_reconnect_interval(config[CONF_RECONNECT_INTERVAL]))
cg.add(var.set_keep_alive(config[CONF_KEEP_ALIVE]))
binary_sensors = binary_sensor.sub_binary_sensors(config)
await binary_sensors(CONF_CONNECTED, var.set_connected_sensor)
@@ -0,0 +1,167 @@
#include "tcp_packet_interface.h"
#include "esphome/core/application.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
namespace esphome::tcp {
using packet_interface::PacketBuffer;
using packet_interface::PacketMetaData;
ESPHOME_LOG_TAG(TAG, "tcp.packet_interface");
// The packet length before each packet
static constexpr size_t LENGTH_SIZE = 2;
// The peer is gone when nothing has been received for this many keep-alive intervals
static constexpr uint32_t KEEP_ALIVE_MISSES = 3;
TcpPacketInterface::TcpPacketInterface() {
this->capabilities_ = packet_interface::CAPABILITY_RELIABILITY | packet_interface::CAPABILITY_SESSION |
packet_interface::CAPABILITY_ADDRESSING;
}
void TcpPacketInterface::setup() {
this->rx_buffer_ = std::make_unique<uint8_t[]>(this->max_packet_size_);
this->link_.begin(TAG);
#ifdef USE_SOCKET_TCP_LISTENER
this->listener_.begin(TAG);
#endif
}
void TcpPacketInterface::dump_config() {
bool server = false;
#ifdef USE_SOCKET_TCP_LISTENER
server = this->server_;
#endif
ESP_LOGCONFIG(TAG,
"TCP Packet Interface '%s':\n"
" Role: %s\n"
" Address: %s:%u\n"
" Reconnect Interval: %" PRIu32 "ms\n"
" Keep Alive: %" PRIu32 "ms\n"
" Max Packet Size: %zu",
this->name_, server ? LOG_STR_LITERAL("server") : LOG_STR_LITERAL("client"),
server ? LOG_STR_LITERAL("*") : this->link_.host(), this->link_.port(),
this->link_.reconnect_interval(), this->keep_alive_ms_, this->max_packet_size_);
#ifdef USE_SOCKET_TCP_LISTENER
this->listener_.dump_config();
#endif
#ifdef USE_BINARY_SENSOR
LOG_BINARY_SENSOR(" ", "Connected", this->connected_sensor_);
#endif
}
void TcpPacketInterface::on_shutdown() {
this->link_.close();
#ifdef USE_SOCKET_TCP_LISTENER
this->listener_.close();
#endif
}
void TcpPacketInterface::loop() {
#ifdef USE_SOCKET_TCP_LISTENER
if (this->server_) {
// A new client is accepted only once the drop of the last has been reported
this->listener_.poll(this->link_, !this->connected_);
} else {
this->link_.poll();
}
#else
this->link_.poll();
#endif
if (this->link_.connected() != this->connected_)
this->on_connection_(this->link_.connected());
if (!this->connected_)
return;
this->read_();
if (!this->connected_)
return;
uint32_t now = App.get_loop_component_start_time();
if (now - this->last_rx_ms_ > KEEP_ALIVE_MISSES * this->keep_alive_ms_) {
ESP_LOGW(TAG, "%s: Nothing received for %" PRIu32 "ms, closing the connection", this->name_,
KEEP_ALIVE_MISSES * this->keep_alive_ms_);
this->link_.close();
this->link_.note_attempt();
this->on_connection_(false);
return;
}
if (now - this->last_tx_ms_ >= this->keep_alive_ms_)
this->queue_frame_(nullptr, 0);
this->link_.flush_tx();
}
void TcpPacketInterface::on_connection_(bool connected) {
ESP_LOGD(TAG, "%s: Connection %s", this->name_, connected ? LOG_STR_LITERAL("up") : LOG_STR_LITERAL("down"));
this->connected_ = connected;
this->rx_have_ = 0;
this->last_rx_ms_ = this->last_tx_ms_ = App.get_loop_component_start_time();
#ifdef USE_BINARY_SENSOR
if (this->connected_sensor_ != nullptr)
this->connected_sensor_->publish_state(connected);
#endif
this->session_callback_.call(connected);
}
void TcpPacketInterface::read_() {
for (;;) {
ssize_t count;
if (this->rx_have_ < LENGTH_SIZE) {
count = this->link_.read(this->rx_header_ + this->rx_have_, LENGTH_SIZE - this->rx_have_);
} else {
count = this->link_.read(this->rx_buffer_.get() + this->rx_have_ - LENGTH_SIZE,
this->rx_length_ + LENGTH_SIZE - this->rx_have_);
}
// Nothing more for now; a dropped link is seen on the next loop
if (count <= 0)
return;
this->last_rx_ms_ = App.get_loop_component_start_time();
this->rx_have_ += count;
if (this->rx_have_ < LENGTH_SIZE)
continue;
if (this->rx_have_ == LENGTH_SIZE) {
this->rx_length_ = encode_uint16(this->rx_header_[1], this->rx_header_[0]);
if (this->rx_length_ > this->max_packet_size_) {
// The stream cannot be resynchronised, so start a new connection
ESP_LOGW(TAG, "%s: Packet of %u bytes is larger than %zu, closing the connection", this->name_,
this->rx_length_, this->max_packet_size_);
this->link_.close();
this->link_.note_attempt();
this->on_connection_(false);
return;
}
}
if (this->rx_have_ == this->rx_length_ + LENGTH_SIZE) {
this->rx_have_ = 0;
// A zero length packet is a keep-alive
if (this->rx_length_ != 0)
this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.get(), this->rx_length_), {});
}
}
}
bool TcpPacketInterface::queue_frame_(const PacketBuffer *data, size_t size) {
if (this->link_.tx_free() < size + LENGTH_SIZE)
return false;
uint8_t *tail = this->link_.tx_tail();
tail[0] = static_cast<uint8_t>(size);
tail[1] = static_cast<uint8_t>(size >> 8);
if (data != nullptr)
data->copy_to(tail + LENGTH_SIZE, size);
this->link_.tx_commit(size + LENGTH_SIZE);
this->last_tx_ms_ = App.get_loop_component_start_time();
return true;
}
bool TcpPacketInterface::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
size_t size = data.size();
if (size > this->max_packet_size_) {
ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_);
return false;
}
if (!this->connected_ || !this->queue_frame_(&data, size))
return false;
this->link_.flush_tx();
return true;
}
} // namespace esphome::tcp
@@ -0,0 +1,84 @@
#pragma once
#include <cstdint>
#include <memory>
#include "esphome/components/packet_interface/packet_interface.h"
#include "esphome/components/socket/tcp_client_link.h"
#ifdef USE_SOCKET_TCP_LISTENER
#include "esphome/components/socket/tcp_listener.h"
#endif
#ifdef USE_BINARY_SENSOR
#include "esphome/components/binary_sensor/binary_sensor.h"
#endif
namespace esphome::tcp {
/**
* A packet interface link over one TCP connection, as a client that connects to a host, or a server that accepts
* one client at a time.
*
* Each packet is sent as a 2 byte length, little endian, then the packet. TCP delivers the stream in order and
* resends what is lost, and the connection joins exactly two devices, so the link provides the reliability,
* addressing and session capabilities itself. Its connection state is the session state.
*
* Each packet is written to the socket as it is sent, so the link takes packets as fast as the socket does and keeps
* the default, unlimited, send capacity; a packet is refused only while the connection is down or the socket is full.
*
* A zero length packet is a keep-alive, sent when nothing else has been sent for keep_alive. TCP can take minutes to
* notice a peer that has gone without closing the connection, so the link closes it, and reconnects or accepts again,
* when nothing has been received for three keep-alive intervals.
*/
class TcpPacketInterface : public packet_interface::PacketInterface {
public:
TcpPacketInterface();
void setup() override;
void loop() override;
void dump_config() override;
void on_shutdown() override;
bool send_to_interface(const packet_interface::PacketBuffer &data,
const packet_interface::PacketMetaData &meta_data) override;
void set_host(const char *host) { this->link_.set_host(host); }
void set_port(uint16_t port) { this->link_.set_port(port); }
void set_reconnect_interval(uint32_t interval_ms) { this->link_.set_reconnect_interval(interval_ms); }
void set_keep_alive(uint32_t keep_alive_ms) { this->keep_alive_ms_ = keep_alive_ms; }
#ifdef USE_SOCKET_TCP_LISTENER
void set_server(bool server) { this->server_ = server; }
#ifdef USE_SOCKET_IPV4_ALLOW
void set_allow(const socket::Ipv4AllowEntry *entries, size_t count) { this->listener_.set_allow(entries, count); }
#endif
#endif
#ifdef USE_BINARY_SENSOR
void set_connected_sensor(binary_sensor::BinarySensor *sensor) { this->connected_sensor_ = sensor; }
#endif
bool is_connected() const { return this->connected_; }
protected:
void on_connection_(bool connected);
void read_();
bool queue_frame_(const packet_interface::PacketBuffer *data, size_t size);
socket::TcpClientLink link_;
#ifdef USE_SOCKET_TCP_LISTENER
socket::TcpListener listener_;
bool server_{false};
#endif
#ifdef USE_BINARY_SENSOR
binary_sensor::BinarySensor *connected_sensor_{nullptr};
#endif
// The packet being received, allocated in setup()
std::unique_ptr<uint8_t[]> rx_buffer_;
uint32_t keep_alive_ms_{10000};
uint32_t last_rx_ms_{0};
uint32_t last_tx_ms_{0};
uint16_t rx_length_{0};
// Bytes of the current frame received so far, counting the 2 byte length
uint16_t rx_have_{0};
uint8_t rx_header_[2]{};
bool connected_{false};
};
} // namespace esphome::tcp
+11
View File
@@ -0,0 +1,11 @@
from esphome.components import socket as socket_component
from esphome.types import ConfigType
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
async def to_code_testing(config: ConfigType) -> None:
socket_component.require_tcp_listener()
manifest.to_code = to_code_testing
manifest.dependencies = manifest.dependencies + ["packet_interface", "socket"]
+4
View File
@@ -0,0 +1,4 @@
packet_interface:
- platform: tcp
role: server
port: 6000
+36
View File
@@ -0,0 +1,36 @@
wifi:
ssid: MySSID
password: password1
packet_interface:
- platform: tcp
id: tcp_client_link
host: 192.168.1.50
port: 6060
keep_alive: 5s
encryption: Johnny Cash
fragmentation:
max_packet_size: 2048
connected:
name: TCP link connected
- platform: tcp
id: tcp_server_link
role: server
port: 6061
allowed_ips:
- 192.168.1.0/24
sensor:
- platform: template
id: tcp_sensor
lambda: return 1.0;
packet_transport:
- platform: packet_interface
packet_interface: tcp_client_link
sensors:
- tcp_sensor
- platform: packet_interface
packet_interface: tcp_server_link
sensors:
- tcp_sensor
@@ -0,0 +1,151 @@
#include <gtest/gtest.h>
#include <csignal>
#include <cstdint>
#include <memory>
#include <sys/socket.h>
#include <unistd.h>
#include <vector>
#include "esphome/components/tcp/packet_interface/tcp_packet_interface.h"
#include "esphome/core/application.h"
#ifdef USE_HOST
namespace esphome::tcp::testing {
using packet_interface::PacketBuffer;
using packet_interface::PacketMetaData;
using Packet = std::vector<uint8_t>;
class TestTcp : public TcpPacketInterface {
public:
TestTcp() {
this->set_max_packet_size(64);
this->set_host("peer");
this->set_port(1);
}
socket::TcpClientLink &link() { return this->link_; }
};
// A layer stacked on the link, to check that the session state reaches it
class UpperLayer : public packet_interface::PacketInterface {
public:
explicit UpperLayer(PacketInterface *lower) { this->stack_on_(lower, packet_interface::CAPABILITY_ENCRYPTION); }
bool send_to_interface(const PacketBuffer &, const PacketMetaData &) override { return true; }
};
struct End {
End() {
this->link.add_packet_interface_listener(
[this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); });
this->link.add_on_session_state_callback([this](bool connected) { this->states.push_back(connected); });
this->link.setup();
}
bool send(const Packet &data) { return this->link.send_to_interface(PacketBuffer(data), {}); }
TestTcp link;
std::vector<Packet> received;
std::vector<bool> states;
};
class TcpPacketInterfaceTest : public ::testing::Test {
protected:
void SetUp() override {
signal(SIGPIPE, SIG_IGN);
new (&App) Application;
App.pre_setup("tcp-tests", 9, "", 0);
set_time(1000);
int fds[2];
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
this->a.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
this->b.link.link().adopt(std::make_unique<socket::Socket>(fds[1]));
this->loops(1);
}
void TearDown() override {
this->a.link.link().close();
this->b.link.link().close();
App.~Application();
}
static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
void loops(int count) {
for (int i = 0; i != count; i++) {
this->a.link.loop();
this->b.link.loop();
}
}
End a, b;
};
TEST_F(TcpPacketInterfaceTest, ProvidesReliabilitySessionAndAddressing) {
for (auto capability : {packet_interface::CAPABILITY_RELIABILITY, packet_interface::CAPABILITY_SESSION,
packet_interface::CAPABILITY_ADDRESSING})
EXPECT_TRUE(this->a.link.has_capability(capability));
EXPECT_EQ(this->a.states, std::vector<bool>{true});
EXPECT_TRUE(this->a.link.is_connected());
}
TEST_F(TcpPacketInterfaceTest, PacketsBothWays) {
Packet big(64);
for (size_t i = 0; i != big.size(); i++)
big[i] = i;
ASSERT_TRUE(this->a.send({1, 2, 3}));
ASSERT_TRUE(this->a.send(big));
ASSERT_TRUE(this->b.send({9}));
this->loops(3);
EXPECT_EQ(this->b.received, (std::vector<Packet>{{1, 2, 3}, big}));
EXPECT_EQ(this->a.received, std::vector<Packet>{{9}});
EXPECT_FALSE(this->a.send(Packet(65)));
}
TEST_F(TcpPacketInterfaceTest, SessionReachesLayersAbove) {
UpperLayer upper(&this->a.link);
std::vector<bool> states;
upper.add_on_session_state_callback([&states](bool connected) { states.push_back(connected); });
EXPECT_TRUE(upper.has_capability(packet_interface::CAPABILITY_SESSION));
this->b.link.link().close(); // the peer goes away
this->loops(3);
EXPECT_EQ(states, std::vector<bool>{false});
}
TEST_F(TcpPacketInterfaceTest, KeepAliveKeepsIdleLinkUp) {
for (uint32_t now = 1000; now <= 61000; now += 5000) {
set_time(now);
this->loops(2);
}
EXPECT_TRUE(this->a.link.is_connected());
EXPECT_TRUE(this->b.link.is_connected());
EXPECT_TRUE(this->a.received.empty()); // keep-alives are not passed up
}
TEST_F(TcpPacketInterfaceTest, SilentPeerClosesLink) {
set_time(1000 + 30001);
this->a.link.loop(); // b has sent nothing for three keep-alive intervals
EXPECT_FALSE(this->a.link.is_connected());
EXPECT_EQ(this->a.states, (std::vector<bool>{true, false}));
}
TEST_F(TcpPacketInterfaceTest, OversizedLengthClosesLink) {
End c;
int fds[2];
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
c.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
c.link.loop();
uint8_t header[2] = {0xFF, 0x00}; // 255 bytes, more than 64
ASSERT_EQ(::write(fds[1], header, sizeof(header)), 2);
c.link.loop();
EXPECT_FALSE(c.link.is_connected());
EXPECT_EQ(c.states, (std::vector<bool>{true, false}));
::close(fds[1]);
}
TEST_F(TcpPacketInterfaceTest, NotSentWhileDown) {
this->a.link.link().close();
this->a.link.loop();
EXPECT_FALSE(this->a.send({1}));
}
} // namespace esphome::tcp::testing
#endif
+2
View File
@@ -0,0 +1,2 @@
packages:
tcp: !include common.yaml
@@ -0,0 +1,35 @@
"""Unit tests for the tcp packet interface platform."""
import pytest
from esphome import config_validation as cv
from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
from esphome.components.tcp import packet_interface as tcp_packet_interface
def test_client_and_server(setup_core) -> None:
"""A client needs a host; a server listens on its port."""
client = tcp_packet_interface.CONFIG_SCHEMA({"host": "192.168.1.50", "port": 6060})
assert client["role"] == "client"
assert client[CONF_MAX_PACKET_SIZE] == 1022
assert tcp_packet_interface.CONFIG_SCHEMA({"role": "server", "port": 6060})
with pytest.raises(cv.Invalid):
tcp_packet_interface.CONFIG_SCHEMA({"port": 6060})
@pytest.mark.parametrize("layer", ["reliability", "session", "addressing"])
def test_provided_layers_rejected(setup_core, layer: str) -> None:
"""TCP provides these itself, so they cannot be configured on it."""
with pytest.raises(cv.Invalid, match=layer):
tcp_packet_interface.CONFIG_SCHEMA(
{"host": "192.168.1.50", "port": 6060, layer: {"address": 1}}
)
def test_auto_load_binary_sensor_for_connected(setup_core) -> None:
"""The binary_sensor component is loaded only for the connected sensor."""
assert tcp_packet_interface.AUTO_LOAD() == ["socket", "binary_sensor"]
config = {"host": "192.168.1.50", "port": 6060}
assert tcp_packet_interface.AUTO_LOAD(config) == ["socket"]
config["connected"] = {"name": "Link"}
assert tcp_packet_interface.AUTO_LOAD(config) == ["socket", "binary_sensor"]