/src/open62541_15/src/client/ua_client_subscriptions.c

Source
/* 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 2015-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
 *    Copyright 2015 (c) Oleksiy Vasylyev
 *    Copyright 2016 (c) Sten Grüner
 *    Copyright 2017-2018 (c) Thomas Stalder, Blue Time Concept SA
 *    Copyright 2016-2017 (c) Florian Palm
 *    Copyright 2017 (c) Frank Meerkötter
 *    Copyright 2017 (c) Stefan Profanter, fortiss GmbH
 *    Copyright 2026 (c) o6 Automation GmbH (Author: Julius Pfrommer)
 */

#include <open62541/client_highlevel.h>
#include <open62541/client_highlevel_async.h>

#include "ua_client_internal.h"

struct UA_Client_MonitoredItem_ForDelete {
    UA_Client *client;
    UA_Client_Subscription *sub;
    UA_UInt32 *monitoredItemId;
};

/*****************/
/* Subscriptions */
/*****************/

/* For ZIP_TREE we use clientHandle comparison */
static enum ZIP_CMP
UA_ClientHandle_cmp(const void *a, const void *b) {
    const UA_Client_MonitoredItem *aa = (const UA_Client_MonitoredItem *)a;
    const UA_Client_MonitoredItem *bb = (const UA_Client_MonitoredItem *)b;
    if(aa->clientHandle < bb->clientHandle)
        return ZIP_CMP_LESS;
    if(aa->clientHandle > bb->clientHandle)
        return ZIP_CMP_MORE;
    return ZIP_CMP_EQ;
}

ZIP_FUNCTIONS(MonitorItemsTree, UA_Client_MonitoredItem, zipfields,
              UA_Client_MonitoredItem, zipfields, UA_ClientHandle_cmp)

static void
MonitoredItem_delete(UA_Client *client, UA_Client_Subscription *sub,
                     UA_Client_MonitoredItem *mon);

static void
Subscription_create(UA_Client *client, UA_Client_Subscription *newSub,
                    UA_CreateSubscriptionResponse *response) {
    UA_LOCK_ASSERT(&client->clientMutex);

    UA_EventLoop *el = client->config.eventLoop;

    newSub->subscriptionId = response->subscriptionId;
    newSub->sequenceNumber = 0;
    newSub->lastActivity = el->dateTime_nowMonotonic(el);
    newSub->publishingInterval = response->revisedPublishingInterval;
    newSub->maxKeepAliveCount = response->revisedMaxKeepAliveCount;
    ZIP_INIT(&newSub->monitoredItems);
    LIST_INSERT_HEAD(&client->subscriptions, newSub, listEntry);

    /* Immediately send the first publish requests if there are none
     * outstanding */
    __Client_Subscriptions_backgroundPublish(client);
}

static void
Subscriptions_create_handler(UA_Client *client, void *data,
                             UA_UInt32 requestId, void *r) {
    UA_LOCK_ASSERT(&client->clientMutex);

    UA_CreateSubscriptionResponse *response = (UA_CreateSubscriptionResponse *)r;
    CustomCallback *cc = (CustomCallback *)data;
    UA_Client_Subscription *newSub = (UA_Client_Subscription *)cc->clientData;
    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
        UA_free(newSub);
        goto cleanup;
    }

    /* Prepare the internal representation */
    Subscription_create(client, newSub, response);

cleanup:
    if(cc->userCallback)
        cc->userCallback(client, cc->userData, requestId, response);
    UA_free(cc);
}

UA_CreateSubscriptionResponse
UA_Client_Subscriptions_create(UA_Client *client,
                               const UA_CreateSubscriptionRequest request,
                               void *subscriptionContext,
                               UA_Client_StatusChangeNotificationCallback statusChangeCallback,
                               UA_Client_DeleteSubscriptionCallback deleteCallback) {
    lockClient(client);

    UA_CreateSubscriptionResponse response;
    UA_Client_Subscription *sub = (UA_Client_Subscription *)
        UA_malloc(sizeof(UA_Client_Subscription));
    if(!sub) {
        UA_CreateSubscriptionResponse_init(&response);
        response.responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
        unlockClient(client);
        return response;
    }
    sub->context = subscriptionContext;
    sub->statusChangeCallback = statusChangeCallback;
    sub->deleteCallback = deleteCallback;

    /* Send the request as a synchronous service call */
    __Client_Service(client, &request, &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONREQUEST],
                     &response, &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONRESPONSE]);
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
        UA_free(sub);
        unlockClient(client);
        return response;
    }

    Subscription_create(client, sub, &response);

    unlockClient(client);
    return response;
}

UA_StatusCode
UA_Client_Subscriptions_create_async(UA_Client *client, const UA_CreateSubscriptionRequest request,
                                     void *subscriptionContext,
                                     UA_Client_StatusChangeNotificationCallback statusChangeCallback,
                                     UA_Client_DeleteSubscriptionCallback deleteCallback,
                                     UA_ClientAsyncCreateSubscriptionCallback createCallback,
                                     void *userdata, UA_UInt32 *requestId) {
    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
    if(!cc)
        return UA_STATUSCODE_BADOUTOFMEMORY;

    UA_Client_Subscription *sub = (UA_Client_Subscription *)
        UA_malloc(sizeof(UA_Client_Subscription));
    if(!sub) {
        UA_free(cc);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }
    sub->context = subscriptionContext;
    sub->statusChangeCallback = statusChangeCallback;
    sub->deleteCallback = deleteCallback;

    cc->userCallback = (UA_ClientAsyncServiceCallback)createCallback;
    cc->userData = userdata;
    cc->clientData = sub;

    /* Send the request as asynchronous service call */
    UA_StatusCode res =
        __UA_Client_AsyncService(client, &request,
                                 &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONREQUEST],
                                 Subscriptions_create_handler,
                                 &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONRESPONSE],
                                 cc, requestId);
    if(res != UA_STATUSCODE_GOOD) {
        UA_free(cc);
        UA_free(sub);
    }
    return res;
}

static UA_Client_Subscription *
findSubscriptionById(const UA_Client *client, UA_UInt32 subscriptionId) {
    UA_Client_Subscription *sub = NULL;
    LIST_FOREACH(sub, &client->subscriptions, listEntry) {
        if(sub->subscriptionId == subscriptionId)
            break;
    }
    return sub;
}

static void
Subscription_modify(UA_Client *client, UA_Client_Subscription *sub,
                    const UA_ModifySubscriptionResponse *response) {
    sub->publishingInterval = response->revisedPublishingInterval;
    sub->maxKeepAliveCount = response->revisedMaxKeepAliveCount;
}

static void
Subscription_modify_handler(UA_Client *client, void *data,
                            UA_UInt32 requestId, void *r) {
    UA_LOCK_ASSERT(&client->clientMutex);

    UA_ModifySubscriptionResponse *response = (UA_ModifySubscriptionResponse *)r;
    CustomCallback *cc = (CustomCallback *)data;
    UA_Client_Subscription *sub =
        findSubscriptionById(client, (UA_UInt32)(uintptr_t)cc->clientData);
    if(sub) {
        Subscription_modify(client, sub, response);
    } else {
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
                    "No internal representation of subscription %" PRIu32,
                    (UA_UInt32)(uintptr_t)cc->clientData);
    }

    if(cc->userCallback)
        cc->userCallback(client, cc->userData, requestId, response);
    UA_free(cc);
}

