open62541/tests/client/check_client_subscriptions.c

/* 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/. */

#include <open62541/client.h>
#include <open62541/client_highlevel_async.h>
#include <open62541/client_config_default.h>
#include <open62541/server.h>
#include <open62541/server_config_default.h>

#include "client/ua_client_internal.h"
#include "server/ua_server_internal.h"
#include "test_helpers.h"

#include <stdio.h>
#include <stdlib.h>

#include "check.h"
#include "testing_clock.h"
#include "thread_wrapper.h"

UA_Server *server;
UA_atomic(uintptr_t) running;
static UA_Boolean serverThreadRunning;
THREAD_HANDLE server_thread;
static UA_Boolean noNewSubscription; /* Don't create a subscription when the
                                        session activates */

THREAD_CALLBACK(serverloop) {
    while(UA_atomic_load(&running))
        UA_Server_run_iterate(server, true);
    return 0;
}

static void runServer(void) {
    if(serverThreadRunning)
        return;
    UA_atomic_store(&running, true);
    THREAD_CREATE(server_thread, serverloop);
    serverThreadRunning = true;
}

static void pauseServer(void) {
    if(!serverThreadRunning)
        return;
    UA_atomic_store(&running, false);
    THREAD_JOIN(server_thread);
    serverThreadRunning = false;
}

static void setup(void) {
    noNewSubscription = false;
    UA_atomic_store(&running, true);
    server = UA_Server_newForUnitTest();
    ck_assert(server != NULL);
    UA_ServerConfig *config = UA_Server_getConfig(server);
    config->maxPublishReqPerSession = 5;
    UA_Server_run_startup(server);
    serverThreadRunning = false;
    runServer();
}

static void teardown(void) {
    if(!server)
        return;

    pauseServer();

    UA_Server_run_shutdown(server);
    UA_Server_delete(server);
    server = NULL;
}

UA_Boolean notificationReceived = false;
UA_UInt32 countNotificationReceived = 0;
UA_Double publishingInterval = 500.0;
UA_StatusCode statusChange = UA_STATUSCODE_GOOD;

static void
statusChangeHandler(UA_Client *client, UA_UInt32 subId, void *subContext,
                    UA_StatusChangeNotification *notification) {
    statusChange = notification->status;
}

static void
dataChangeHandler(UA_Client *client, UA_UInt32 subId, void *subContext,
                  UA_UInt32 monId, void *monContext, UA_DataValue *value) {
    notificationReceived = true;
    countNotificationReceived++;
}

static void
iterateUntilNotification(UA_Client *client, UA_UInt32 maxIterations) {
    UA_StatusCode retval = UA_STATUSCODE_GOOD;
    for(UA_UInt32 i = 0; i < maxIterations && !notificationReceived; ++i) {
        UA_Server_run_iterate(server, false);
        /* Wait 1ms per iteration. A zero timeout makes this a busy-poll that
         * can return before the PublishResponse is readable on the socket. */
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
        UA_fakeSleep(1);
    }
}

static void
clearSubscriptionsDataChangeHandler(UA_Client *client, UA_UInt32 subId,
                                    void *subContext, UA_UInt32 monId,
                                    void *monContext, UA_DataValue *value) {
    __Client_Subscriptions_clear(client);
}

static void
clearSubscriptionsInactivityHandler(UA_Client *client, UA_UInt32 subId,
                                    void *subContext) {
    __Client_Subscriptions_clear(client);
}

static size_t clearedSubscriptions;

static void
clearSubscriptionsDeleteHandler(UA_Client *client, UA_UInt32 subId,
                                void *subContext) {
    clearedSubscriptions++;
    __Client_Subscriptions_clear(client);
}

static void
clearSubscriptionsMonitoredItemDeleteHandler(UA_Client *client,
                                             UA_UInt32 subId,
                                             void *subContext,
                                             UA_UInt32 monId,
                                             void *monContext) {
    __Client_Subscriptions_clear(client);
}

static UA_Client *
newConnectedClient(void) {
    /* Use the same fake clock as the server and the Valgrind-safe timeout. */
    UA_Client *client = UA_Client_newForUnitTest();
    ck_assert(client != NULL);
    ck_assert_uint_eq(UA_Client_connect(client, "opc.tcp://localhost:4840"),
                      UA_STATUSCODE_GOOD);
    return client;
}

static UA_UInt32
createTestSubscription(UA_Client *client,
                       UA_Client_DeleteSubscriptionCallback deleteCallback) {
    UA_CreateSubscriptionRequest request =
        UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL,
                                       deleteCallback);
    ck_assert_uint_eq(response.responseHeader.serviceResult,
                      UA_STATUSCODE_GOOD);
    return response.subscriptionId;
}

static UA_UInt32
createTestMonitoredItem(UA_Client *client, UA_UInt32 subId,
                        UA_Client_DataChangeNotificationCallback callback,
                        UA_Client_DeleteMonitoredItemCallback deleteCallback) {
    UA_MonitoredItemCreateRequest request =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    UA_MonitoredItemCreateResult response =
        UA_Client_MonitoredItems_createDataChange(
            client, subId, UA_TIMESTAMPSTORETURN_BOTH, request, NULL,
            callback, deleteCallback);
    ck_assert_uint_eq(response.statusCode, UA_STATUSCODE_GOOD);
    return response.monitoredItemId;
}

static void
createSubscriptionCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                           UA_CreateSubscriptionResponse *r) {
    UA_CreateSubscriptionResponse_copy(r, (UA_CreateSubscriptionResponse *)userdata);
}

static void
modifySubscriptionCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                           UA_ModifySubscriptionResponse *r) {
    UA_ModifySubscriptionResponse_copy(r, (UA_ModifySubscriptionResponse *)userdata);
}

static void
clearSubscriptionsOnRead(UA_Client *client, void *userdata,
                         UA_UInt32 requestId, void *response) {
    __Client_Subscriptions_clear(client);
}

static void
createDataChangesCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                          UA_CreateMonitoredItemsResponse *r) {
    UA_CreateMonitoredItemsResponse_copy(r, (UA_CreateMonitoredItemsResponse *)userdata);
}

static void
deleteMonitoredItemsCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                             UA_DeleteMonitoredItemsResponse *r) {
    UA_DeleteMonitoredItemsResponse_copy(r, (UA_DeleteMonitoredItemsResponse *)userdata);
}

static void
modifyMonitoredItemsCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                             UA_ModifyMonitoredItemsResponse *response) {
    UA_ModifyMonitoredItemsResponse_copy(
        response, (UA_ModifyMonitoredItemsResponse*)userdata);
}

static void
injectNotification(UA_Client *client, UA_Client_Subscription *sub,
                   const UA_DataType *notificationType, void *notification) {
    UA_ExtensionObject notificationData;
    UA_ExtensionObject_init(&notificationData);
    notificationData.encoding = UA_EXTENSIONOBJECT_DECODED;
    notificationData.content.decoded.type = notificationType;
    notificationData.content.decoded.data = notification;

    UA_PublishRequest request;
    UA_PublishRequest_init(&request);
    UA_PublishResponse response;
    UA_PublishResponse_init(&response);
    response.responseHeader.serviceResult = UA_STATUSCODE_GOOD;
    response.subscriptionId = sub->subscriptionId;
    response.notificationMessage.sequenceNumber = sub->sequenceNumber + 1;
    response.notificationMessage.notificationData = &notificationData;
    response.notificationMessage.notificationDataSize = 1;

    lockClient(client);
    client->currentlyOutStandingPublishRequests++;
    __Client_Subscriptions_processPublishResponse(client, &request, &response);
    unlockClient(client);
}

static void
injectDataChangeNotification(UA_Client *client, UA_Client_Subscription *sub,
                             UA_UInt32 clientHandle) {
    UA_MonitoredItemNotification min;
    UA_MonitoredItemNotification_init(&min);
    min.clientHandle = clientHandle;

    UA_DataChangeNotification dcn;
    UA_DataChangeNotification_init(&dcn);
    dcn.monitoredItems = &min;
    dcn.monitoredItemsSize = 1;
    injectNotification(client, sub,
                       &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION], &dcn);
}

static size_t eventFieldsSize;

static void
eventHandler(UA_Client *client, UA_UInt32 subId, void *subContext,
             UA_UInt32 monId, void *monContext,
             const UA_KeyValueMap eventFields) {
    eventFieldsSize = eventFields.mapSize;
}

static void
injectEventNotification(UA_Client *client, UA_Client_Subscription *sub,
                        UA_UInt32 clientHandle, size_t fieldsSize) {
    UA_STACKARRAY(UA_Variant, fields, fieldsSize);
    for(size_t i = 0; i < fieldsSize; i++)
        UA_Variant_init(&fields[i]);

    UA_EventFieldList efl;
    UA_EventFieldList_init(&efl);
    efl.clientHandle = clientHandle;
    efl.eventFields = fields;
    efl.eventFieldsSize = fieldsSize;

    UA_EventNotificationList enl;
    UA_EventNotificationList_init(&enl);
    enl.events = &efl;
    enl.eventsSize = 1;
    injectNotification(client, sub,
                       &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST], &enl);
}

static UA_Client_MonitoredItem *
findTestMonitoredItem(UA_Client_MonitoredItem *mon, UA_UInt32 monitoredItemId) {
    if(!mon || mon->monitoredItemId == monitoredItemId)
        return mon;
    UA_Client_MonitoredItem *found =
        findTestMonitoredItem(ZIP_LEFT(mon, zipfields), monitoredItemId);
    if(found)
        return found;
    return findTestMonitoredItem(ZIP_RIGHT(mon, zipfields), monitoredItemId);
}

static size_t
countTestMonitoredItems(UA_Client_MonitoredItem *mon) {
    if(!mon)
        return 0;
    return 1 + countTestMonitoredItems(ZIP_LEFT(mon, zipfields)) +
        countTestMonitoredItems(ZIP_RIGHT(mon, zipfields));
}

static void
deleteSubscriptionsCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                            UA_DeleteSubscriptionsResponse *r) {
    UA_DeleteSubscriptionsResponse_copy(r, (UA_DeleteSubscriptionsResponse *)userdata);
}

