/* Copyright (C) 2020-2023 Oxan van Leeuwen
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see .
*/
#include "stream_server.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
#include "esphome/core/util.h"
#include "esphome/components/network/util.h"
#include "esphome/components/socket/socket.h"
static const char *TAG = "stream_server";
using namespace esphome;
void StreamServerComponent::setup() {
ESP_LOGCONFIG(TAG, "Setting up stream server...");
struct sockaddr_storage bind_addr;
socklen_t bind_addrlen = socket::set_sockaddr_any(reinterpret_cast(&bind_addr), sizeof(bind_addr), htons(this->port_));
this->socket_ = socket::socket_ip(SOCK_STREAM, PF_INET);
this->socket_->bind(reinterpret_cast(&bind_addr), bind_addrlen);
this->socket_->listen(8);
}
void StreamServerComponent::loop() {
this->accept();
this->read();
this->write();
this->cleanup();
}
void StreamServerComponent::dump_config() {
ESP_LOGCONFIG(TAG, "Stream Server:");
ESP_LOGCONFIG(TAG, " Address: %s:%u", esphome::network::get_ip_address().str().c_str(), this->port_);
}
void StreamServerComponent::on_shutdown() {
for (const Client &client : this->clients_)
client.socket->shutdown(SHUT_RDWR);
}
void StreamServerComponent::accept() {
struct sockaddr_storage client_addr;
socklen_t client_addrlen = sizeof(client_addr);
std::unique_ptr socket = this->socket_->accept(reinterpret_cast(&client_addr), &client_addrlen);
if (!socket)
return;
socket->setblocking(false);
std::string identifier = socket->getpeername();
this->clients_.emplace_back(std::move(socket), identifier);
ESP_LOGD(TAG, "New client connected from %s", identifier.c_str());
}
void StreamServerComponent::cleanup() {
auto discriminator = [](const Client &client) { return !client.disconnected; };
auto last_client = std::partition(this->clients_.begin(), this->clients_.end(), discriminator);
this->clients_.erase(last_client, this->clients_.end());
}
void StreamServerComponent::read() {
int len;
while ((len = this->stream_->available()) > 0) {
char buf[128];
len = std::min(len, 128);
this->stream_->read_array(reinterpret_cast(buf), len);
for (const Client &client : this->clients_)
client.socket->write(buf, len);
}
}
void StreamServerComponent::write() {
uint8_t buf[128];
ssize_t len;
for (Client &client : this->clients_) {
while ((len = client.socket->read(&buf, sizeof(buf))) > 0)
this->stream_->write_array(buf, len);
if (len == 0) {
ESP_LOGD(TAG, "Client %s disconnected", client.identifier.c_str());
client.disconnected = true;
continue;
}
}
}
StreamServerComponent::Client::Client(std::unique_ptr socket, std::string identifier)
: socket(std::move(socket)), identifier{identifier} {}