UA_StatusCode
UA_Client_Subscriptions_getContext(UA_Client *client, UA_UInt32 subscriptionId,
                                   void **subContext) {
    if(!client || !subContext)
        return UA_STATUSCODE_BADINVALIDARGUMENT;

    lockClient(client);
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
    if(!sub) {
        unlockClient(client);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    *subContext = sub->context;
    unlockClient(client);
    return UA_STATUSCODE_GOOD;
}

UA_StatusCode
UA_Client_Subscriptions_setContext(UA_Client *client, UA_UInt32 subscriptionId,
                                   void *subContext) {
    if(!client)
        return UA_STATUSCODE_BADINVALIDARGUMENT;

    lockClient(client);
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
    if(!sub) {
        unlockClient(client);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    sub->context = subContext;
    unlockClient(client);
    return UA_STATUSCODE_GOOD;
}

UA_ModifySubscriptionResponse
UA_Client_Subscriptions_modify(UA_Client *client,
                               const UA_ModifySubscriptionRequest request) {
    UA_ModifySubscriptionResponse response;
    UA_ModifySubscriptionResponse_init(&response);

    /* Find the internal representation */
    lockClient(client);
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        unlockClient(client);
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
        return response;
    }

    /* Call the service */
    __Client_Service(client,
                     &request, &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONREQUEST],
                     &response, &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONRESPONSE]);

    /* Adjust the internal representation. Lookup again for thread-safety. */
    sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
        unlockClient(client);
        return response;
    }
    Subscription_modify(client, sub, &response);
    unlockClient(client);
    return response;
}

UA_StatusCode
UA_Client_Subscriptions_modify_async(UA_Client *client,
                                     const UA_ModifySubscriptionRequest request,
                                     UA_ClientAsyncModifySubscriptionCallback callback,
                                     void *userdata, UA_UInt32 *requestId) {
    lockClient(client);

    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        unlockClient(client);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
    if(!cc) {
        unlockClient(client);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }

    cc->clientData = (void *)(uintptr_t)request.subscriptionId;
    cc->userData = userdata;
    cc->userCallback = (UA_ClientAsyncServiceCallback)callback;

    UA_StatusCode res =
        __Client_AsyncService(client, &request,
                              &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONREQUEST],
                              Subscription_modify_handler,
                              &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONRESPONSE],
                              cc, requestId);

    unlockClient(client);
    return res;
}

static void *
MonitoredItem_delete_wrapper(void *data, UA_Client_MonitoredItem *mon) {
    struct UA_Client_MonitoredItem_ForDelete *deleteMonitoredItem =
        (struct UA_Client_MonitoredItem_ForDelete *)data;
    if(deleteMonitoredItem != NULL) {
        if(deleteMonitoredItem->monitoredItemId != NULL &&
           (mon->monitoredItemId != *deleteMonitoredItem->monitoredItemId)) {
            return NULL;
        }
        MonitoredItem_delete(deleteMonitoredItem->client, deleteMonitoredItem->sub, mon);
    }
    return NULL;
}

static void
__Client_Subscription_deleteInternal(UA_Client *client,
                                     UA_Client_Subscription *sub) {
    /* Remove the MonitoredItems */
    struct UA_Client_MonitoredItem_ForDelete deleteMonitoredItem;
    memset(&deleteMonitoredItem, 0, sizeof(struct UA_Client_MonitoredItem_ForDelete));
    deleteMonitoredItem.client = client;
    deleteMonitoredItem.sub = sub;
    ZIP_ITER(MonitorItemsTree, &sub->monitoredItems,
             MonitoredItem_delete_wrapper, &deleteMonitoredItem);

    /* Call the delete callback */
    if(sub->deleteCallback) {
        void *subC = sub->context;
        UA_UInt32 subId = sub->subscriptionId;
        sub->deleteCallback(client, subId, subC);
    }

    /* Remove */
    LIST_REMOVE(sub, listEntry);
    UA_free(sub);
}

static void
__Client_Subscription_processDelete(UA_Client *client,
                                    const UA_DeleteSubscriptionsRequest *request,
                                    const UA_DeleteSubscriptionsResponse *response)  {
    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD)
        return;

    /* Check that the request and response size -- use the same index for both */
    if(request->subscriptionIdsSize != response->resultsSize)
        return;

    for(size_t i = 0; i < request->subscriptionIdsSize; i++) {
        if(response->results[i] != UA_STATUSCODE_GOOD &&
           response->results[i] != UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID)
            continue;

        /* Get the Subscription */
        UA_Client_Subscription *sub =
            findSubscriptionById(client, request->subscriptionIds[i]);
        if(!sub) {
            UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
                        "No internal representation of subscription %" PRIu32,
                        request->subscriptionIds[i]);
            continue;
        }

        /* Delete the Subscription */
        __Client_Subscription_deleteInternal(client, sub);
    }
}

typedef struct {
    UA_DeleteSubscriptionsRequest request;
    UA_ClientAsyncServiceCallback userCallback;
    void *userData;
} DeleteSubscriptionCallback;

static void
Subscriptions_delete_handler(UA_Client *client, void *data,
                             UA_UInt32 requestId, void *r) {
    UA_DeleteSubscriptionsResponse *response =
        (UA_DeleteSubscriptionsResponse *)r;
    DeleteSubscriptionCallback *dsc =
        (DeleteSubscriptionCallback*)data;

    lockClient(client);

    /* Delete */
    __Client_Subscription_processDelete(client, &dsc->request, response);

    /* Userland Callback */
    if(dsc->userCallback)
        dsc->userCallback(client, dsc->userData, requestId, response);

    /* Cleanup */
    UA_DeleteSubscriptionsRequest_clear(&dsc->request);
    UA_free(dsc);

    unlockClient(client);
}

UA_StatusCode
UA_Client_Subscriptions_delete_async(UA_Client *client,
                                     const UA_DeleteSubscriptionsRequest request,
                                     UA_ClientAsyncDeleteSubscriptionsCallback callback,
                                     void *userdata, UA_UInt32 *requestId) {
    /* Make a copy of the request that persists into the async callback */
    DeleteSubscriptionCallback *dsc = (DeleteSubscriptionCallback*)
        UA_malloc(sizeof(DeleteSubscriptionCallback));
    if(!dsc)
        return UA_STATUSCODE_BADOUTOFMEMORY;
    dsc->userCallback = (UA_ClientAsyncServiceCallback)callback;
    dsc->userData = userdata;
    UA_StatusCode res = UA_DeleteSubscriptionsRequest_copy(&request, &dsc->request);
    if(res != UA_STATUSCODE_GOOD) {
        UA_free(dsc);
        return res;
    }

    /* Make the async call */
    res = __UA_Client_AsyncService(client, &request,
                                   &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSREQUEST],
                                   Subscriptions_delete_handler,
                                   &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSRESPONSE],
                                   dsc, requestId);
    if(res != UA_STATUSCODE_GOOD) {
        UA_DeleteSubscriptionsRequest_clear(&dsc->request);
        UA_free(dsc);
    }
    return res;
}

UA_DeleteSubscriptionsResponse
UA_Client_Subscriptions_delete(UA_Client *client,
                               const UA_DeleteSubscriptionsRequest request) {
    lockClient(client);

    /* Send the request */
    UA_DeleteSubscriptionsResponse response;
    __Client_Service(client, &request,
                     &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSREQUEST],
                     &response, &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSRESPONSE]);

    /* Process */
    __Client_Subscription_processDelete(client, &request, &response);

    unlockClient(client);
    return response;
}

UA_StatusCode
UA_Client_Subscriptions_deleteSingle(UA_Client *client, UA_UInt32 subscriptionId) {
    UA_DeleteSubscriptionsRequest request;
    UA_DeleteSubscriptionsRequest_init(&request);
    request.subscriptionIds = &subscriptionId;
    request.subscriptionIdsSize = 1;

    UA_DeleteSubscriptionsResponse response =
        UA_Client_Subscriptions_delete(client, request);

    UA_StatusCode retval = response.responseHeader.serviceResult;
    if(retval != UA_STATUSCODE_GOOD) {
        UA_DeleteSubscriptionsResponse_clear(&response);
        return retval;
    }

    if(response.resultsSize != 1) {
        UA_DeleteSubscriptionsResponse_clear(&response);
        return UA_STATUSCODE_BADINTERNALERROR;
    }

    retval = response.results[0];
    UA_DeleteSubscriptionsResponse_clear(&response);
    return retval;
}