START_TEST(Client_notification_callback_clears_subscription) {
    UA_Client *client = newConnectedClient();
    UA_UInt32 subId = createTestSubscription(client, NULL);
    createTestMonitoredItem(client, subId,
                            clearSubscriptionsDataChangeHandler, NULL);

    UA_StatusCode retval;
    pauseServer();
    UA_Server_run_iterate(server, true);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, true);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert(LIST_EMPTY(&client->subscriptions));

    runServer();
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_monitoreditem_delete_callback_clears_subscriptions) {
    UA_Client *client = newConnectedClient();
    UA_UInt32 subId = createTestSubscription(client, NULL);
    UA_UInt32 monIds[2];
    monIds[0] = createTestMonitoredItem(
        client, subId, dataChangeHandler,
        clearSubscriptionsMonitoredItemDeleteHandler);
    monIds[1] = createTestMonitoredItem(client, subId,
                                        dataChangeHandler, NULL);
    UA_DeleteMonitoredItemsRequest deleteRequest;
    UA_DeleteMonitoredItemsRequest_init(&deleteRequest);
    deleteRequest.subscriptionId = subId;
    deleteRequest.monitoredItemIdsSize = 2;
    deleteRequest.monitoredItemIds = monIds;
    UA_DeleteMonitoredItemsResponse deleteResponse =
        UA_Client_MonitoredItems_delete(client, deleteRequest);
    ck_assert_uint_eq(deleteResponse.responseHeader.serviceResult,
                      UA_STATUSCODE_GOOD);
    UA_DeleteMonitoredItemsResponse_clear(&deleteResponse);
    ck_assert(LIST_EMPTY(&client->subscriptions));

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_monitoreditem_create_callback_clears_subscriptions) {
    UA_Client *client = newConnectedClient();
    UA_UInt32 subId = createTestSubscription(client, NULL);
    UA_MonitoredItemCreateRequest items[2];
    items[0] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UINT32_MAX));
    items[1] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    void *contexts[2] = {NULL, NULL};
    UA_Client_DataChangeNotificationCallback callbacks[2] = {
        dataChangeHandler, dataChangeHandler};
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks[2] = {
        clearSubscriptionsMonitoredItemDeleteHandler, NULL};
    UA_CreateMonitoredItemsRequest request;
    UA_CreateMonitoredItemsRequest_init(&request);
    request.subscriptionId = subId;
    request.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    request.itemsToCreate = items;
    request.itemsToCreateSize = 2;
    UA_CreateMonitoredItemsResponse response;
    UA_CreateMonitoredItemsResponse_init(&response);
    if(_i == 0) {
        response = UA_Client_MonitoredItems_createDataChanges(
            client, request, contexts, callbacks, deleteCallbacks);
    } else {
        /* Drive both peers on this thread. A fast client-only loop can exhaust
         * its iterations before the server is scheduled under Valgrind. */
        pauseServer();
        response.responseHeader.serviceResult = UA_STATUSCODE_BADWAITINGFORINITIALDATA;
        UA_StatusCode res = UA_Client_MonitoredItems_createDataChanges_async(
            client, request, contexts, callbacks, deleteCallbacks,
            createDataChangesCallback, &response, NULL);
        ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);
        for(size_t i = 0; i < 1000 && response.responseHeader.serviceResult ==
                UA_STATUSCODE_BADWAITINGFORINITIALDATA; i++) {
            UA_Server_run_iterate(server, false);
            res = UA_Client_run_iterate(client, 10);
            ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);
        }
        runServer();
        ck_assert_uint_eq(response.resultsSize, 2);
    }
    ck_assert_uint_eq(response.responseHeader.serviceResult,
                      UA_STATUSCODE_GOOD);
    UA_CreateMonitoredItemsResponse_clear(&response);
    ck_assert(LIST_EMPTY(&client->subscriptions));

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_delete_callback_clears_subscriptions) {
    UA_Client *client = newConnectedClient();
    createTestSubscription(client, clearSubscriptionsDeleteHandler);
    createTestSubscription(client, clearSubscriptionsDeleteHandler);
    clearedSubscriptions = 0;

    lockClient(client);
    __Client_Subscriptions_clear(client);
    unlockClient(client);
    ck_assert(LIST_EMPTY(&client->subscriptions));
    ck_assert_uint_eq(clearedSubscriptions, 2);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_inactivity_callback_clears_subscription) {
    UA_Client *client = newConnectedClient();
    UA_ClientConfig *config = UA_Client_getConfig(client);
    config->subscriptionInactivityCallback =
        clearSubscriptionsInactivityHandler;
    createTestSubscription(client, NULL);

    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert(sub != NULL);
    sub->lastActivity = 0;
    client->currentlyOutStandingPublishRequests = 1;
    lockClient(client);
    __Client_Subscriptions_backgroundPublishInactivityCheck(client);
    unlockClient(client);
    ck_assert(LIST_EMPTY(&client->subscriptions));

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    // a valid UA_Client_Subscriptions_modify
    UA_ModifySubscriptionRequest modifySubscriptionRequest;
    UA_ModifySubscriptionRequest_init(&modifySubscriptionRequest);
    modifySubscriptionRequest.subscriptionId = response.subscriptionId;
    modifySubscriptionRequest.requestedPublishingInterval = response.revisedPublishingInterval;
    modifySubscriptionRequest.requestedLifetimeCount = response.revisedLifetimeCount;
    modifySubscriptionRequest.requestedMaxKeepAliveCount = response.revisedMaxKeepAliveCount;
    UA_ModifySubscriptionResponse modifySubscriptionResponse =
        UA_Client_Subscriptions_modify(client,modifySubscriptionRequest);
    ck_assert_int_eq(modifySubscriptionResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    // an invalid UA_Client_Subscriptions_modify
    modifySubscriptionRequest.subscriptionId = 99999; // invalid
    modifySubscriptionResponse = UA_Client_Subscriptions_modify(client,modifySubscriptionRequest);
    ck_assert_int_eq(modifySubscriptionResponse.responseHeader.serviceResult,
                     UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);
    UA_UInt32 monId = monResponse.monitoredItemId;

    /* manually control the server thread */
    pauseServer();

    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    notificationReceived = false;
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);

    /* run the server in an independent thread again */
    runServer();

    retval = UA_Client_MonitoredItems_deleteSingle(client, subId, monId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_async) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();

    /* manually control the server thread */
    pauseServer();

    UA_UInt32 requestId = 0;
    UA_CreateSubscriptionResponse response;
    retval = UA_Client_Subscriptions_create_async(client, request, NULL, NULL, NULL,
                                                  createSubscriptionCallback, &response,
                                                  &requestId);

    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    response.responseHeader.serviceResult = 1;
    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(response.responseHeader.serviceResult == 1);

    UA_UInt32 subId = response.subscriptionId;

    UA_ModifySubscriptionRequest modifySubscriptionRequest;
    UA_ModifySubscriptionRequest_init(&modifySubscriptionRequest);
    modifySubscriptionRequest.subscriptionId = response.subscriptionId;
    modifySubscriptionRequest.requestedPublishingInterval =
        response.revisedPublishingInterval;
    modifySubscriptionRequest.requestedLifetimeCount = response.revisedLifetimeCount;
    modifySubscriptionRequest.requestedMaxKeepAliveCount =
        response.revisedMaxKeepAliveCount;

    UA_ModifySubscriptionResponse modifySubscriptionResponse;
    retval = UA_Client_Subscriptions_modify_async(client, modifySubscriptionRequest,
                                                  modifySubscriptionCallback,
                                                  &modifySubscriptionResponse, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    modifySubscriptionResponse.responseHeader.serviceResult = 1;
    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(modifySubscriptionResponse.responseHeader.serviceResult == 1);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest singleMonRequest =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    void *contexts = NULL;
    UA_Client_DataChangeNotificationCallback notifications = dataChangeHandler;
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks = NULL;

    UA_CreateMonitoredItemsRequest monRequest;
    UA_CreateMonitoredItemsRequest_init(&monRequest);
    monRequest.subscriptionId = subId;
    monRequest.itemsToCreate = &singleMonRequest;
    monRequest.itemsToCreateSize = 1;
    UA_CreateMonitoredItemsResponse monResponse;
    UA_CreateMonitoredItemsResponse_init(&monResponse);
    retval = UA_Client_MonitoredItems_createDataChanges_async(client, monRequest,
                                                              &contexts, &notifications, &deleteCallbacks,
                                                              createDataChangesCallback, &monResponse, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 0);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(monResponse.resultsSize == 0 &&
            monResponse.responseHeader.serviceResult == UA_STATUSCODE_GOOD);

    ck_assert_uint_eq(monResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(monResponse.resultsSize, 1);
    ck_assert_uint_eq(monResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_UInt32 monId = monResponse.results[0].monitoredItemId;

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    notificationReceived = false;
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);

    UA_DeleteMonitoredItemsRequest monDeleteRequest;
    UA_DeleteMonitoredItemsRequest_init(&monDeleteRequest);
    monDeleteRequest.subscriptionId = subId;
    monDeleteRequest.monitoredItemIds = &monId;
    monDeleteRequest.monitoredItemIdsSize = 1;
    UA_DeleteMonitoredItemsResponse monDeleteResponse;
    UA_DeleteMonitoredItemsResponse_init(&monDeleteResponse);

    retval = UA_Client_MonitoredItems_delete_async(client, monDeleteRequest,
                                                   deleteMonitoredItemsCallback,
                                                   &monDeleteResponse, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    monDeleteResponse.responseHeader.serviceResult = 1;
    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(monDeleteResponse.responseHeader.serviceResult == 1);

    ck_assert_uint_eq(monDeleteResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(monDeleteResponse.resultsSize, 1);
    ck_assert_uint_eq(monDeleteResponse.results[0], UA_STATUSCODE_GOOD);

    UA_DeleteSubscriptionsRequest subDeleteRequest;
    UA_DeleteSubscriptionsRequest_init(&subDeleteRequest);
    subDeleteRequest.subscriptionIds = &monId;
    subDeleteRequest.subscriptionIdsSize = 1;
    UA_DeleteSubscriptionsResponse subDeleteResponse;
    UA_DeleteSubscriptionsResponse_init(&subDeleteResponse);
    retval = UA_Client_Subscriptions_delete_async(client, subDeleteRequest,
                                                  deleteSubscriptionsCallback,
                                                  &subDeleteResponse, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    subDeleteResponse.responseHeader.serviceResult = 1;
    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(subDeleteResponse.responseHeader.serviceResult == 1);

    ck_assert_uint_eq(subDeleteResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(subDeleteResponse.resultsSize, 1);
    ck_assert_uint_eq(subDeleteResponse.results[0], UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionResponse_clear(&response);
    UA_ModifySubscriptionResponse_clear(&modifySubscriptionResponse);
    UA_CreateMonitoredItemsResponse_clear(&monResponse);
    UA_DeleteMonitoredItemsResponse_clear(&monDeleteResponse);
    UA_DeleteSubscriptionsResponse_clear(&subDeleteResponse);

    /* run the server in an independent thread again */
    runServer();

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_delete_async_noCallback) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create a subscription synchronously */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* manually control the server thread */
    pauseServer();

    /* Delete the subscription asynchronously without a userland callback.
     * This must not dereference a NULL callback pointer. */
    UA_UInt32 requestId = 0;
    UA_DeleteSubscriptionsRequest subDeleteRequest;
    UA_DeleteSubscriptionsRequest_init(&subDeleteRequest);
    subDeleteRequest.subscriptionIds = &subId;
    subDeleteRequest.subscriptionIdsSize = 1;
    retval = UA_Client_Subscriptions_delete_async(client, subDeleteRequest,
                                                  NULL, NULL, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Run until the async response has been processed and the subscription
     * has been removed from the internal representation. A missing NULL-check
     * on the userland callback would crash here. */
    void *ctx = NULL;
    UA_DateTime maxStop = UA_DateTime_nowMonotonic() + (UA_DateTime)(2 * UA_DATETIME_SEC);
    do {
        UA_Server_run_iterate(server, true);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(UA_Client_Subscriptions_getContext(client, subId, &ctx) ==
            UA_STATUSCODE_GOOD &&
            UA_DateTime_nowMonotonic() < maxStop);

    /* The subscription has been removed from the internal representation */
    retval = UA_Client_Subscriptions_getContext(client, subId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    /* run the server in an independent thread again */
    runServer();

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_createDataChanges) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    UA_MonitoredItemCreateRequest items[3];
    UA_UInt32 newMonitoredItemIds[3];
    UA_Client_DataChangeNotificationCallback callbacks[3];
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks[3];
    void *contexts[3];

    /* monitor the server state */
    items[0] = UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    callbacks[0] = dataChangeHandler;
    contexts[0] = NULL;
    deleteCallbacks[0] = NULL;

    /* monitor invalid node */
    items[1] = UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, 999999));
    callbacks[1] = dataChangeHandler;
    contexts[1] = NULL;
    deleteCallbacks[1] = NULL;

    /* monitor current time */
    items[2] = UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    callbacks[2] = dataChangeHandler;
    contexts[2] = NULL;
    deleteCallbacks[2] = NULL;

    UA_CreateMonitoredItemsRequest createRequest;
    UA_CreateMonitoredItemsRequest_init(&createRequest);
    createRequest.subscriptionId = subId;
    createRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    createRequest.itemsToCreate = items;
    createRequest.itemsToCreateSize = 3;
    UA_CreateMonitoredItemsResponse createResponse =
       UA_Client_MonitoredItems_createDataChanges(client, createRequest, contexts,
                                                   callbacks, deleteCallbacks);

    ck_assert_uint_eq(createResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(createResponse.resultsSize, 3);
    ck_assert_uint_eq(createResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    newMonitoredItemIds[0] = createResponse.results[0].monitoredItemId;
    ck_assert_uint_eq(createResponse.results[1].statusCode, UA_STATUSCODE_BADNODEIDUNKNOWN);
    newMonitoredItemIds[1] = createResponse.results[1].monitoredItemId;
    ck_assert_uint_eq(newMonitoredItemIds[1], 0);
    ck_assert_uint_eq(createResponse.results[2].statusCode, UA_STATUSCODE_GOOD);
    newMonitoredItemIds[2] = createResponse.results[2].monitoredItemId;
    ck_assert_uint_eq(createResponse.results[2].statusCode, UA_STATUSCODE_GOOD);
    UA_CreateMonitoredItemsResponse_clear(&createResponse);

    /* manually control the server thread */
    pauseServer();

    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    countNotificationReceived = 0;
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 2);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 3);

    /* run the server in an independent thread again */
    runServer();

    UA_DeleteMonitoredItemsRequest deleteRequest;
    UA_DeleteMonitoredItemsRequest_init(&deleteRequest);
    deleteRequest.subscriptionId = subId;
    deleteRequest.monitoredItemIds = newMonitoredItemIds;
    deleteRequest.monitoredItemIdsSize = 3;

    UA_DeleteMonitoredItemsResponse deleteResponse =
        UA_Client_MonitoredItems_delete(client, deleteRequest);

    ck_assert_uint_eq(deleteResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(deleteResponse.resultsSize, 3);
    ck_assert_uint_eq(deleteResponse.results[0], UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(deleteResponse.results[1], UA_STATUSCODE_BADMONITOREDITEMIDINVALID);
    ck_assert_uint_eq(deleteResponse.results[2], UA_STATUSCODE_GOOD);

    UA_DeleteMonitoredItemsResponse_clear(&deleteResponse);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

/* An interval of -1 links the subscription to the publishing interval of the
 * server */
START_TEST(Client_subscription_createDataChanges_negativeInterval) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* monitor current time */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    monRequest.requestedParameters.samplingInterval = -2.0;

    UA_MonitoredItemCreateResult monResult =
        UA_Client_MonitoredItems_createDataChange(client, subId, UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);

    ck_assert_uint_eq(monResult.statusCode, UA_STATUSCODE_GOOD);
    ck_assert(monResult.revisedSamplingInterval == response.revisedPublishingInterval);
    UA_MonitoredItemCreateResult_clear(&monResult);
    UA_CreateSubscriptionResponse_clear(&response);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

/* An unchanged value shall not be published after a ModifyMonitoredItem */
START_TEST(Client_subscription_modifyMonitoredItem) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    UA_MonitoredItemCreateRequest items[1];
    UA_UInt32 newMonitoredItemIds[1];
    UA_Client_DataChangeNotificationCallback callbacks[1];
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks[1];
    void *contexts[1];

    /* Monitor the server state. Does not change during the unit test. */
    items[0] = UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    items[0].requestedParameters.samplingInterval = publishingInterval * 0.2;
    callbacks[0] = dataChangeHandler;
    contexts[0] = NULL;
    deleteCallbacks[0] = NULL;

    UA_CreateMonitoredItemsRequest createRequest;
    UA_CreateMonitoredItemsRequest_init(&createRequest);
    createRequest.subscriptionId = subId;
    createRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_NEITHER;
    createRequest.itemsToCreate = items;
    createRequest.itemsToCreateSize = 1;
    UA_CreateMonitoredItemsResponse createResponse =
       UA_Client_MonitoredItems_createDataChanges(client, createRequest, contexts,
                                                  callbacks, deleteCallbacks);

    ck_assert_uint_eq(createResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(createResponse.resultsSize, 1);
    ck_assert_uint_eq(createResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    newMonitoredItemIds[0] = createResponse.results[0].monitoredItemId;
    UA_CreateMonitoredItemsResponse_clear(&createResponse);

    /* manually control the publish cycles */
    pauseServer();

    /* Receive the initial value */
    notificationReceived = false;
    countNotificationReceived = 0;
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    iterateUntilNotification(client, 100);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 1);

    /* No further update */
    notificationReceived = false;
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, false);
    ck_assert_uint_eq(countNotificationReceived, 1);

    /* Modify the MonitoredItem and change the sampling interval */
    UA_MonitoredItemModifyRequest modify1;
    UA_MonitoredItemModifyRequest_init(&modify1);
    modify1.monitoredItemId = newMonitoredItemIds[0];
    modify1.requestedParameters.samplingInterval = publishingInterval * 1.5;

    UA_ModifyMonitoredItemsRequest modifyRequest;
    UA_ModifyMonitoredItemsRequest_init(&modifyRequest);
    modifyRequest.subscriptionId = subId;
    modifyRequest.itemsToModify = &modify1;
    modifyRequest.itemsToModifySize = 1;

    runServer();

    UA_ModifyMonitoredItemsResponse modifyResponse =
        UA_Client_MonitoredItems_modify(client, modifyRequest);
    ck_assert_uint_eq(modifyResponse.resultsSize, 1);
    ck_assert_uint_eq(modifyResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&modifyResponse);

    pauseServer();

    /* Sleep longer than the publishing interval */
    notificationReceived = false;
    countNotificationReceived = 0;
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, false);
    ck_assert_uint_eq(countNotificationReceived, 0);

    /* Modify the MonitoredItem and change the trigger. We want to see an
     * update. But not immediately. */
    UA_DataChangeFilter filter;
    UA_DataChangeFilter_init(&filter);
    filter.trigger = UA_DATACHANGETRIGGER_STATUSVALUETIMESTAMP;
    modify1.requestedParameters.filter.content.decoded.data = &filter;
    modify1.requestedParameters.filter.content.decoded.type =
        &UA_TYPES[UA_TYPES_DATACHANGEFILTER];
    modify1.requestedParameters.filter.encoding = UA_EXTENSIONOBJECT_DECODED;

    runServer();

    modifyResponse = UA_Client_MonitoredItems_modify(client, modifyRequest);
    ck_assert_uint_eq(modifyResponse.resultsSize, 1);
    ck_assert_uint_eq(modifyResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&modifyResponse);

    pauseServer();

    /* Sleep longer than the publishing interval */
    notificationReceived = false;
    countNotificationReceived = 0;
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, false);
    ck_assert_uint_eq(countNotificationReceived, 0);

    /* Sleep long enough to trigger the next sampling. */
    UA_fakeSleep((UA_UInt32)(publishingInterval * 0.6));
    UA_Server_run_iterate(server, false);

    /* Sleep long enough to trigger the publish callback */
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    /* No change here, as the value subscribed value has no source timestamp. */
    ck_assert_uint_eq(notificationReceived, false);
    ck_assert_uint_eq(countNotificationReceived, 0);

    runServer();

    /* Delete and clean up */
    UA_DeleteMonitoredItemsRequest deleteRequest;
    UA_DeleteMonitoredItemsRequest_init(&deleteRequest);
    deleteRequest.subscriptionId = subId;
    deleteRequest.monitoredItemIds = newMonitoredItemIds;
    deleteRequest.monitoredItemIdsSize = 1;

    UA_DeleteMonitoredItemsResponse deleteResponse =
        UA_Client_MonitoredItems_delete(client, deleteRequest);

    ck_assert_uint_eq(deleteResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(deleteResponse.resultsSize, 1);
    UA_DeleteMonitoredItemsResponse_clear(&deleteResponse);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_modifyMonitoredItem_doubleBuffer) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Notifications are injected below in a deterministic order. */
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;

    UA_CreateSubscriptionRequest subRequest =
        UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse subResponse =
        UA_Client_Subscriptions_create(client, subRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(subResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(
            client, subResponse.subscriptionId, UA_TIMESTAMPSTORETURN_BOTH,
            monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert_ptr_ne(sub, NULL);
    UA_Client_MonitoredItem *mon = ZIP_ROOT(&sub->monitoredItems);
    ck_assert_ptr_ne(mon, NULL);
    UA_UInt32 oldHandle =
        mon->parameters.clientHandle;
    UA_Double oldSamplingInterval = mon->parameters.samplingInterval;
    ck_assert_uint_eq(sub->pendingRekeys, 0);
    /* A notification with an unknown clientHandle while nothing is pending
     * must be dropped without scanning the tree (the O(N*n) guard). */
    notificationReceived = false;
    injectDataChangeNotification(client, sub, 0xDEADBEEFu);
    ck_assert(!notificationReceived);
    ck_assert_uint_eq(sub->pendingRekeys, 0);

    UA_MonitoredItemModifyRequest item;
    UA_MonitoredItemModifyRequest_init(&item);
    item.monitoredItemId = monResponse.monitoredItemId;
    item.requestedParameters.samplingInterval = 1000.0;
    item.requestedParameters.queueSize = 1;
    item.requestedParameters.discardOldest = true;

    UA_ModifyMonitoredItemsRequest request;
    UA_ModifyMonitoredItemsRequest_init(&request);
    request.subscriptionId = subResponse.subscriptionId;
    request.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    request.itemsToModify = &item;
    request.itemsToModifySize = 1;

    /* First exercise response-before-notification. */
    UA_ModifyMonitoredItemsResponse response =
        UA_Client_MonitoredItems_modify(client, request);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(response.resultsSize, 1);
    ck_assert_uint_eq(response.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&response);

    ck_assert(mon->pendingParameters.clientHandle);
    ck_assert(mon->parameters.samplingInterval == oldSamplingInterval);
    ck_assert(mon->pendingParameters.samplingInterval == 1000.0);
    UA_UInt32 newHandle =
        mon->pendingParameters.clientHandle;
    ck_assert_uint_ne(oldHandle, newHandle);
    ck_assert_uint_eq(sub->pendingRekeys, 1);

    notificationReceived = false;
    countNotificationReceived = 0;
    injectDataChangeNotification(client, sub, oldHandle);
    ck_assert(notificationReceived);
    ck_assert_uint_eq(countNotificationReceived, 1);
    ck_assert_uint_eq(mon->parameters.clientHandle, oldHandle);
    ck_assert(mon->pendingParameters.clientHandle);

    injectDataChangeNotification(client, sub, newHandle);
    ck_assert_uint_eq(countNotificationReceived, 2);
    ck_assert_uint_eq(mon->parameters.clientHandle, newHandle);
    ck_assert(!mon->pendingParameters.clientHandle);
    ck_assert(mon->parameters.samplingInterval == 1000.0);
    ck_assert_uint_eq(sub->pendingRekeys, 0);

    /* Now deliver the new handle before the asynchronous modify response. */
    pauseServer();
    item.requestedParameters.samplingInterval = 2000.0;
    UA_ModifyMonitoredItemsResponse asyncResponse;
    UA_ModifyMonitoredItemsResponse_init(&asyncResponse);
    UA_UInt32 requestId = 0;
    retval = UA_Client_MonitoredItems_modify_async(
        client, request, modifyMonitoredItemsCallback, &asyncResponse, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert(mon->pendingParameters.clientHandle);
    UA_UInt32 newestHandle =
        mon->pendingParameters.clientHandle;
    ck_assert_uint_ne(newHandle, newestHandle);

    injectDataChangeNotification(client, sub, newestHandle);
    ck_assert_uint_eq(mon->parameters.clientHandle,
                      newestHandle);
    ck_assert(!mon->pendingParameters.clientHandle);
    ck_assert(mon->parameters.samplingInterval == 2000.0);

    runServer();
    for(size_t i = 0; i < 1000 && asyncResponse.resultsSize == 0; i++)
        UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(asyncResponse.responseHeader.serviceResult,
                      UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(asyncResponse.resultsSize, 1);
    ck_assert_uint_eq(asyncResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(mon->parameters.clientHandle,
                      newestHandle);
    UA_ModifyMonitoredItemsResponse_clear(&asyncResponse);

    /* A newer pending generation supersedes the previous one. */
    item.requestedParameters.samplingInterval = 3000.0;
    response = UA_Client_MonitoredItems_modify(client, request);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&response);
    UA_UInt32 supersededHandle =
        mon->pendingParameters.clientHandle;

    item.requestedParameters.samplingInterval = 4000.0;
    response = UA_Client_MonitoredItems_modify(client, request);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&response);
    UA_UInt32 latestHandle =
        mon->pendingParameters.clientHandle;
    ck_assert_uint_ne(supersededHandle, latestHandle);

    UA_UInt32 notificationCount = countNotificationReceived;
    injectDataChangeNotification(client, sub, supersededHandle);
    ck_assert_uint_eq(countNotificationReceived, notificationCount);
    ck_assert_uint_eq(mon->parameters.clientHandle,
                      newestHandle);
    ck_assert(mon->pendingParameters.clientHandle);

    injectDataChangeNotification(client, sub, latestHandle);
    ck_assert_uint_eq(countNotificationReceived, notificationCount + 1);
    ck_assert_uint_eq(mon->parameters.clientHandle,
                      latestHandle);
    ck_assert(!mon->pendingParameters.clientHandle);

    /* A rejected modification drops only its matching pending slot. */
    request.timestampsToReturn = (UA_TimestampsToReturn)100;
    response = UA_Client_MonitoredItems_modify(client, request);
    ck_assert_uint_eq(response.responseHeader.serviceResult,
                      UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID);
    ck_assert(!mon->pendingParameters.clientHandle);
    ck_assert_uint_eq(mon->parameters.clientHandle,
                      latestHandle);
    UA_ModifyMonitoredItemsResponse_clear(&response);

    retval = UA_Client_Subscriptions_deleteSingle(client,
                                                   subResponse.subscriptionId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_modifyEventFilter_doubleBuffer) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;

    UA_CreateSubscriptionRequest subRequest =
        UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse subResponse =
        UA_Client_Subscriptions_create(client, subRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(subResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    UA_QualifiedName oldBrowsePath = UA_QUALIFIEDNAME(0, "Message");
    UA_SimpleAttributeOperand oldClause;
    UA_SimpleAttributeOperand_init(&oldClause);
    oldClause.typeDefinitionId =
        UA_NODEID_NUMERIC(0, UA_NS0ID_BASEEVENTTYPE);
    oldClause.browsePath = &oldBrowsePath;
    oldClause.browsePathSize = 1;
    oldClause.attributeId = UA_ATTRIBUTEID_VALUE;

    UA_EventFilter oldFilter;
    UA_EventFilter_init(&oldFilter);
    oldFilter.selectClauses = &oldClause;
    oldFilter.selectClausesSize = 1;

    UA_MonitoredItemCreateRequest monRequest;
    UA_MonitoredItemCreateRequest_init(&monRequest);
    monRequest.itemToMonitor.nodeId =
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER);
    monRequest.itemToMonitor.attributeId = UA_ATTRIBUTEID_EVENTNOTIFIER;
    monRequest.monitoringMode = UA_MONITORINGMODE_REPORTING;
    monRequest.requestedParameters.queueSize = 1;
    monRequest.requestedParameters.discardOldest = true;
    UA_ExtensionObject_setValueNoDelete(
        &monRequest.requestedParameters.filter, &oldFilter,
        &UA_TYPES[UA_TYPES_EVENTFILTER]);

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createEvent(
            client, subResponse.subscriptionId, UA_TIMESTAMPSTORETURN_BOTH,
            monRequest, NULL, eventHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert_ptr_ne(sub, NULL);
    UA_Client_MonitoredItem *mon = ZIP_ROOT(&sub->monitoredItems);
    ck_assert_ptr_ne(mon, NULL);
    ck_assert_uint_eq(mon->eventFields.mapSize, 0);
    UA_UInt32 oldHandle =
        mon->parameters.clientHandle;

    UA_QualifiedName newBrowsePaths[2] = {
        UA_QUALIFIEDNAME(0, "Severity"),
        UA_QUALIFIEDNAME(0, "SourceName")
    };
    UA_SimpleAttributeOperand newClauses[2];
    for(size_t i = 0; i < 2; i++) {
        UA_SimpleAttributeOperand_init(&newClauses[i]);
        newClauses[i].typeDefinitionId =
            UA_NODEID_NUMERIC(0, UA_NS0ID_BASEEVENTTYPE);
        newClauses[i].browsePath = &newBrowsePaths[i];
        newClauses[i].browsePathSize = 1;
        newClauses[i].attributeId = UA_ATTRIBUTEID_VALUE;
    }

    UA_EventFilter newFilter;
    UA_EventFilter_init(&newFilter);
    newFilter.selectClauses = newClauses;
    newFilter.selectClausesSize = 2;

    UA_MonitoredItemModifyRequest item;
    UA_MonitoredItemModifyRequest_init(&item);
    item.monitoredItemId = monResponse.monitoredItemId;
    item.requestedParameters.queueSize = 1;
    item.requestedParameters.discardOldest = true;
    UA_ExtensionObject_setValueNoDelete(
        &item.requestedParameters.filter, &newFilter,
        &UA_TYPES[UA_TYPES_EVENTFILTER]);

    UA_ModifyMonitoredItemsRequest request;
    UA_ModifyMonitoredItemsRequest_init(&request);
    request.subscriptionId = subResponse.subscriptionId;
    request.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    request.itemsToModify = &item;
    request.itemsToModifySize = 1;
    UA_ModifyMonitoredItemsResponse response =
        UA_Client_MonitoredItems_modify(client, request);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(response.resultsSize, 1);
    ck_assert_uint_eq(response.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&response);

    ck_assert(mon->pendingParameters.clientHandle);
    ck_assert_uint_eq(mon->eventFields.mapSize, 0);
    ck_assert_ptr_eq(mon->parameters.filter.content.decoded.type,
                     &UA_TYPES[UA_TYPES_EVENTFILTER]);
    ck_assert_ptr_eq(mon->pendingParameters.filter.content.decoded.type,
                     &UA_TYPES[UA_TYPES_EVENTFILTER]);
    UA_EventFilter *activeFilter = (UA_EventFilter*)
        mon->parameters.filter.content.decoded.data;
    UA_EventFilter *pendingFilter = (UA_EventFilter*)
        mon->pendingParameters.filter.content.decoded.data;
    ck_assert_uint_eq(activeFilter->selectClausesSize, 1);
    ck_assert_uint_eq(pendingFilter->selectClausesSize, 2);
    ck_assert_ptr_ne(pendingFilter, &newFilter);
    UA_UInt32 newHandle =
        mon->pendingParameters.clientHandle;

    eventFieldsSize = 0;
    injectEventNotification(client, sub, oldHandle, 1);
    ck_assert_uint_eq(eventFieldsSize, 1);
    ck_assert_uint_eq(mon->eventFields.mapSize, 1);
    ck_assert_uint_eq(mon->parameters.clientHandle, oldHandle);
    ck_assert(mon->pendingParameters.clientHandle);

    injectEventNotification(client, sub, newHandle, 2);
    ck_assert_uint_eq(eventFieldsSize, 2);
    ck_assert_uint_eq(mon->eventFields.mapSize, 2);
    ck_assert_uint_eq(mon->parameters.clientHandle, newHandle);
    ck_assert(!mon->pendingParameters.clientHandle);
    activeFilter = (UA_EventFilter*)mon->parameters.filter.content.decoded.data;
    ck_assert_uint_eq(activeFilter->selectClausesSize, 2);

    retval = UA_Client_Subscriptions_deleteSingle(client,
                                                   subResponse.subscriptionId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_modifyMonitoredItem_edgeCases) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;

    UA_CreateSubscriptionRequest subRequest =
        UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse subResponse =
        UA_Client_Subscriptions_create(client, subRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(subResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = subResponse.subscriptionId;

    UA_MonitoredItemCreateRequest items[3] = {
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE)),
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME)),
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_SECONDSTILLSHUTDOWN))
    };
    UA_Client_DataChangeNotificationCallback callbacks[3] = {
        dataChangeHandler, dataChangeHandler, dataChangeHandler
    };
    void *contexts[3] = {NULL, NULL, NULL};
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks[3] = {
        NULL, NULL, NULL
    };

    UA_CreateMonitoredItemsRequest createRequest;
    UA_CreateMonitoredItemsRequest_init(&createRequest);
    createRequest.subscriptionId = subId;
    createRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    createRequest.itemsToCreate = items;
    createRequest.itemsToCreateSize = 3;
    UA_CreateMonitoredItemsResponse createResponse =
        UA_Client_MonitoredItems_createDataChanges(
            client, createRequest, contexts, callbacks, deleteCallbacks);
    ck_assert_uint_eq(createResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(createResponse.resultsSize, 3);

    UA_UInt32 monIds[3];
    for(size_t i = 0; i < 3; i++) {
        ck_assert_uint_eq(createResponse.results[i].statusCode, UA_STATUSCODE_GOOD);
        monIds[i] = createResponse.results[i].monitoredItemId;
    }
    UA_CreateMonitoredItemsResponse_clear(&createResponse);

    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert_ptr_ne(sub, NULL);
    UA_Client_MonitoredItem *mons[3];
    for(size_t i = 0; i < 3; i++) {
        mons[i] = findTestMonitoredItem(ZIP_ROOT(&sub->monitoredItems), monIds[i]);
        ck_assert_ptr_ne(mons[i], NULL);
    }
    ck_assert_uint_eq(countTestMonitoredItems(ZIP_ROOT(&sub->monitoredItems)), 3);

    /* A mixed response keeps the successful item's pending state and rolls
     * back only the failed item. */
    UA_MonitoredItemModifyRequest modifyItems[2];
    for(size_t i = 0; i < 2; i++) {
        UA_MonitoredItemModifyRequest_init(&modifyItems[i]);
        modifyItems[i].monitoredItemId = monIds[i];
        modifyItems[i].requestedParameters.samplingInterval = 1000.0;
        modifyItems[i].requestedParameters.queueSize = 1;
        modifyItems[i].requestedParameters.discardOldest = true;
    }
    UA_DataChangeFilter unsupportedFilter;
    UA_DataChangeFilter_init(&unsupportedFilter);
    unsupportedFilter.deadbandType = (UA_DeadbandType)100;
    UA_ExtensionObject_setValueNoDelete(
        &modifyItems[1].requestedParameters.filter, &unsupportedFilter,
        &UA_TYPES[UA_TYPES_DATACHANGEFILTER]);

    UA_ModifyMonitoredItemsRequest modifyRequest;
    UA_ModifyMonitoredItemsRequest_init(&modifyRequest);
    modifyRequest.subscriptionId = subId;
    modifyRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    modifyRequest.itemsToModify = modifyItems;
    modifyRequest.itemsToModifySize = 2;
    UA_ModifyMonitoredItemsResponse modifyResponse =
        UA_Client_MonitoredItems_modify(client, modifyRequest);
    ck_assert_uint_eq(modifyResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(modifyResponse.resultsSize, 2);
    ck_assert_uint_eq(modifyResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    ck_assert_uint_ne(modifyResponse.results[1].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&modifyResponse);
    ck_assert(mons[0]->pendingParameters.clientHandle);
    ck_assert(!mons[1]->pendingParameters.clientHandle);
    ck_assert(!mons[2]->pendingParameters.clientHandle);

    /* Find and promote a pending handle while several monitored items exist. */
    UA_UInt32 pendingHandle =
        mons[0]->pendingParameters.clientHandle;
    UA_UInt32 otherHandles[2] = {
        mons[1]->parameters.clientHandle,
        mons[2]->parameters.clientHandle
    };
    injectDataChangeNotification(client, sub, pendingHandle);
    ck_assert_uint_eq(mons[0]->parameters.clientHandle,
                      pendingHandle);
    ck_assert_uint_eq(mons[1]->parameters.clientHandle,
                      otherHandles[0]);
    ck_assert_uint_eq(mons[2]->parameters.clientHandle,
                      otherHandles[1]);

    /* Deletion clears both embedded slots while settings are pending. */
    modifyRequest.itemsToModify = &modifyItems[1];
    modifyRequest.itemsToModifySize = 1;
    UA_ExtensionObject_init(&modifyItems[1].requestedParameters.filter);
    modifyResponse = UA_Client_MonitoredItems_modify(client, modifyRequest);
    ck_assert_uint_eq(modifyResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(modifyResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    UA_ModifyMonitoredItemsResponse_clear(&modifyResponse);
    ck_assert(mons[1]->pendingParameters.clientHandle);
    retval = UA_Client_MonitoredItems_deleteSingle(client, subId, monIds[1]);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_ptr_eq(findTestMonitoredItem(ZIP_ROOT(&sub->monitoredItems), monIds[1]),
                     NULL);
    ck_assert_uint_eq(countTestMonitoredItems(ZIP_ROOT(&sub->monitoredItems)), 2);

    /* A local asynchronous submission failure rolls back the prepared slot. */
    UA_MonitoredItemModifyRequest asyncItem;
    UA_MonitoredItemModifyRequest_init(&asyncItem);
    asyncItem.monitoredItemId = monIds[2];
    asyncItem.requestedParameters.samplingInterval = 2000.0;
    asyncItem.requestedParameters.queueSize = 1;
    asyncItem.requestedParameters.discardOldest = true;
    modifyRequest.itemsToModify = &asyncItem;
    modifyRequest.itemsToModifySize = 1;
    UA_SecureChannelState channelState = client->channel.state;
    client->channel.state = UA_SECURECHANNELSTATE_CLOSED;
    retval = UA_Client_MonitoredItems_modify_async(
        client, modifyRequest, NULL, NULL, NULL);
    client->channel.state = channelState;
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSERVERNOTCONNECTED);
    ck_assert(!mons[2]->pendingParameters.clientHandle);
    ck_assert_uint_eq(mons[2]->parameters.clientHandle,
                      otherHandles[1]);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_createDataChanges_async) {
    UA_UInt32 reqId = 0;
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    // Async subscription creation is tested in Client_subscription_async
    // simplify test case using synchronous here
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    UA_MonitoredItemCreateRequest items[3];
    UA_UInt32 newMonitoredItemIds[3];
    UA_Client_DataChangeNotificationCallback callbacks[3];
    UA_Client_DeleteMonitoredItemCallback deleteCallbacks[3];
    void *contexts[3];

    /* monitor the server state */
    items[0] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    callbacks[0] = dataChangeHandler;
    contexts[0] = NULL;
    deleteCallbacks[0] = NULL;

    /* monitor invalid node */
    items[1] = UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, 999999));
    callbacks[1] = dataChangeHandler;
    contexts[1] = NULL;
    deleteCallbacks[1] = NULL;

    /* monitor current time */
    items[2] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    callbacks[2] = dataChangeHandler;
    contexts[2] = NULL;
    deleteCallbacks[2] = NULL;

    UA_CreateMonitoredItemsRequest createRequest;
    UA_CreateMonitoredItemsRequest_init(&createRequest);
    createRequest.subscriptionId = subId;
    createRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    createRequest.itemsToCreate = items;
    createRequest.itemsToCreateSize = 3;
    UA_CreateMonitoredItemsResponse createResponse;

    /* manually control the server thread */
    pauseServer();

    retval = UA_Client_MonitoredItems_createDataChanges_async(client, createRequest,
                                                              contexts, callbacks, deleteCallbacks,
                                                              createDataChangesCallback,
                                                              &createResponse, &reqId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    createResponse.responseHeader.serviceResult = 1;
    do {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 1);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    } while(createResponse.responseHeader.serviceResult == 1);

    ck_assert_uint_eq(createResponse.resultsSize, 3);
    ck_assert_uint_eq(createResponse.results[0].statusCode, UA_STATUSCODE_GOOD);
    newMonitoredItemIds[0] = createResponse.results[0].monitoredItemId;
    ck_assert_uint_eq(createResponse.results[1].statusCode,
                      UA_STATUSCODE_BADNODEIDUNKNOWN);
    newMonitoredItemIds[1] = createResponse.results[1].monitoredItemId;
    ck_assert_uint_eq(newMonitoredItemIds[1], 0);
    ck_assert_uint_eq(createResponse.results[2].statusCode, UA_STATUSCODE_GOOD);
    newMonitoredItemIds[2] = createResponse.results[2].monitoredItemId;
    ck_assert_uint_eq(createResponse.results[2].statusCode, UA_STATUSCODE_GOOD);
    UA_CreateMonitoredItemsResponse_clear(&createResponse);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    countNotificationReceived = 0;
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 2);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 3);

    UA_DeleteMonitoredItemsRequest deleteRequest;
    UA_DeleteMonitoredItemsRequest_init(&deleteRequest);
    deleteRequest.subscriptionId = subId;
    deleteRequest.monitoredItemIds = newMonitoredItemIds;
    deleteRequest.monitoredItemIdsSize = 3;

    UA_DeleteMonitoredItemsResponse deleteResponse;
    UA_DeleteMonitoredItemsResponse_init(&deleteResponse);
    retval = UA_Client_MonitoredItems_delete_async(client, deleteRequest,
                                                   deleteMonitoredItemsCallback, &deleteResponse, &reqId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    pauseServer();
    for(size_t i = 0; i < 1000 &&
        deleteResponse.responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
        deleteResponse.resultsSize == 0; ++i) {
        UA_Server_run_iterate(server, false);
        retval = UA_Client_run_iterate(client, 0);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    }

    ck_assert_uint_eq(deleteResponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(deleteResponse.resultsSize, 3);
    ck_assert_uint_eq(deleteResponse.results[0], UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(deleteResponse.results[1], UA_STATUSCODE_BADMONITOREDITEMIDINVALID);
    ck_assert_uint_eq(deleteResponse.results[2], UA_STATUSCODE_GOOD);

    UA_DeleteMonitoredItemsResponse_clear(&deleteResponse);

    /* run the server in an independent thread again */
    runServer();

    // Async subscription deletion is tested in Client_subscription_async
    // simplify test case using synchronous here
    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

typedef struct {
    UA_UInt32 requestId;
    UA_UInt32 requestHandle;
    UA_Boolean received;
} CancelledPublish;

static void
cancelledPublishCallback(UA_Client *client, void *userdata,
                         UA_UInt32 requestId, void *response) {
    CancelledPublish *expected = (CancelledPublish*)userdata;
    UA_PublishResponse *pr = (UA_PublishResponse*)response;
    ck_assert(!expected->received);
    ck_assert_uint_eq(requestId, expected->requestId);
    ck_assert_uint_eq(pr->responseHeader.requestHandle, expected->requestHandle);
    ck_assert_uint_eq(pr->responseHeader.serviceResult,
                      UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT);
    expected->received = true;
}

START_TEST(Client_cancelQueuedPublish) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;
    ck_assert_uint_eq(UA_Client_connect(client, "opc.tcp://localhost:4840"),
                      UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_CreateSubscriptionResponse_clear(&response);

    /* Interleave matching and non-matching handles to exercise removal from
     * both the head and the tail, while preserving the middle request. */
    CancelledPublish pending[3] = {{0, 1234, false}, {0, 5678, false}, {0, 1234, false}};
    for(size_t i = 0; i < 3; i++) {
        UA_PublishRequest pr;
        UA_PublishRequest_init(&pr);
        pr.requestHeader.requestHandle = pending[i].requestHandle;
        pr.requestHeader.timeoutHint = 60000;
        ck_assert_uint_eq(__UA_Client_AsyncService(
            client, &pr, &UA_TYPES[UA_TYPES_PUBLISHREQUEST], cancelledPublishCallback,
            &UA_TYPES[UA_TYPES_PUBLISHRESPONSE], &pending[i], &pending[i].requestId),
            UA_STATUSCODE_GOOD);
    }

    UA_UInt32 count = 0;
    ck_assert_uint_eq(UA_Client_cancelByRequestHandle(client, 1234, &count),
                      UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(count, 2);
    ck_assert(pending[0].received);
    ck_assert(!pending[1].received);
    ck_assert(pending[2].received);

    ck_assert_uint_eq(UA_Client_cancelByRequestHandle(client, 5678, &count),
                      UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(count, 1);
    ck_assert(pending[1].received);
    ck_assert_uint_eq(UA_Client_cancelByRequestHandle(client, 1234, &count),
                      UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(count, 0);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
} END_TEST

START_TEST(Client_subscription_keepAlive) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* For the keepalive tests disable automatically sending publish requests */
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedMaxKeepAliveCount = 1;
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);
    UA_UInt32 monId = monResponse.monitoredItemId;

    /* Ensure that the subscription is late */
    pauseServer();
    UA_fakeSleep((UA_UInt32)(publishingInterval + 1));
    runServer();

    /* Manually send a publish request */
    UA_PublishRequest pr;
    UA_PublishRequest_init(&pr);
    pr.subscriptionAcknowledgementsSize = 0;
    UA_PublishResponse presponse;
    UA_PublishResponse_init(&presponse);
    __UA_Client_Service(client, &pr, &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
                        &presponse, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
    ck_assert_uint_eq(presponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(presponse.notificationMessage.notificationDataSize, 1);
    UA_PublishResponse_clear(&presponse);
    UA_PublishRequest_clear(&pr);

    /* Ensure that the subscription is late */
    pauseServer();
    UA_fakeSleep((UA_UInt32)(publishingInterval + 1));
    runServer();

    UA_PublishRequest_init(&pr);
    pr.subscriptionAcknowledgementsSize = 0;
    UA_PublishResponse_init(&presponse);
    __UA_Client_Service(client, &pr, &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
                        &presponse, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);

    ck_assert_uint_eq(presponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(presponse.notificationMessage.notificationDataSize, 0);
    UA_PublishResponse_clear(&presponse);
    UA_PublishRequest_clear(&pr);

    retval = UA_Client_MonitoredItems_deleteSingle(client, subId, monId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

typedef struct {
    UA_UInt32 count;
    UA_DateTime publishTime;
    size_t notificationDataSize;
    UA_Boolean moreNotifications;
} KeepAliveResponse;

static void
keepAliveResponseCallback(UA_Client *client, void *userdata,
                          UA_UInt32 requestId, void *response) {
    KeepAliveResponse *received = (KeepAliveResponse*)userdata;
    UA_PublishResponse *pr = (UA_PublishResponse*)response;
    ck_assert_uint_eq(pr->responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    received->count++;
    received->publishTime = pr->notificationMessage.publishTime;
    received->notificationDataSize = pr->notificationMessage.notificationDataSize;
    received->moreNotifications = pr->moreNotifications;
}

static void
sendKeepAlivePublishRequest(UA_Client *client, KeepAliveResponse *received) {
    UA_PublishRequest request;
    UA_PublishRequest_init(&request);
    UA_StatusCode res = __UA_Client_AsyncService(
        client, &request, &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
        keepAliveResponseCallback, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE],
        received, NULL);
    ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);
}

static void
iterateKeepAliveClientServer(UA_Client *client) {
    /* Drain network traffic and delayed callbacks without advancing the clock. */
    for(unsigned i = 0; i < 10; i++) {
        UA_Server_run_iterate(server, false);
        ck_assert_uint_eq(UA_Client_run_iterate(client, 1), UA_STATUSCODE_GOOD);
    }
}

static void
checkKeepAlivePublishingDisabled(unsigned itemCase, UA_UInt32 keepAliveCount,
                                 UA_Boolean firstKeepAliveLate) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;
    ck_assert_uint_eq(UA_Client_connect(client, "opc.tcp://localhost:4840"),
                      UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedPublishingInterval = 1000;
    request.requestedMaxKeepAliveCount = keepAliveCount;
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_double_eq(response.revisedPublishingInterval, 1000);
    ck_assert_uint_eq(response.revisedMaxKeepAliveCount, keepAliveCount);
    UA_UInt32 subId = response.subscriptionId;
    UA_CreateSubscriptionResponse_clear(&response);

    /* Empty, changing, disabled and static MonitoredItem cases. */
    if(itemCase != 0) {
        UA_UInt32 nodeId = (itemCase == 3) ? UA_NS0ID_SERVER_SERVERSTATUS_STARTTIME :
                                           UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME;
        UA_MonitoredItemCreateRequest monRequest =
            UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, nodeId));
        monRequest.requestedParameters.samplingInterval = -1;
        monRequest.requestedParameters.queueSize = 1;
        if(itemCase == 2)
            monRequest.monitoringMode = UA_MONITORINGMODE_DISABLED;
        UA_MonitoredItemCreateResult monResponse =
            UA_Client_MonitoredItems_createDataChange(
                client, subId, UA_TIMESTAMPSTORETURN_NEITHER,
                monRequest, NULL, dataChangeHandler, NULL);
        ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);
        UA_MonitoredItemCreateResult_clear(&monResponse);
    }

    /* Deliver the initial notification before disabling publishing. In the
     * static case this leaves no notifications waiting in the queue. */
    pauseServer();
    KeepAliveResponse received = {0, 0, 0, false};
    sendKeepAlivePublishRequest(client, &received);
    iterateKeepAliveClientServer(client);
    UA_fakeSleep(1000);
    iterateKeepAliveClientServer(client);
    ck_assert_uint_eq(received.count, 1);
    ck_assert_uint_eq(received.notificationDataSize,
                      (itemCase == 1 || itemCase == 3) ? 1 : 0);

    runServer();
    UA_SetPublishingModeRequest mode;
    UA_SetPublishingModeRequest_init(&mode);
    mode.subscriptionIds = &subId;
    mode.subscriptionIdsSize = 1;
    mode.publishingEnabled = false;
    UA_SetPublishingModeResponse changed =
        UA_Client_Subscriptions_setPublishingMode(client, mode);
    ck_assert_uint_eq(changed.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(changed.resultsSize, 1);
    ck_assert_uint_eq(changed.results[0], UA_STATUSCODE_GOOD);
    UA_SetPublishingModeResponse_clear(&changed);
    pauseServer();

    UA_Boolean moreNotifications = false;
    for(unsigned round = 0; round < 3; round++) {
        UA_UInt32 previousCount = received.count;
        UA_DateTime previousTime = received.publishTime;
        UA_Boolean late = firstKeepAliveLate && round == 0;
        if(!late) {
            sendKeepAlivePublishRequest(client, &received);
            iterateKeepAliveClientServer(client);
            ck_assert_uint_eq(received.count, previousCount);
        }
        for(UA_UInt32 cycle = 1; cycle <= keepAliveCount; cycle++) {
            UA_fakeSleep(1000);
            iterateKeepAliveClientServer(client);
            ck_assert_uint_eq(received.count,
                              previousCount + (!late && cycle == keepAliveCount));
        }
        if(late) {
            /* A previously late subscription must also return to normal after
             * sending its keep-alive, despite the queued data changes. */
            sendKeepAlivePublishRequest(client, &received);
            iterateKeepAliveClientServer(client);
        }
        ck_assert_uint_eq(received.count, previousCount + 1);
        ck_assert_int_eq(received.publishTime - previousTime,
                         keepAliveCount * UA_DATETIME_SEC);
        ck_assert_uint_eq(received.notificationDataSize, 0);
        moreNotifications |= received.moreNotifications;
    }
    ck_assert(!moreNotifications);

    if(itemCase == 1) {
        /* Data delivery must resume when publishing is re-enabled. */
        runServer();
        mode.publishingEnabled = true;
        changed = UA_Client_Subscriptions_setPublishingMode(client, mode);
        ck_assert_uint_eq(changed.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
        ck_assert_uint_eq(changed.resultsSize, 1);
        ck_assert_uint_eq(changed.results[0], UA_STATUSCODE_GOOD);
        UA_SetPublishingModeResponse_clear(&changed);
        pauseServer();
        UA_UInt32 previousCount = received.count;
        sendKeepAlivePublishRequest(client, &received);
        iterateKeepAliveClientServer(client);
        UA_fakeSleep(1000);
        iterateKeepAliveClientServer(client);
        ck_assert_uint_eq(received.count, previousCount + 1);
        ck_assert_uint_eq(received.notificationDataSize, 1);
    }

    runServer();
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}

START_TEST(Client_subscription_keepAlive_publishingDisabled) {
    checkKeepAlivePublishingDisabled((unsigned)_i, 10, false);
} END_TEST

START_TEST(Client_subscription_keepAlive_publishingDisabled_late) {
    checkKeepAlivePublishingDisabled(1, 10, true);
} END_TEST

START_TEST(Client_subscription_keepAlive_publishingDisabled_oneCycle) {
    checkKeepAlivePublishingDisabled(1, 1, false);
} END_TEST

START_TEST(Client_subscription_priority) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* For the test disable automatically sending publish requests */
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;

    // prio = 0
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedMaxKeepAliveCount = 1;
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId1 = response.subscriptionId;

    // prio = 255 (highest)
    request.priority = 255;
    response = UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId2 = response.subscriptionId;

    // prio = 0
    request.priority = 0;
    response = UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId3 = response.subscriptionId;

    /* manually control the server thread */
    pauseServer();

    /* Ensure that the subscription is late */
    UA_fakeSleep((UA_UInt32)(publishingInterval + 1));
    UA_Server_run_iterate(server, false);

    /* run the server in an independent thread again */
    runServer();

    /* Manually send a publish request */
    UA_PublishRequest pr;
    UA_PublishRequest_init(&pr);
    pr.subscriptionAcknowledgementsSize = 0;
    UA_PublishResponse presponse;
    UA_PublishResponse_init(&presponse);
    __UA_Client_Service(client, &pr, &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
                        &presponse, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
    ck_assert_uint_eq(presponse.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(presponse.notificationMessage.notificationDataSize, 0);
    ck_assert_uint_eq(presponse.subscriptionId, subId2); /* the highest prio */
    UA_PublishResponse_clear(&presponse);
    UA_PublishRequest_clear(&pr);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    retval = UA_Client_Subscriptions_deleteSingle(client, subId2);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    retval = UA_Client_Subscriptions_deleteSingle(client, subId3);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_connectionClose) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Manually close the connection. The connection is internally closed at the
     * next iteration of the EventLoop. Hence the next request is sent out. But
     * the connection "actually closes" before receiving the response. */
    UA_ConnectionManager *cm = client->channel.connectionManager;
    uintptr_t connId = client->channel.connectionId;
    cm->closeConnection(cm, connId);

    notificationReceived = false;

    /* Reconnect a new SecureChannel and recover the Session */
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_SessionState ss;
    UA_Client_getState(client, NULL, &ss, NULL);
    ck_assert_uint_eq(ss, UA_SESSIONSTATE_CREATED);
    while(ss != UA_SESSIONSTATE_ACTIVATED) {
        UA_Client_run_iterate(client, 1);
        UA_Client_getState(client, NULL, &ss, NULL);
    }

    /* manually control the server thread */
    pauseServer();

    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    UA_Server_run_iterate(server, false);

    /* Send Publish requests */
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Still receiving on the MonitoredItem */
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    UA_Server_run_iterate(server, false);
    countNotificationReceived = 0;
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 1);

    /* run the server in an independent thread again */
    runServer();

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

/* The server closes the Session while PublishRequests are queued. The first
 * BadSessionClosed response cleans up the Session and resets the counter of
 * outstanding PublishRequests before the response itself is processed. */
START_TEST(Client_subscription_publishCounterAfterSessionClosed) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* Send the PublishRequests */
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_gt(client->currentlyOutStandingPublishRequests, 1);

    /* Close the Session on the server. The queued PublishRequests are answered
     * with BadSessionClosed. */
    pauseServer();
    retval = UA_Server_closeSession(server, &client->sessionId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    for(size_t i = 0; i < 100 &&
            client->sessionState == UA_SESSIONSTATE_ACTIVATED; i++) {
        UA_Server_run_iterate(server, false);
        UA_Client_run_iterate(client, 1);
    }
    ck_assert(client->sessionState != UA_SESSIONSTATE_ACTIVATED);

    /* The counter did not wrap around */
    ck_assert_uint_le(client->currentlyOutStandingPublishRequests,
                      cc->outStandingPublishRequests);

    runServer();
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_statusChange) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedLifetimeCount = 5;
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request,
                                       NULL, statusChangeHandler, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_NodeId monTarget = UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME);
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(monTarget);

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH, monRequest,
                                                  NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    /* Manually control the server thread */
    pauseServer();

    /* Manually set the StatusChange */
    UA_Subscription *sub = getSubscriptionById(server, response.subscriptionId);
    sub->statusChange = 1234; /* some statuscode */

    /* Send publish requests and receive them on the server side */
    UA_Client_run_iterate(client, 1);
    UA_Server_run_iterate(server, false);

    /* Server sends a StatusChange notification */
    UA_fakeSleep((UA_UInt32)response.revisedPublishingInterval + 1);
    UA_Server_run_iterate(server, false);

    /* Client receives the StatusChange */
    UA_Client_run_iterate(client, 1);

    /* The same status has been received */
    ck_assert_uint_eq(statusChange, 1234);

    UA_Server_run_shutdown(server);
    UA_Client_disconnect(client);
    UA_Client_delete(client);

    runServer();
}
END_TEST

/* Write to the variable that is being monitored at a high rate */
START_TEST(Client_subscription_writeBurst) {
    /* add a variable node to the address space */
    UA_VariableAttributes attr = UA_VariableAttributes_default;
    UA_Int32 myInteger = 42;
    UA_Variant_setScalar(&attr.value, &myInteger, &UA_TYPES[UA_TYPES_INT32]);
    attr.description = UA_LOCALIZEDTEXT("en-US","the answer");
    attr.displayName = UA_LOCALIZEDTEXT("en-US","the answer");
    attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
    UA_NodeId myIntegerNodeId = UA_NODEID_STRING(1, "the.answer");
    UA_QualifiedName myIntegerName = UA_QUALIFIEDNAME(1, "the answer");
    UA_NodeId parentNodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER);
    UA_NodeId parentReferenceNodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES);
    UA_StatusCode retval = UA_Server_addVariableNode(server, myIntegerNodeId, parentNodeId,
                                                     parentReferenceNodeId, myIntegerName,
                                                     UA_NODEID_NULL, attr, NULL, NULL);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client *client = UA_Client_newForUnitTest();
    /* This test intentionally generates an unconstrained asynchronous burst. */
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->maxAsyncServiceCalls = 0;
    retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(myIntegerNodeId);

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    pauseServer();

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_DateTime origTime = UA_DateTime_nowMonotonic();
    do {
        myInteger++;
        retval = UA_Client_writeValueAttribute_async(client, myIntegerNodeId, &attr.value,
                                                     NULL, NULL, NULL);
        ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
        UA_Server_run_iterate(server, false);
        UA_Client_run_iterate(client, 1);
        UA_fakeSleep(2);
    } while(UA_DateTime_nowMonotonic() - origTime < 1 * UA_DATETIME_SEC);

    printf("done\n");

    runServer();

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_timeout) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->reconnectInterval = 10;
    cc->reconnectIntervalMax = 40;
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);

    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Still receiving on the MonitoredItem */
    UA_DateTime renewSleep = UA_DateTime_nowMonotonic() - client->nextChannelRenewal;
    UA_fakeSleep((UA_UInt32)(renewSleep / UA_DATETIME_MSEC) + 1);

    /* manually control the server thread */
    pauseServer();

    /* Shut down the server */
    UA_Server_run_shutdown(server);

    /* The client tries to reconnect, but has to fail eventually as the server
     * is down. Advance the clock for the delay between the re-dials. */
    do {
        UA_fakeSleep(10);
        UA_Client_run_iterate(client, 1);
    } while(client->connectStatus == UA_STATUSCODE_GOOD);

    /* The refused re-dials were repeated before giving up */
    ck_assert_uint_eq(client->connectStatus, UA_STATUSCODE_BADCONNECTIONREJECTED);
    ck_assert_uint_gt(client->reconnectAttempts, 1);

    UA_Client_disconnect(client);
    UA_Client_delete(client);

    /* Run the server in an independent thread again for the shutdown after the
     * unit test */
    runServer();
}
END_TEST

START_TEST(Client_subscription_detach) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);


    /* Close the session but detach the subscription */
    UA_CloseSessionRequest closeRequest;
    UA_CloseSessionRequest_init(&closeRequest);
    closeRequest.deleteSubscriptions = false;
    UA_CloseSessionResponse closeResponse;

    __UA_Client_Service(client,
                        &closeRequest, &UA_TYPES[UA_TYPES_CLOSESESSIONREQUEST],
                        &closeResponse, &UA_TYPES[UA_TYPES_CLOSESESSIONRESPONSE]);

    /* manually control the server thread */
    pauseServer();

    /* Let the subscription run its course */
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    /* run the server in an independent thread again */
    runServer();

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_without_notification) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedMaxKeepAliveCount = 1;
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    monRequest.requestedParameters.samplingInterval = 99999999.0;

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    UA_UInt32 monId = monResponse.monitoredItemId;
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    /* manually control the server thread */
    pauseServer();

    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    notificationReceived = false;
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, true);

    notificationReceived = false;
    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    retval = UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(notificationReceived, false);

    /* Get the server back up */
    runServer();

    retval = UA_Client_MonitoredItems_deleteSingle(client, subId, monId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

static UA_SecureChannelState chanState;
static UA_SessionState sessState;
static UA_Boolean hasMon;

static void
monCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
            UA_CreateMonitoredItemsResponse *r) {
    hasMon = true;
}

static void
createSubscriptionCallback2(UA_Client *client, void *userdata, UA_UInt32 requestId,
                            UA_CreateSubscriptionResponse *r) {
    UA_CreateSubscriptionResponse *rr = (UA_CreateSubscriptionResponse*)r;

    ck_assert_uint_ne(rr->subscriptionId, 0);

    /* Add a MonitoredItem */
    UA_NodeId currentTime =
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME);
    UA_CreateMonitoredItemsRequest req;
    UA_CreateMonitoredItemsRequest_init(&req);
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(currentTime);
    req.itemsToCreate = &monRequest;
    req.itemsToCreateSize = 1;
    req.subscriptionId = rr->subscriptionId;
    
    UA_Client_DataChangeNotificationCallback callbacks[1] = {dataChangeHandler};
    UA_StatusCode res =
        UA_Client_MonitoredItems_createDataChanges_async(client, req,
                                                         NULL, callbacks, NULL,
                                                         monCallback, NULL, NULL);
    ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);
}

static void
stateCallback(UA_Client *client, UA_SecureChannelState channelState,
              UA_SessionState sessionState, UA_StatusCode recoveryStatus) {
    chanState = channelState;
    sessState = sessionState;

    if(noNewSubscription)
        return;

    if(sessionState == UA_SESSIONSTATE_ACTIVATED) {
        /* A new session was created. We need to create the subscription. */
        UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
        request.requestedMaxKeepAliveCount = 1;
        UA_StatusCode res =
            UA_Client_Subscriptions_create_async(client, request, NULL, NULL, NULL,
                                                 createSubscriptionCallback2, NULL, NULL);

        ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);
    }
}

static UA_Boolean inactivityCallbackCalled = false;

static void
subscriptionInactivityCallback (UA_Client *client, UA_UInt32 subId, void *subContext) {
    inactivityCallbackCalled = true;
}

START_TEST(Client_subscription_async_sub) {
    UA_Client *client = UA_Client_newForUnitTest();

    /* Set stateCallback */
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->stateCallback = stateCallback;
    cc->subscriptionInactivityCallback = subscriptionInactivityCallback;
    inactivityCallbackCalled = false;

    /* Activate background publish request */
    cc->outStandingPublishRequests = 10;

    ck_assert_uint_eq(chanState, UA_SECURECHANNELSTATE_CLOSED);

    hasMon = false;
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(sessState, UA_SESSIONSTATE_ACTIVATED);

    UA_Client_run_iterate(client, 1);

    /* manually control the server thread */
    pauseServer();

    countNotificationReceived = 0;
    notificationReceived = false;

    while(!hasMon) {
        UA_Client_run_iterate(client, 1);
        UA_Server_run_iterate(server, false);
    }

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);
    UA_Client_run_iterate(client, 1);

    UA_Server_run_iterate(server, false);
    UA_Client_run_iterate(client, 1);

    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 1);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 2);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 3);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 4);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    notificationReceived = false;
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 5);

    /* Simulate network cable unplugged (no response from server) */
    UA_ConnectionManager *cm = client->channel.connectionManager;
    cm->closeConnection(cm, client->channel.connectionId);
    UA_fakeSleep((UA_UInt32)cc->timeout * 2);

    ck_assert_uint_lt(client->config.outStandingPublishRequests, 10);
    ck_assert_uint_eq(inactivityCallbackCalled, false);

    UA_Client_run_iterate(client, 0);
    ck_assert_uint_eq(inactivityCallbackCalled, true);
    ck_assert_uint_eq(sessState, UA_SESSIONSTATE_CREATED);

    /* Get the server back up */
    runServer();

    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_reconnect) {
    UA_Client *client = UA_Client_newForUnitTest();

    /* Set stateCallback */
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->stateCallback = stateCallback;
    cc->subscriptionInactivityCallback = subscriptionInactivityCallback;
    inactivityCallbackCalled = false;

    /* Activate background publish request */
    cc->outStandingPublishRequests = 10;

    hasMon = false;
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(sessState, UA_SESSIONSTATE_ACTIVATED);

    UA_Client_run_iterate(client, 1);

    /* manually control the server thread */
    pauseServer();

    while(!hasMon) {
        UA_Server_run_iterate(server, false);
        UA_Client_run_iterate(client, 1);
    }

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Server_run_iterate(server, false);

    countNotificationReceived = 0;

    notificationReceived = false;
    UA_Client_run_iterate(client, 1);
    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 1);

    runServer();
    UA_Client_disconnectSecureChannel(client);

    /* Reconnect to the old session and see if the old subscription still works */
    noNewSubscription = true;
    retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(sessState, UA_SESSIONSTATE_ACTIVATED);

    /* manually control the server thread */
    pauseServer();

    notificationReceived = false;

    while(!notificationReceived) {
        UA_fakeSleep((UA_UInt32)publishingInterval + 1);
        UA_Client_run_iterate(client, 1);
        UA_Server_run_iterate(server, false);
    }

    ck_assert_uint_eq(notificationReceived, true);
    ck_assert_uint_eq(countNotificationReceived, 2);

    /* Get the server back up */
    runServer();

    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_server_disappears) {
    UA_Client *client = UA_Client_newForUnitTest();

    /* Set stateCallback */
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->stateCallback = stateCallback;
    cc->subscriptionInactivityCallback = subscriptionInactivityCallback;
    inactivityCallbackCalled = false;

    /* Activate background publish request */
    cc->outStandingPublishRequests = 10;

    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(sessState, UA_SESSIONSTATE_ACTIVATED);

    /* Create a Subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* Create a MonitoredItem */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Client_run_iterate(client, 1);

    /* Shut down the server */
    teardown();

    UA_fakeSleep((UA_UInt32)publishingInterval + 1);
    UA_Client_run_iterate(client, 1);

    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_transfer) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    request.requestedLifetimeCount = 5;
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                              NULL, statusChangeHandler, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));

    for(size_t i = 0; i < 5; i++) {
    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, dataChangeHandler, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);
    }

    /* Create a second client */
    UA_Client *client2 = UA_Client_newForUnitTest();
    retval = UA_Client_connect(client2, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Move the subscription to the second client */
    UA_TransferSubscriptionsRequest trequest;
    UA_TransferSubscriptionsRequest_init(&trequest);
    trequest.subscriptionIds = &response.subscriptionId;
    trequest.subscriptionIdsSize = 1;

    UA_TransferSubscriptionsResponse tresponse;
    __UA_Client_Service(client2,
                        &trequest, &UA_TYPES[UA_TYPES_TRANSFERSUBSCRIPTIONSREQUEST],
                        &tresponse,  &UA_TYPES[UA_TYPES_TRANSFERSUBSCRIPTIONSRESPONSE]);

    UA_TransferSubscriptionsResponse_clear(&tresponse);

    /* Iterate the clients some more to see what happens */
    for(size_t i = 0; i < 10; i++) {
        UA_Client_run_iterate(client, 1);
        UA_Client_run_iterate(client2, 1);

        UA_fakeSleep(100);
    }

    /* Delete */
    UA_Client_disconnect(client);
    UA_Client_delete(client);
    UA_Client_disconnect(client2);
    UA_Client_delete(client2);
}
END_TEST

