module.yaml

@@ -0,0 +1,14 @@
+esphome:
+  # ...
+  includes:
+    - stream_server.h
+    - stream_server.cpp
+
+uart:
+  id: uart_bus
+  #  ...
+  
+custom_component:
+  - lambda: |-
+      auto stream_server = new StreamServerComponent(id(uart_bus));
+      return {stream_server};

stream_server.cpp

@@ -0,0 +1,86 @@
+#include "stream_server.h"
+
+#include "esphome/core/log.h"
+#include "esphome/core/util.h"
+
+static const char *TAG = "streamserver";
+
+using namespace esphome;
+
+void StreamServerComponent::setup() {
+    ESP_LOGCONFIG(TAG, "Setting up stream server...");
+    this->recv_buf_.reserve(128);
+
+    this->server_ = AsyncServer(this->port_);
+    this->server_.begin();
+    this->server_.onClient([this](void *h, AsyncClient *tcpClient) {
+        if(tcpClient == nullptr)
+            return;
+
+        this->clients_.push_back(std::unique_ptr<Client>(new Client(tcpClient, this->recv_buf_)));
+    }, this);
+}
+
+void StreamServerComponent::loop() {
+    this->cleanup();
+    this->read();
+    this->write();
+}
+
+void StreamServerComponent::cleanup() {
+    auto discriminator = [](std::unique_ptr<Client> &client) { return !client->disconnected; };
+    auto last_client = std::partition(this->clients_.begin(), this->clients_.end(), discriminator);
+    for (auto it = last_client; it != this->clients_.end(); it++)
+        ESP_LOGD(TAG, "Client %s disconnected", (*it)->identifier.c_str());
+
+    this->clients_.erase(last_client, this->clients_.end());
+}
+
+void StreamServerComponent::read() {
+    int len;
+    while ((len = this->stream_->available()) > 0) {
+        char *buf = new char[len];
+        size_t read = this->stream_->readBytes(buf, len);
+        for (auto const& client : this->clients_)
+            client->tcp_client->write(buf, read);
+    }
+}
+
+void StreamServerComponent::write() {
+    size_t len;
+    while ((len = this->recv_buf_.size()) > 0) {
+        this->stream_->write(this->recv_buf_.data(), len);
+        this->recv_buf_.erase(this->recv_buf_.begin(), this->recv_buf_.begin() + len);
+    }
+}
+
+void StreamServerComponent::dump_config() {
+    ESP_LOGCONFIG(TAG, "Stream Server:");
+    ESP_LOGCONFIG(TAG, "  Address: %s:%u", network_get_address().c_str(), this->port_);
+}
+
+void StreamServerComponent::on_shutdown() {
+    for (auto &client : this->clients_)
+        client->tcp_client->close(true);
+}
+
+StreamServerComponent::Client::Client(AsyncClient *client, std::vector<uint8_t> &recv_buf) :
+        tcp_client{client}, identifier{client->remoteIP().toString().c_str()}, disconnected{false} {
+    ESP_LOGD(TAG, "New client connected from %s", this->identifier.c_str());
+
+    this->tcp_client->onError(     [this](void *h, AsyncClient *client, int8_t error)  { this->disconnected = true; });
+    this->tcp_client->onDisconnect([this](void *h, AsyncClient *client)                { this->disconnected = true; });
+    this->tcp_client->onTimeout(   [this](void *h, AsyncClient *client, uint32_t time) { this->disconnected = true; });
+
+    this->tcp_client->onData([&](void *h, AsyncClient *client, void *data, size_t len) {
+        if (len == 0 || data == nullptr)
+            return;
+
+        auto buf = static_cast<uint8_t *>(data);
+        recv_buf.insert(recv_buf.end(), buf, buf + len);
+    }, nullptr);
+}
+
+StreamServerComponent::Client::~Client() {
+    delete this->tcp_client;
+}

stream_server.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include "esphome/core/component.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+#include <Stream.h>
+#include <ESPAsyncTCP.h>
+
+class StreamServerComponent : public esphome::Component {
+public:
+    explicit StreamServerComponent(Stream *stream) : stream_{stream} {}
+
+    void setup() override;
+    void loop() override;
+    void dump_config() override;
+    void on_shutdown() override;
+
+    void set_port(uint16_t port) { this->port_ = port; }
+
+protected:
+    void cleanup();
+    void read();
+    void write();
+
+    struct Client {
+        Client(AsyncClient *client, std::vector<uint8_t> &recv_buf);
+        ~Client();
+
+        AsyncClient *tcp_client{nullptr};
+        std::string identifier{};
+        bool disconnected{false};
+    };
+
+    Stream *stream_{nullptr};
+    AsyncServer server_{0};
+    uint16_t port_{6638};
+    std::vector<uint8_t> recv_buf_{};
+    std::vector<std::unique_ptr<Client>> clients_{};
+};