MARTe-Integrated-Components/Source/Components/DataSources/TCPStreamer/TCPStreamer.cpp

/**
 * @file TCPStreamer.cpp
 * @brief Source file for class TCPStreamer
 * @date 05/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.
 *
 * @details This source file contains the definition of all the methods for
 * the class TCPStreamer (public, protected, and private).
 */

#define DLL_API

#include <string.h>

#include "AdvancedErrorManagement.h"
#include "ConfigurationDatabase.h"
#include "TCPStreamer.h"

namespace MARTe {

/* Owned by the TCPStreamService when Service is set. */
static const char8 *const tcpStreamerServiceKeys[] = {
    "Port", "MaxClients", "ClientQueueDepth", "ClientStallTimeout", "SendBufferSize"
};

static const char8 *const tcpStreamerRejectedKeys[] = {
    "MulticastGroup", "Interface", "DataPort", "ClientTimeout"
};

TCPStreamer::TCPStreamer() : UDPStreamer(), serviceStream(NULL_PTR(TCPSServiceStream *)) {
    tcpServer = new TCPSServer();
    SetServer(tcpServer);
}

TCPStreamer::~TCPStreamer() {
    /* tcpServer is owned (and deleted) by UDPStreamer. */
    tcpServer = NULL_PTR(TCPSServer *);
    serviceStream = NULL_PTR(TCPSServiceStream *);
}

bool TCPStreamer::Initialise(StructuredDataI &data) {
    bool ok = true;
    const uint32 numKeys = data.GetNumberOfChildren();
    for (uint32 i = 0u; (i < numKeys) && ok; i++) {
        const char8 *name = data.GetChildName(i);
        if (name == NULL_PTR(const char8 *)) {
            continue;
        }
        for (uint32 k = 0u; k < (sizeof(tcpStreamerRejectedKeys) / sizeof(tcpStreamerRejectedKeys[0])); k++) {
            if (strcmp(name, tcpStreamerRejectedKeys[k]) == 0) {
                REPORT_ERROR(ErrorManagement::ParametersError,
                             "Parameter '%s' is UDP-only and not supported by TCPStreamer.", name);
                ok = false;
            }
        }
    }
    StreamString serviceName;
    if (ok && data.Read("Service", serviceName) && (serviceName.Size() > 0u)) {
        for (uint32 i = 0u; (i < numKeys) && ok; i++) {
            const char8 *name = data.GetChildName(i);
            for (uint32 k = 0u; (name != NULL_PTR(const char8 *)) &&
                                (k < (sizeof(tcpStreamerServiceKeys) / sizeof(tcpStreamerServiceKeys[0]))); k++) {
                if (strcmp(name, tcpStreamerServiceKeys[k]) == 0) {
                    REPORT_ERROR(ErrorManagement::ParametersError,
                                 "Parameter '%s' belongs to the TCPStreamService and cannot be set together with Service.",
                                 name);
                    ok = false;
                }
            }
        }
        StreamString streamName;
        if (ok && (!data.Read("StreamName", streamName) || (streamName.Size() == 0u))) {
            streamName = GetName();
        }
        if (ok && (streamName.Size() > TCPS_MAX_STREAM_NAME)) {
            REPORT_ERROR(ErrorManagement::ParametersError,
                         "StreamName is longer than %u characters.", TCPS_MAX_STREAM_NAME);
            ok = false;
        }
        if (ok) {
            serviceStream = new TCPSServiceStream(serviceName.Buffer(), streamName.Buffer());
            SetServer(serviceStream); /* replaces (and deletes) the private TCPSServer */
            tcpServer = NULL_PTR(TCPSServer *);
        }
    }
    if (ok) {
        ok = UDPStreamer::Initialise(data);
    }
    return ok;
}

bool TCPStreamer::UsesService() const {
    return serviceStream != NULL_PTR(TCPSServiceStream *);
}

uint16 TCPStreamer::GetBoundPort() const {
    return UsesService() ? serviceStream->GetPort() : tcpServer->GetPort();
}

uint32 TCPStreamer::GetClientCount() const {
    return UsesService() ? serviceStream->GetClientCount() : tcpServer->GetClientCount();
}

uint64 TCPStreamer::GetClientDrops() const {
    return UsesService() ? serviceStream->GetClientDrops() : tcpServer->GetClientDrops();
}

uint32 TCPStreamer::GetStallEvictions() const {
    return UsesService() ? 0u : tcpServer->GetStallEvictions();
}

uint32 TCPStreamer::DefaultMaxPayloadSize() const {
    return TCPSServer::TCPS_DEFAULT_MAX_PAYLOAD;
}

bool TCPStreamer::InitialiseTransport(StructuredDataI &data) {
    if (UsesService()) {
        ConfigurationDatabase scdb;
        bool sok = scdb.Write("MaxPayloadSize", GetMaxPayloadSize());
        if (sok) {
            sok = scdb.MoveToRoot();
        }
        if (sok) {
            sok = serviceStream->Initialise(scdb);
        }
        if (!sok) {
            REPORT_ERROR(ErrorManagement::ParametersError, "Could not configure the service stream.");
        }
        return sok;
    }
    ConfigurationDatabase cdb;
    /* The base class has already resolved Port and MaxPayloadSize. */
    bool ok = cdb.Write("Port", static_cast<uint32>(GetPort()));
    if (ok) {
        ok = cdb.Write("MaxPayloadSize", GetMaxPayloadSize());
    }
    const char8 *const optional[] = {
        "MaxClients", "ClientQueueDepth", "ClientStallTimeout", "SendBufferSize"
    };
    for (uint32 k = 0u; (k < (sizeof(optional) / sizeof(optional[0]))) && ok; k++) {
        uint32 v = 0u;
        if (data.Read(optional[k], v)) {
            ok = cdb.Write(optional[k], v);
        }
    }
    if (ok) {
        ok = cdb.MoveToRoot();
    }
    if (ok) {
        ok = tcpServer->Initialise(cdb);
    }
    if (!ok) {
        REPORT_ERROR(ErrorManagement::ParametersError, "Could not configure the TCPSServer.");
    }
    return ok;
}

CLASS_REGISTER(TCPStreamer, "1.0")

} /* namespace MARTe */