/**
 * @file TCPStreamService.cpp
 * @brief Source file for class TCPStreamService
 * @date 06/11/2026
 * @author Martino Ferrari
 *
 * @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and
 * the Development of Fusion Energy ('Fusion for Energy').
 * Licensed under the EUPL, Version 1.1 or - as soon they will be approved
 * by the European Commission - subsequent versions of the EUPL (the "Licence")
 * You may not use this work except in compliance with the Licence.
 * You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl
 *
 * @warning Unless required by applicable law or agreed to in writing,
 * software distributed under the Licence is distributed on an "AS IS"
 * basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the Licence permissions and limitations under the Licence.
 */

#define DLL_API

#include "TCPStreamService.h"

#include "AdvancedErrorManagement.h"
#include "Atomic.h"
#include "HighResolutionTimer.h"
#include "TCPSConn.h"
#include "TCPSServiceStream.h"
#include "TCPSWorker.h"

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

namespace MARTe {

static const int TCPS_ACCEPT_SLICE_MS = 50;

TCPStreamService::TCPStreamService()
    : Object(), EmbeddedServiceMethodBinderI(), port(44500u), boundPort(0u),
      numWorkers(2u), maxClients(64u), maxStreams(32u), queueDepth(8u),
      stallTimeoutS(5u), handshakeTimeoutS(5u), sendBufferSize(0u),
      cpuMask(0xFFFFFFFFu), stackSize(65536u), stallTicks(0u), handshakeTicks(0u),
      listenFd(-1), acceptThread(*this), acceptorStarted(false),
      workers(NULL_PTR(TCPSWorker **)), workersCreated(0u),
      numObservers(0u),
      streams(NULL_PTR(TCPSServiceStream **)), numStreams(0u),
      connCount(0), evictions(0), rejected(0) {
    (void)notifyMutex.Create(true);
    for (uint32 i = 0u; i < TCPS_MAX_OBSERVERS; i++) {
        observers[i] = NULL_PTR(TCPSStreamObserverI *);
    }
}

TCPStreamService::~TCPStreamService() {
    Shutdown();
}

void TCPStreamService::Shutdown() {
    if (acceptorStarted) {
        (void)acceptThread.Stop();
        acceptorStarted = false;
    }
    for (uint32 i = 0u; i < workersCreated; i++) {
        workers[i]->Stop();
    }
    if (listenFd >= 0) {
        (void)close(listenFd);
        listenFd = -1;
    }
    for (uint32 i = 0u; i < workersCreated; i++) {
        delete workers[i];
    }
    workersCreated = 0u;
    if (workers != NULL_PTR(TCPSWorker **)) {
        delete[] workers;
        workers = NULL_PTR(TCPSWorker **);
    }
    if (streams != NULL_PTR(TCPSServiceStream **)) {
        delete[] streams;
        streams = NULL_PTR(TCPSServiceStream **);
    }
    numStreams = 0u;
}

bool TCPStreamService::Initialise(StructuredDataI &data) {
    bool ok = Object::Initialise(data);

    port = 44500u;
    (void)data.Read("Port", port);
    if (port > 65535u) {
        REPORT_ERROR(ErrorManagement::ParametersError, "Port %u out of range.", port);
        ok = false;
    }
    numWorkers = 2u;
    (void)data.Read("Workers", numWorkers);
    if ((numWorkers == 0u) || (numWorkers > 32u)) {
        REPORT_ERROR(ErrorManagement::ParametersError, "Workers %u must be in 1..32.", numWorkers);
        ok = false;
    }
    maxClients = 64u;
    (void)data.Read("MaxClients", maxClients);
    if ((maxClients == 0u) || (maxClients > 1024u)) {
        REPORT_ERROR(ErrorManagement::ParametersError, "MaxClients %u must be in 1..1024.", maxClients);
        ok = false;
    }
    maxStreams = 32u;
    (void)data.Read("MaxStreams", maxStreams);
    if ((maxStreams == 0u) || (maxStreams > 256u)) {
        REPORT_ERROR(ErrorManagement::ParametersError, "MaxStreams %u must be in 1..256.", maxStreams);
        ok = false;
    }
    queueDepth = 8u;
    (void)data.Read("ClientQueueDepth", queueDepth);
    if ((queueDepth == 0u) || (queueDepth > 4096u)) {
        REPORT_ERROR(ErrorManagement::ParametersError, "ClientQueueDepth %u must be in 1..4096.", queueDepth);
        ok = false;
    }
    stallTimeoutS = 5u;
    (void)data.Read("ClientStallTimeout", stallTimeoutS);
    handshakeTimeoutS = 5u;
    (void)data.Read("HandshakeTimeout", handshakeTimeoutS);
    if (handshakeTimeoutS == 0u) {
        REPORT_ERROR(ErrorManagement::ParametersError, "HandshakeTimeout must be at least 1 second.");
        ok = false;
    }
    sendBufferSize = 0u;
    (void)data.Read("SendBufferSize", sendBufferSize);
    (void)data.Read("CPUMask", cpuMask);
    (void)data.Read("StackSize", stackSize);
    stallTicks = static_cast<uint64>(stallTimeoutS) * HighResolutionTimer::Frequency();
    handshakeTicks = static_cast<uint64>(handshakeTimeoutS) * HighResolutionTimer::Frequency();

    if (!ok) {
        return false;
    }

    /* Listening socket. */
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    ok = (fd >= 0);
    if (!ok) {
        REPORT_ERROR(ErrorManagement::FatalError, "socket() failed: %s.", strerror(errno));
    }
    if (ok) {
        int one = 1;
        (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
        struct sockaddr_in addr;
        memset(&addr, 0, sizeof(addr));
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = htonl(INADDR_ANY);
        addr.sin_port = htons(static_cast<uint16>(port));
        if (bind(fd, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr)) != 0) {
            REPORT_ERROR(ErrorManagement::FatalError, "bind(%u) failed: %s.", port, strerror(errno));
            ok = false;
        }
    }
    if (ok && (listen(fd, 32) != 0)) {
        REPORT_ERROR(ErrorManagement::FatalError, "listen() failed: %s.", strerror(errno));
        ok = false;
    }
    if (ok) {
        int flags = fcntl(fd, F_GETFL, 0);
        ok = (flags >= 0) && (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
    }
    if (ok) {
        struct sockaddr_in addr;
        socklen_t len = static_cast<socklen_t>(sizeof(addr));
        memset(&addr, 0, sizeof(addr));
        boundPort = static_cast<uint16>(port);
        if (getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &len) == 0) {
            boundPort = ntohs(addr.sin_port);
        }
    }
    if (!ok) {
        if (fd >= 0) {
            (void)close(fd);
        }
        return false;
    }
    listenFd = fd;

    streams = new TCPSServiceStream *[maxStreams];
    workers = new TCPSWorker *[numWorkers];
    for (uint32 i = 0u; (i < numWorkers) && ok; i++) {
        workers[i] = new TCPSWorker(*this, maxClients);
        workersCreated = i + 1u;
        ok = workers[i]->Start(cpuMask, stackSize);
        if (!ok) {
            REPORT_ERROR(ErrorManagement::FatalError, "Could not start worker %u.", i);
        }
    }
    if (ok) {
        acceptThread.SetCPUMask(cpuMask);
        acceptThread.SetStackSize(stackSize);
        ok = (acceptThread.Start() == ErrorManagement::NoError);
        acceptorStarted = ok;
        if (!ok) {
            REPORT_ERROR(ErrorManagement::FatalError, "Could not start the acceptor thread.");
        }
    }
    if (!ok) {
        Shutdown();
    }
    return ok;
}

ErrorManagement::ErrorType TCPStreamService::Execute(ExecutionInfo &info) {
    if (info.GetStage() != ExecutionInfo::MainStage) {
        return ErrorManagement::NoError;
    }
    struct pollfd p;
    p.fd = listenFd;
    p.events = POLLIN;
    p.revents = 0;
    if (poll(&p, 1, TCPS_ACCEPT_SLICE_MS) <= 0) {
        return ErrorManagement::NoError;
    }
    for (;;) {
        int fd = accept(listenFd, NULL_PTR(struct sockaddr *), NULL_PTR(socklen_t *));
        if (fd < 0) {
            break;
        }
        if (static_cast<uint32>(connCount) >= maxClients) {
            REPORT_ERROR(ErrorManagement::Warning,
                         "MaxClients (%u) reached; connection refused.", maxClients);
            (void)close(fd);
            continue;
        }
        int flags = fcntl(fd, F_GETFL, 0);
        if ((flags < 0) || (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)) {
            (void)close(fd);
            continue;
        }
        int one = 1;
        (void)setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
        if (sendBufferSize > 0u) {
            int sz = static_cast<int>(sendBufferSize);
            (void)setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz));
        }

