open62541-sys 0.6.1

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Copyright (c) 2017-2025 Fraunhofer IOSB (Author: Andreas Ebner)
 * Copyright (c) 2018 Fraunhofer IOSB (Author: Julius Pfrommer)
 * Copyright (c) 2021 Fraunhofer IOSB (Author: Jan Hermes)
 * Copyright (c) 2022 Siemens AG (Author: Thomas Fischer)
 * Copyright (c) 2022 Fraunhofer IOSB (Author: Noel Graf)
 * Copyright (c) 2022 Linutronix GmbH (Author: Muddasir Shakil)
 */

#include "ua_pubsub_internal.h"

#ifdef UA_ENABLE_PUBSUB /* conditional compilation */

#ifdef UA_ENABLE_PUBSUB_SKS
#include "ua_pubsub_keystorage.h"
#endif

#define UA_DATETIMESTAMP_2000 125911584000000000
#define UA_RESERVEID_FIRST_ID 0x8000

static const char *pubSubStateNames[6] = {
    "Disabled", "Paused", "Operational", "Error", "PreOperational", "Invalid"
};

static void
UA_PubSubManager_stop(UA_ServerComponent *sc);

static UA_StatusCode
UA_PubSubManager_start(UA_ServerComponent *sc, UA_Server *server);

const char *
UA_PubSubState_name(UA_PubSubState state) {
    if(state < UA_PUBSUBSTATE_DISABLED || state > UA_PUBSUBSTATE_PREOPERATIONAL)
        return pubSubStateNames[5];
    return pubSubStateNames[state];
}

void
UA_PubSubComponentHead_clear(UA_PubSubComponentHead *psch) {
    UA_NodeId_clear(&psch->identifier);
    UA_String_clear(&psch->logIdString);
    memset(psch, 0, sizeof(UA_PubSubComponentHead));
}

/* See declaration in ua_pubsub_internal.h for full documentation. */
void
UA_PubSubComponent_freeWithoutLifecycleCallback(UA_PubSubManager *psm,
                                                void *component,
                                                UA_PubSubComponentType type) {
    UA_Server *server = psm->sc.server;
    UA_StatusCode (*savedCb)(UA_Server*, const UA_NodeId,
                             const UA_PubSubComponentType, UA_Boolean) =
        server->config.pubSubConfig.componentLifecycleCallback;
    server->config.pubSubConfig.componentLifecycleCallback = NULL;
    switch(type) {
    case UA_PUBSUBCOMPONENT_CONNECTION:
        UA_PubSubConnection_delete(psm, (UA_PubSubConnection*)component);
        break;
    case UA_PUBSUBCOMPONENT_WRITERGROUP:
        UA_WriterGroup_remove(psm, (UA_WriterGroup*)component);
        break;
    case UA_PUBSUBCOMPONENT_READERGROUP:
        UA_ReaderGroup_remove(psm, (UA_ReaderGroup*)component);
        break;
    case UA_PUBSUBCOMPONENT_DATASETWRITER:
        UA_DataSetWriter_remove(psm, (UA_DataSetWriter*)component);
        break;
    case UA_PUBSUBCOMPONENT_DATASETREADER:
        UA_DataSetReader_remove(psm, (UA_DataSetReader*)component);
        break;
    default:
        break;
    }
    server->config.pubSubConfig.componentLifecycleCallback = savedCb;
}

UA_StatusCode
UA_PublisherId_copy(const UA_PublisherId *src,
                    UA_PublisherId *dst) {
    memcpy(dst, src, sizeof(UA_PublisherId));
    if(src->idType == UA_PUBLISHERIDTYPE_STRING)
        return UA_String_copy(&src->id.string, &dst->id.string);
    return UA_STATUSCODE_GOOD;
}

void
UA_PublisherId_clear(UA_PublisherId *p) {
    if(p->idType == UA_PUBLISHERIDTYPE_STRING)
        UA_String_clear(&p->id.string);
    memset(p, 0, sizeof(UA_PublisherId));
}