START_TEST(Client_subscription_getSetContext) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Get context — initially NULL */
    void *ctx = (void*)0xDEAD;
    retval = UA_Client_Subscriptions_getContext(client, subId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_ptr_eq(ctx, NULL);

    /* Set context */
    int myData = 42;
    retval = UA_Client_Subscriptions_setContext(client, subId, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Get context again */
    retval = UA_Client_Subscriptions_getContext(client, subId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_ptr_eq(ctx, &myData);

    /* Invalid subscription id */
    retval = UA_Client_Subscriptions_getContext(client, 99999, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    retval = UA_Client_Subscriptions_setContext(client, 99999, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    /* NULL args */
    retval = UA_Client_Subscriptions_getContext(client, subId, NULL);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADINVALIDARGUMENT);

    retval = UA_Client_Subscriptions_getContext(NULL, subId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADINVALIDARGUMENT);

    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_deleteSingle) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Delete it */
    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Delete again — should fail */
    retval = UA_Client_Subscriptions_deleteSingle(client, subId);
    ck_assert(retval != UA_STATUSCODE_GOOD);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_setPublishingMode) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Disable publishing */
    UA_SetPublishingModeRequest spmReq;
    UA_SetPublishingModeRequest_init(&spmReq);
    spmReq.publishingEnabled = false;
    spmReq.subscriptionIds = &subId;
    spmReq.subscriptionIdsSize = 1;
    UA_SetPublishingModeResponse spmResp =
        UA_Client_Subscriptions_setPublishingMode(client, spmReq);
    ck_assert_uint_eq(spmResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(spmResp.resultsSize, 1);
    ck_assert_uint_eq(spmResp.results[0], UA_STATUSCODE_GOOD);
    UA_SetPublishingModeResponse_clear(&spmResp);

    /* Re-enable publishing */
    spmReq.publishingEnabled = true;
    spmResp = UA_Client_Subscriptions_setPublishingMode(client, spmReq);
    ck_assert_uint_eq(spmResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_SetPublishingModeResponse_clear(&spmResp);

    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

static void
dataChangeCallback_ext(UA_Client *c, UA_UInt32 subId, void *subContext,
                       UA_UInt32 monId, void *monContext,
                       UA_DataValue *value) {
    /* no-op: just need a valid callback */
}

static UA_UInt32 monDeleteCount;

static void
monDeleteCallback(UA_Client *c, UA_UInt32 subId, void *subContext,
                  UA_UInt32 monId, void *monContext) {
    monDeleteCount++;
}

/* Close the Session on the server side. The client then receives
 * BadSessionClosed for its queued Publish requests or BadSessionIdInvalid for
 * the next request. Both remove the local Subscriptions while the synchronous
 * call still waits for its response. */
static void
closeSessionOnServer(UA_Client *client) {
    pauseServer();
    UA_StatusCode retval = UA_Server_closeSession(server, &client->sessionId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    runServer();
}

START_TEST(Client_monitoredItems_create_sessionClosed) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    closeSessionOnServer(client);

    monDeleteCount = 0;
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    UA_MonitoredItemCreateResult monResult =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL,
                                                  dataChangeCallback_ext,
                                                  monDeleteCallback);
    ck_assert(monResult.statusCode == UA_STATUSCODE_BADSESSIONIDINVALID ||
              monResult.statusCode == UA_STATUSCODE_BADSESSIONCLOSED);

    /* The MonitoredItem was removed once, together with the Subscription */
    ck_assert_uint_eq(monDeleteCount, 1);

    UA_MonitoredItemCreateResult_clear(&monResult);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_monitoredItems_modify_sessionClosed) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    UA_MonitoredItemCreateResult monResult =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL,
                                                  dataChangeCallback_ext, NULL);
    ck_assert_uint_eq(monResult.statusCode, UA_STATUSCODE_GOOD);

    closeSessionOnServer(client);

    UA_MonitoredItemModifyRequest itemToModify;
    UA_MonitoredItemModifyRequest_init(&itemToModify);
    itemToModify.monitoredItemId = monResult.monitoredItemId;
    itemToModify.requestedParameters.samplingInterval = 100.0;
    itemToModify.requestedParameters.queueSize = 1;

    UA_ModifyMonitoredItemsRequest modifyRequest;
    UA_ModifyMonitoredItemsRequest_init(&modifyRequest);
    modifyRequest.subscriptionId = response.subscriptionId;
    modifyRequest.timestampsToReturn = UA_TIMESTAMPSTORETURN_BOTH;
    modifyRequest.itemsToModify = &itemToModify;
    modifyRequest.itemsToModifySize = 1;

    UA_ModifyMonitoredItemsResponse modifyResponse =
        UA_Client_MonitoredItems_modify(client, modifyRequest);
    ck_assert(modifyResponse.responseHeader.serviceResult ==
              UA_STATUSCODE_BADSESSIONIDINVALID ||
              modifyResponse.responseHeader.serviceResult ==
              UA_STATUSCODE_BADSESSIONCLOSED);

    UA_ModifyMonitoredItemsResponse_clear(&modifyResponse);
    UA_MonitoredItemCreateResult_clear(&monResult);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_monitoredItem_getSetContext) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Create a monitored item */
    UA_MonitoredItemCreateRequest monReq =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    UA_MonitoredItemCreateResult monResult =
        UA_Client_MonitoredItems_createDataChange(client, subId,
            UA_TIMESTAMPSTORETURN_BOTH, monReq, NULL, dataChangeCallback_ext, NULL);
    ck_assert_uint_eq(monResult.statusCode, UA_STATUSCODE_GOOD);
    UA_UInt32 monId = monResult.monitoredItemId;

    /* Get context — initially NULL */
    void *ctx = (void*)0xDEAD;
    retval = UA_Client_MonitoredItem_getContext(client, subId, monId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_ptr_eq(ctx, NULL);

    /* Set context */
    int myData = 123;
    retval = UA_Client_MonitoredItem_setContext(client, subId, monId, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Get context again */
    retval = UA_Client_MonitoredItem_getContext(client, subId, monId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_ptr_eq(ctx, &myData);

    /* Invalid subscription id */
    retval = UA_Client_MonitoredItem_getContext(client, 99999, monId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    /* Invalid monitored item id */
    retval = UA_Client_MonitoredItem_getContext(client, subId, 99999, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADMONITOREDITEMIDINVALID);

    /* NULL args */
    retval = UA_Client_MonitoredItem_getContext(client, subId, monId, NULL);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADINVALIDARGUMENT);

    retval = UA_Client_MonitoredItem_getContext(NULL, subId, monId, &ctx);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADINVALIDARGUMENT);

    /* Set context with invalid sub */
    retval = UA_Client_MonitoredItem_setContext(client, 99999, monId, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    retval = UA_Client_MonitoredItem_setContext(client, subId, 99999, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADMONITOREDITEMIDINVALID);

    retval = UA_Client_MonitoredItem_setContext(NULL, subId, monId, &myData);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADINVALIDARGUMENT);

    UA_Client_MonitoredItems_deleteSingle(client, subId, monId);
    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_setMonitoringMode) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription + monitored item */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    UA_MonitoredItemCreateRequest monReq =
        UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    UA_MonitoredItemCreateResult monResult =
        UA_Client_MonitoredItems_createDataChange(client, subId,
            UA_TIMESTAMPSTORETURN_BOTH, monReq, NULL, dataChangeCallback_ext, NULL);
    ck_assert_uint_eq(monResult.statusCode, UA_STATUSCODE_GOOD);
    UA_UInt32 monId = monResult.monitoredItemId;

    /* Set monitoring mode to Disabled */
    UA_SetMonitoringModeRequest smmReq;
    UA_SetMonitoringModeRequest_init(&smmReq);
    smmReq.subscriptionId = subId;
    smmReq.monitoringMode = UA_MONITORINGMODE_DISABLED;
    smmReq.monitoredItemIds = &monId;
    smmReq.monitoredItemIdsSize = 1;
    UA_SetMonitoringModeResponse smmResp =
        UA_Client_MonitoredItems_setMonitoringMode(client, smmReq);
    ck_assert_uint_eq(smmResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_SetMonitoringModeResponse_clear(&smmResp);

    /* Set back to Reporting */
    smmReq.monitoringMode = UA_MONITORINGMODE_REPORTING;
    smmResp = UA_Client_MonitoredItems_setMonitoringMode(client, smmReq);
    ck_assert_uint_eq(smmResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_SetMonitoringModeResponse_clear(&smmResp);

    UA_Client_MonitoredItems_deleteSingle(client, subId, monId);
    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_setTriggering) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Create trigger and triggering items */
    UA_MonitoredItemCreateRequest monReqs[2];
    monReqs[0] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
    monReqs[1] = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));

    UA_MonitoredItemCreateResult monResults[2];
    monResults[0] = UA_Client_MonitoredItems_createDataChange(client, subId,
        UA_TIMESTAMPSTORETURN_BOTH, monReqs[0], NULL, dataChangeCallback_ext, NULL);
    ck_assert_uint_eq(monResults[0].statusCode, UA_STATUSCODE_GOOD);
    monResults[1] = UA_Client_MonitoredItems_createDataChange(client, subId,
        UA_TIMESTAMPSTORETURN_BOTH, monReqs[1], NULL, dataChangeCallback_ext, NULL);
    ck_assert_uint_eq(monResults[1].statusCode, UA_STATUSCODE_GOOD);

    UA_UInt32 triggeringItemId = monResults[0].monitoredItemId;
    UA_UInt32 linkedItemId = monResults[1].monitoredItemId;

    /* SetTriggering: add a link */
    UA_SetTriggeringRequest stReq;
    UA_SetTriggeringRequest_init(&stReq);
    stReq.subscriptionId = subId;
    stReq.triggeringItemId = triggeringItemId;
    stReq.linksToAdd = &linkedItemId;
    stReq.linksToAddSize = 1;
    UA_SetTriggeringResponse stResp =
        UA_Client_MonitoredItems_setTriggering(client, stReq);
    ck_assert_uint_eq(stResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_SetTriggeringResponse_clear(&stResp);

    /* Remove the link */
    UA_SetTriggeringRequest stReq2;
    UA_SetTriggeringRequest_init(&stReq2);
    stReq2.subscriptionId = subId;
    stReq2.triggeringItemId = triggeringItemId;
    stReq2.linksToRemove = &linkedItemId;
    stReq2.linksToRemoveSize = 1;
    stResp = UA_Client_MonitoredItems_setTriggering(client, stReq2);
    ck_assert_uint_eq(stResp.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_SetTriggeringResponse_clear(&stResp);

    UA_Client_MonitoredItems_deleteSingle(client, subId, monResults[0].monitoredItemId);
    UA_Client_MonitoredItems_deleteSingle(client, subId, monResults[1].monitoredItemId);
    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_modifyAsync) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Create subscription */
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    UA_UInt32 subId = response.subscriptionId;

    /* Modify via async */
    UA_ModifySubscriptionRequest modReq;
    UA_ModifySubscriptionRequest_init(&modReq);
    modReq.subscriptionId = subId;
    modReq.requestedPublishingInterval = 200.0;
    modReq.requestedMaxKeepAliveCount = 20;
    modReq.requestedLifetimeCount = 600;
    modReq.maxNotificationsPerPublish = 100;

    UA_UInt32 reqId = 0;
    retval = UA_Client_Subscriptions_modify_async(client, modReq,
                                                  NULL, NULL, &reqId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert(reqId != 0);

    /* Drive iterations to get response */
    for(int i = 0; i < 20; i++) {
        UA_fakeSleep(10);
        UA_Client_run_iterate(client, 1);
    }

    UA_Client_Subscriptions_deleteSingle(client, subId);
    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_methodcall) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
                                                                            NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);

    /* monitor the server state */
    UA_MonitoredItemCreateRequest monRequest =
        UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));

    UA_MonitoredItemCreateResult monResponse =
        UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
                                                  UA_TIMESTAMPSTORETURN_BOTH,
                                                  monRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(monResponse.statusCode, UA_STATUSCODE_GOOD);