        TCPSWorker *best = workers[0];
        for (uint32 i = 1u; i < workersCreated; i++) {
            if (workers[i]->Load() < best->Load()) {
                best = workers[i];
            }
        }
        TCPSConn *c = new TCPSConn(fd, best, queueDepth,
                                   HighResolutionTimer::Counter() + handshakeTicks);
        Atomic::Increment(&connCount);
        if (!best->AddIncoming(c)) {
            delete c;
            Atomic::Decrement(&connCount);
        }
    }
    return ErrorManagement::NoError;
}

uint16 TCPStreamService::GetBoundPort() const {
    return boundPort;
}

uint32 TCPStreamService::GetClientCount() const {
    uint32 n = 0u;
    topology.FastLock();
    for (uint32 i = 0u; i < numStreams; i++) {
        n += streams[i]->GetClientCount();
    }
    topology.FastUnLock();
    return n;
}

uint32 TCPStreamService::GetStreamCount() const {
    return numStreams;
}

uint32 TCPStreamService::GetWorkerCount() const {
    return workersCreated;
}

uint64 TCPStreamService::GetClientDrops() const {
    uint64 n = 0u;
    topology.FastLock();
    for (uint32 i = 0u; i < numStreams; i++) {
        n += streams[i]->GetClientDrops();
    }
    topology.FastUnLock();
    return n;
}