UA_StatusCode
UA_PublisherId_fromVariant(UA_PublisherId *p, const UA_Variant *src) {
    if(!UA_Variant_isScalar(src))
        return UA_STATUSCODE_BADINTERNALERROR;

    memset(p, 0, sizeof(UA_PublisherId));

    const void *data = (const void*)src->data;
    if(src->type == &UA_TYPES[UA_TYPES_BYTE]) {
        p->idType = UA_PUBLISHERIDTYPE_BYTE;
        p->id.byte = *(const UA_Byte*)data;
    } else if(src->type == &UA_TYPES[UA_TYPES_UINT16]) {
        p->idType  = UA_PUBLISHERIDTYPE_UINT16;
        p->id.uint16 = *(const UA_UInt16*)data;
    } else if(src->type == &UA_TYPES[UA_TYPES_UINT32]) {
        p->idType  = UA_PUBLISHERIDTYPE_UINT32;
        p->id.uint32 = *(const UA_UInt32*)data;
    } else if(src->type == &UA_TYPES[UA_TYPES_UINT64]) {
        p->idType  = UA_PUBLISHERIDTYPE_UINT64;
        p->id.uint64 = *(const UA_UInt64*)data;
    } else if(src->type == &UA_TYPES[UA_TYPES_STRING]) {
        p->idType  = UA_PUBLISHERIDTYPE_STRING;
        return UA_String_copy((const UA_String *)data, &p->id.string);
    } else {
        return UA_STATUSCODE_BADINTERNALERROR;
    }
    return UA_STATUSCODE_GOOD;
}

void
UA_PublisherId_toVariant(const UA_PublisherId *p, UA_Variant *dst) {
    UA_PublisherId *p2 = (UA_PublisherId*)(uintptr_t)p;
    switch(p->idType) {
    case UA_PUBLISHERIDTYPE_BYTE:
        UA_Variant_setScalar(dst, &p2->id.byte, &UA_TYPES[UA_TYPES_BYTE]); break;
    case UA_PUBLISHERIDTYPE_UINT16:
        UA_Variant_setScalar(dst, &p2->id.uint16, &UA_TYPES[UA_TYPES_UINT16]); break;
    case UA_PUBLISHERIDTYPE_UINT32:
        UA_Variant_setScalar(dst, &p2->id.uint32, &UA_TYPES[UA_TYPES_UINT32]); break;
    case UA_PUBLISHERIDTYPE_UINT64:
        UA_Variant_setScalar(dst, &p2->id.uint64, &UA_TYPES[UA_TYPES_UINT64]); break;
    case UA_PUBLISHERIDTYPE_STRING:
        UA_Variant_setScalar(dst, &p2->id.string, &UA_TYPES[UA_TYPES_STRING]); break;
    default: break; /* This is not possible if the PublisherId is well-defined */
    }
}

UA_ConnectionManager *
getCM(UA_EventLoop *el, UA_String protocol) {
    for(UA_EventSource *es = el->eventSources; es != NULL; es = es->next) {
        if(es->eventSourceType != UA_EVENTSOURCETYPE_CONNECTIONMANAGER)
            continue;
        UA_ConnectionManager *cm = (UA_ConnectionManager*)es;
        if(UA_String_equal(&protocol, &cm->protocol))
            return cm;
    }
    return NULL;
}

static enum ZIP_CMP
cmpReserveId(const void *a, const void *b) {
    const UA_ReserveId *aa = (const UA_ReserveId*)a;
    const UA_ReserveId *bb = (const UA_ReserveId*)b;
    if(aa->id != bb->id)
        return (aa->id < bb->id) ? ZIP_CMP_LESS : ZIP_CMP_MORE;
    if(aa->reserveIdType != bb->reserveIdType)
        return (aa->reserveIdType < bb->reserveIdType) ? ZIP_CMP_LESS : ZIP_CMP_MORE;
    return (enum ZIP_CMP)UA_order(&aa->transportProfileUri,
                                  &bb->transportProfileUri, &UA_TYPES[UA_TYPES_STRING]);
}

ZIP_FUNCTIONS(UA_ReserveIdTree, UA_ReserveId, treeEntry, UA_ReserveId, id, cmpReserveId)

static UA_ReserveId *
UA_ReserveId_new(UA_UInt16 id, UA_String transportProfileUri,
                 UA_ReserveIdType reserveIdType, UA_NodeId sessionId) {
    UA_ReserveId *reserveId = (UA_ReserveId *)UA_calloc(1, sizeof(UA_ReserveId));
    if(!reserveId)
        return NULL;
    reserveId->id = id;
    reserveId->reserveIdType = reserveIdType;
    UA_String_copy(&transportProfileUri, &reserveId->transportProfileUri);
    reserveId->sessionId = sessionId;
    return reserveId;
}