/* Minimal nodeset does not contain any method we can call here */
#ifdef UA_GENERATED_NAMESPACE_ZERO
    UA_UInt32 monId = monResponse.monitoredItemId;
    UA_UInt32 subId = response.subscriptionId;

    /* call a method to get monitored item id */
    UA_Variant input;
    UA_Variant_init(&input);
    UA_Variant_setScalarCopy(&input, &subId, &UA_TYPES[UA_TYPES_UINT32]);
    size_t outputSize;
    UA_Variant *output;
    retval = UA_Client_call(client, UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER),
                            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_GETMONITOREDITEMS),
                            1, &input, &outputSize, &output);

    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(outputSize, 2);

    ck_assert_uint_eq(output[0].arrayLength, 1);

    ck_assert_uint_eq(*((UA_UInt32*)output[0].data), monId);

    UA_Array_delete(output, outputSize, &UA_TYPES[UA_TYPES_VARIANT]);
    UA_Variant_clear(&input);

    /* call with invalid subscription id */
    UA_Variant_init(&input);
    subId = 0;
    UA_Variant_setScalarCopy(&input, &subId, &UA_TYPES[UA_TYPES_UINT32]);
    retval = UA_Client_call(client, UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER),
                            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_GETMONITOREDITEMS),
                            1, &input, &outputSize, &output);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);
    UA_Variant_clear(&input);

    /* The method callback must not trust its mutable OutputArguments metadata.
     * Removing the concrete property makes the Call service allocate a result
     * array with the inherited metadata size. */
    retval = UA_Client_deleteNode(
        client,
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_GETMONITOREDITEMS_OUTPUTARGUMENTS),
        true);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    UA_Variant_init(&input);
    subId = response.subscriptionId;
    UA_Variant_setScalarCopy(&input, &subId, &UA_TYPES[UA_TYPES_UINT32]);
    outputSize = 0;
    output = NULL;
    retval = UA_Client_call(client, UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER),
                            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_GETMONITOREDITEMS),
                            1, &input, &outputSize, &output);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADARGUMENTSMISSING);
    UA_Array_delete(output, outputSize, &UA_TYPES[UA_TYPES_VARIANT]);
    UA_Variant_clear(&input);