uint32 TCPStreamService::GetStallEvictions() const {
    return static_cast<uint32>(evictions);
}

uint32 TCPStreamService::GetRejectedSubscriptions() const {
    return static_cast<uint32>(rejected);
}

bool TCPStreamService::RegisterStream(TCPSServiceStream *s) {
    bool ok = true;
    // Held across the registration so observers see (un)registrations in order.
    (void)notifyMutex.Lock();
    topology.FastLock();
    if (numStreams >= maxStreams) {
        REPORT_ERROR(ErrorManagement::ParametersError,
                     "Cannot register stream '%s': MaxStreams (%u) reached.",
                     s->GetStreamName(), maxStreams);
        ok = false;
    }
    for (uint32 i = 0u; (i < numStreams) && ok; i++) {
        if (strcmp(streams[i]->GetStreamName(), s->GetStreamName()) == 0) {
            REPORT_ERROR(ErrorManagement::ParametersError,
                         "A stream named '%s' is already registered.", s->GetStreamName());
            ok = false;
        }
    }
    if (ok) {
        streams[numStreams] = s;
        numStreams++;
    }
    topology.FastUnLock();
    if (ok) {
        for (uint32 i = 0u; i < numObservers; i++) {
            observers[i]->OnStreamRegistered(s);
        }
    }
    (void)notifyMutex.UnLock();
    return ok;
}

void TCPStreamService::UnregisterStream(TCPSServiceStream *s) {
    (void)notifyMutex.Lock();
    bool wasRegistered = false;
    topology.FastLock();
    for (uint32 i = 0u; i < numStreams; i++) {
        if (streams[i] == s) {
            streams[i] = streams[numStreams - 1u];
            numStreams--;
            wasRegistered = true;
            break;
        }
    }
    s->DropAllConns();
    topology.FastUnLock();
    if (wasRegistered) {
        for (uint32 i = 0u; i < numObservers; i++) {
            observers[i]->OnStreamUnregistered(s);
        }
    }
    (void)notifyMutex.UnLock();
}

