<commit_before>/* 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 2017-2018 (c) Fraunhofer IOSB (Author: <NAME>)
 *    Copyright 2017 (c) <NAME>, fortiss GmbH
 *    Copyright 2018 (c) <NAME>, fortiss GmbH
 *    Copyright 2018 (c) <NAME>, Blue Time Concept SA
 *    Copyright 2018 (c) <NAME>, Root-Core
 */

#include "ua_server_internal.h"
#include "ua_subscription.h"

#ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */

/****************/
/* Notification */
/****************/

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS

static const UA_NodeId overflowEventType =
    {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_EVENTQUEUEOVERFLOWEVENTTYPE}};
static const UA_NodeId simpleOverflowEventType =
    {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_SIMPLEOVERFLOWEVENTTYPE}};

static UA_Boolean
UA_Notification_isOverflowEvent(UA_Server *server, UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;
    if(mon->monitoredItemType != UA_MONITOREDITEMTYPE_EVENTNOTIFY)
        return false;

    UA_EventFieldList *efl = &n->data.event.fields;
    if(efl->eventFieldsSize == 1 &&
       efl->eventFields[0].type == &UA_TYPES[UA_TYPES_NODEID] &&
       isNodeInTree(server->nsCtx, (const UA_NodeId *)efl->eventFields[0].data,
                    &overflowEventType, &subtypeId, 1)) {
        return true;
    }

    return false;
}

static UA_Notification *
createEventOverflowNotification(UA_MonitoredItem *mon) {
    UA_Notification *overflowNotification = (UA_Notification *) UA_malloc(sizeof(UA_Notification));
    if(!overflowNotification)
        return NULL;

    overflowNotification->mon = mon;
    UA_EventFieldList_init(&overflowNotification->data.event.fields);
    overflowNotification->data.event.fields.eventFields = UA_Variant_new();
    if(!overflowNotification->data.event.fields.eventFields) {
        UA_EventFieldList_deleteMembers(&overflowNotification->data.event.fields);
        UA_free(overflowNotification);
        return NULL;
    }

    overflowNotification->data.event.fields.eventFieldsSize = 1;
    UA_StatusCode retval =
        UA_Variant_setScalarCopy(overflowNotification->data.event.fields.eventFields,
                                 &simpleOverflowEventType, &UA_TYPES[UA_TYPES_NODEID]);
    if(retval != UA_STATUSCODE_GOOD) {
        UA_EventFieldList_deleteMembers(&overflowNotification->data.event.fields);
        UA_free(overflowNotification);
        return NULL;
    }

    return overflowNotification;
}

#endif

void
UA_Notification_enqueue(UA_Server *server, UA_Subscription *sub,
                        UA_MonitoredItem *mon, UA_Notification *n) {
    /* Add to the MonitoredItem */
    TAILQ_INSERT_TAIL(&mon->queue, n, listEntry);
    ++mon->queueSize;

    /* Add to the subscription */
    TAILQ_INSERT_TAIL(&sub->notificationQueue, n, globalEntry);
    ++sub->notificationQueueSize;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        ++sub->dataChangeNotifications;
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        ++sub->eventNotifications;
        if(UA_Notification_isOverflowEvent(server, n))
            ++mon->eventOverflows;
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        ++sub->statusChangeNotifications;
    }

    /* Ensure enough space is available in the MonitoredItem. Do this only after
     * adding the new Notification. */
    UA_MonitoredItem_ensureQueueSpace(server, mon);
}

void
UA_Notification_dequeue(UA_Server *server, UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;
    UA_Subscription *sub = mon->subscription;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        --sub->dataChangeNotifications;
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        --sub->eventNotifications;
        if(UA_Notification_isOverflowEvent(server, n))
            --mon->eventOverflows;
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        --sub->statusChangeNotifications;
    }

    TAILQ_REMOVE(&mon->queue, n, listEntry);
    --mon->queueSize;

    TAILQ_REMOVE(&sub->notificationQueue, n, globalEntry);
    --sub->notificationQueueSize;
}

void
UA_Notification_delete(UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        UA_DataValue_deleteMembers(&n->data.value);
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        UA_EventFieldList_deleteMembers(&n->data.event.fields);
        /* EventFilterResult currently isn't being used
         * UA_EventFilterResult_delete(notification->data.event->result); */
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        /* Nothing to do */
    }

    UA_free(n);
}

/*****************/
/* MonitoredItem */
/*****************/