#endif

    UA_Client_disconnect(client);
    UA_Client_delete(client);
}
END_TEST

START_TEST(Client_subscription_modifyAsync_invalidSubscriptionId) {
    /* src/client/ua_client_subscriptions.c:280-283:
     *   sub = findSubscriptionById(client, request.subscriptionId);
     *   if(!sub) { unlockClient; return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID; }
     * The sync UA_Client_Subscriptions_modify is tested for this path;
     * the async variant is not. */
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* Build a request with a clearly invalid subscriptionId */
    UA_ModifySubscriptionRequest req;
    UA_ModifySubscriptionRequest_init(&req);
    req.subscriptionId = 99999;
    req.requestedPublishingInterval = 500.0;
    req.requestedLifetimeCount = 100;
    req.requestedMaxKeepAliveCount = 10;

    UA_ModifySubscriptionResponse resp;
    UA_UInt32 requestId = 0;
    retval = UA_Client_Subscriptions_modify_async(
        client, req,
        (UA_ClientAsyncModifySubscriptionCallback)NULL,
        &resp, &requestId);
    ck_assert_uint_eq(retval, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
} END_TEST

START_TEST(Client_subscription_admission_before_state_lookup) {
    UA_Client *client = UA_Client_newForUnitTest();
    UA_ClientConfig *cc = UA_Client_getConfig(client);
    cc->outStandingPublishRequests = 0;
    cc->maxAsyncServiceCalls = 1;
    cc->asyncServiceCallRule = UA_RULEHANDLING_ACCEPT;

    UA_StatusCode res = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);

    UA_CreateSubscriptionRequest createRequest =
        UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse createResponse =
        UA_Client_Subscriptions_create(client, createRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(createResponse.responseHeader.serviceResult,
                      UA_STATUSCODE_GOOD);
    UA_UInt32 subId = createResponse.subscriptionId;
    UA_CreateSubscriptionResponse_clear(&createResponse);

    UA_ReadValueId rvid;
    UA_ReadValueId_init(&rvid);
    rvid.attributeId = UA_ATTRIBUTEID_VALUE;
    rvid.nodeId = UA_NODEID_NUMERIC(
        0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME);
    UA_ReadRequest readRequest;
    UA_ReadRequest_init(&readRequest);
    readRequest.nodesToRead = &rvid;
    readRequest.nodesToReadSize = 1;

    res = __UA_Client_AsyncService(
        client, &readRequest, &UA_TYPES[UA_TYPES_READREQUEST],
        clearSubscriptionsOnRead, &UA_TYPES[UA_TYPES_READRESPONSE],
        &subId, NULL);
    ck_assert_uint_eq(res, UA_STATUSCODE_GOOD);

    UA_ModifySubscriptionRequest modifyRequest;
    UA_ModifySubscriptionRequest_init(&modifyRequest);
    modifyRequest.subscriptionId = subId;
    modifyRequest.requestedPublishingInterval = 500.0;
    modifyRequest.requestedLifetimeCount = 100;
    modifyRequest.requestedMaxKeepAliveCount = 10;

    /* Admission processes the read callback first. The callback deletes the
     * subscription, so the state lookup afterwards must observe its removal. */
    res = UA_Client_Subscriptions_modify_async(
        client, modifyRequest, NULL, NULL, NULL);
    ck_assert_uint_eq(res, UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID);
    ck_assert_uint_eq(client->outstandingAsyncServiceCalls, 0);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
} END_TEST

START_TEST(Client_renewSecureChannel_returnsGoodCallAgain) {
    /* src/client/ua_client_connect.c:712-715:
     *   if(channel.state != OPEN || renewState == SENT || nextRenewal > now)
     *     return UA_STATUSCODE_GOODCALLAGAIN;
     * Right after a fresh connect, the channel is OPEN and the renewal
     * timer is in the future, so the call returns GOODCALLAGAIN without
     * sending anything. None of the existing tests call the public
     * UA_Client_renewSecureChannel directly. */
    UA_Client *client = UA_Client_newForUnitTest();
    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);

    /* First call: channel is fresh, so GOODCALLAGAIN */
    retval = UA_Client_renewSecureChannel(client);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOODCALLAGAIN);

    /* Second call: same -- channel is still fresh */
    retval = UA_Client_renewSecureChannel(client);
    ck_assert_uint_eq(retval, UA_STATUSCODE_GOODCALLAGAIN);

    UA_Client_disconnect(client);
    UA_Client_delete(client);
} END_TEST