/******************/
/* MonitoredItems */
/******************/

static void
MonitoredItem_delete(UA_Client *client, UA_Client_Subscription *sub,
                     UA_Client_MonitoredItem *mon) {
    UA_LOCK_ASSERT(&client->clientMutex);

    ZIP_REMOVE(MonitorItemsTree, &sub->monitoredItems, mon);
    if(mon->deleteCallback)
        mon->deleteCallback(client, sub->subscriptionId, sub->context,
                            mon->monitoredItemId, mon->context);
    for(size_t i = 0; i < mon->eventFields.mapSize; i++) {
        UA_Variant_init(&mon->eventFields.map[i].value);
    }
    UA_KeyValueMap_clear(&mon->eventFields);
    UA_free(mon);
}

static UA_StatusCode
prepareEventFieldsMap(UA_Client_MonitoredItem *newMon,
                      UA_MonitoringParameters *params) {
    /* Get the EventFilter */
    UA_ExtensionObject *eo = &params->filter;
    if(eo->content.decoded.type != &UA_TYPES[UA_TYPES_EVENTFILTER])
        return UA_STATUSCODE_GOOD;
    UA_EventFilter *ef = (UA_EventFilter*)eo->content.decoded.data;
    UA_StatusCode res = UA_STATUSCODE_GOOD;

    /* Check whether there are fields */
    if(ef->selectClausesSize == 0)
        return UA_STATUSCODE_GOOD;

    /* Allocate the map */
    newMon->eventFields.map = (UA_KeyValuePair*)
        UA_calloc(ef->selectClausesSize, sizeof(UA_KeyValuePair));
    if(!newMon->eventFields.map)
        return UA_STATUSCODE_BADOUTOFMEMORY;
    newMon->eventFields.mapSize = ef->selectClausesSize;

    /* Create the key-strings for the fields */
    for(size_t i = 0; i < newMon->eventFields.mapSize; i++) {
        res |= UA_SimpleAttributeOperand_print(&ef->selectClauses[i],
                                               &newMon->eventFields.map[i].key.name);
    }

    return res;
}

static UA_StatusCode
MonitoredItem_createBegin(UA_Client *client, UA_Client_Subscription *sub,
                          UA_MonitoredItemCreateRequest *item,
                          UA_Client_DeleteMonitoredItemCallback deleteCallback,
                          void *context, void *handlingCallback,
                          UA_Client_MonitoredItem **outMon) {
    /* Allocate MonitoredItem */
    UA_Client_MonitoredItem *mon = (UA_Client_MonitoredItem *)
        UA_calloc(1, sizeof(UA_Client_MonitoredItem));
    if(!mon)
        return UA_STATUSCODE_BADOUTOFMEMORY;

    /* Cache the field name map */
    mon->isEventMonitoredItem =
        (item->itemToMonitor.attributeId == UA_ATTRIBUTEID_EVENTNOTIFIER);
    if(mon->isEventMonitoredItem) {
        UA_StatusCode res = prepareEventFieldsMap(mon, &item->requestedParameters);
        if(res != UA_STATUSCODE_GOOD) {
            UA_free(mon);
            return res;
        }
    }

    /* Set a unique ClientHandle */
    item->requestedParameters.clientHandle = ++client->monitoredItemHandles;

    /* Fill in members and add to the client  */
    mon->clientHandle = item->requestedParameters.clientHandle;
    mon->context = context;
    mon->deleteCallback = deleteCallback;
    mon->handler.dataChangeCallback =
        (UA_Client_DataChangeNotificationCallback)(uintptr_t)handlingCallback;
    ZIP_INSERT(MonitorItemsTree, &sub->monitoredItems, mon);

    UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                 "Subscription %" PRIu32 " | Added a MonitoredItem with handle %" PRIu32,
                 sub->subscriptionId, mon->clientHandle);

    *outMon = mon;
    return UA_STATUSCODE_GOOD;
}

static void
MonitoredItem_createFinish(UA_Client *client, UA_Client_Subscription *sub,
                           UA_Client_MonitoredItem *mon,
                           UA_MonitoredItemCreateResult *result) {
    UA_assert(result->statusCode == UA_STATUSCODE_GOOD);

    mon->monitoredItemId = result->monitoredItemId;
    /* revisedSamplingInterval; */
    /* revisedQueueSize; */
    /* filterResult; */
}

/************************************/
/* CreateMonitoredItems Synchronous */
/************************************/

static void
Client_MonitoredItems_create(UA_Client *client,
                             const UA_CreateMonitoredItemsRequest *constRequest,
                             void **contexts, void **handlingCallbacks,
                             UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
                             UA_CreateMonitoredItemsResponse *response) {
    UA_LOCK_ASSERT(&client->clientMutex);
    UA_CreateMonitoredItemsResponse_init(response);

    /* Any items? */
    if(constRequest->itemsToCreateSize == 0) {
        response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
        return;
    }

    /* Get the subscription */
    UA_Client_Subscription *sub =
        findSubscriptionById(client, constRequest->subscriptionId);
    if(!sub) {
        response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
        return;
    }

    /* Make a mutable copy. We modify the request to set the internal
     * clientHandle. */
    UA_CreateMonitoredItemsRequest request;
    UA_StatusCode res =
        UA_CreateMonitoredItemsRequest_copy(constRequest, &request);
    if(res != UA_STATUSCODE_GOOD) {
        response->responseHeader.serviceResult = res;
        return;
    }

    /* Create the MonitoredItems */
    UA_STACKARRAY(UA_Client_MonitoredItem*, mons, request.itemsToCreateSize);
    memset(mons, 0, sizeof(UA_Client_MonitoredItem*) * request.itemsToCreateSize);
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
        void *context = (contexts) ? contexts[i] : NULL;
        void *handlingCallback = (handlingCallbacks) ? handlingCallbacks[i] : NULL;
        UA_Client_DeleteMonitoredItemCallback deleteCallback =
            (deleteCallbacks) ? deleteCallbacks[i] : NULL;
        res |= MonitoredItem_createBegin(client, sub, &request.itemsToCreate[i],
                                         deleteCallback, context,
                                         handlingCallback, &mons[i]);
    }

    /* Failure -> Delete created MonitoredItems. Directly call deleteCallback if
     * creation failed. The MonitoredItemId is not yet known, use zero. */
    if(res != UA_STATUSCODE_GOOD) {
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
            if(mons[i])
                MonitoredItem_delete(client, sub, mons[i]);
            else if(deleteCallbacks && contexts)
                deleteCallbacks[i](client, request.subscriptionId,
                                   sub->context, 0, contexts[i]);
        }
        UA_CreateMonitoredItemsRequest_clear(&request);
        response->responseHeader.serviceResult = res;
        return;
    }

    /* Call the service */
    __Client_Service(client, &request,
                     &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSREQUEST],
                     response, &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSRESPONSE]);

    /* Check that the response size is good */
    if(response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
       response->resultsSize != request.itemsToCreateSize)
        response->responseHeader.serviceResult = UA_STATUSCODE_BADINTERNALERROR;

    /* Update the MonitoredItems */
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
        UA_assert(mons[i]);
        UA_MonitoredItemCreateResult *item = &response->results[i];
        if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD ||
           item->statusCode != UA_STATUSCODE_GOOD) {
            MonitoredItem_delete(client, sub, mons[i]);
            continue;
        }
        MonitoredItem_createFinish(client, sub, mons[i], item);
    }

    UA_CreateMonitoredItemsRequest_clear(&request);
}