static UA_Boolean
UA_ReserveId_isFree(UA_PubSubManager *psm, UA_UInt16 id, UA_String transportProfileUri,
                    UA_ReserveIdType reserveIdType) {
    /* Is the id already in use? */
    UA_ReserveId compare;
    compare.id = id;
    compare.reserveIdType = reserveIdType;
    compare.transportProfileUri = transportProfileUri;
    if(ZIP_FIND(UA_ReserveIdTree, &psm->reserveIds, &compare))
        return false;

    UA_PubSubConnection *tmpConnection;
    TAILQ_FOREACH(tmpConnection, &psm->connections, listEntry) {
        UA_WriterGroup *writerGroup;
        LIST_FOREACH(writerGroup, &tmpConnection->writerGroups, listEntry) {
            if(reserveIdType == UA_WRITER_GROUP) {
                if(UA_String_equal(&tmpConnection->config.transportProfileUri,
                                   &transportProfileUri) &&
                   writerGroup->config.writerGroupId == id)
                    return false;
            /* reserveIdType == UA_DATA_SET_WRITER */
            } else {
                UA_DataSetWriter *currentWriter;
                LIST_FOREACH(currentWriter, &writerGroup->writers, listEntry) {
                    if(UA_String_equal(&tmpConnection->config.transportProfileUri,
                                       &transportProfileUri) &&
                       currentWriter->config.dataSetWriterId == id)
                        return false;
                }
            }
        }
    }
    return true;
}

static UA_UInt16
UA_ReserveId_createId(UA_PubSubManager *psm,  UA_NodeId sessionId,
                      UA_String transportProfileUri, UA_ReserveIdType reserveIdType) {
    /* Total number of possible Ids */
    UA_UInt16 numberOfIds = 0x8000;
    /* Contains next possible free Id */
    static UA_UInt16 next_id_writerGroup = UA_RESERVEID_FIRST_ID;
    static UA_UInt16 next_id_writer = UA_RESERVEID_FIRST_ID;
    UA_UInt16 next_id;
    UA_Boolean is_free = false;

    if(reserveIdType == UA_WRITER_GROUP)
        next_id = next_id_writerGroup;
    else
        next_id = next_id_writer;

    for(;numberOfIds > 0;numberOfIds--) {
        if(next_id < UA_RESERVEID_FIRST_ID)
            next_id = UA_RESERVEID_FIRST_ID;
        is_free = UA_ReserveId_isFree(psm, next_id, transportProfileUri, reserveIdType);
        if(is_free)
            break;
        next_id++;
    }
    if(!is_free) {
        UA_LOG_ERROR(psm->logging, UA_LOGCATEGORY_PUBSUB,
                     "PubSub ReserveId creation failed. No free ID could be found.");
        return 0;
    }

    if(reserveIdType == UA_WRITER_GROUP)
        next_id_writerGroup = (UA_UInt16)(next_id + 1);
    else
        next_id_writer = (UA_UInt16)(next_id + 1);

    UA_ReserveId *reserveId =
        UA_ReserveId_new(next_id, transportProfileUri, reserveIdType, sessionId);
    if(!reserveId)
        return 0;

    ZIP_INSERT(UA_ReserveIdTree, &psm->reserveIds, reserveId);
    psm->reserveIdsSize++;
    return next_id;
}

static void *
removeReserveId(void *context, UA_ReserveId *elem) {
    UA_String_clear(&elem->transportProfileUri);
    UA_free(elem);
    return NULL;
}

struct RemoveInactiveReserveIdContext {
    UA_PubSubManager *psm;
    UA_ReserveIdTree newTree;
};

/* Remove ReserveIds that are not attached to any session */
static void *
removeInactiveReserveId(void *context, UA_ReserveId *elem) {
    struct RemoveInactiveReserveIdContext *ctx =
        (struct RemoveInactiveReserveIdContext*)context;

    if(UA_NodeId_equal(&ctx->psm->sc.server->adminSession.sessionId, &elem->sessionId))
        goto still_active;

    session_list_entry *session;
    LIST_FOREACH(session, &ctx->psm->sc.server->sessions, pointers) {
        if(UA_NodeId_equal(&session->session.sessionId, &elem->sessionId))
            goto still_active;
    }

    ctx->psm->reserveIdsSize--;
    UA_String_clear(&elem->transportProfileUri);
    UA_free(elem);
    return NULL;

 still_active:
    ZIP_INSERT(UA_ReserveIdTree, &ctx->newTree, elem);
    return NULL;
}