void
UA_MonitoredItem_init(UA_MonitoredItem *mon, UA_Subscription *sub) {
    memset(mon, 0, sizeof(UA_MonitoredItem));
    mon->subscription = sub;
    TAILQ_INIT(&mon->queue);
}

void
UA_MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem) {
    if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        /* Remove the sampling callback */
        UA_MonitoredItem_unregisterSampleCallback(server, monitoredItem);
    } else if (monitoredItem->monitoredItemType != UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        /* TODO: Access val data.event */
        UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
                     "MonitoredItemTypes other than ChangeNotify or EventNotify "
                     "are not supported yet");
    }

    /* Remove the queued notifications if attached to a subscription (not a
     * local MonitoredItem) */
    if(monitoredItem->subscription) {
        UA_Notification *notification, *notification_tmp;
        TAILQ_FOREACH_SAFE(notification, &monitoredItem->queue,
                           listEntry, notification_tmp) {
            /* Remove the item from the queues and free the memory */
            UA_Notification_dequeue(server, notification);
            UA_Notification_delete(notification);
        }
    }

    /* if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY)
     * -> UA_DataChangeFilter does not hold dynamic content we need to free */

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        /* Remove the monitored item from the node queue */
        UA_Server_editNode(server, NULL, &monitoredItem->monitoredNodeId,
                           UA_MonitoredItem_removeNodeEventCallback, monitoredItem);
        /* Delete the event filter */
        UA_EventFilter_deleteMembers(&monitoredItem->filter.eventFilter);
    }
#endif /* UA_ENABLE_SUBSCRIPTIONS_EVENTS */

    /* Deregister MonitoredItem in userland */
    if(server->config.monitoredItemRegisterCallback && monitoredItem->registered) {
        /* Get the session context. Local MonitoredItems don't have a subscription. */
        UA_Session *session = NULL;
        if(monitoredItem->subscription)
            session = monitoredItem->subscription->session;<commit_msg>Stack: Clear DataChangeFilter to avoid potential memory leaks<commit_after>/* 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 2017-2018 (c) Fraunhofer IOSB (Author: <NAME>)
 *    Copyright 2017 (c) <NAME>, fortiss GmbH
 *    Copyright 2018 (c) <NAME>, fortiss GmbH
 *    Copyright 2018 (c) <NAME>, Blue Time Concept SA
 *    Copyright 2018 (c) <NAME>, Root-Core
 */

#include "ua_server_internal.h"
#include "ua_subscription.h"

#ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */

/****************/
/* Notification */
/****************/

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS

static const UA_NodeId overflowEventType =
    {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_EVENTQUEUEOVERFLOWEVENTTYPE}};
static const UA_NodeId simpleOverflowEventType =
    {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_SIMPLEOVERFLOWEVENTTYPE}};

static UA_Boolean
UA_Notification_isOverflowEvent(UA_Server *server, UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;
    if(mon->monitoredItemType != UA_MONITOREDITEMTYPE_EVENTNOTIFY)
        return false;

    UA_EventFieldList *efl = &n->data.event.fields;
    if(efl->eventFieldsSize == 1 &&
       efl->eventFields[0].type == &UA_TYPES[UA_TYPES_NODEID] &&
       isNodeInTree(server->nsCtx, (const UA_NodeId *)efl->eventFields[0].data,
                    &overflowEventType, &subtypeId, 1)) {
        return true;
    }

    return false;
}

static UA_Notification *
createEventOverflowNotification(UA_MonitoredItem *mon) {
    UA_Notification *overflowNotification = (UA_Notification *) UA_malloc(sizeof(UA_Notification));
    if(!overflowNotification)
        return NULL;

    overflowNotification->mon = mon;
    UA_EventFieldList_init(&overflowNotification->data.event.fields);
    overflowNotification->data.event.fields.eventFields = UA_Variant_new();
    if(!overflowNotification->data.event.fields.eventFields) {
        UA_EventFieldList_deleteMembers(&overflowNotification->data.event.fields);
        UA_free(overflowNotification);
        return NULL;
    }

    overflowNotification->data.event.fields.eventFieldsSize = 1;
    UA_StatusCode retval =
        UA_Variant_setScalarCopy(overflowNotification->data.event.fields.eventFields,
                                 &simpleOverflowEventType, &UA_TYPES[UA_TYPES_NODEID]);
    if(retval != UA_STATUSCODE_GOOD) {
        UA_EventFieldList_deleteMembers(&overflowNotification->data.event.fields);
        UA_free(overflowNotification);
        return NULL;
    }

    return overflowNotification;
}