UA_CreateMonitoredItemsResponse
UA_Client_MonitoredItems_createDataChanges(UA_Client *client,
                                           const UA_CreateMonitoredItemsRequest request,
                                           void **contexts,
                                           UA_Client_DataChangeNotificationCallback *callbacks,
                                           UA_Client_DeleteMonitoredItemCallback *deleteCallbacks) {
    UA_CreateMonitoredItemsResponse response;
    lockClient(client);
    Client_MonitoredItems_create(client, &request, contexts, (void **)callbacks,
                                 deleteCallbacks, &response);
    unlockClient(client);
    return response;
}

UA_MonitoredItemCreateResult
UA_Client_MonitoredItems_createDataChange(UA_Client *client, UA_UInt32 subscriptionId,
                                          UA_TimestampsToReturn timestampsToReturn,
                                          const UA_MonitoredItemCreateRequest item,
                                          void *context,
                                          UA_Client_DataChangeNotificationCallback callback,
                                          UA_Client_DeleteMonitoredItemCallback deleteCallback) {
    UA_CreateMonitoredItemsRequest request;
    UA_CreateMonitoredItemsRequest_init(&request);
    request.subscriptionId = subscriptionId;
    request.timestampsToReturn = timestampsToReturn;
    request.itemsToCreate = (UA_MonitoredItemCreateRequest*)(uintptr_t)&item;
    request.itemsToCreateSize = 1;
    UA_CreateMonitoredItemsResponse response =
       UA_Client_MonitoredItems_createDataChanges(client, request, &context,
                                                   &callback, &deleteCallback);
    UA_MonitoredItemCreateResult result;
    UA_MonitoredItemCreateResult_init(&result);
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
        result.statusCode = response.responseHeader.serviceResult;
    if(result.statusCode == UA_STATUSCODE_GOOD &&
       response.resultsSize != 1)
        result.statusCode = UA_STATUSCODE_BADINTERNALERROR;
    if(result.statusCode == UA_STATUSCODE_GOOD) {
        result = response.results[0];
        UA_MonitoredItemCreateResult_init(&response.results[0]);
    }
    UA_CreateMonitoredItemsResponse_clear(&response);
    return result;
}

UA_CreateMonitoredItemsResponse
UA_Client_MonitoredItems_createEvents(UA_Client *client,
                                      const UA_CreateMonitoredItemsRequest request,
                                      void **contexts,
                                      UA_Client_EventNotificationCallback *callback,
                                      UA_Client_DeleteMonitoredItemCallback *deleteCallback) {
    UA_CreateMonitoredItemsResponse response;
    lockClient(client);
    Client_MonitoredItems_create(client, &request, contexts, (void **)callback,
                                 deleteCallback, &response);
    unlockClient(client);
    return response;
}

UA_MonitoredItemCreateResult
UA_Client_MonitoredItems_createEvent(UA_Client *client, UA_UInt32 subscriptionId,
                                     UA_TimestampsToReturn timestampsToReturn,
                                     const UA_MonitoredItemCreateRequest item, void *context,
                                     UA_Client_EventNotificationCallback callback,
                                     UA_Client_DeleteMonitoredItemCallback deleteCallback) {
    UA_CreateMonitoredItemsRequest request;
    UA_CreateMonitoredItemsRequest_init(&request);
    request.subscriptionId = subscriptionId;
    request.timestampsToReturn = timestampsToReturn;
    request.itemsToCreate = (UA_MonitoredItemCreateRequest*)(uintptr_t)&item;
    request.itemsToCreateSize = 1;
    UA_CreateMonitoredItemsResponse response =
       UA_Client_MonitoredItems_createEvents(client, request, &context,
                                             &callback, &deleteCallback);
    UA_MonitoredItemCreateResult result;
    UA_MonitoredItemCreateResult_init(&result);
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
        result.statusCode = response.responseHeader.serviceResult;
    if(result.statusCode == UA_STATUSCODE_GOOD &&
       response.resultsSize != 1)
        result.statusCode = UA_STATUSCODE_BADINTERNALERROR;
    if(result.statusCode == UA_STATUSCODE_GOOD) {
        result = response.results[0];
        UA_MonitoredItemCreateResult_init(&response.results[0]);
    }
    UA_CreateMonitoredItemsResponse_clear(&response);
    return result;
}

/*************************************/
/* CreateMonitoredItems Asynchronous */
/*************************************/

/* The handles are an array of [subId, monSize, monHandleId1, monHandle2, ...] */
static void
MonitoredItems_create_async_handler(UA_Client *client, void *data,
                                    UA_UInt32 requestId, void *resp) {
    CustomCallback *cc = (CustomCallback*)data;
    UA_UInt32 *handles = (UA_UInt32*)cc->clientData;
    UA_CreateMonitoredItemsResponse *response =
        (UA_CreateMonitoredItemsResponse *)resp;

    lockClient(client);

    /* Extract the first elements from the handles */
    UA_UInt32 subId = handles[0];
    UA_UInt32 monSize = handles[1];
    UA_UInt32 *monHandles = handles + 2;

    /* Check that the response size is good */
    if(response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
       response->resultsSize != monSize)
        response->responseHeader.serviceResult = UA_STATUSCODE_BADINTERNALERROR;

    /* Get the Subscription from the SubscriptionId */
    UA_Client_Subscription *sub = findSubscriptionById(client, subId);
    if(!sub)
        response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;

    /* Update the MonitoredItems */
    UA_Client_MonitoredItem dummy;
    for(size_t i = 0; sub && i < monSize; i++) {
        /* Get the MonitoredItem from the ClientHandle */
        dummy.clientHandle = monHandles[i];
        UA_Client_MonitoredItem *mon =
            ZIP_FIND(MonitorItemsTree, &sub->monitoredItems, &dummy);
        if(!mon)
            continue;

        /* Delete MonitoredItem if the creation failed */
        UA_MonitoredItemCreateResult *item = &response->results[i];
        if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD ||
           item->statusCode != UA_STATUSCODE_GOOD) {
            MonitoredItem_delete(client, sub, mon);
            continue;
        }

        /* Update the MonitoredItem with the server's response  */
        MonitoredItem_createFinish(client, sub, mon, item);
    }

    /* Notify the application */
    if(cc->userCallback) {
        UA_ClientAsyncCreateMonitoredItemsCallback cb =
            (UA_ClientAsyncCreateMonitoredItemsCallback)cc->userCallback;
        cb(client, cc->userData, requestId, response);
    }

    /* Clean up */
    UA_free(handles);
    UA_free(cc);

    unlockClient(client);
}