void
UA_PubSubManager_freeIds(UA_PubSubManager *psm) {
    struct RemoveInactiveReserveIdContext removeCtx;
    removeCtx.psm = psm;
    removeCtx.newTree.root = NULL;
    ZIP_ITER(UA_ReserveIdTree, &psm->reserveIds,
             removeInactiveReserveId, &removeCtx);
    psm->reserveIds = removeCtx.newTree;
}

UA_StatusCode
UA_PubSubManager_reserveIds(UA_PubSubManager *psm, UA_NodeId sessionId, UA_UInt16 numRegWriterGroupIds,
                            UA_UInt16 numRegDataSetWriterIds, UA_String transportProfileUri,
                            UA_UInt16 **writerGroupIds, UA_UInt16 **dataSetWriterIds) {
    UA_PubSubManager_freeIds(psm);

    /* Check the validation of the transportProfileUri */
    UA_String profile_1 = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt-uadp");
    UA_String profile_2 = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt-json");
    UA_String profile_3 = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
    if(!UA_String_equal(&transportProfileUri, &profile_1) &&
       !UA_String_equal(&transportProfileUri, &profile_2) &&
       !UA_String_equal(&transportProfileUri, &profile_3)) {
        UA_LOG_ERROR(psm->logging, UA_LOGCATEGORY_PUBSUB,
                     "PubSub ReserveId creation failed. No valid transport profile uri.");
        return UA_STATUSCODE_BADINVALIDARGUMENT;
    }
    *writerGroupIds = (UA_UInt16*)UA_Array_new(numRegWriterGroupIds, &UA_TYPES[UA_TYPES_UINT16]);
    *dataSetWriterIds = (UA_UInt16*)UA_Array_new(numRegDataSetWriterIds, &UA_TYPES[UA_TYPES_UINT16]);

    for(int i = 0; i < numRegWriterGroupIds; i++) {
        (*writerGroupIds)[i] =
            UA_ReserveId_createId(psm, sessionId, transportProfileUri, UA_WRITER_GROUP);
    }
    for(int i = 0; i < numRegDataSetWriterIds; i++) {
        (*dataSetWriterIds)[i] =
            UA_ReserveId_createId(psm, sessionId, transportProfileUri, UA_DATA_SET_WRITER);
    }
    return UA_STATUSCODE_GOOD;
}

/* Calculate the time difference between current time and UTC (00:00) on January
 * 1, 2000. */
UA_UInt32
UA_PubSubConfigurationVersionTimeDifference(UA_DateTime now) {
    UA_UInt32 timeDiffSince2000 = (UA_UInt32)(now - UA_DATETIMESTAMP_2000);
    return timeDiffSince2000;
}

/* Generate a new unique NodeId. This NodeId will be used for the information
 * model representation of PubSub entities. */
#ifndef UA_ENABLE_PUBSUB_INFORMATIONMODEL
void
UA_PubSubManager_generateUniqueNodeId(UA_PubSubManager *psm, UA_NodeId *nodeId) {
    *nodeId = UA_NODEID_NUMERIC(1, ++psm->uniqueIdCount);
}
#endif

UA_Guid
UA_PubSubManager_generateUniqueGuid(UA_PubSubManager *psm) {
    while(true) {
        UA_NodeId testId = UA_NODEID_GUID(1, UA_Guid_random());
        const UA_Node *testNode = UA_NODESTORE_GET(psm->sc.server, &testId);
        if(!testNode)
            return testId.identifier.guid;
        UA_NODESTORE_RELEASE(psm->sc.server, testNode);
    }
}

static UA_UInt64
generateRandomUInt64(void) {
    UA_UInt64 id = 0;
    UA_Guid ident = UA_Guid_random();

    id = id + ident.data1;
    id = (id << 32) + ident.data2;
    id = (id << 16) + ident.data3;
    return id;
}