typedef struct {
    UA_Client *client;
    UA_UInt32 requestId;
    unsigned calls;
    UA_SetMonitoringModeResponse monitoring;
    UA_SetTriggeringResponse triggering;
} AsyncSettingsResult;

static void
monitoringModeCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                       UA_SetMonitoringModeResponse *response) {
    AsyncSettingsResult *result = (AsyncSettingsResult *)userdata;
    ck_assert_ptr_eq(client, result->client);
    ck_assert_uint_eq(result->calls++, 0);
    result->requestId = requestId;
    ck_assert_uint_eq(UA_SetMonitoringModeResponse_copy(response, &result->monitoring),
                      UA_STATUSCODE_GOOD);
}

static void
triggeringCallback(UA_Client *client, void *userdata, UA_UInt32 requestId,
                    UA_SetTriggeringResponse *response) {
    AsyncSettingsResult *result = (AsyncSettingsResult *)userdata;
    ck_assert_ptr_eq(client, result->client);
    ck_assert_uint_eq(result->calls++, 0);
    result->requestId = requestId;
    ck_assert_uint_eq(UA_SetTriggeringResponse_copy(response, &result->triggering),
                      UA_STATUSCODE_GOOD);
}

static UA_Client *
createClientForAsyncSettings(UA_UInt32 *subscriptionId, UA_UInt32 monitoredIds[2]) {
    UA_Client *client = UA_Client_newForUnitTest();
    ck_assert_ptr_nonnull(client);
    UA_Client_getConfig(client)->outStandingPublishRequests = 0;
    ck_assert_uint_eq(UA_Client_connect(client, "opc.tcp://localhost:4840"), UA_STATUSCODE_GOOD);
    UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    *subscriptionId = response.subscriptionId;
    UA_CreateSubscriptionResponse_clear(&response);
    for(size_t i = 0; i < 2; i++) {
        UA_MonitoredItemCreateRequest item = UA_MonitoredItemCreateRequest_default(
            UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
        item.monitoringMode = i == 0 ? UA_MONITORINGMODE_REPORTING : UA_MONITORINGMODE_SAMPLING;
        UA_MonitoredItemCreateResult created = UA_Client_MonitoredItems_createDataChange(
            client, *subscriptionId, UA_TIMESTAMPSTORETURN_NEITHER, item, NULL, NULL, NULL);
        ck_assert_uint_eq(created.statusCode, UA_STATUSCODE_GOOD);
        monitoredIds[i] = created.monitoredItemId;
        UA_MonitoredItemCreateResult_clear(&created);
    }
    return client;
}

static void
awaitAsyncSettings(UA_Client *client) {
    /* The Read follows the settings request on the same connection. Waiting
     * for its response also dispatches the preceding asynchronous response,
     * including when the caller did not supply a callback. */
    UA_Variant value;
    UA_Variant_init(&value);
    ck_assert_uint_eq(UA_Client_readValueAttribute(client,
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE), &value), UA_STATUSCODE_GOOD);
    UA_Variant_clear(&value);
}