static UA_StatusCode
Client_MonitoredItems_createAsync(UA_Client *client,
                                  const UA_CreateMonitoredItemsRequest *constRequest,
                                  void **contexts, void **handlingCallbacks,
                                  UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
                                  UA_ClientAsyncServiceCallback createCallback,
                                  void *userdata, UA_UInt32 *requestId) {
    UA_LOCK_ASSERT(&client->clientMutex);

    /* Get the Subscription */
    UA_Client_Subscription *sub =
        findSubscriptionById(client, constRequest->subscriptionId);
    if(!sub)
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;

    /* Any items? */
    if(constRequest->itemsToCreateSize == 0)
        return UA_STATUSCODE_BADNOTHINGTODO;

    /* Make a mutable copy. We modify the request to set the internal
     * clientHandle. */
    UA_CreateMonitoredItemsRequest request;
    UA_StatusCode res =
        UA_CreateMonitoredItemsRequest_copy(constRequest, &request);
    if(res != UA_STATUSCODE_GOOD)
        return res;

    /* Allocate context for the async handling */
    CustomCallback *cc = (CustomCallback*)UA_calloc(1, sizeof(CustomCallback));
    if(!cc) {
        UA_CreateMonitoredItemsRequest_clear(&request);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }
    UA_UInt32 *handles = (UA_UInt32*)
        UA_malloc(sizeof(UA_UInt32) * (request.itemsToCreateSize + 2));
    if(!handles) {
        UA_free(cc);
        UA_CreateMonitoredItemsRequest_clear(&request);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }

    /* Create the MonitoredItems locally */
    UA_STACKARRAY(UA_Client_MonitoredItem*, mons, request.itemsToCreateSize);
    memset(mons, 0, sizeof(UA_Client_MonitoredItem*) * request.itemsToCreateSize);
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
        void *context = (contexts) ? contexts[i] : NULL;
        void *handlingCallback = (handlingCallbacks) ? handlingCallbacks[i] : NULL;
        UA_Client_DeleteMonitoredItemCallback deleteCallback =
            (deleteCallbacks) ? deleteCallbacks[i] : NULL;
        res |= MonitoredItem_createBegin(client, sub, &request.itemsToCreate[i],
                                         deleteCallback, context,
                                         handlingCallback, &mons[i]);
    }

    /* Failure -> Delete created MonitoredItems. Directly call deleteCallback if
     * creation failed. The MonitoredItemId is not yet known, use zero. */
    if(res != UA_STATUSCODE_GOOD) {
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
            if(mons[i])
                MonitoredItem_delete(client, sub, mons[i]);
            else if(deleteCallbacks && contexts)
                deleteCallbacks[i](client, request.subscriptionId,
                                   sub->context, 0, contexts[i]);
        }
        UA_free(handles);
        UA_free(cc);
        UA_CreateMonitoredItemsRequest_clear(&request);
        return res;
    }

    /* Set the async handler context */
    handles[0] = sub->subscriptionId;
    handles[1] = (UA_UInt32)request.itemsToCreateSize;
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
        handles[i+2] = mons[i]->clientHandle;
    }
    cc->clientData = handles;
    cc->userCallback = createCallback;
    cc->userData = userdata;

    /* Call the service asynchronously */
    res = __Client_AsyncService(client, &request,
                                &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSREQUEST],
                                MonitoredItems_create_async_handler,
                                &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSRESPONSE],
                                cc, requestId);

    /* Manually clean up the context in the failure case */
    if(res != UA_STATUSCODE_GOOD) {
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
            UA_assert(mons[i]);
            MonitoredItem_delete(client, sub, mons[i]);
        }
        UA_free(handles);
        UA_free(cc);
    }

    /* Clean up */
    UA_CreateMonitoredItemsRequest_clear(&request);
    return res;
}

UA_StatusCode
UA_Client_MonitoredItems_createDataChanges_async(UA_Client *client,
                                                 const UA_CreateMonitoredItemsRequest request,
                                                 void **contexts,
                                                 UA_Client_DataChangeNotificationCallback *callbacks,
                                                 UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
                                                 UA_ClientAsyncCreateMonitoredItemsCallback createCallback,
                                                 void *userdata, UA_UInt32 *requestId) {
    lockClient(client);
    UA_StatusCode res =
        Client_MonitoredItems_createAsync(client, &request, contexts,
                                          (void **)callbacks, deleteCallbacks,
                                          (UA_ClientAsyncServiceCallback)createCallback,
                                          userdata, requestId);
    unlockClient(client);
    return res;
}

UA_StatusCode
UA_Client_MonitoredItems_createEvents_async(UA_Client *client,
                                            const UA_CreateMonitoredItemsRequest request,
                                            void **contexts,
                                            UA_Client_EventNotificationCallback *callbacks,
                                            UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
                                            UA_ClientAsyncCreateMonitoredItemsCallback createCallback,
                                            void *userdata, UA_UInt32 *requestId) {
    lockClient(client);
    UA_StatusCode res =
        Client_MonitoredItems_createAsync(client, &request, contexts,
                                          (void **)callbacks, deleteCallbacks,
                                          (UA_ClientAsyncServiceCallback)createCallback,
                                          userdata, requestId);
    unlockClient(client);
    return res;
}

static void
MonitoredItems_delete(UA_Client *client, UA_Client_Subscription *sub,
                      const UA_DeleteMonitoredItemsRequest *request,
                      const UA_DeleteMonitoredItemsResponse *response) {
#ifdef __clang_analyzer__
    return;
#endif

    /* Loop over deleted MonitoredItems */
    struct UA_Client_MonitoredItem_ForDelete deleteMonitoredItem;
    memset(&deleteMonitoredItem, 0, sizeof(struct UA_Client_MonitoredItem_ForDelete));
    deleteMonitoredItem.client = client;
    deleteMonitoredItem.sub = sub;

    for(size_t i = 0; i < response->resultsSize; i++) {
        if(response->results[i] != UA_STATUSCODE_GOOD &&
           response->results[i] != UA_STATUSCODE_BADMONITOREDITEMIDINVALID) {
            continue;
        }
        deleteMonitoredItem.monitoredItemId = &request->monitoredItemIds[i];
        /* Delete the internal representation */
        ZIP_ITER(MonitorItemsTree,&sub->monitoredItems,
                 MonitoredItem_delete_wrapper, &deleteMonitoredItem);
    }
}

static void
MonitoredItems_delete_handler(UA_Client *client, void *d, UA_UInt32 requestId, void *r) {
    UA_Client_Subscription *sub = NULL;
    CustomCallback *cc = (CustomCallback *)d;
    UA_DeleteMonitoredItemsResponse *response = (UA_DeleteMonitoredItemsResponse *)r;
    UA_DeleteMonitoredItemsRequest *request =
        (UA_DeleteMonitoredItemsRequest *)cc->clientData;

    lockClient(client);

    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD)
        goto cleanup;

    sub = findSubscriptionById(client, request->subscriptionId);
    if(!sub) {
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
                    "No internal representation of subscription %" PRIu32,
                    request->subscriptionId);
        goto cleanup;
    }

    /* Delete MonitoredItems from the internal representation */
    MonitoredItems_delete(client, sub, request, response);

cleanup:
    if(cc->userCallback)
        cc->userCallback(client, cc->userData, requestId, response);
    UA_DeleteMonitoredItemsRequest_delete(request);
    UA_free(cc);

    unlockClient(client);
}

UA_DeleteMonitoredItemsResponse
UA_Client_MonitoredItems_delete(UA_Client *client,
                                const UA_DeleteMonitoredItemsRequest request) {
    /* Send the request */
    UA_DeleteMonitoredItemsResponse response;
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSREQUEST],
                        &response, &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSRESPONSE]);

    /* A problem occured remote? */
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
        return response;

    lockClient(client);

    /* Find the internal subscription representation */
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
                    "No internal representation of subscription %" PRIu32,
                    request.subscriptionId);
        unlockClient(client);
        return response;
    }

    /* Remove MonitoredItems in the internal representation */
    MonitoredItems_delete(client, sub, &request, &response);

    unlockClient(client);

    return response;
}

UA_StatusCode
UA_Client_MonitoredItems_delete_async(UA_Client *client,
                                      const UA_DeleteMonitoredItemsRequest request,
                                      UA_ClientAsyncDeleteMonitoredItemsCallback callback,
                                      void *userdata, UA_UInt32 *requestId) {
    /* Send the request */
    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
    if(!cc)
        return UA_STATUSCODE_BADOUTOFMEMORY;

    UA_DeleteMonitoredItemsRequest *req_copy = UA_DeleteMonitoredItemsRequest_new();
    if(!req_copy) {
        UA_free(cc);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }

    UA_DeleteMonitoredItemsRequest_copy(&request, req_copy);
    cc->clientData = req_copy;
    cc->userCallback = (UA_ClientAsyncServiceCallback)callback;
    cc->userData = userdata;

    UA_StatusCode res =
        __UA_Client_AsyncService(client, &request,
                                 &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSREQUEST],
                                 MonitoredItems_delete_handler,
                                 &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSRESPONSE],
                                 cc, requestId);
    if(res != UA_STATUSCODE_GOOD) {
        UA_DeleteMonitoredItemsRequest_delete(req_copy);
        UA_free(cc);
    }
    return res;
}