bool TCPStreamService::AddObserver(TCPSStreamObserverI *obs) {
    bool ok = false;
    (void)notifyMutex.Lock();
    if (numObservers < TCPS_MAX_OBSERVERS) {
        observers[numObservers] = obs;
        numObservers++;
        ok = true;
        // Replay: streams cannot (un)register while notifyMutex is held.
        topology.FastLock();
        const uint32 n = numStreams;
        TCPSServiceStream **snap = new TCPSServiceStream *[(n > 0u) ? n : 1u];
        for (uint32 i = 0u; i < n; i++) {
            snap[i] = streams[i];
        }
        topology.FastUnLock();
        for (uint32 i = 0u; i < n; i++) {
            obs->OnStreamRegistered(snap[i]);
        }
        delete[] snap;
    }
    (void)notifyMutex.UnLock();
    return ok;
}

void TCPStreamService::RemoveObserver(TCPSStreamObserverI *obs) {
    (void)notifyMutex.Lock();
    for (uint32 i = 0u; i < numObservers; i++) {
        if (observers[i] == obs) {
            observers[i] = observers[numObservers - 1u];
            numObservers--;
            break;
        }
    }
    (void)notifyMutex.UnLock();
}

bool TCPStreamService::Attach(TCPSConn *c, char8 *streamList) {
    TCPSServiceStream *found[TCPS_MAX_CONN_STREAMS];
    uint32 n = 0u;
    bool ok = true;

    topology.FastLock();
    char8 *p = streamList;
    while (ok) {
        char8 *comma = strchr(p, ',');
        if (comma != NULL_PTR(char8 *)) {
            *comma = '\0';
        }
        const uint32 len = static_cast<uint32>(strlen(p));
        TCPSServiceStream *match = NULL_PTR(TCPSServiceStream *);
        if ((len == 0u) || (len > TCPS_MAX_STREAM_NAME) || (n >= TCPS_MAX_CONN_STREAMS)) {
            ok = false;
        }
        for (uint32 i = 0u; ok && (i < numStreams) && (match == NULL_PTR(TCPSServiceStream *)); i++) {
            if (strcmp(streams[i]->GetStreamName(), p) == 0) {
                match = streams[i];
            }
        }
        for (uint32 i = 0u; ok && (i < n); i++) {
            if (found[i] == match) {
                ok = false; // the same stream listed twice
            }
        }
        if (ok && (match == NULL_PTR(TCPSServiceStream *))) {
            ok = false;
        }
        if (ok) {
            found[n] = match;
            n++;
            if (comma == NULL_PTR(char8 *)) {
                break;
            }
            p = comma + 1;
        }
    }

    if (ok) {
        c->qMutex.FastLock();
        c->q.Reserve(n);
        c->qMutex.FastUnLock();
        uint32 attached = 0u;
        for (uint32 i = 0u; ok && (i < n); i++) {
            ok = found[i]->AttachConn(c, static_cast<uint16>(i));
            if (ok) {
                attached++;
            }
        }
        if (!ok) {
            for (uint32 i = 0u; i < attached; i++) {
                found[i]->DetachConn(c);
            }
        } else {
            for (uint32 i = 0u; i < n; i++) {
                c->streams[i] = found[i];
            }
            c->numStreams = n;
        }
    }
    topology.FastUnLock();
    return ok;
}

void TCPStreamService::Detach(TCPSConn *c) {
    topology.FastLock();
    for (uint32 i = 0u; i < c->numStreams; i++) {
        if (c->streams[i] != NULL_PTR(TCPSServiceStream *)) {
            c->streams[i]->DetachConn(c);
            c->streams[i] = NULL_PTR(TCPSServiceStream *);
        }
    }
    topology.FastUnLock();
}

void TCPStreamService::ConnClosed() {
    Atomic::Decrement(&connCount);
}

void TCPStreamService::NoteEviction() {
    Atomic::Increment(&evictions);
}

void TCPStreamService::NoteRejected() {
    Atomic::Increment(&rejected);
}

uint32 TCPStreamService::QueueDepth() const {
    return queueDepth;
}

uint64 TCPStreamService::StallTicks() const {
    return stallTicks;
}

uint64 TCPStreamService::HandshakeTicks() const {
    return handshakeTicks;
}

uint32 TCPStreamService::MaxClients() const {
    return maxClients;
}

CLASS_REGISTER(TCPStreamService, "1.0")

} // namespace MARTe