UA_StatusCode
UA_Server_enableAllPubSubComponents(UA_Server *server) {
    lockServer(server);
    UA_PubSubManager *psm = getPSM(server);
    if(!psm) {
        unlockServer(server);
        return UA_STATUSCODE_BADINTERNALERROR;
    }

    UA_StatusCode res = UA_PubSubManager_start(&psm->sc, server);
    if(res != UA_STATUSCODE_GOOD)
        return res;

    UA_PubSubConnection *c;
    TAILQ_FOREACH(c, &psm->connections, listEntry) {
        UA_WriterGroup *wg;
        LIST_FOREACH(wg, &c->writerGroups, listEntry) {
            UA_DataSetWriter *dsw;
            LIST_FOREACH(dsw, &wg->writers, listEntry) {
                res |= UA_DataSetWriter_setPubSubState(psm, dsw, UA_PUBSUBSTATE_OPERATIONAL);
            }
            res |= UA_WriterGroup_setPubSubState(psm, wg, UA_PUBSUBSTATE_OPERATIONAL);
        }

        UA_ReaderGroup *rg;
        LIST_FOREACH(rg, &c->readerGroups, listEntry) {
            UA_DataSetReader *dsr;
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
                UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_OPERATIONAL,
                                                UA_STATUSCODE_GOOD);
            }
            res |= UA_ReaderGroup_setPubSubState(psm, rg, UA_PUBSUBSTATE_OPERATIONAL);
        }

        res |= UA_PubSubConnection_setPubSubState(psm, c, UA_PUBSUBSTATE_OPERATIONAL);
    }

    unlockServer(server);
    return res;
}

static void
disableAllPubSubComponents(UA_PubSubManager *psm) {
    UA_PubSubConnection *c;
    TAILQ_FOREACH(c, &psm->connections, listEntry) {
        UA_ReaderGroup *rg;
        LIST_FOREACH(rg, &c->readerGroups, listEntry) {
            UA_DataSetReader *dsr;
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
                UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_DISABLED,
                                                UA_STATUSCODE_BADSHUTDOWN);
            }
            UA_ReaderGroup_setPubSubState(psm, rg, UA_PUBSUBSTATE_DISABLED);
        }

        UA_WriterGroup *wg;
        LIST_FOREACH(wg, &c->writerGroups, listEntry) {
            UA_DataSetWriter *dsw;
            LIST_FOREACH(dsw, &wg->writers, listEntry) {
                UA_DataSetWriter_setPubSubState(psm, dsw, UA_PUBSUBSTATE_DISABLED);
            }
            UA_WriterGroup_setPubSubState(psm, wg, UA_PUBSUBSTATE_DISABLED);
        }

        UA_PubSubConnection_setPubSubState(psm, c, UA_PUBSUBSTATE_DISABLED);
    }
}

void
UA_Server_disableAllPubSubComponents(UA_Server *server) {
    lockServer(server);
    UA_PubSubManager *psm = getPSM(server);
    if(psm)
        UA_PubSubManager_stop(&psm->sc); /* Calls disableAll internally */
    unlockServer(server);
}

static UA_StatusCode
getPubSubComponentType(UA_PubSubManager *psm, UA_NodeId componentId,
                       UA_PubSubComponentType *outType) {
    UA_PubSubConnection *c;
    TAILQ_FOREACH(c, &psm->connections, listEntry) {
        if(UA_NodeId_equal(&componentId, &c->head.identifier)) {
            *outType = c->head.componentType;
            return UA_STATUSCODE_GOOD;
        }

        UA_WriterGroup *wg;
        LIST_FOREACH(wg, &c->writerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &wg->head.identifier)) {
                *outType = wg->head.componentType;
                return UA_STATUSCODE_GOOD;
            }

            UA_DataSetWriter *dsw;
            LIST_FOREACH(dsw, &wg->writers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsw->head.identifier)) {
                    *outType = dsw->head.componentType;
                    return UA_STATUSCODE_GOOD;
                }
            }
        }

        UA_ReaderGroup *rg;
        LIST_FOREACH(rg, &c->readerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &rg->head.identifier)) {
                *outType = rg->head.componentType;
                return UA_STATUSCODE_GOOD;
            }

            UA_DataSetReader *dsr;
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsr->head.identifier)) {
                    *outType = dsr->head.componentType;
                    return UA_STATUSCODE_GOOD;
                }
            }
        }
    }

    return UA_STATUSCODE_BADNOTFOUND;
}

UA_StatusCode
UA_Server_getPubSubComponentType(UA_Server *server, UA_NodeId componentId,
                                 UA_PubSubComponentType *outType) {
    if(!outType)
        return UA_STATUSCODE_BADINVALIDARGUMENT;
    lockServer(server);
    UA_PubSubManager *psm = getPSM(server);
    UA_StatusCode res = (psm) ?
        getPubSubComponentType(psm, componentId, outType) : UA_STATUSCODE_BADINTERNALERROR;
    unlockServer(server);
    return res;
}