UA_StatusCode
UA_Client_MonitoredItems_deleteSingle(UA_Client *client, UA_UInt32 subscriptionId,
                                      UA_UInt32 monitoredItemId) {
    UA_DeleteMonitoredItemsRequest request;
    UA_DeleteMonitoredItemsRequest_init(&request);
    request.subscriptionId = subscriptionId;
    request.monitoredItemIds = &monitoredItemId;
    request.monitoredItemIdsSize = 1;

    UA_DeleteMonitoredItemsResponse response =
        UA_Client_MonitoredItems_delete(client, request);

    UA_StatusCode retval = response.responseHeader.serviceResult;
    if(retval != UA_STATUSCODE_GOOD) {
        UA_DeleteMonitoredItemsResponse_clear(&response);
        return retval;
    }

    if(response.resultsSize != 1) {
        UA_DeleteMonitoredItemsResponse_clear(&response);
        return UA_STATUSCODE_BADINTERNALERROR;
    }

    retval = response.results[0];
    UA_DeleteMonitoredItemsResponse_clear(&response);
    return retval;
}

static void *
MonitoredItem_findByID(void *data, UA_Client_MonitoredItem *mon) {
    UA_UInt32 monitorId = *(UA_UInt32*)data;
    if(monitorId && (mon->monitoredItemId == monitorId))
        return mon;
    return NULL;
}

static UA_Client_MonitoredItem *
findMonitoredItemById(UA_Client_Subscription *sub, UA_UInt32 monitoredItemId) {
    return (UA_Client_MonitoredItem *)
        ZIP_ITER(MonitorItemsTree, &sub->monitoredItems,
                 MonitoredItem_findByID, &monitoredItemId);
}

static void
setExistingClientHandles(UA_Client_Subscription *sub,
                         UA_ModifyMonitoredItemsRequest *request) {
    for(size_t i = 0; i < request->itemsToModifySize; ++i) {
        UA_MonitoredItemModifyRequest *mimr = &request->itemsToModify[i];
        UA_Client_MonitoredItem *mon =
            findMonitoredItemById(sub, mimr->monitoredItemId);
        if(mon)
            mimr->requestedParameters.clientHandle = mon->clientHandle;
    }
}

UA_ModifyMonitoredItemsResponse
UA_Client_MonitoredItems_modify(UA_Client *client,
                                const UA_ModifyMonitoredItemsRequest request) {
    UA_ModifyMonitoredItemsResponse response;
    UA_ModifyMonitoredItemsResponse_init(&response);

    /* Make a modifiable copy of the request */
    UA_ModifyMonitoredItemsRequest modifiedRequest;
    UA_StatusCode res = UA_ModifyMonitoredItemsRequest_copy(&request, &modifiedRequest);
    if(res != UA_STATUSCODE_GOOD) {
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
        return response;
    }

    lockClient(client);

    /* Get the subscription */
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        unlockClient(client);
        UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
        return response;
    }

    /* Reuse the existing ClientHandles for the MonitoredItems */
    setExistingClientHandles(sub, &modifiedRequest);

    /* Call the service */
    __Client_Service(client, &modifiedRequest,
                     &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSREQUEST], &response,
                     &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSRESPONSE]);

    unlockClient(client);

    /* Clean up */
    UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
    return response;
}

UA_StatusCode
UA_Client_MonitoredItems_modify_async(UA_Client *client,
                                      const UA_ModifyMonitoredItemsRequest request,
                                      UA_ClientAsyncModifyMonitoredItemsCallback callback,
                                      void *userdata, UA_UInt32 *requestId) {
    /* Make a modifiable copy of the request */
    UA_ModifyMonitoredItemsRequest modifiedRequest;
    UA_StatusCode res = UA_ModifyMonitoredItemsRequest_copy(&request, &modifiedRequest);
    if(res != UA_STATUSCODE_GOOD)
        return res;

    lockClient(client);

    /* Get the subscription */
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
    if(!sub) {
        unlockClient(client);
        UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    /* Reuse the existing ClientHandles for the MonitoredItems */
    setExistingClientHandles(sub, &modifiedRequest);

    /* Call the service */
    UA_StatusCode statusCode = __Client_AsyncService(
        client, &modifiedRequest, &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSREQUEST],
        (UA_ClientAsyncServiceCallback)callback,
        &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSRESPONSE], userdata, requestId);

    unlockClient(client);

    /* Clean up */
    UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
    return statusCode;
}

UA_StatusCode
UA_Client_MonitoredItem_getContext(UA_Client *client, UA_UInt32 subscriptionId,
                                   UA_UInt32 monitoredItemId, void **monContext) {
    if(!client || !monContext)
        return UA_STATUSCODE_BADINVALIDARGUMENT;

    *monContext = NULL;

    lockClient(client);
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
    if(!sub) {
        unlockClient(client);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    UA_StatusCode itemstatus = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
    UA_Client_MonitoredItem *monItem = findMonitoredItemById(sub, monitoredItemId);
    if(monItem) {
        *monContext = monItem->context;
        itemstatus = UA_STATUSCODE_GOOD;
    }
    unlockClient(client);
    return itemstatus;
}

UA_StatusCode
UA_Client_MonitoredItem_setContext(UA_Client *client, UA_UInt32 subscriptionId,
                                   UA_UInt32 monitoredItemId, void *monContext) {
    if(!client)
        return UA_STATUSCODE_BADINVALIDARGUMENT;

    lockClient(client);
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
    if(!sub) {
        unlockClient(client);
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
    }

    UA_StatusCode itemstatus = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
    UA_Client_MonitoredItem *monItem = findMonitoredItemById(sub, monitoredItemId);
    if(monItem) {
        monItem->context = monContext;
        itemstatus = UA_STATUSCODE_GOOD;
    }
    unlockClient(client);
    return itemstatus;
}

/*************************************/
/* Async Processing of Notifications */
/*************************************/

/* Assume the request is already initialized */
UA_StatusCode
__Client_preparePublishRequest(UA_Client *client, UA_PublishRequest *request) {
    UA_LOCK_ASSERT(&client->clientMutex);

    /* Count acks */
    UA_Client_NotificationsAckNumber *ack;
    LIST_FOREACH(ack, &client->pendingNotificationsAcks, listEntry)
        ++request->subscriptionAcknowledgementsSize;

    /* Create the array. Returns a sentinel pointer if the length is zero. */
    request->subscriptionAcknowledgements = (UA_SubscriptionAcknowledgement*)
        UA_Array_new(request->subscriptionAcknowledgementsSize,
                     &UA_TYPES[UA_TYPES_SUBSCRIPTIONACKNOWLEDGEMENT]);
    if(!request->subscriptionAcknowledgements) {
        request->subscriptionAcknowledgementsSize = 0;
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }

    size_t i = 0;
    UA_Client_NotificationsAckNumber *ack_tmp;
    LIST_FOREACH_SAFE(ack, &client->pendingNotificationsAcks, listEntry, ack_tmp) {
        LIST_REMOVE(ack, listEntry);
        UA_SubscriptionAcknowledgement *reqAck = &request->subscriptionAcknowledgements[i];
        reqAck->sequenceNumber = ack->subAck.sequenceNumber;
        reqAck->subscriptionId = ack->subAck.subscriptionId;
        UA_free(ack);
        i++;
    }
    return UA_STATUSCODE_GOOD;
}

/* According to specification, Part 4 5.13.1, the value 0 is never used for the
 * sequence number */
static UA_UInt32
__nextSequenceNumber(UA_UInt32 sequenceNumber) {
    UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
    if(nextSequenceNumber == 0)
        nextSequenceNumber = 1;
    return nextSequenceNumber;
}

static void
processDataChangeNotification(UA_Client *client, UA_Client_Subscription *sub,
                              UA_DataChangeNotification *dataChangeNotification) {
    UA_LOCK_ASSERT(&client->clientMutex);

    for(size_t j = 0; j < dataChangeNotification->monitoredItemsSize; ++j) {
        UA_MonitoredItemNotification *min = &dataChangeNotification->monitoredItems[j];

        /* Find the MonitoredItem */
        UA_Client_MonitoredItem *mon;
        UA_Client_MonitoredItem dummy;
        dummy.clientHandle = min->clientHandle;
        mon = ZIP_FIND(MonitorItemsTree, &sub->monitoredItems, &dummy);

        if(!mon) {
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "Could not process a notification with clienthandle %" PRIu32
                           " on subscription %" PRIu32, min->clientHandle, sub->subscriptionId);
            continue;
        }

        if(mon->isEventMonitoredItem) {
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "MonitoredItem is configured for Events. But received a "
                           "DataChangeNotification.");
            continue;
        }

        if(mon->handler.dataChangeCallback) {
            void *subC = sub->context;
            void *monC = mon->context;
            UA_UInt32 subId = sub->subscriptionId;
            UA_UInt32 monId = mon->monitoredItemId;
            mon->handler.dataChangeCallback(client, subId, subC, monId, monC, &min->value);
        }
    }
}