static void
deleteAsyncSettingsClient(UA_Client *client, UA_UInt32 subscriptionId) {
    ck_assert_uint_eq(UA_Client_Subscriptions_deleteSingle(client, subscriptionId),
                      UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(UA_Client_disconnect(client), UA_STATUSCODE_GOOD);
    UA_Client_delete(client);
}

START_TEST(Client_asyncMonitoringModeReportsResults) {
    UA_UInt32 subscriptionId, monitoredIds[2];
    UA_Client *client = createClientForAsyncSettings(&subscriptionId, monitoredIds);
    UA_UInt32 ids[2] = {monitoredIds[0], UA_UINT32_MAX};
    UA_SetMonitoringModeRequest request;
    UA_SetMonitoringModeRequest_init(&request);
    request.subscriptionId = _i == 2 ? UA_UINT32_MAX : subscriptionId;
    request.monitoringMode = _i == 3 ? (UA_MonitoringMode)3 : UA_MONITORINGMODE_DISABLED;
    request.monitoredItemIdsSize = 2;
    request.monitoredItemIds = ids;
    AsyncSettingsResult result = {0};
    result.client = client;
    UA_UInt32 requestId = 0;
    ck_assert_uint_eq(UA_Client_MonitoredItems_setMonitoringMode_async(client, request,
        _i == 1 ? NULL : monitoringModeCallback, &result, &requestId), UA_STATUSCODE_GOOD);
    ck_assert_uint_ne(requestId, 0);
    awaitAsyncSettings(client);
    ck_assert_uint_eq(result.calls, _i == 1 ? 0 : 1);
    if(_i != 1) {
        ck_assert_uint_eq(result.requestId, requestId);
        UA_StatusCode expected = _i == 2 ? UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID :
                                         UA_STATUSCODE_GOOD;
        ck_assert_uint_eq(result.monitoring.responseHeader.serviceResult, expected);
        ck_assert_uint_eq(result.monitoring.resultsSize, _i == 2 ? 0 : 2);
        if(_i != 2) {
            ck_assert_uint_eq(result.monitoring.results[0],
                _i == 3 ? UA_STATUSCODE_BADMONITORINGMODEINVALID : UA_STATUSCODE_GOOD);
            ck_assert_uint_eq(result.monitoring.results[1], UA_STATUSCODE_BADMONITOREDITEMIDINVALID);
        }
    }
    pauseServer();
    lockServer(server);
    UA_Subscription *sub = getSubscriptionById(server, subscriptionId);
    ck_assert_ptr_nonnull(sub);
    UA_MonitoredItem *mon = UA_Subscription_getMonitoredItem(sub, monitoredIds[0]);
    ck_assert_ptr_nonnull(mon);
    ck_assert_uint_eq(mon->monitoringMode,
                      _i < 2 ? UA_MONITORINGMODE_DISABLED : UA_MONITORINGMODE_REPORTING);
    mon = UA_Subscription_getMonitoredItem(sub, monitoredIds[1]);
    ck_assert_ptr_nonnull(mon);
    ck_assert_uint_eq(mon->monitoringMode, UA_MONITORINGMODE_SAMPLING);
    unlockServer(server);
    runServer();
    UA_SetMonitoringModeResponse_clear(&result.monitoring);
    deleteAsyncSettingsClient(client, subscriptionId);
} END_TEST

static void
assertServerTriggerLink(UA_UInt32 subscriptionId, const UA_UInt32 monitoredIds[2],
                        UA_Boolean exists) {
    pauseServer();
    lockServer(server);
    UA_Subscription *sub = getSubscriptionById(server, subscriptionId);
    ck_assert_ptr_nonnull(sub);
    UA_MonitoredItem *mon = UA_Subscription_getMonitoredItem(sub, monitoredIds[0]);
    ck_assert_ptr_nonnull(mon);
    ck_assert_uint_eq(mon->triggeringLinksSize, exists ? 1 : 0);
    if(exists)
        ck_assert_uint_eq(mon->triggeringLinks[0], monitoredIds[1]);
    unlockServer(server);
    runServer();
}

START_TEST(Client_asyncTriggeringReportsResults) {
    UA_UInt32 subscriptionId, monitoredIds[2];
    UA_Client *client = createClientForAsyncSettings(&subscriptionId, monitoredIds);
    UA_UInt32 links[2] = {monitoredIds[1], UA_UINT32_MAX};
    UA_SetTriggeringRequest request;
    UA_SetTriggeringRequest_init(&request);
    request.subscriptionId = _i == 2 ? UA_UINT32_MAX : subscriptionId;
    request.triggeringItemId = _i == 3 ? UA_UINT32_MAX : monitoredIds[0];
    request.linksToAddSize = 2;
    request.linksToAdd = links;
    AsyncSettingsResult result = {0};
    result.client = client;
    UA_UInt32 requestId = 0;
    ck_assert_uint_eq(UA_Client_MonitoredItems_setTriggering_async(client, request,
        _i == 1 ? NULL : triggeringCallback, &result, &requestId), UA_STATUSCODE_GOOD);
    ck_assert_uint_ne(requestId, 0);
    awaitAsyncSettings(client);
    ck_assert_uint_eq(result.calls, _i == 1 ? 0 : 1);
    if(_i != 1) {
        ck_assert_uint_eq(result.requestId, requestId);
        UA_StatusCode expected = _i == 2 ? UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID :
            (_i == 3 ? UA_STATUSCODE_BADMONITOREDITEMIDINVALID : UA_STATUSCODE_GOOD);
        ck_assert_uint_eq(result.triggering.responseHeader.serviceResult, expected);
        ck_assert_uint_eq(result.triggering.addResultsSize, _i == 0 ? 2 : 0);
        if(_i == 0) {
            ck_assert_uint_eq(result.triggering.addResults[0], UA_STATUSCODE_GOOD);
            ck_assert_uint_eq(result.triggering.addResults[1], UA_STATUSCODE_BADMONITOREDITEMIDINVALID);
        }
    }
    UA_SetTriggeringResponse_clear(&result.triggering);
    assertServerTriggerLink(subscriptionId, monitoredIds, _i < 2);
    if(_i < 2) {
        result.calls = 0;
        request.linksToAddSize = 0;
        request.linksToAdd = NULL;
        request.linksToRemoveSize = 1;
        request.linksToRemove = &monitoredIds[1];
        ck_assert_uint_eq(UA_Client_MonitoredItems_setTriggering_async(client, request,
            _i == 1 ? NULL : triggeringCallback, &result, &requestId), UA_STATUSCODE_GOOD);
        awaitAsyncSettings(client);
        ck_assert_uint_eq(result.calls, _i == 1 ? 0 : 1);
        if(_i == 0) {
            ck_assert_uint_eq(result.requestId, requestId);
            ck_assert_uint_eq(result.triggering.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
            ck_assert_uint_eq(result.triggering.addResultsSize, 0);
            ck_assert_uint_eq(result.triggering.removeResultsSize, 1);
            ck_assert_uint_eq(result.triggering.removeResults[0], UA_STATUSCODE_GOOD);
        }
        UA_SetTriggeringResponse_clear(&result.triggering);
        assertServerTriggerLink(subscriptionId, monitoredIds, false);
    }
    deleteAsyncSettingsClient(client, subscriptionId);
} END_TEST

typedef struct {
    UA_Client *client;
    UA_UInt32 subscriptionId, monitoredItemId;
    unsigned subscriptionsDeleted, itemsDeleted, inactivityCalls, statusCalls;
} PublishObserver;

static void
publishSubscriptionDeleted(UA_Client *client, UA_UInt32 subId, void *context) {
    PublishObserver *observer = (PublishObserver *)context;
    ck_assert_ptr_eq(client, observer->client);
    ck_assert_uint_eq(subId, observer->subscriptionId);
    observer->subscriptionsDeleted++;
}

static void
publishItemDeleted(UA_Client *client, UA_UInt32 subId, void *subContext,
                   UA_UInt32 monId, void *monContext) {
    PublishObserver *observer = (PublishObserver *)subContext;
    ck_assert_ptr_eq(client, observer->client);
    ck_assert_ptr_eq(monContext, observer);
    ck_assert_uint_eq(subId, observer->subscriptionId);
    ck_assert_uint_eq(monId, observer->monitoredItemId);
    observer->itemsDeleted++;
}

static void
publishInactivity(UA_Client *client, UA_UInt32 subId, void *context) {
    PublishObserver *observer = (PublishObserver *)context;
    ck_assert_ptr_eq(client, observer->client);
    ck_assert_uint_eq(subId, observer->subscriptionId);
    observer->inactivityCalls++;
}

static void
publishStatusChanged(UA_Client *client, UA_UInt32 subId, void *context,
                      UA_StatusChangeNotification *notification) {
    PublishObserver *observer = (PublishObserver *)context;
    ck_assert_ptr_eq(client, observer->client);
    ck_assert_uint_eq(subId, observer->subscriptionId);
    ck_assert_uint_eq(notification->status, UA_STATUSCODE_GOODSUBSCRIPTIONTRANSFERRED);
    observer->statusCalls++;
}

static UA_Client *
createPublishClient(PublishObserver *observer, UA_Boolean statusCallback) {
    memset(observer, 0, sizeof(*observer));
    UA_Client *client = UA_Client_newForUnitTest();
    ck_assert_ptr_nonnull(client);
    observer->client = client;
    UA_ClientConfig *config = UA_Client_getConfig(client);
    config->outStandingPublishRequests = 0;
    config->subscriptionInactivityCallback = publishInactivity;
    ck_assert_uint_eq(UA_Client_connect(client, "opc.tcp://localhost:4840"), UA_STATUSCODE_GOOD);
    UA_CreateSubscriptionResponse sub = UA_Client_Subscriptions_create(client,
        UA_CreateSubscriptionRequest_default(), observer,
        statusCallback ? publishStatusChanged : NULL, publishSubscriptionDeleted);
    ck_assert_uint_eq(sub.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    observer->subscriptionId = sub.subscriptionId;
    UA_CreateSubscriptionResponse_clear(&sub);
    UA_MonitoredItemCreateRequest item = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_STATE));
    UA_MonitoredItemCreateResult created = UA_Client_MonitoredItems_createDataChange(
        client, observer->subscriptionId, UA_TIMESTAMPSTORETURN_NEITHER, item,
        observer, NULL, publishItemDeleted);
    ck_assert_uint_eq(created.statusCode, UA_STATUSCODE_GOOD);
    observer->monitoredItemId = created.monitoredItemId;
    UA_MonitoredItemCreateResult_clear(&created);
    return client;
}

static void
dispatchPublishResponse(UA_Client *client, UA_PublishResponse *response) {
    UA_PublishRequest request;
    UA_PublishRequest_init(&request);
    lockClient(client);
    client->currentlyOutStandingPublishRequests++;
    __Client_Subscriptions_processPublishResponse(client, &request, response);
    unlockClient(client);
}

START_TEST(Client_publishErrorsPreserveOrRemoveSubscription) {
    static const UA_StatusCode errors[] = {
        UA_STATUSCODE_BADSHUTDOWN, UA_STATUSCODE_BADNOSUBSCRIPTION,
        UA_STATUSCODE_BADTIMEOUT, UA_STATUSCODE_BADTIMEOUT,
        UA_STATUSCODE_BADSESSIONCLOSED, UA_STATUSCODE_BADINTERNALERROR,
        UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS, UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS,
        UA_STATUSCODE_GOOD
    };
    PublishObserver observer;
    UA_Client *client = createPublishClient(&observer, true);
    UA_ClientConfig *config = UA_Client_getConfig(client);
    if(_i == 3)
        config->subscriptionInactivityCallback = NULL;
    if(_i == 6 || _i == 7)
        config->outStandingPublishRequests = _i == 6 ? 2 : 1;
    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert_ptr_nonnull(sub);
    UA_DateTime activity = sub->lastActivity;
    UA_UInt32 sequence = sub->sequenceNumber;
    UA_PublishResponse response;
    UA_PublishResponse_init(&response);
    response.subscriptionId = _i == 8 ? UA_UINT32_MAX : observer.subscriptionId;
    response.responseHeader.serviceResult = errors[_i];
    dispatchPublishResponse(client, &response);
    UA_Boolean removed = _i == 4 || _i == 7;
    ck_assert_uint_eq(observer.subscriptionsDeleted, removed ? 1 : 0);
    ck_assert_uint_eq(observer.itemsDeleted, removed ? 1 : 0);
    ck_assert_uint_eq(observer.inactivityCalls, _i == 2 ? 1 : 0);
    ck_assert_uint_eq(observer.statusCalls, 0);
    ck_assert_uint_eq(response.responseHeader.serviceResult,
                      _i == 8 ? UA_STATUSCODE_BADNOSUBSCRIPTION : errors[_i]);
    if(_i == 6 || _i == 7)
        ck_assert_uint_eq(config->outStandingPublishRequests, 1);
    config->outStandingPublishRequests = 0;
    if(removed) {
        ck_assert(LIST_EMPTY(&client->subscriptions));
    } else {
        ck_assert_ptr_eq(LIST_FIRST(&client->subscriptions), sub);
        ck_assert(sub->lastActivity == activity);
        ck_assert_uint_eq(sub->sequenceNumber, sequence);
        ck_assert_uint_eq(countTestMonitoredItems(ZIP_ROOT(&sub->monitoredItems)), 1);
        ck_assert_uint_eq(client->currentlyOutStandingPublishRequests, 0);
        ck_assert_uint_eq(UA_Client_Subscriptions_deleteSingle(client, observer.subscriptionId),
                          UA_STATUSCODE_GOOD);
    }
    UA_PublishResponse_clear(&response);
    ck_assert_uint_eq(UA_Client_disconnect(client), UA_STATUSCODE_GOOD);
    UA_Client_delete(client);
    ck_assert_uint_eq(observer.subscriptionsDeleted, 1);
    ck_assert_uint_eq(observer.itemsDeleted, 1);
} END_TEST

START_TEST(Client_publishWrapsSequenceAndClearsAcknowledgements) {
    PublishObserver observer;
    UA_Client *client = createPublishClient(&observer, _i == 0);
    UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
    ck_assert_ptr_nonnull(sub);
    sub->sequenceNumber = UA_UINT32_MAX;
    UA_StatusChangeNotification status;
    UA_StatusChangeNotification_init(&status);
    status.status = UA_STATUSCODE_GOODSUBSCRIPTIONTRANSFERRED;
    UA_DateTime unknown = 0;
    UA_ExtensionObject notification;
    UA_ExtensionObject_init(&notification);
    if(_i < 2) {
        notification.encoding = UA_EXTENSIONOBJECT_DECODED;
        notification.content.decoded.type = &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION];
        notification.content.decoded.data = &status;
    } else if(_i == 2) {
        notification.encoding = UA_EXTENSIONOBJECT_DECODED;
        notification.content.decoded.type = &UA_TYPES[UA_TYPES_DATETIME];
        notification.content.decoded.data = &unknown;
    }
    /* The injected response borrows these stack objects; it is not cleared. */
    UA_UInt32 available[3] = {17, 1, 23};
    UA_PublishResponse response;
    UA_PublishResponse_init(&response);
    response.subscriptionId = observer.subscriptionId;
    response.notificationMessage.sequenceNumber = 1;
    response.notificationMessage.notificationData = &notification;
    response.notificationMessage.notificationDataSize = 1;
    response.availableSequenceNumbers = available;
    response.availableSequenceNumbersSize = 3;
    dispatchPublishResponse(client, &response);
    ck_assert_uint_eq(response.responseHeader.serviceResult, UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(sub->sequenceNumber, 1);
    ck_assert_uint_eq(observer.statusCalls, _i == 0 ? 1 : 0);
    ck_assert_uint_eq(client->currentlyOutStandingPublishRequests, 0);
    UA_Client_NotificationsAckNumber *ack = LIST_FIRST(&client->pendingNotificationsAcks);
    ck_assert_ptr_nonnull(ack);
    ck_assert_uint_eq(ack->subAck.subscriptionId, observer.subscriptionId);
    ck_assert_uint_eq(ack->subAck.sequenceNumber, 1);
    ck_assert_ptr_null(LIST_NEXT(ack, listEntry));
    lockClient(client);
    __Client_Subscriptions_clear(client);
    unlockClient(client);
    ck_assert(LIST_EMPTY(&client->pendingNotificationsAcks));
    ck_assert(LIST_EMPTY(&client->subscriptions));
    ck_assert_uint_eq(observer.subscriptionsDeleted, 1);
    ck_assert_uint_eq(observer.itemsDeleted, 1);
    ck_assert_uint_eq(UA_Client_disconnect(client), UA_STATUSCODE_GOOD);
    UA_Client_delete(client);
    ck_assert_uint_eq(observer.subscriptionsDeleted, 1);
    ck_assert_uint_eq(observer.itemsDeleted, 1);
} END_TEST

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
typedef struct {
    PublishObserver *owner;
    UA_CreateMonitoredItemsResponse response;
    UA_UInt32 requestId, monitoredItemId, deletedItemId;
    unsigned createCalls, deleteCalls, notifications;
} AsyncEventObserver;

static void
asyncEventCreated(UA_Client *client, void *context, UA_UInt32 requestId,
                   UA_CreateMonitoredItemsResponse *response) {
    AsyncEventObserver *observer = (AsyncEventObserver *)context;
    ck_assert_ptr_eq(client, observer->owner->client);
    ck_assert_uint_eq(observer->createCalls++, 0);
    observer->requestId = requestId;
    ck_assert_uint_eq(UA_CreateMonitoredItemsResponse_copy(response, &observer->response),
                      UA_STATUSCODE_GOOD);
}

static void
asyncEventDeleted(UA_Client *client, UA_UInt32 subId, void *subContext,
                   UA_UInt32 monId, void *context) {
    AsyncEventObserver *observer = (AsyncEventObserver *)context;
    ck_assert_ptr_eq(client, observer->owner->client);
    ck_assert_ptr_eq(subContext, observer->owner);
    ck_assert_uint_eq(subId, observer->owner->subscriptionId);
    ck_assert_uint_eq(observer->deleteCalls++, 0);
    observer->deletedItemId = monId;
}

static void
asyncEventReceived(UA_Client *client, UA_UInt32 subId, void *subContext,
                    UA_UInt32 monId, void *context, const UA_KeyValueMap fields) {
    AsyncEventObserver *observer = (AsyncEventObserver *)context;
    ck_assert_ptr_eq(client, observer->owner->client);
    ck_assert_ptr_eq(subContext, observer->owner);
    ck_assert_uint_eq(subId, observer->owner->subscriptionId);
    ck_assert_uint_eq(monId, observer->monitoredItemId);
    ck_assert_uint_eq(fields.mapSize, 1);
    const UA_LocalizedText *message = (const UA_LocalizedText *)UA_KeyValueMap_getScalar(
        &fields, UA_QUALIFIEDNAME(0, "/Message"), &UA_TYPES[UA_TYPES_LOCALIZEDTEXT]);
    ck_assert_ptr_nonnull(message);
    UA_LocalizedText expected = UA_LOCALIZEDTEXT("en", "accepted event");
    ck_assert(UA_LocalizedText_equal(message, &expected));
    observer->notifications++;
}

static UA_Client_MonitoredItem *
findEventByContext(UA_Client_MonitoredItem *mon, void *context) {
    if(!mon || mon->context == context)
        return mon;
    UA_Client_MonitoredItem *found = findEventByContext(ZIP_LEFT(mon, zipfields), context);
    return found ? found : findEventByContext(ZIP_RIGHT(mon, zipfields), context);
}

START_TEST(Client_asyncEventCreationOwnsPendingItems) {
    PublishObserver owner;
    UA_Client *client = createPublishClient(&owner, false);
    AsyncEventObserver observer;
    memset(&observer, 0, sizeof(observer));
    observer.owner = &owner;
    UA_QualifiedName path = UA_QUALIFIEDNAME(0, "Message");
    UA_SimpleAttributeOperand clause;
    UA_SimpleAttributeOperand_init(&clause);
    clause.typeDefinitionId = UA_NODEID_NUMERIC(0, UA_NS0ID_BASEEVENTTYPE);
    clause.browsePath = &path;
    clause.browsePathSize = 1;
    clause.attributeId = UA_ATTRIBUTEID_VALUE;
    UA_EventFilter filter;
    UA_EventFilter_init(&filter);
    filter.selectClauses = &clause;
    filter.selectClausesSize = _i == 3 ? 0 : 1;
    UA_MonitoredItemCreateRequest item = UA_MonitoredItemCreateRequest_default(
        UA_NODEID_NUMERIC(0, _i == 2 ? UA_UINT32_MAX : UA_NS0ID_SERVER));
    item.itemToMonitor.attributeId = UA_ATTRIBUTEID_EVENTNOTIFIER;
    UA_ExtensionObject_setValueNoDelete(&item.requestedParameters.filter,
                                        &filter, &UA_TYPES[UA_TYPES_EVENTFILTER]);
    UA_CreateMonitoredItemsRequest request;
    UA_CreateMonitoredItemsRequest_init(&request);
    request.subscriptionId = _i == 4 ? UA_UINT32_MAX : owner.subscriptionId;
    request.timestampsToReturn = _i == 7 ? (UA_TimestampsToReturn)999 : UA_TIMESTAMPSTORETURN_BOTH;
    request.itemsToCreate = &item;
    request.itemsToCreateSize = _i == 5 ? 0 : 1;
    void *context = &observer;
    UA_Client_EventNotificationCallback notificationCallback = asyncEventReceived;
    UA_Client_DeleteMonitoredItemCallback deleteCallback = asyncEventDeleted;
    UA_UInt32 requestId = UA_UINT32_MAX;
    UA_StatusCode result = UA_Client_MonitoredItems_createEvents_async(client, request,
        &context, &notificationCallback, &deleteCallback,
        _i == 1 ? NULL : asyncEventCreated, &observer, &requestId);
    UA_StatusCode expected = _i == 4 ? UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID :
        (_i == 5 ? UA_STATUSCODE_BADNOTHINGTODO : UA_STATUSCODE_GOOD);
    ck_assert_uint_eq(result, expected);
    ck_assert_uint_eq(observer.createCalls, 0);
    if(_i == 6) {
        /* Drop the local subscription before the create response arrives. Its
         * pending item must be deleted exactly once, using the unknown id 0. */
        lockClient(client);
        __Client_Subscriptions_clear(client);
        unlockClient(client);
        ck_assert_uint_eq(observer.deleteCalls, 1);
        ck_assert_uint_eq(observer.deletedItemId, 0);
    }
    if(result == UA_STATUSCODE_GOOD) {
        awaitAsyncSettings(client);
        if(_i != 1) {
            ck_assert_uint_eq(observer.createCalls, 1);
            ck_assert_uint_eq(observer.requestId, requestId);
            UA_StatusCode serviceResult = _i == 6 ? UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID :
                (_i == 7 ? UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID : UA_STATUSCODE_GOOD);
            ck_assert_uint_eq(observer.response.responseHeader.serviceResult, serviceResult);
            ck_assert_uint_eq(observer.response.resultsSize, _i == 7 ? 0 : 1);
            if(_i != 7) {
                UA_StatusCode itemResult = _i == 2 ? UA_STATUSCODE_BADNODEIDUNKNOWN :
                    (_i == 3 ? UA_STATUSCODE_BADEVENTFILTERINVALID : UA_STATUSCODE_GOOD);
                ck_assert_uint_eq(observer.response.results[0].statusCode, itemResult);
            }
        }
    } else {
        ck_assert_uint_eq(requestId, UA_UINT32_MAX);
        ck_assert_uint_eq(observer.deleteCalls, 0);
    }
    if(_i == 6) {
        ck_assert(LIST_EMPTY(&client->subscriptions));
    } else {
        UA_Client_Subscription *sub = LIST_FIRST(&client->subscriptions);
        ck_assert_ptr_nonnull(sub);
        UA_Boolean created = _i < 2;
        ck_assert_uint_eq(countTestMonitoredItems(ZIP_ROOT(&sub->monitoredItems)), created ? 2 : 1);
        if(created) {
            UA_Client_MonitoredItem *mon = findEventByContext(ZIP_ROOT(&sub->monitoredItems), &observer);
            ck_assert_ptr_nonnull(mon);
            ck_assert(mon->isEventMonitoredItem);
            ck_assert_uint_ne(mon->monitoredItemId, 0);
            observer.monitoredItemId = mon->monitoredItemId;
            if(_i == 0)
                ck_assert_uint_eq(observer.response.results[0].monitoredItemId, mon->monitoredItemId);
            UA_Client_MonitoredItem *dataItem = findTestMonitoredItem(
                ZIP_ROOT(&sub->monitoredItems), owner.monitoredItemId);
            ck_assert_ptr_nonnull(dataItem);
            /* Ignore unknown handles, wrong notification kinds and wrong
             * field counts. A later valid event must still be delivered. */
            injectEventNotification(client, sub, UA_UINT32_MAX, 1);
            injectEventNotification(client, sub, dataItem->parameters.clientHandle, 1);
            injectDataChangeNotification(client, sub, mon->parameters.clientHandle);
            injectEventNotification(client, sub, mon->parameters.clientHandle, 2);
            ck_assert_uint_eq(observer.notifications, 0);
            UA_LocalizedText message = UA_LOCALIZEDTEXT("en", "accepted event");
            UA_Variant value;
            UA_Variant_setScalar(&value, &message, &UA_TYPES[UA_TYPES_LOCALIZEDTEXT]);
            UA_EventFieldList fields;
            UA_EventFieldList_init(&fields);
            fields.clientHandle = mon->parameters.clientHandle;
            fields.eventFields = &value;
            fields.eventFieldsSize = 1;
            UA_EventNotificationList notification;
            UA_EventNotificationList_init(&notification);
            notification.events = &fields;
            notification.eventsSize = 1;
            injectNotification(client, sub, &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST], &notification);
            ck_assert_uint_eq(observer.notifications, 1);
            ck_assert_uint_eq(observer.deleteCalls, 0);
        }
        ck_assert_uint_eq(UA_Client_Subscriptions_deleteSingle(client, owner.subscriptionId),
                          UA_STATUSCODE_GOOD);
    }
    ck_assert_uint_eq(observer.deleteCalls, (_i == 4 || _i == 5) ? 0 : 1);
    ck_assert_uint_eq(observer.deletedItemId, observer.monitoredItemId);
    ck_assert_uint_eq(owner.subscriptionsDeleted, 1);
    ck_assert_uint_eq(owner.itemsDeleted, 1);
    UA_CreateMonitoredItemsResponse_clear(&observer.response);
    ck_assert_uint_eq(UA_Client_disconnect(client), UA_STATUSCODE_GOOD);
    UA_Client_delete(client);
} END_TEST
#endif