static UA_StatusCode
getPubSubComponentParent(UA_PubSubManager *psm, UA_NodeId componentId,
                         UA_NodeId *outParent) {
    UA_PubSubConnection *c;
    TAILQ_FOREACH(c, &psm->connections, listEntry) {
        if(UA_NodeId_equal(&componentId, &c->head.identifier))
            return UA_STATUSCODE_BADNOTSUPPORTED;

        UA_WriterGroup *wg;
        LIST_FOREACH(wg, &c->writerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &wg->head.identifier))
                return UA_NodeId_copy(&c->head.identifier, outParent);

            UA_DataSetWriter *dsw;
            LIST_FOREACH(dsw, &wg->writers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsw->head.identifier))
                    return UA_NodeId_copy(&wg->head.identifier, outParent);
            }
        }

        UA_ReaderGroup *rg;
        LIST_FOREACH(rg, &c->readerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &rg->head.identifier))
                return UA_NodeId_copy(&c->head.identifier, outParent);

            UA_DataSetReader *dsr;
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsr->head.identifier))
                    return UA_NodeId_copy(&rg->head.identifier, outParent);
            }
        }
    }

    return UA_STATUSCODE_BADNOTFOUND;
}

UA_StatusCode
UA_Server_getPubSubComponentParent(UA_Server *server, UA_NodeId componentId,
                                   UA_NodeId *outParent) {
    if(!outParent)
        return UA_STATUSCODE_BADINVALIDARGUMENT;
    lockServer(server);
    UA_PubSubManager *psm = getPSM(server);
    UA_StatusCode res = (psm) ?
        getPubSubComponentParent(psm, componentId, outParent) : UA_STATUSCODE_BADINTERNALERROR;
    unlockServer(server);
    return res;
}

static UA_StatusCode
getPubSubComponentChildren(UA_PubSubManager *psm, UA_NodeId componentId,
                           size_t *outChildrenSize, UA_NodeId **outChildren) {
    UA_WriterGroup *wg;
    UA_ReaderGroup *rg;
    UA_DataSetWriter *dsw;
    UA_DataSetReader *dsr;
    UA_PubSubConnection *c;

    UA_StatusCode res = UA_STATUSCODE_GOOD;
    TAILQ_FOREACH(c, &psm->connections, listEntry) {
        if(UA_NodeId_equal(&componentId, &c->head.identifier)) {
            /* Count the children */
            size_t children = 0;
            LIST_FOREACH(wg, &c->writerGroups, listEntry)
                children++;
            LIST_FOREACH(rg, &c->readerGroups, listEntry)
                children++;

            /* Empty array? */
            if(children == 0) {
                *outChildren = NULL;
                *outChildrenSize = 0;
                return UA_STATUSCODE_GOOD;
            }

            /* Allocate the array */
            *outChildren = (UA_NodeId*)UA_calloc(children, sizeof(UA_NodeId));
            if(!*outChildren)
                return UA_STATUSCODE_BADOUTOFMEMORY;
            *outChildrenSize = children;

            /* Copy the NodeIds */
            size_t pos = 0;
            LIST_FOREACH(wg, &c->writerGroups, listEntry) {
                res |= UA_NodeId_copy(&wg->head.identifier, (*outChildren) + pos);
                pos++;
            }
            LIST_FOREACH(rg, &c->readerGroups, listEntry) {
                res |= UA_NodeId_copy(&rg->head.identifier, (*outChildren) + pos);
                pos++;
            }
            goto out;
        }

        LIST_FOREACH(wg, &c->writerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &wg->head.identifier)) {
                /* Count the children */
                size_t children = 0;
                LIST_FOREACH(dsw, &wg->writers, listEntry)
                    children++;

                /* Empty array? */
                if(children == 0) {
                    *outChildren = NULL;
                    *outChildrenSize = 0;
                    return UA_STATUSCODE_GOOD;
                }

                /* Allocate the array */
                *outChildren = (UA_NodeId*)UA_calloc(children, sizeof(UA_NodeId));
                if(!*outChildren)
                    return UA_STATUSCODE_BADOUTOFMEMORY;
                *outChildrenSize = children;

                /* Copy the NodeIds */
                size_t pos = 0;
                LIST_FOREACH(dsw, &wg->writers, listEntry) {
                    res |= UA_NodeId_copy(&dsw->head.identifier, (*outChildren) + pos);
                    pos++;
                }
                goto out;
            }

            /* DataSetWriter have no children (with a state machine) */
            LIST_FOREACH(dsw, &wg->writers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsw->head.identifier))
                    return UA_STATUSCODE_BADNOTSUPPORTED;
            }
        }

        LIST_FOREACH(rg, &c->readerGroups, listEntry) {
            if(UA_NodeId_equal(&componentId, &rg->head.identifier)) {
                /* Count the children */
                size_t children = 0;
                LIST_FOREACH(dsr, &rg->readers, listEntry)
                    children++;

                /* Empty array? */
                if(children == 0) {
                    *outChildren = NULL;
                    *outChildrenSize = 0;
                    return UA_STATUSCODE_GOOD;
                }

                /* Allocate the array */
                *outChildren = (UA_NodeId*)UA_calloc(children, sizeof(UA_NodeId));
                if(!*outChildren)
                    return UA_STATUSCODE_BADOUTOFMEMORY;
                *outChildrenSize = children;

                /* Copy the NodeIds */
                size_t pos = 0;
                LIST_FOREACH(dsr, &rg->readers, listEntry) {
                    res |= UA_NodeId_copy(&dsr->head.identifier, (*outChildren) + pos);
                    pos++;
                }
                goto out;
            }

            /* DataSetReader have no children (with a state machine) */
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
                if(UA_NodeId_equal(&componentId, &dsr->head.identifier))
                    return UA_STATUSCODE_BADNOTSUPPORTED;
            }
        }
    }

    return UA_STATUSCODE_BADNOTFOUND;

 out:
    if(res != UA_STATUSCODE_GOOD)
        UA_Array_delete(*outChildren, *outChildrenSize, &UA_TYPES[UA_TYPES_NODEID]);
    return res;
}