#endif

void
UA_Notification_enqueue(UA_Server *server, UA_Subscription *sub,
                        UA_MonitoredItem *mon, UA_Notification *n) {
    /* Add to the MonitoredItem */
    TAILQ_INSERT_TAIL(&mon->queue, n, listEntry);
    ++mon->queueSize;

    /* Add to the subscription */
    TAILQ_INSERT_TAIL(&sub->notificationQueue, n, globalEntry);
    ++sub->notificationQueueSize;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        ++sub->dataChangeNotifications;
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        ++sub->eventNotifications;
        if(UA_Notification_isOverflowEvent(server, n))
            ++mon->eventOverflows;
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        ++sub->statusChangeNotifications;
    }

    /* Ensure enough space is available in the MonitoredItem. Do this only after
     * adding the new Notification. */
    UA_MonitoredItem_ensureQueueSpace(server, mon);
}

void
UA_Notification_dequeue(UA_Server *server, UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;
    UA_Subscription *sub = mon->subscription;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        --sub->dataChangeNotifications;
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        --sub->eventNotifications;
        if(UA_Notification_isOverflowEvent(server, n))
            --mon->eventOverflows;
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        --sub->statusChangeNotifications;
    }

    TAILQ_REMOVE(&mon->queue, n, listEntry);
    --mon->queueSize;

    TAILQ_REMOVE(&sub->notificationQueue, n, globalEntry);
    --sub->notificationQueueSize;
}

void
UA_Notification_delete(UA_Notification *n) {
    UA_MonitoredItem *mon = n->mon;

    if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        UA_DataValue_deleteMembers(&n->data.value);
#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        UA_EventFieldList_deleteMembers(&n->data.event.fields);
        /* EventFilterResult currently isn't being used
         * UA_EventFilterResult_delete(notification->data.event->result); */
#endif
    } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
        /* Nothing to do */
    }

    UA_free(n);
}

/*****************/
/* MonitoredItem */
/*****************/

void
UA_MonitoredItem_init(UA_MonitoredItem *mon, UA_Subscription *sub) {
    memset(mon, 0, sizeof(UA_MonitoredItem));
    mon->subscription = sub;
    TAILQ_INIT(&mon->queue);
}

void
UA_MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem) {
    if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
        /* Remove the sampling callback */
        UA_MonitoredItem_unregisterSampleCallback(server, monitoredItem);
    } else if (monitoredItem->monitoredItemType != UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        /* TODO: Access val data.event */
        UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
                     "MonitoredItemTypes other than ChangeNotify or EventNotify "
                     "are not supported yet");
    }

    /* Remove the queued notifications if attached to a subscription (not a
     * local MonitoredItem) */
    if(monitoredItem->subscription) {
        UA_Notification *notification, *notification_tmp;
        TAILQ_FOREACH_SAFE(notification, &monitoredItem->queue,
                           listEntry, notification_tmp) {
            /* Remove the item from the queues and free the memory */
            UA_Notification_dequeue(server, notification);
            UA_Notification_delete(notification);
        }
    }

    /* if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY)
     * -> UA_DataChangeFilter does not hold dynamic content we need to free */

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    if(monitoredItem->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
        /* Remove the monitored item from the node queue */
        UA_Server_editNode(server, NULL, &monitoredItem->monitoredNodeId,
                           UA_MonitoredItem_removeNodeEventCallback, monitoredItem);
		UA_EventFilter_clear(&monitoredItem->filter.eventFilter);
    } else
#endif /* UA_ENABLE_SUBSCRIPTIONS_EVENTS */
	{
		UA_DataChangeFilter_clear(&monitoredItem->filter.dataChangeFilter);
    }

    /* Deregister MonitoredItem in userland */
    if(server->config.monitoredItemRegisterCallback && monitoredItem->registered) {
        /* Get the session context. Local MonitoredItems don't have a subscription. */
        UA_Session *session = NULL;
        if(monitoredItem->subscription)
            session = monitoredItem->subscription->session;
{'fraction_non_alphanumeric': 0.1081689394942407, 'fraction_numerical': 0.005031113464848405, 'mean_word_length': 3.701836290071584, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 142, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2878567', 'n_tokens_mistral': 4880, 'n_tokens_neox': 4597, 'n_words': 848}