static Suite* testSuite_Client(void) {
    Suite *s = suite_create("Client Subscription");

    TCase *tc_client = tcase_create("Client Subscription Basic");
    tcase_add_checked_fixture(tc_client, setup, teardown);
    tcase_add_test(tc_client, Client_subscription);
    tcase_add_test(tc_client, Client_subscription_async);
    tcase_add_test(tc_client,
                   Client_subscription_admission_before_state_lookup);
    tcase_add_test(tc_client, Client_subscription_delete_async_noCallback);
    tcase_add_test(tc_client, Client_subscription_statusChange);
    tcase_add_test(tc_client, Client_subscription_timeout);
    tcase_add_test(tc_client, Client_subscription_detach);
    tcase_add_test(tc_client, Client_subscription_connectionClose);
    tcase_add_test(tc_client, Client_subscription_publishCounterAfterSessionClosed);
    tcase_add_test(tc_client, Client_subscription_createDataChanges);
    tcase_add_test(tc_client, Client_subscription_createDataChanges_negativeInterval);
    tcase_add_test(tc_client, Client_subscription_modifyMonitoredItem);
    tcase_add_test(tc_client, Client_subscription_createDataChanges_async);
    tcase_add_test(tc_client, Client_cancelQueuedPublish);
    tcase_add_test(tc_client, Client_subscription_keepAlive);
    tcase_add_test(tc_client, Client_subscription_priority);
    tcase_add_test(tc_client, Client_subscription_without_notification);
    tcase_add_test(tc_client, Client_subscription_async_sub);
    tcase_add_test(tc_client, Client_subscription_reconnect);
    tcase_add_test(tc_client, Client_subscription_server_disappears);
    tcase_add_test(tc_client, Client_subscription_transfer);
    tcase_add_test(tc_client, Client_subscription_writeBurst);
    tcase_add_test(tc_client, Client_subscription_getSetContext);
    tcase_add_test(tc_client, Client_subscription_deleteSingle);
    tcase_add_test(tc_client, Client_subscription_setPublishingMode);
    tcase_add_test(tc_client, Client_monitoredItems_create_sessionClosed);
    tcase_add_test(tc_client, Client_monitoredItems_modify_sessionClosed);
    tcase_add_test(tc_client, Client_monitoredItem_getSetContext);
    tcase_add_test(tc_client, Client_subscription_setMonitoringMode);
    tcase_add_test(tc_client, Client_subscription_setTriggering);
    tcase_add_test(tc_client, Client_subscription_modifyAsync);
    tcase_add_test(tc_client, Client_subscription_modifyAsync_invalidSubscriptionId);
    tcase_add_test(tc_client, Client_renewSecureChannel_returnsGoodCallAgain);
    suite_add_tcase(s,tc_client);

    TCase *tc_disabledPublishing = tcase_create("Disabled Publishing");
    tcase_add_checked_fixture(tc_disabledPublishing, setup, teardown);
    tcase_add_loop_test(tc_disabledPublishing,
                       Client_subscription_keepAlive_publishingDisabled, 0, 4);
    tcase_add_test(tc_disabledPublishing,
                   Client_subscription_keepAlive_publishingDisabled_late);
    tcase_add_test(tc_disabledPublishing,
                   Client_subscription_keepAlive_publishingDisabled_oneCycle);
    suite_add_tcase(s, tc_disabledPublishing);

    TCase *tc_doubleBuffer = tcase_create("MonitoredItem Double Buffer");
    tcase_add_checked_fixture(tc_doubleBuffer, setup, teardown);
    tcase_add_test(tc_doubleBuffer,
                   Client_subscription_modifyMonitoredItem_doubleBuffer);
    tcase_add_test(tc_doubleBuffer,
                   Client_subscription_modifyEventFilter_doubleBuffer);
    tcase_add_test(tc_doubleBuffer,
                   Client_subscription_modifyMonitoredItem_edgeCases);
    suite_add_tcase(s, tc_doubleBuffer);
    TCase *tc_callbackLifetime = tcase_create("Notification Callback Lifetime");
    tcase_add_checked_fixture(tc_callbackLifetime, setup, teardown);
    tcase_add_test(tc_callbackLifetime,
                   Client_notification_callback_clears_subscription);
    tcase_add_test(tc_callbackLifetime,
                   Client_inactivity_callback_clears_subscription);
    suite_add_tcase(s, tc_callbackLifetime);

    TCase *tc_deleteCallbackLifetime = tcase_create("Delete Callback Lifetime");
    tcase_add_checked_fixture(tc_deleteCallbackLifetime, setup, teardown);
    tcase_add_test(tc_deleteCallbackLifetime,
                   Client_delete_callback_clears_subscriptions);
    tcase_add_test(tc_deleteCallbackLifetime,
                   Client_monitoreditem_delete_callback_clears_subscriptions);
    tcase_add_loop_test(tc_deleteCallbackLifetime,
                        Client_monitoreditem_create_callback_clears_subscriptions, 0, 2);
    suite_add_tcase(s, tc_deleteCallbackLifetime);

    TCase *tc_asyncSettings = tcase_create("Async monitoring settings");
    tcase_add_checked_fixture(tc_asyncSettings, setup, teardown);
    tcase_add_loop_test(tc_asyncSettings, Client_asyncMonitoringModeReportsResults, 0, 4);
    tcase_add_loop_test(tc_asyncSettings, Client_asyncTriggeringReportsResults, 0, 4);
    suite_add_tcase(s, tc_asyncSettings);

    TCase *tc_publish = tcase_create("Publish response lifecycle");
    tcase_add_checked_fixture(tc_publish, setup, teardown);
    tcase_add_loop_test(tc_publish, Client_publishErrorsPreserveOrRemoveSubscription, 0, 9);
    tcase_add_loop_test(tc_publish, Client_publishWrapsSequenceAndClearsAcknowledgements, 0, 4);
    suite_add_tcase(s, tc_publish);

#ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
    TCase *tc_asyncEvents = tcase_create("Async event monitoring");
    tcase_add_checked_fixture(tc_asyncEvents, setup, teardown);
    tcase_add_loop_test(tc_asyncEvents, Client_asyncEventCreationOwnsPendingItems, 0, 8);
    suite_add_tcase(s, tc_asyncEvents);
#endif

    TCase *tc_client2 = tcase_create("Client Subscription + Method Call of GetMonitoredItmes");
    tcase_add_checked_fixture(tc_client2, setup, teardown);
    tcase_add_test(tc_client2, Client_methodcall);
    suite_add_tcase(s,tc_client2);

    return s;
}

int main(void) {
    Suite *s = testSuite_Client();
    SRunner *sr = srunner_create(s);
    srunner_set_fork_status(sr, CK_NOFORK);
    srunner_run_all(sr,CK_NORMAL);
    int number_failed = srunner_ntests_failed(sr);
    srunner_free(sr);
    return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
}