UA_StatusCode
UA_Server_getPubSubComponentChildren(UA_Server *server, UA_NodeId componentId,
                                     size_t *outChildrenSize, UA_NodeId **outChildren) {
    if(!outChildrenSize || !outChildren)
        return UA_STATUSCODE_BADINVALIDARGUMENT;
    lockServer(server);
    UA_PubSubManager *psm = getPSM(server);
    UA_StatusCode res = (psm) ?
        getPubSubComponentChildren(psm, componentId,
                                   outChildrenSize, outChildren) : UA_STATUSCODE_BADINTERNALERROR;
    unlockServer(server);
    return res;
}

void
UA_PubSubManager_setState(UA_PubSubManager *psm, UA_LifecycleState state) {
    if(state == UA_LIFECYCLESTATE_STOPPED)
        state = UA_LIFECYCLESTATE_STOPPING;

    /* Check if all connections are closed if we are started */
    if(state == UA_LIFECYCLESTATE_STOPPING) {
        UA_PubSubConnection *c;
        TAILQ_FOREACH(c, &psm->connections, listEntry) {
            if(c->sendChannel != 0 || c->recvChannelsSize > 0)
                goto set_state;

            UA_WriterGroup *wg;
            LIST_FOREACH(wg, &c->writerGroups, listEntry) {
                if(wg->sendChannel > 0)
                    goto set_state;
            }

            UA_ReaderGroup *rg;
            LIST_FOREACH(rg, &c->readerGroups, listEntry) {
                if(rg->recvChannelsSize > 0)
                    goto set_state;
            }
        }

        /* No open connections -> stopped */
        state = UA_LIFECYCLESTATE_STOPPED;
    }

 set_state:
    if(state == psm->sc.state)
        return;
    psm->sc.state = state;
    if(psm->sc.notifyState)
        psm->sc.notifyState(&psm->sc, state);

    /* When we just started, trigger all connections to go from PAUSED to
     * OPERATIONAL */
    if(state == UA_LIFECYCLESTATE_STARTED) {
        UA_PubSubConnection *c;
        TAILQ_FOREACH(c, &psm->connections, listEntry) {
            if (psm->pubSubInitialSetupMode) {
                UA_PubSubConnection_setPubSubState(psm, c, UA_PUBSUBSTATE_OPERATIONAL);
            } else {
                UA_PubSubConnection_setPubSubState(psm, c, c->head.state);
            }
        }
    }
}

