<commit_before>// Copyright (c) 2019, 2021 by <NAME> GmbH. All rights reserved.
// Copyright (c) 2021 by Apex.AI Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

#include "iceoryx_posh/internal/roudi/roudi.hpp"
#include "iceoryx_posh/internal/log/posh_logging.hpp"
#include "iceoryx_posh/internal/runtime/node_property.hpp"
#include "iceoryx_posh/popo/subscriber_options.hpp"
#include "iceoryx_posh/roudi/introspection_types.hpp"
#include "iceoryx_posh/roudi/memory/roudi_memory_manager.hpp"
#include "iceoryx_posh/runtime/port_config_info.hpp"
#include "iceoryx_utils/cxx/convert.hpp"
#include "iceoryx_utils/cxx/helplets.hpp"
#include "iceoryx_utils/posix_wrapper/thread.hpp"

namespace iox
{
namespace roudi
{
RouDi::RouDi(RouDiMemoryInterface& roudiMemoryInterface,
             PortManager& portManager,
             RoudiStartupParameters roudiStartupParameters)
    : m_killProcessesInDestructor(roudiStartupParameters.m_killProcessesInDestructor)
    , m_runMonitoringAndDiscoveryThread(true)
    , m_runHandleRuntimeMessageThread(true)
    , m_roudiMemoryInterface(&roudiMemoryInterface)
    , m_portManager(&portManager)
    , m_prcMgr(concurrent::ForwardArgsToCTor,
               *m_roudiMemoryInterface,
               portManager,
               roudiStartupParameters.m_compatibilityCheckLevel)
    , m_mempoolIntrospection(*m_roudiMemoryInterface->introspectionMemoryManager()
                                  .value(), /// @todo create a RouDiMemoryManagerData struct with all the pointer
                             *m_roudiMemoryInterface->segmentManager().value(),
                             PublisherPortUserType(m_prcMgr->addIntrospectionPublisherPort(IntrospectionMempoolService,
                                                                                           IPC_CHANNEL_ROUDI_NAME)))
    , m_monitoringMode(roudiStartupParameters.m_monitoringMode)
    , m_processKillDelay(roudiStartupParameters.m_processKillDelay)
{
    if (cxx::isCompiledOn32BitSystem())
    {
        LogWarn() << "Runnning RouDi on 32-bit architectures is not supported! Use at your own risk!";
    }
    m_processIntrospection.registerPublisherPort(PublisherPortUserType(
        m_prcMgr->addIntrospectionPublisherPort(IntrospectionProcessService, IPC_CHANNEL_ROUDI_NAME)));
    m_prcMgr->initIntrospection(&m_processIntrospection);
    m_processIntrospection.run();
    m_mempoolIntrospection.run();

    // since RouDi offers the introspection services, also add it to the list of processes
    m_processIntrospection.addProcess(getpid(), IPC_CHANNEL_ROUDI_NAME);

    // run the threads
    m_monitoringAndDiscoveryThread = std::thread(&RouDi::monitorAndDiscoveryUpdate, this);
    posix::setThreadName(m_monitoringAndDiscoveryThread.native_handle(), "Mon+Discover");

    if (roudiStartupParameters.m_runtimesMessagesThreadStart == RuntimeMessagesThreadStart::IMMEDIATE)
    {
        startProcessRuntimeMessagesThread();
    }
}

RouDi::~RouDi()
{
    shutdown();
}

void RouDi::startProcessRuntimeMessagesThread()
{
    m_handleRuntimeMessageThread = std::thread(&RouDi::processRuntimeMessages, this);
    posix::setThreadName(m_handleRuntimeMessageThread.native_handle(), "IPC-msg-process");
}

void RouDi::shutdown()
{
    m_processIntrospection.stop();
    m_portManager->stopPortIntrospection();

    // stop the process management thread in order to prevent application to register while shutting down
    m_runMonitoringAndDiscoveryThread = false;
    if (m_monitoringAndDiscoveryThread.joinable())
    {
        LogDebug() << "Joining 'Mon+Discover' thread...";
        m_monitoringAndDiscoveryThread.join();
        LogDebug() << "...'Mon+Discover' thread joined.";
    }

    if (m_killProcessesInDestructor)
    {
        cxx::DeadlineTimer finalKillTimer(m_processKillDelay);

        m_prcMgr->requestShutdownOfAllProcesses();

        using namespace units::duration_literals;
        auto remainingDurationForWarnPrint = m_processKillDelay - 2_s;
        while (m_prcMgr->isAnyRegisteredProcessStillRunning() && !finalKillTimer.hasExpired())
        {
            if (remainingDurationForWarnPrint > finalKillTimer.remainingTime())
            {
                LogWarn() << "Some applications seem to not shutdown gracefully! Time until hard shutdown: "
                          << finalKillTimer.remainingTime().toSeconds() << "s!";
                remainingDurationForWarnPrint = remainingDurationForWarnPrint - 5_s;
            }
            // give processes some time to terminate
            std::this_thread::sleep_for(std::chrono::milliseconds(PROCESS_TERMINATED_CHECK_INTERVAL.toMilliseconds()));
        }

        // Is any processes still alive?
        if (m_prcMgr->isAnyRegisteredProcessStillRunning() && finalKillTimer.hasExpired())
        {
            // Time to kill them
            m_prcMgr->killAllProcesses();
        }

        if (m_prcMgr->isAnyRegisteredProcessStillRunning())
        {
            m_prcMgr->printWarningForRegisteredProcessesAndClearProcessList();
        }
    }

    // Postpone the IpcChannelThread in order to receive TERMINATION
    m_runHandleRuntimeMessageThread = false;

    if (m_handleRuntimeMessageThread.joinable())
    {
        LogDebug() << "Joining 'IPC-msg-process' thread...";
        m_handleRuntimeMessageThread.join();
        LogDebug() << "...'IPC-msg-process' thread joined.";
    }
}

void RouDi::cyclicUpdateHook()
{
    // default implementation; do nothing
}

void RouDi::monitorAndDiscoveryUpdate()
{
    while (m_runMonitoringAndDiscoveryThread)
    {
        m_prcMgr->run();

        cyclicUpdateHook();

        std::this_thread::sleep_for(std::chrono::milliseconds(DISCOVERY_INTERVAL.toMilliseconds()));
    }
}

void RouDi::processRuntimeMessages()
{
    runtime::IpcInterfaceCreator roudiIpcInterface{IPC_CHANNEL_ROUDI_NAME};

    // the logger is intentionally not used, to ensure that this message is always printed
    std::cout << "RouDi is ready for clients" << std::endl;

    while (m_runHandleRuntimeMessageThread)
    {
        // read RouDi's IPC channel
        runtime::IpcMessage message;
        if (roudiIpcInterface.timedReceive(m_runtimeMessagesThreadTimeout, message))
        {
            auto cmd = runtime::stringToIpcMessageType(message.getElementAtIndex(0).c_str());
            std::string runtimeName = message.getElementAtIndex(1);

            processMessage(message, cmd, RuntimeName_t(cxx::TruncateToCapacity, runtimeName));
        }
    }
}

version::VersionInfo RouDi::parseRegisterMessage(const runtime::IpcMessage& message,
                                                 uint32_t& pid,
                                                 uid_t& userId,
                                                 int64_t& transmissionTimestamp)
{
    cxx::convert::fromString(message.getElementAtIndex(2).c_str(), pid);
    cxx::convert::fromString(message.getElementAtIndex(3).c_str(), userId);
    cxx::convert::fromString(message.getElementAtIndex(4).c_str(), transmissionTimestamp);
    cxx::Serialization serializationVersionInfo(message.getElementAtIndex(5));
    return serializationVersionInfo;
}

void RouDi::processMessage(const runtime::IpcMessage& message,
                           const iox::runtime::IpcMessageType& cmd,
                           const RuntimeName_t& runtimeName)
{
    switch (cmd)
    {
    case runtime::IpcMessageType::SERVICE_REGISTRY_CHANGE_COUNTER:
    {
        m_prcMgr->sendServiceRegistryChangeCounterToProcess(runtimeName);
        break;
    }
    case runtime::IpcMessageType::REG:
    {
        if (message.getNumberOfElements() != 6)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::REG\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            uint32_t pid;
            uid_t userId;
            int64_t transmissionTimestamp;
            version::VersionInfo versionInfo = parseRegisterMessage(message, pid, userId, transmissionTimestamp);

            registerProcess(
                runtimeName, pid, {userId}, transmissionTimestamp, getUniqueSessionIdForProcess(), versionInfo);
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_PUBLISHER:
    {
        if (message.getNumberOfElements() != 8)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_PUBLISHER\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            capro::ServiceDescription service(cxx::Serialization(message.getElementAtIndex(2)));
            cxx::Serialization portConfigInfoSerialization(message.getElementAtIndex(7));

            popo::PublisherOptions options;
            uint64_t historyCapacity{};
            if (!cxx::convert::fromString(message.getElementAtIndex(3).c_str(), historyCapacity))
            {
                LogError() << message.getElementAtIndex(3).c_str() << "cannot be extracted from string\n";
            }
            options.historyCapacity = historyCapacity;
            options.nodeName = NodeName_t(cxx::TruncateToCapacity, message.getElementAtIndex(4));

            uint64_t offerOnCreate{};
            if (!cxx::convert::fromString(message.getElementAtIndex(5).c_str(), offerOnCreate))
            {
                LogError() << message.getElementAtIndex(5).c_str() << "cannot be extracted from string\n";
            }
            options.offerOnCreate = (0U == offerOnCreate) ? false : true;

            uint8_t subscriberTooSlowPolicy{};
            if (!cxx::convert::fromString(message.getElementAtIndex(6).c_str(), subscriberTooSlowPolicy))
            {
                LogError() << message.getElementAtIndex(6).c_str() << "cannot be extracted from string\n";
            }
            options.subscriberTooSlowPolicy = static_cast<popo::SubscriberTooSlowPolicy>(subscriberTooSlowPolicy);

            m_prcMgr->addPublisherForProcess(
                runtimeName, service, options, iox::runtime::PortConfigInfo(portConfigInfoSerialization));
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_SUBSCRIBER:
    {
        if (message.getNumberOfElements() != 9)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_SUBSCRIBER\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            capro::ServiceDescription service(cxx::Serialization(message.getElementAtIndex(2)));
            cxx::Serialization portConfigInfoSerialization(message.getElementAtIndex(8));


            popo::SubscriberOptions options;
            uint64_t historyRequest;
            if (!cxx::convert::fromString(message.getElementAtIndex(3).c_str(), historyRequest))
            {
                LogError() << message.getElementAtIndex(3).c_str() << "cannot be extracted from string\n";
            }
            options.historyRequest = historyRequest;
            uint64_t queueCapacity;
            if (!cxx::convert::fromString(message.getElementAtIndex(4).c_str(), queueCapacity))
            {
                LogError() << message.getElementAtIndex(4).c_str() << "cannot be extracted from string\n";
            }
            options.queueCapacity = queueCapacity;
            options.nodeName = NodeName_t(cxx::TruncateToCapacity, message.getElementAtIndex(5));

            uint32_t subscribeOnCreate;
            if (!cxx::convert::fromString(message.getElementAtIndex(6).c_str(), subscribeOnCreate))
            {
                LogError() << message.getElementAtIndex(6).c_str() << "cannot be extracted from string\n";
            }
            options.subscribeOnCreate = (0U == subscribeOnCreate ? false : true);

            uint8_t queueFullPolicy{};
            if (!cxx::convert::fromString(message.getElementAtIndex(7).c_str(), queueFullPolicy))
            {
                LogError() << message.getElementAtIndex(7).c_str() << "cannot be extracted from string\n";
            }
            options.queueFullPolicy = static_cast<popo::QueueFullPolicy>(queueFullPolicy);

            m_prcMgr->addSubscriberForProcess(
                runtimeName, service, options, iox::runtime::PortConfigInfo(portConfigInfoSerialization));
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_CONDITION_VARIABLE:
    {
        if (message.getNumberOfElements() != 2)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_CONDITION_VARIABLE\" from \""
                       << runtimeName << "\"received!";
        }
        else
        {
            m_prcMgr->addConditionVariableForProcess(runtimeName);
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_INTERFACE:
    {<commit_msg>iox-#260 introduced break statement to return from method when cxx::convert::fromString fails<commit_after>// Copyright (c) 2019, 2021 by <NAME>. All rights reserved.
// Copyright (c) 2021 by Apex.AI Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

#include "iceoryx_posh/internal/roudi/roudi.hpp"
#include "iceoryx_posh/internal/log/posh_logging.hpp"
#include "iceoryx_posh/internal/runtime/node_property.hpp"
#include "iceoryx_posh/popo/subscriber_options.hpp"
#include "iceoryx_posh/roudi/introspection_types.hpp"
#include "iceoryx_posh/roudi/memory/roudi_memory_manager.hpp"
#include "iceoryx_posh/runtime/port_config_info.hpp"
#include "iceoryx_utils/cxx/convert.hpp"
#include "iceoryx_utils/cxx/helplets.hpp"
#include "iceoryx_utils/posix_wrapper/thread.hpp"

namespace iox
{
namespace roudi
{
RouDi::RouDi(RouDiMemoryInterface& roudiMemoryInterface,
             PortManager& portManager,
             RoudiStartupParameters roudiStartupParameters)
    : m_killProcessesInDestructor(roudiStartupParameters.m_killProcessesInDestructor)
    , m_runMonitoringAndDiscoveryThread(true)
    , m_runHandleRuntimeMessageThread(true)
    , m_roudiMemoryInterface(&roudiMemoryInterface)
    , m_portManager(&portManager)
    , m_prcMgr(concurrent::ForwardArgsToCTor,
               *m_roudiMemoryInterface,
               portManager,
               roudiStartupParameters.m_compatibilityCheckLevel)
    , m_mempoolIntrospection(*m_roudiMemoryInterface->introspectionMemoryManager()
                                  .value(), /// @todo create a RouDiMemoryManagerData struct with all the pointer
                             *m_roudiMemoryInterface->segmentManager().value(),
                             PublisherPortUserType(m_prcMgr->addIntrospectionPublisherPort(IntrospectionMempoolService,
                                                                                           IPC_CHANNEL_ROUDI_NAME)))
    , m_monitoringMode(roudiStartupParameters.m_monitoringMode)
    , m_processKillDelay(roudiStartupParameters.m_processKillDelay)
{
    if (cxx::isCompiledOn32BitSystem())
    {
        LogWarn() << "Runnning RouDi on 32-bit architectures is not supported! Use at your own risk!";
    }
    m_processIntrospection.registerPublisherPort(PublisherPortUserType(
        m_prcMgr->addIntrospectionPublisherPort(IntrospectionProcessService, IPC_CHANNEL_ROUDI_NAME)));
    m_prcMgr->initIntrospection(&m_processIntrospection);
    m_processIntrospection.run();
    m_mempoolIntrospection.run();

    // since RouDi offers the introspection services, also add it to the list of processes
    m_processIntrospection.addProcess(getpid(), IPC_CHANNEL_ROUDI_NAME);

    // run the threads
    m_monitoringAndDiscoveryThread = std::thread(&RouDi::monitorAndDiscoveryUpdate, this);
    posix::setThreadName(m_monitoringAndDiscoveryThread.native_handle(), "Mon+Discover");

    if (roudiStartupParameters.m_runtimesMessagesThreadStart == RuntimeMessagesThreadStart::IMMEDIATE)
    {
        startProcessRuntimeMessagesThread();
    }
}

RouDi::~RouDi()
{
    shutdown();
}

void RouDi::startProcessRuntimeMessagesThread()
{
    m_handleRuntimeMessageThread = std::thread(&RouDi::processRuntimeMessages, this);
    posix::setThreadName(m_handleRuntimeMessageThread.native_handle(), "IPC-msg-process");
}

void RouDi::shutdown()
{
    m_processIntrospection.stop();
    m_portManager->stopPortIntrospection();

    // stop the process management thread in order to prevent application to register while shutting down
    m_runMonitoringAndDiscoveryThread = false;
    if (m_monitoringAndDiscoveryThread.joinable())
    {
        LogDebug() << "Joining 'Mon+Discover' thread...";
        m_monitoringAndDiscoveryThread.join();
        LogDebug() << "...'Mon+Discover' thread joined.";
    }

    if (m_killProcessesInDestructor)
    {
        cxx::DeadlineTimer finalKillTimer(m_processKillDelay);

        m_prcMgr->requestShutdownOfAllProcesses();

        using namespace units::duration_literals;
        auto remainingDurationForWarnPrint = m_processKillDelay - 2_s;
        while (m_prcMgr->isAnyRegisteredProcessStillRunning() && !finalKillTimer.hasExpired())
        {
            if (remainingDurationForWarnPrint > finalKillTimer.remainingTime())
            {
                LogWarn() << "Some applications seem to not shutdown gracefully! Time until hard shutdown: "
                          << finalKillTimer.remainingTime().toSeconds() << "s!";
                remainingDurationForWarnPrint = remainingDurationForWarnPrint - 5_s;
            }
            // give processes some time to terminate
            std::this_thread::sleep_for(std::chrono::milliseconds(PROCESS_TERMINATED_CHECK_INTERVAL.toMilliseconds()));
        }

        // Is any processes still alive?
        if (m_prcMgr->isAnyRegisteredProcessStillRunning() && finalKillTimer.hasExpired())
        {
            // Time to kill them
            m_prcMgr->killAllProcesses();
        }

        if (m_prcMgr->isAnyRegisteredProcessStillRunning())
        {
            m_prcMgr->printWarningForRegisteredProcessesAndClearProcessList();
        }
    }

    // Postpone the IpcChannelThread in order to receive TERMINATION
    m_runHandleRuntimeMessageThread = false;

    if (m_handleRuntimeMessageThread.joinable())
    {
        LogDebug() << "Joining 'IPC-msg-process' thread...";
        m_handleRuntimeMessageThread.join();
        LogDebug() << "...'IPC-msg-process' thread joined.";
    }
}

void RouDi::cyclicUpdateHook()
{
    // default implementation; do nothing
}

void RouDi::monitorAndDiscoveryUpdate()
{
    while (m_runMonitoringAndDiscoveryThread)
    {
        m_prcMgr->run();

        cyclicUpdateHook();

        std::this_thread::sleep_for(std::chrono::milliseconds(DISCOVERY_INTERVAL.toMilliseconds()));
    }
}

void RouDi::processRuntimeMessages()
{
    runtime::IpcInterfaceCreator roudiIpcInterface{IPC_CHANNEL_ROUDI_NAME};

    // the logger is intentionally not used, to ensure that this message is always printed
    std::cout << "RouDi is ready for clients" << std::endl;

    while (m_runHandleRuntimeMessageThread)
    {
        // read RouDi's IPC channel
        runtime::IpcMessage message;
        if (roudiIpcInterface.timedReceive(m_runtimeMessagesThreadTimeout, message))
        {
            auto cmd = runtime::stringToIpcMessageType(message.getElementAtIndex(0).c_str());
            std::string runtimeName = message.getElementAtIndex(1);

            processMessage(message, cmd, RuntimeName_t(cxx::TruncateToCapacity, runtimeName));
        }
    }
}

version::VersionInfo RouDi::parseRegisterMessage(const runtime::IpcMessage& message,
                                                 uint32_t& pid,
                                                 uid_t& userId,
                                                 int64_t& transmissionTimestamp)
{
    cxx::convert::fromString(message.getElementAtIndex(2).c_str(), pid);
    cxx::convert::fromString(message.getElementAtIndex(3).c_str(), userId);
    cxx::convert::fromString(message.getElementAtIndex(4).c_str(), transmissionTimestamp);
    cxx::Serialization serializationVersionInfo(message.getElementAtIndex(5));
    return serializationVersionInfo;
}

void RouDi::processMessage(const runtime::IpcMessage& message,
                           const iox::runtime::IpcMessageType& cmd,
                           const RuntimeName_t& runtimeName)
{
    switch (cmd)
    {
    case runtime::IpcMessageType::SERVICE_REGISTRY_CHANGE_COUNTER:
    {
        m_prcMgr->sendServiceRegistryChangeCounterToProcess(runtimeName);
        break;
    }
    case runtime::IpcMessageType::REG:
    {
        if (message.getNumberOfElements() != 6)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::REG\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            uint32_t pid;
            uid_t userId;
            int64_t transmissionTimestamp;
            version::VersionInfo versionInfo = parseRegisterMessage(message, pid, userId, transmissionTimestamp);

            registerProcess(
                runtimeName, pid, {userId}, transmissionTimestamp, getUniqueSessionIdForProcess(), versionInfo);
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_PUBLISHER:
    {
        if (message.getNumberOfElements() != 8)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_PUBLISHER\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            capro::ServiceDescription service(cxx::Serialization(message.getElementAtIndex(2)));
            cxx::Serialization portConfigInfoSerialization(message.getElementAtIndex(7));

            popo::PublisherOptions options;
            uint64_t historyCapacity{};
            if (!cxx::convert::fromString(message.getElementAtIndex(3).c_str(), historyCapacity))
            {
                LogError() << message.getElementAtIndex(3).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.historyCapacity = historyCapacity;
            options.nodeName = NodeName_t(cxx::TruncateToCapacity, message.getElementAtIndex(4));

            uint64_t offerOnCreate{};
            if (!cxx::convert::fromString(message.getElementAtIndex(5).c_str(), offerOnCreate))
            {
                LogError() << message.getElementAtIndex(5).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.offerOnCreate = (0U == offerOnCreate) ? false : true;

            uint8_t subscriberTooSlowPolicy{};
            if (!cxx::convert::fromString(message.getElementAtIndex(6).c_str(), subscriberTooSlowPolicy))
            {
                LogError() << message.getElementAtIndex(6).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.subscriberTooSlowPolicy = static_cast<popo::SubscriberTooSlowPolicy>(subscriberTooSlowPolicy);

            m_prcMgr->addPublisherForProcess(
                runtimeName, service, options, iox::runtime::PortConfigInfo(portConfigInfoSerialization));
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_SUBSCRIBER:
    {
        if (message.getNumberOfElements() != 9)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_SUBSCRIBER\" from \"" << runtimeName
                       << "\"received!";
        }
        else
        {
            capro::ServiceDescription service(cxx::Serialization(message.getElementAtIndex(2)));
            cxx::Serialization portConfigInfoSerialization(message.getElementAtIndex(8));


            popo::SubscriberOptions options;
            uint64_t historyRequest;
            if (!cxx::convert::fromString(message.getElementAtIndex(3).c_str(), historyRequest))
            {
                LogError() << message.getElementAtIndex(3).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.historyRequest = historyRequest;
            uint64_t queueCapacity;
            if (!cxx::convert::fromString(message.getElementAtIndex(4).c_str(), queueCapacity))
            {
                LogError() << message.getElementAtIndex(4).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.queueCapacity = queueCapacity;
            options.nodeName = NodeName_t(cxx::TruncateToCapacity, message.getElementAtIndex(5));

            uint32_t subscribeOnCreate;
            if (!cxx::convert::fromString(message.getElementAtIndex(6).c_str(), subscribeOnCreate))
            {
                LogError() << message.getElementAtIndex(6).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.subscribeOnCreate = (0U == subscribeOnCreate ? false : true);

            uint8_t queueFullPolicy{};
            if (!cxx::convert::fromString(message.getElementAtIndex(7).c_str(), queueFullPolicy))
            {
                LogError() << message.getElementAtIndex(7).c_str() << "cannot be extracted from string\n";
                break;
            }
            options.queueFullPolicy = static_cast<popo::QueueFullPolicy>(queueFullPolicy);

            m_prcMgr->addSubscriberForProcess(
                runtimeName, service, options, iox::runtime::PortConfigInfo(portConfigInfoSerialization));
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_CONDITION_VARIABLE:
    {
        if (message.getNumberOfElements() != 2)
        {
            LogError() << "Wrong number of parameters for \"IpcMessageType::CREATE_CONDITION_VARIABLE\" from \""
                       << runtimeName << "\"received!";
        }
        else
        {
            m_prcMgr->addConditionVariableForProcess(runtimeName);
        }
        break;
    }
    case runtime::IpcMessageType::CREATE_INTERFACE:
    {
{'fraction_non_alphanumeric': 0.10330118694362017, 'fraction_numerical': 0.005749258160237389, 'mean_word_length': 2.80160744500846, 'pattern_counts': {'":': 0, '<': 153, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7539303', 'n_tokens_mistral': 7592, 'n_tokens_neox': 7349, 'n_words': 1314}