/* 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 #include #include #include #include #include "client/ua_client_internal.h" #include "server/ua_server_internal.h" #include "test_helpers.h" #include #include #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 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(¬ificationData); 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 = ¬ificationData; 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_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, ¬ifications, &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(¬ification); 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 = ¬ification; 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, ¬ificationCallback, &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(¬ification); notification.events = &fields; notification.eventsSize = 1; injectNotification(client, sub, &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST], ¬ification); 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_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; }