RetiTransportServer.cpp

// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (C) 2026 Dobrev IT Ltd
//
// This file is part of RetiMesh Node.
//
// RetiMesh Node 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.
//
// RetiMesh Node 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 RetiMesh Node. If not, see <https://www.gnu.org/licenses/>.

// ============================================================================
//  RetiTransportServer.cpp — see RetiTransportServer.h for the data flow.
// ============================================================================
#include "RetiTransportServer.h"
#include "RnsTransport.h"

RetiTransportServer transportServer;

// ---------------------------------------------------------------------------
void RetiTransportServer::begin(RingbufHandle_t tcpInRing) {
  _tcpInRing = tcpInRing;
  _lock = xSemaphoreCreateMutex();

  _server = new AsyncServer(RNS_TCP_PORT);
  _server->setNoDelay(true);
  _server->onClient([](void* self, AsyncClient* client) {
    static_cast<RetiTransportServer*>(self)->onClient(client);
  }, this);
  _server->begin();

  log_i("Reticulum transport listening on 0.0.0.0:%d", RNS_TCP_PORT);
}

size_t RetiTransportServer::clientCount() {
  xSemaphoreTake(_lock, portMAX_DELAY);
  size_t n = _clients.size();
  xSemaphoreGive(_lock);
  return n;
}

// ---------------------------------------------------------------------------
// Connection lifecycle (all callbacks run on the AsyncTCP event task).
// ---------------------------------------------------------------------------
void RetiTransportServer::onClient(AsyncClient* client) {
  if (client == nullptr) return;

  if (clientCount() >= RNS_MAX_CLIENTS) {
    log_w("Rejecting %s: client limit (%d) reached",
          client->remoteIP().toString().c_str(), RNS_MAX_CLIENTS);
    client->close();
    return;
  }

  auto* ctx = new ClientCtx{ _nextId++, client, {} };
  client->setNoDelay(true);              // packets are latency-sensitive
  client->setKeepAlive(10000, 3);        // detect vanished phones

  client->onData([](void* arg, AsyncClient*, void* data, size_t len) {
    auto* c = static_cast<ClientCtx*>(arg);
    transportServer.onData(c, (const uint8_t*)data, len);
  }, ctx);

  client->onDisconnect([](void* arg, AsyncClient*) {
    transportServer.onDisconnect(static_cast<ClientCtx*>(arg));
  }, ctx);

  client->onError([](void* arg, AsyncClient*, int8_t error) {
    log_w("TCP client error %d", error);
  }, ctx);

  xSemaphoreTake(_lock, portMAX_DELAY);
  _clients.push_back(ctx);
  g_stats.tcpClients = _clients.size();
  xSemaphoreGive(_lock);

  String ip = client->remoteIP().toString();
  RnsTransport::clientConnected(ctx->id, ip.c_str());
  log_i("Reticulum peer connected: %s (#%lu, %d total)", ip.c_str(),
        (unsigned long)ctx->id, (int)g_stats.tcpClients);
}

void RetiTransportServer::onDisconnect(ClientCtx* ctx) {
  xSemaphoreTake(_lock, portMAX_DELAY);
  for (auto it = _clients.begin(); it != _clients.end(); ++it) {
    if (*it == ctx) { _clients.erase(it); break; }
  }
  g_stats.tcpClients = _clients.size();
  xSemaphoreGive(_lock);

  RnsTransport::clientDisconnected(ctx->id);
  AsyncClient* client = ctx->client;
  delete ctx;
  delete client;                         // server-accepted clients are ours
  log_i("Reticulum peer disconnected (%d left)", (int)g_stats.tcpClients);
}

// ---------------------------------------------------------------------------
// Inbound: TCP bytes -> HDLC deframer -> [client id | packet] -> RNS task.
// Runs on the AsyncTCP task (core 0); it only copies bytes.
// ---------------------------------------------------------------------------
void RetiTransportServer::onData(ClientCtx* ctx, const uint8_t* data, size_t len) {
  for (size_t i = 0; i < len; i++) {
    ctx->deframer.feed(data[i], [this, ctx](const uint8_t* pkt, size_t pktLen) {
      g_stats.tcpRxPackets++;
      uint8_t item[sizeof(RnsTransport::TcpItemHeader) + RNS_MTU];
      RnsTransport::TcpItemHeader h{ ctx->id };
      memcpy(item, &h, sizeof(h));
      memcpy(item + sizeof(h), pkt, pktLen);
      // Drop rather than back-pressure the socket: a stalled AsyncTCP task
      // takes the web server down with it, and Reticulum tolerates loss.
      if (xRingbufferSend(_tcpInRing, item, sizeof(h) + pktLen, pdMS_TO_TICKS(20)) != pdTRUE)
        log_w("TCP-in ring full, dropping %u-byte packet", (unsigned)pktLen);
    });
  }
}

// ---------------------------------------------------------------------------
// Outbound: one packet to one client. Called from the RNS task; _lock
// serialises the client list and the framing scratch buffer.
// ---------------------------------------------------------------------------
bool RetiTransportServer::sendTo(uint32_t clientId, const uint8_t* packet, size_t len) {
  bool ok = false;
  xSemaphoreTake(_lock, portMAX_DELAY);
  size_t frameLen = HDLC::frame(packet, len, _frameBuf, sizeof(_frameBuf));
  if (frameLen > 0) {
    for (auto* ctx : _clients) {
      if (ctx->id != clientId) continue;
      AsyncClient* c = ctx->client;
      // Slow-consumer policy: if the socket's window can't take the whole
      // frame right now, drop it for that client instead of blocking.
      if (c->connected() && c->canSend() && c->space() >= frameLen) {
        c->write((const char*)_frameBuf, frameLen);
        ok = true;
      }
      break;
    }
  }
  xSemaphoreGive(_lock);
  return ok;
}