static void
processEventNotification(UA_Client *client, UA_Client_Subscription *sub,
                         UA_EventNotificationList *eventNotificationList) {
    UA_LOCK_ASSERT(&client->clientMutex);

    for(size_t j = 0; j < eventNotificationList->eventsSize; ++j) {
        UA_EventFieldList *efl = &eventNotificationList->events[j];

        /* Find the MonitoredItem */
        UA_Client_MonitoredItem *mon;
        UA_Client_MonitoredItem dummy;
        dummy.clientHandle = efl->clientHandle;
        mon = ZIP_FIND(MonitorItemsTree, &sub->monitoredItems, &dummy);

        if(!mon) {
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                         "Could not process a notification with clienthandle %" PRIu32
                         " on subscription %" PRIu32, efl->clientHandle,
                         sub->subscriptionId);
            continue;
        }

        if(!mon->isEventMonitoredItem) {
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                         "MonitoredItem is configured for DataChanges. But received a "
                         "EventNotification");
            continue;
        }

        if(mon->eventFields.mapSize != efl->eventFieldsSize) {
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                         "MonitoredItem received a EventNotification with the "
                          "wrong number of event fields");
            continue;
        }

        /* Prepare the key-value map and call the callback  */
        for(size_t i = 0; i < mon->eventFields.mapSize; i++) {
            mon->eventFields.map[i].value = efl->eventFields[i];
        }
        mon->handler.eventCallback(client, sub->subscriptionId, sub->context,
                                   mon->monitoredItemId, mon->context, mon->eventFields);
    }
}

static void
processNotificationMessage(UA_Client *client, UA_Client_Subscription *sub,
                           UA_ExtensionObject *msg) {
    UA_LOCK_ASSERT(&client->clientMutex);

    if(msg->encoding != UA_EXTENSIONOBJECT_DECODED)
        return;

    /* Handle DataChangeNotification */
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION]) {
        UA_DataChangeNotification *dataChangeNotification =
            (UA_DataChangeNotification *)msg->content.decoded.data;
        processDataChangeNotification(client, sub, dataChangeNotification);
        return;
    }

    /* Handle EventNotification */
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST]) {
        UA_EventNotificationList *eventNotificationList =
            (UA_EventNotificationList *)msg->content.decoded.data;
        processEventNotification(client, sub, eventNotificationList);
        return;
    }

    /* Handle StatusChangeNotification */
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION]) {
        if(sub->statusChangeCallback) {
            void *subC = sub->context;
            UA_UInt32 subId = sub->subscriptionId;
            sub->statusChangeCallback(client, subId, subC,
                                      (UA_StatusChangeNotification*)msg->content.decoded.data);
        } else {
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "Dropped a StatusChangeNotification since no "
                           "callback is registered");
        }
        return;
    }

    UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                   "Unknown notification message type");
}

static void
__Client_Subscriptions_processPublishResponse(UA_Client *client, UA_PublishRequest *request,
                                              UA_PublishResponse *response) {
    UA_LOCK_ASSERT(&client->clientMutex);

    /* Reduce the number of "in-flight" PublishRequests */
    client->currentlyOutStandingPublishRequests--;

    /* Process ServiceResult for bad StatusCodes without referring to a
     * SubscriptionId */
    switch(response->responseHeader.serviceResult) {
    case UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS:
        /* Correct the assumed number of max outstanding requests */
        if(client->config.outStandingPublishRequests > 1) {
            client->config.outStandingPublishRequests--;
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "PublishResponse: Too many PublishRequest, reduce "
                           "outStandingPublishRequests to %" PRId16,
                           client->config.outStandingPublishRequests);
        } else {
            UA_LOG_ERROR(client->config.logging, UA_LOGCATEGORY_CLIENT,
                         "PublishResponse: Too many PublishRequests when "
                         "outStandingPublishRequests = 1");
            UA_Client_Subscriptions_deleteSingle(client, response->subscriptionId);
        }
        return;

    case UA_STATUSCODE_BADSHUTDOWN:
        /* If the remote server shuts down, DEBUG-log to avoid a warning-storm
         * for normal operations */
        UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                     "PublishResponse: Received BadShutdown status");
        return;

    case UA_STATUSCODE_BADNOSUBSCRIPTION:
        /* There is no Subscription configured, the server expects no
         * PublishRequests. We demote this to debug-logging, as it can occur
         * during regular shutdown when the Subscriptions are removed. */
        UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
                     "PublishResponse: Received BadNoSubscription status");
        return;

    default:
        break;
    }

    /* Get the Subscription */
    UA_Client_Subscription *sub = findSubscriptionById(client, response->subscriptionId);
    if(!sub) {
        response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                       "PublishResponse: Received response for an unknown Subscription");
        return;
    }

    /* Process ServiceResult */
    switch(response->responseHeader.serviceResult) {
    case UA_STATUSCODE_BADSESSIONCLOSED:
        /* The Session no longer exists on the server - remove the Subscription */
        __Client_Subscription_deleteInternal(client, sub);
        return;

    case UA_STATUSCODE_BADTIMEOUT:
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                       "PublishResponse: Aborted with BadTimeout status");
        if(client->config.subscriptionInactivityCallback) {
            void *subC = sub->context;
            UA_UInt32 subId = sub->subscriptionId;
            client->config.subscriptionInactivityCallback(client, subId, subC);
        }
        return;

    case UA_STATUSCODE_GOOD:
        break; /* Continue below */

    default:
        /* Catch-all for other bad StatusCodes */
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                       "PublishResponse: Received %s status",
                       UA_StatusCode_name(response->responseHeader.serviceResult));
        return;
    }

    /* Update the LastActivity for the Subscription */
    UA_EventLoop *el = client->config.eventLoop;
    sub->lastActivity = el->dateTime_nowMonotonic(el);

    /* Detect missing message - OPC Unified Architecture, Part 4 5.13.1.1 e) */
    UA_NotificationMessage *msg = &response->notificationMessage;
    if(__nextSequenceNumber(sub->sequenceNumber) != msg->sequenceNumber) {
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                       "PublishResponse: Invalid subscription sequence number: "
                       "Expected %" PRIu32 " but got %" PRIu32,
                       __nextSequenceNumber(sub->sequenceNumber), msg->sequenceNumber);
        /* This is an error. But we do not abort the connection. Some server
         * SDKs misbehave from time to time and send out-of-order sequence
         * numbers. (Probably some multi-threading synchronization issue.) */
        /* UA_Client_disconnect(client);
           return; */
    }

    /* According to f), a keep-alive message contains no notifications and has
     * the sequence number of the next NotificationMessage that is to be sent =>
     * More than one consecutive keep-alive message or a NotificationMessage
     * following a keep-alive message will share the same sequence number. */
    if(msg->notificationDataSize)
        sub->sequenceNumber = msg->sequenceNumber;

    /* Process the notification messages */
    for(size_t k = 0; k < msg->notificationDataSize; ++k)
        processNotificationMessage(client, sub, &msg->notificationData[k]);

    /* Add the current NotificationMessage (SequenceNumber) to the list of
     * pending acks to be acknowledged. But only if it is in the list of
     * sequence numbers the server has available. */
    for(size_t i = 0; i < response->availableSequenceNumbersSize; i++) {
        if(response->availableSequenceNumbers[i] != msg->sequenceNumber)
            continue;
        UA_Client_NotificationsAckNumber *tmpAck = (UA_Client_NotificationsAckNumber*)
            UA_malloc(sizeof(UA_Client_NotificationsAckNumber));
        if(!tmpAck) {
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "PublishResponse: Not enough memory to store the pending "
                           "acknowledgement for Subscription %" PRIu32, sub->subscriptionId);
            break;
        }
        tmpAck->subAck.sequenceNumber = msg->sequenceNumber;
        tmpAck->subAck.subscriptionId = sub->subscriptionId;
        LIST_INSERT_HEAD(&client->pendingNotificationsAcks, tmpAck, listEntry);
        break;
    }
}