static UA_StatusCode
UA_PubSubManager_start(UA_ServerComponent *sc, UA_Server *server) {
    UA_PubSubManager *psm = (UA_PubSubManager*)sc;
    if(psm->sc.state == UA_LIFECYCLESTATE_STOPPING) {
        UA_LOG_ERROR(psm->logging, UA_LOGCATEGORY_PUBSUB,
                     "The PubSubManager is still stopping");
        return UA_STATUSCODE_BADINTERNALERROR;
    }

    /* Re-cache for the case that the configuration has been updated */
    psm->logging = server->config.logging;

    UA_PubSubManager_setState(psm, UA_LIFECYCLESTATE_STARTED);

    return UA_STATUSCODE_GOOD;
}

static void
UA_PubSubManager_stop(UA_ServerComponent *sc) {
    UA_PubSubManager *psm = (UA_PubSubManager*)sc;
    disableAllPubSubComponents(psm);
    UA_PubSubManager_setState(psm, UA_LIFECYCLESTATE_STOPPED);
}

UA_StatusCode
UA_PubSubManager_clear(UA_PubSubManager *psm) {
    if(psm->sc.state != UA_LIFECYCLESTATE_STOPPED) {
        UA_LOG_ERROR(psm->logging, UA_LOGCATEGORY_PUBSUB,
                     "Cannot delete the PubSubManager because "
                     "it is not stopped");
        return UA_STATUSCODE_BADINTERNALERROR;
    }

    UA_LOCK_ASSERT(&psm->sc.server->serviceMutex);

    /* Remove Connections - this also remove WriterGroups and ReaderGroups */
    UA_PubSubConnection *c, *tmpC;
    TAILQ_FOREACH_SAFE(c, &psm->connections, listEntry, tmpC) {
        UA_PubSubConnection_delete(psm, c);
    }

    /* Remove the DataSets */
    UA_PublishedDataSet *tmpPDS1, *tmpPDS2;
    TAILQ_FOREACH_SAFE(tmpPDS1, &psm->publishedDataSets, listEntry, tmpPDS2) {
        UA_PublishedDataSet_remove(psm, tmpPDS1);
    }

    /* Remove the ReserveIds*/
    ZIP_ITER(UA_ReserveIdTree, &psm->reserveIds, removeReserveId, NULL);
    psm->reserveIdsSize = 0;

    /* Delete subscribed datasets */
    UA_SubscribedDataSet *tmpSDS1, *tmpSDS2;
    TAILQ_FOREACH_SAFE(tmpSDS1, &psm->subscribedDataSets, listEntry, tmpSDS2) {
        UA_SubscribedDataSet_remove(psm, tmpSDS1);
    }

#ifdef UA_ENABLE_PUBSUB_SKS
    /* Remove the SecurityGroups */
    UA_SecurityGroup *tmpSG1, *tmpSG2;
    TAILQ_FOREACH_SAFE(tmpSG1, &psm->securityGroups, listEntry, tmpSG2) {
        UA_SecurityGroup_remove(psm, tmpSG1);
    }

    /* Remove the keyStorages */
    UA_PubSubKeyStorage *ks, *ksTmp;
    LIST_FOREACH_SAFE(ks, &psm->pubSubKeyList, keyStorageList, ksTmp) {
        UA_PubSubKeyStorage_delete(psm, ks);
    }
#endif

    return UA_STATUSCODE_GOOD;
}

UA_ServerComponent *
UA_PubSubManager_new(UA_Server *server) {
    UA_PubSubManager *psm = (UA_PubSubManager*)UA_calloc(1, sizeof(UA_PubSubManager));
    if(!psm)
        return NULL;

    psm->sc.server = server;
    psm->sc.name = UA_STRING("pubsub");
    psm->sc.start = UA_PubSubManager_start;
    psm->sc.stop = UA_PubSubManager_stop;
    psm->sc.clear = (UA_StatusCode (*)(UA_ServerComponent *))UA_PubSubManager_clear;

    /* Set the logging shortcut */
    psm->logging = server->config.logging;

    /* TODO: Using the Mac address to generate the defaultPublisherId.
     * In the future, this can be retrieved from the Eventloop. */
    psm->defaultPublisherId = generateRandomUInt64();

    TAILQ_INIT(&psm->connections);
    TAILQ_INIT(&psm->publishedDataSets);
    TAILQ_INIT(&psm->subscribedDataSets);

#ifdef UA_ENABLE_PUBSUB_SKS
    TAILQ_INIT(&psm->securityGroups);
#endif

#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
    /* Build PubSub information model */
    initPubSubNS0(server);
#endif

    return &psm->sc;
}

#endif /* UA_ENABLE_PUBSUB */