static void
processPublishResponseAsync(UA_Client *client, void *userdata,
                            UA_UInt32 requestId, void *response) {
    UA_PublishRequest *req = (UA_PublishRequest*)userdata;
    UA_PublishResponse *res = (UA_PublishResponse*)response;

    lockClient(client);

    /* Process the response */
    __Client_Subscriptions_processPublishResponse(client, req, res);

    /* Delete the cached request */
    UA_PublishRequest_delete(req);

    /* Fill up the outstanding publish requests */
    __Client_Subscriptions_backgroundPublish(client);

    unlockClient(client);
}

void
__Client_Subscriptions_clear(UA_Client *client) {
    UA_Client_NotificationsAckNumber *n;
    UA_Client_NotificationsAckNumber *tmp;
    LIST_FOREACH_SAFE(n, &client->pendingNotificationsAcks, listEntry, tmp) {
        LIST_REMOVE(n, listEntry);
        UA_free(n);
    }

    UA_Client_Subscription *sub;
    UA_Client_Subscription *tmps;
    LIST_FOREACH_SAFE(sub, &client->subscriptions, listEntry, tmps)
        __Client_Subscription_deleteInternal(client, sub); /* force local removal */

    client->monitoredItemHandles = 0;
}

void
__Client_Subscriptions_backgroundPublishInactivityCheck(UA_Client *client) {
    UA_LOCK_ASSERT(&client->clientMutex);

    UA_EventLoop *el = client->config.eventLoop;
    UA_DateTime nowm = el->dateTime_nowMonotonic(el);

    UA_Client_Subscription *sub;
    LIST_FOREACH(sub, &client->subscriptions, listEntry) {
        UA_DateTime maxSilence = (UA_DateTime)
            ((sub->publishingInterval * sub->maxKeepAliveCount) +
             client->config.timeout) * UA_DATETIME_MSEC;
        if(maxSilence + sub->lastActivity < nowm) {
            /* Reset activity */
            sub->lastActivity = nowm;

            if(client->config.subscriptionInactivityCallback) {
                void *subC = sub->context;
                UA_UInt32 subId = sub->subscriptionId;
                client->config.subscriptionInactivityCallback(client, subId, subC);
            }
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
                           "Inactivity for Subscription %" PRIu32 ".",
                           sub->subscriptionId);
        }
    }
}

void
__Client_Subscriptions_backgroundPublish(UA_Client *client) {
    UA_LOCK_ASSERT(&client->clientMutex);

    if(client->sessionState != UA_SESSIONSTATE_ACTIVATED)
        return;

    /* The session must have at least one subscription */
    if(!LIST_FIRST(&client->subscriptions))
        return;

    while(client->currentlyOutStandingPublishRequests < client->config.outStandingPublishRequests) {
        UA_PublishRequest *request = UA_PublishRequest_new();
        if(!request)
            return;

        /* Publish requests are valid for 10 minutes */
        request->requestHeader.timeoutHint = 10 * 60 * 1000;

        UA_StatusCode retval = __Client_preparePublishRequest(client, request);
        if(retval != UA_STATUSCODE_GOOD) {
            UA_PublishRequest_delete(request);
            return;
        }

        retval = __Client_AsyncService(client, request,
                                         &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
                                         processPublishResponseAsync,
                                         &UA_TYPES[UA_TYPES_PUBLISHRESPONSE],
                                         (void*)request, NULL);
        if(retval != UA_STATUSCODE_GOOD) {
            UA_PublishRequest_delete(request);
            return;
        }

        client->currentlyOutStandingPublishRequests++;
    }
}

UA_SetPublishingModeResponse
UA_Client_Subscriptions_setPublishingMode(UA_Client *client,
                                          const UA_SetPublishingModeRequest request) {
    UA_SetPublishingModeResponse response;
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETPUBLISHINGMODEREQUEST],
                        &response, &UA_TYPES[UA_TYPES_SETPUBLISHINGMODERESPONSE]);
    return response;
}

UA_SetMonitoringModeResponse
UA_Client_MonitoredItems_setMonitoringMode(UA_Client *client,
                                           const UA_SetMonitoringModeRequest request) {
    UA_SetMonitoringModeResponse response;
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETMONITORINGMODEREQUEST],
                        &response, &UA_TYPES[UA_TYPES_SETMONITORINGMODERESPONSE]);
    return response;
}

UA_StatusCode
UA_Client_MonitoredItems_setMonitoringMode_async(UA_Client *client,
                                                 const UA_SetMonitoringModeRequest request,
                                                 UA_ClientAsyncSetMonitoringModeCallback callback,
                                                 void *userdata, UA_UInt32 *requestId) {
    return __UA_Client_AsyncService(client, &request,
                                    &UA_TYPES[UA_TYPES_SETMONITORINGMODEREQUEST],
                                    (UA_ClientAsyncServiceCallback)callback,
                                    &UA_TYPES[UA_TYPES_SETMONITORINGMODERESPONSE],
                                    userdata, requestId);
}

UA_SetTriggeringResponse
UA_Client_MonitoredItems_setTriggering(UA_Client *client,
                                       const UA_SetTriggeringRequest request) {
    UA_SetTriggeringResponse response;
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETTRIGGERINGREQUEST],
                        &response, &UA_TYPES[UA_TYPES_SETTRIGGERINGRESPONSE]);
    return response;
}

UA_StatusCode
UA_Client_MonitoredItems_setTriggering_async(UA_Client *client,
                                             const UA_SetTriggeringRequest request,
                                             UA_ClientAsyncSetTriggeringCallback callback,
                                             void *userdata, UA_UInt32 *requestId) {
    return __UA_Client_AsyncService(client, &request,
                                    &UA_TYPES[UA_TYPES_SETTRIGGERINGREQUEST],
                                    (UA_ClientAsyncServiceCallback)callback,
                                    &UA_TYPES[UA_TYPES_SETTRIGGERINGRESPONSE],
                                    userdata, requestId);
}

Coverage Report

Created: 2026-06-30 06:45