Merge pull request #664 from open62541/test_client_subscriptions

Test client subscriptions

src/client/ua_client.c

@@ -66,15 +66,8 @@ static void UA_Client_deleteMembers(UA_Client* client) {
         free(n);
     }
     UA_Client_Subscription *sub, *tmps;
-    LIST_FOREACH_SAFE(sub, &client->subscriptions, listEntry, tmps) {
-        UA_Client_MonitoredItem *mon, *tmpmon;
-        LIST_FOREACH_SAFE(mon, &sub->MonitoredItems, listEntry, tmpmon) {
-            UA_Client_Subscriptions_removeMonitoredItem(client, sub->SubscriptionID,
-                                                        mon->MonitoredItemId);
-        }
-        LIST_REMOVE(sub, listEntry);
-        free(sub);
-    }
+    LIST_FOREACH_SAFE(sub, &client->subscriptions, listEntry, tmps)
+        UA_Client_Subscriptions_forceDelete(client, sub); /* force local removal */
 #endif
 }
 
@@ -84,8 +77,7 @@ void UA_Client_reset(UA_Client* client){
 }
 
 void UA_Client_delete(UA_Client* client){
-    if(client->state != UA_CLIENTSTATE_READY)
-        UA_Client_deleteMembers(client);
+    UA_Client_deleteMembers(client);
     UA_free(client);
 }
 

src/client/ua_client_highlevel_subscriptions.c

@@ -24,70 +24,85 @@ UA_StatusCode UA_Client_Subscriptions_new(UA_Client *client, UA_SubscriptionSett
     request.maxNotificationsPerPublish = settings.maxNotificationsPerPublish;
     request.publishingEnabled = settings.publishingEnabled;
     request.priority = settings.priority;
-    
+
     UA_CreateSubscriptionResponse response = UA_Client_Service_createSubscription(client, request);
     UA_StatusCode retval = response.responseHeader.serviceResult;
-    if(retval == UA_STATUSCODE_GOOD) {
-        UA_Client_Subscription *newSub = UA_malloc(sizeof(UA_Client_Subscription));
-        LIST_INIT(&newSub->MonitoredItems);
-        newSub->LifeTime = response.revisedLifetimeCount;
-        newSub->KeepAliveCount = response.revisedMaxKeepAliveCount;
-        newSub->PublishingInterval = response.revisedPublishingInterval;
-        newSub->SubscriptionID = response.subscriptionId;
-        newSub->NotificationsPerPublish = request.maxNotificationsPerPublish;
-        newSub->Priority = request.priority;
-        if(newSubscriptionId)
-            *newSubscriptionId = newSub->SubscriptionID;
-        LIST_INSERT_HEAD(&client->subscriptions, newSub, listEntry);
+    if(retval != UA_STATUSCODE_GOOD)
+        goto cleanup;
+
+    UA_Client_Subscription *newSub = UA_malloc(sizeof(UA_Client_Subscription));
+    if(!newSub) {
+        retval = UA_STATUSCODE_BADOUTOFMEMORY;
+        goto cleanup;
     }
-    
+
+    LIST_INIT(&newSub->MonitoredItems);
+    newSub->LifeTime = response.revisedLifetimeCount;
+    newSub->KeepAliveCount = response.revisedMaxKeepAliveCount;
+    newSub->PublishingInterval = response.revisedPublishingInterval;
+    newSub->SubscriptionID = response.subscriptionId;
+    newSub->NotificationsPerPublish = request.maxNotificationsPerPublish;
+    newSub->Priority = request.priority;
+    LIST_INSERT_HEAD(&client->subscriptions, newSub, listEntry);
+
+    if(newSubscriptionId)
+        *newSubscriptionId = newSub->SubscriptionID;
+
+ cleanup:
     UA_CreateSubscriptionResponse_deleteMembers(&response);
     return retval;
 }
 
+/* remove the subscription remotely */
 UA_StatusCode UA_Client_Subscriptions_remove(UA_Client *client, UA_UInt32 subscriptionId) {
     UA_Client_Subscription *sub;
-    UA_StatusCode retval = UA_STATUSCODE_GOOD;
-    
     LIST_FOREACH(sub, &client->subscriptions, listEntry) {
         if(sub->SubscriptionID == subscriptionId)
             break;
     }
-    
-    // Problem? We do not have this subscription registeres. Maybe the server should
-    // be consulted at this point?
     if(!sub)
         return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
-    
-    UA_DeleteSubscriptionsRequest request;
-    UA_DeleteSubscriptionsRequest_init(&request);
-    request.subscriptionIdsSize = 1;
-    request.subscriptionIds = (UA_UInt32 *) UA_malloc(sizeof(UA_UInt32));
-    *request.subscriptionIds = sub->SubscriptionID;
-    
+
+    UA_StatusCode retval = UA_STATUSCODE_GOOD;
     UA_Client_MonitoredItem *mon, *tmpmon;
     LIST_FOREACH_SAFE(mon, &sub->MonitoredItems, listEntry, tmpmon) {
-        retval |= UA_Client_Subscriptions_removeMonitoredItem(client, sub->SubscriptionID,
-                                                              mon->MonitoredItemId);
-    }
-    if(retval != UA_STATUSCODE_GOOD) {
-        UA_DeleteSubscriptionsRequest_deleteMembers(&request);
-        return retval;
+        retval = UA_Client_Subscriptions_removeMonitoredItem(client, sub->SubscriptionID,
+                                                             mon->MonitoredItemId);
+        if(retval != UA_STATUSCODE_GOOD)
+            return retval;
     }
-    
+
+    /* remove the subscription remotely */
+    UA_DeleteSubscriptionsRequest request;
+    UA_DeleteSubscriptionsRequest_init(&request);
+    request.subscriptionIdsSize = 1;
+    request.subscriptionIds = &sub->SubscriptionID;
     UA_DeleteSubscriptionsResponse response = UA_Client_Service_deleteSubscriptions(client, request);
-    if(response.resultsSize > 0)
+    retval = response.responseHeader.serviceResult;
+    if(retval == UA_STATUSCODE_GOOD && response.resultsSize > 0)
         retval = response.results[0];
-    else
-        retval = response.responseHeader.serviceResult;
-    
-    if(retval == UA_STATUSCODE_GOOD) {
-        LIST_REMOVE(sub, listEntry);
-        UA_free(sub);
-    }
-    UA_DeleteSubscriptionsRequest_deleteMembers(&request);
     UA_DeleteSubscriptionsResponse_deleteMembers(&response);
-    return retval;
+
+    if(retval != UA_STATUSCODE_GOOD && retval != UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID) {
+        UA_LOG_INFO(client->config.logger, UA_LOGCATEGORY_CLIENT,
+                    "Could not remove subscription %u with statuscode 0x%08x",
+                    sub->SubscriptionID, retval);
+        return retval;
+    }
+
+    UA_Client_Subscriptions_forceDelete(client, sub);
+    return UA_STATUSCODE_GOOD;
+}
+
+void UA_Client_Subscriptions_forceDelete(UA_Client *client, UA_Client_Subscription *sub) {
+    UA_Client_MonitoredItem *mon, *mon_tmp;
+    LIST_FOREACH_SAFE(mon, &sub->MonitoredItems, listEntry, mon_tmp) {
+        UA_NodeId_deleteMembers(&mon->monitoredNodeId);
+        LIST_REMOVE(mon, listEntry);
+        UA_free(mon);
+    }
+    LIST_REMOVE(sub, listEntry);
+    UA_free(sub);
 }
 
 UA_StatusCode
@@ -134,7 +149,7 @@ UA_Client_Subscriptions_addMonitoredItem(UA_Client *client, UA_UInt32 subscripti
     /* Create the handler */
     UA_Client_MonitoredItem *newMon = UA_malloc(sizeof(UA_Client_MonitoredItem));
     newMon->MonitoringMode = UA_MONITORINGMODE_REPORTING;
-    UA_NodeId_copy(&nodeId, &newMon->monitoredNodeId); 
+    UA_NodeId_copy(&nodeId, &newMon->monitoredNodeId);
     newMon->AttributeID = attributeID;
     newMon->ClientHandle = client->monitoredItemHandles;
     newMon->SamplingInterval = sub->PublishingInterval;
@@ -148,7 +163,7 @@ UA_Client_Subscriptions_addMonitoredItem(UA_Client *client, UA_UInt32 subscripti
 
     UA_LOG_DEBUG(client->config.logger, UA_LOGCATEGORY_CLIENT,
                  "Created a monitored item with client handle %u", client->monitoredItemHandles);
-    
+
     UA_CreateMonitoredItemsResponse_deleteMembers(&response);
     return UA_STATUSCODE_GOOD;
 }
@@ -172,34 +187,29 @@ UA_Client_Subscriptions_removeMonitoredItem(UA_Client *client, UA_UInt32 subscri
     if(!mon)
         return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
 
-    UA_StatusCode retval = UA_STATUSCODE_GOOD;
-
-    if(client->state == UA_CLIENTSTATE_CONNECTED) {
-        UA_DeleteMonitoredItemsRequest request;
-        UA_DeleteMonitoredItemsRequest_init(&request);
-        request.subscriptionId = sub->SubscriptionID;
-        request.monitoredItemIdsSize = 1;
-        request.monitoredItemIds = (UA_UInt32 *) UA_malloc(sizeof(UA_UInt32));
-        request.monitoredItemIds[0] = mon->MonitoredItemId;
+    /* remove the monitoreditem remotely */
+    UA_DeleteMonitoredItemsRequest request;
+    UA_DeleteMonitoredItemsRequest_init(&request);
+    request.subscriptionId = sub->SubscriptionID;
+    request.monitoredItemIdsSize = 1;
+    request.monitoredItemIds = &mon->MonitoredItemId;
+    UA_DeleteMonitoredItemsResponse response = UA_Client_Service_deleteMonitoredItems(client, request);
 
-        UA_DeleteMonitoredItemsResponse response = UA_Client_Service_deleteMonitoredItems(client, request);
-
-        if(response.resultsSize > 1)
-            retval = response.results[0];
-        else
-            retval = response.responseHeader.serviceResult;
-
-        UA_DeleteMonitoredItemsRequest_deleteMembers(&request);
-        UA_DeleteMonitoredItemsResponse_deleteMembers(&response);
-    }
-    
-    if(retval == UA_STATUSCODE_GOOD) {
-        LIST_REMOVE(mon, listEntry);
-        UA_NodeId_deleteMembers(&mon->monitoredNodeId);
-        UA_free(mon);
+    UA_StatusCode retval = response.responseHeader.serviceResult;
+    if(retval == UA_STATUSCODE_GOOD && response.resultsSize > 1)
+        retval = response.results[0];
+    UA_DeleteMonitoredItemsResponse_deleteMembers(&response);
+    if(retval != UA_STATUSCODE_GOOD && retval != UA_STATUSCODE_BADMONITOREDITEMIDINVALID) {
+        UA_LOG_INFO(client->config.logger, UA_LOGCATEGORY_CLIENT,
+                    "Could not remove monitoreditem %u with statuscode 0x%08x",
+                    monitoredItemId, retval);
+        return retval;
     }
-    
-    return retval;
+
+    LIST_REMOVE(mon, listEntry);
+    UA_NodeId_deleteMembers(&mon->monitoredNodeId);
+    UA_free(mon);
+    return UA_STATUSCODE_GOOD;
 }
 
 static void

src/client/ua_client_internal.h

@@ -16,27 +16,27 @@ typedef struct UA_Client_NotificationsAckNumber_s {
 } UA_Client_NotificationsAckNumber;
 
 typedef struct UA_Client_MonitoredItem_s {
-    UA_UInt32  MonitoredItemId;
-    UA_UInt32  MonitoringMode;
-    UA_NodeId  monitoredNodeId;
-    UA_UInt32  AttributeID;
-    UA_UInt32  ClientHandle;
-    UA_Double  SamplingInterval;
-    UA_UInt32  QueueSize;
-    UA_Boolean DiscardOldest;
-    void       (*handler)(UA_UInt32 monId, UA_DataValue *value, void *context);
-    void       *handlerContext;
     LIST_ENTRY(UA_Client_MonitoredItem_s)  listEntry;
+    UA_UInt32 MonitoredItemId;
+    UA_UInt32 MonitoringMode;
+    UA_NodeId monitoredNodeId;
+    UA_UInt32 AttributeID;
+    UA_UInt32 ClientHandle;
+    UA_Double SamplingInterval;
+    UA_UInt32 QueueSize;
+    UA_Boolean DiscardOldest;
+    void (*handler)(UA_UInt32 monId, UA_DataValue *value, void *context);
+    void *handlerContext;
 } UA_Client_MonitoredItem;
 
 typedef struct UA_Client_Subscription_s {
+    LIST_ENTRY(UA_Client_Subscription_s) listEntry;
     UA_UInt32 LifeTime;
     UA_UInt32 KeepAliveCount;
     UA_Double PublishingInterval;
     UA_UInt32 SubscriptionID;
     UA_UInt32 NotificationsPerPublish;
     UA_UInt32 Priority;
-    LIST_ENTRY(UA_Client_Subscription_s) listEntry;
     LIST_HEAD(UA_ListOfClientMonitoredItems, UA_Client_MonitoredItem_s) MonitoredItems;
 } UA_Client_Subscription;
 
@@ -82,4 +82,6 @@ struct UA_Client {
     UA_DateTime scRenewAt;
 };
 
+void UA_Client_Subscriptions_forceDelete(UA_Client *client, UA_Client_Subscription *sub);
+
 #endif /* UA_CLIENT_INTERNAL_H_ */

src/server/ua_server_worker.c

@@ -200,7 +200,7 @@ static UA_StatusCode addRepeatedJob(UA_Server *server, struct AddRepeatedJob * U
     UA_StatusCode retval = UA_STATUSCODE_GOOD;
 
     /* search for matching entry */
-    UA_DateTime firstTime = UA_DateTime_nowMonotonic();
+    UA_DateTime firstTime = UA_DateTime_nowMonotonic() + arw->interval;
     tempTw = LIST_FIRST(&server->repeatedJobs);
     while(tempTw) {
         if(arw->interval == tempTw->interval) {
@@ -310,18 +310,18 @@ static UA_DateTime processRepeatedJobs(UA_Server *server, UA_DateTime current) {
             jobsCopy[i] = tw->jobs[i].job;
         dispatchJobs(server, jobsCopy, tw->jobsSize); // frees the job pointer
 #else
-		size_t size = tw->jobsSize;
+        size_t size = tw->jobsSize;
         for(size_t i = 0; i < size; i++)
             processJobs(server, &tw->jobs[i].job, 1); // does not free the job ptr
 #endif
 
-		/* Elements are removed only here. Check if empty. */
-		if(tw->jobsSize == 0) {
-			LIST_REMOVE(tw, pointers);
-			UA_free(tw);
+        /* Elements are removed only here. Check if empty. */
+        if(tw->jobsSize == 0) {
+            LIST_REMOVE(tw, pointers);
+            UA_free(tw);
             UA_assert(LIST_FIRST(&server->repeatedJobs) != tw); /* Assert for static code checkers */
-			continue;
-		}
+            continue;
+        }
 
         /* Set the time for the next execution */
         tw->nextTime += tw->interval;
@@ -358,7 +358,7 @@ static void removeRepeatedJob(UA_Server *server, UA_Guid *jobId) {
         for(size_t i = 0; i < tw->jobsSize; i++) {
             if(!UA_Guid_equal(jobId, &tw->jobs[i].id))
                 continue;
-			tw->jobsSize--; /* if size == 0, tw is freed during the next processing */
+            tw->jobsSize--; /* if size == 0, tw is freed during the next processing */
             if(tw->jobsSize > 0)
                 tw->jobs[i] = tw->jobs[tw->jobsSize]; // move the last entry to overwrite
             goto finish;

src/server/ua_services_subscription.c

@@ -189,6 +189,9 @@ Service_CreateMonitoredItems_single(UA_Server *server, UA_Session *session, UA_S
                              request->requestedParameters.discardOldest);
     LIST_INSERT_HEAD(&sub->MonitoredItems, newMon, listEntry);
 
+    /* Create the first sample */
+    UA_MoniteredItem_SampleCallback(server, newMon);
+
     /* Prepare the response */
     UA_String_copy(&request->itemToMonitor.indexRange, &newMon->indexRange);
     result->revisedSamplingInterval = newMon->samplingInterval;

src/server/ua_subscription.c

@@ -43,7 +43,7 @@ void MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem) {
     UA_free(monitoredItem);
 }
 
-static void SampleCallback(UA_Server *server, UA_MonitoredItem *monitoredItem) {
+void UA_MoniteredItem_SampleCallback(UA_Server *server, UA_MonitoredItem *monitoredItem) {
     UA_Subscription *sub = monitoredItem->subscription;
     if(monitoredItem->monitoredItemType != UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
         UA_LOG_DEBUG_SESSION(server->config.logger, sub->session, "MonitoredItem %i | "
@@ -125,7 +125,7 @@ static void SampleCallback(UA_Server *server, UA_MonitoredItem *monitoredItem) {
 
 UA_StatusCode MonitoredItem_registerSampleJob(UA_Server *server, UA_MonitoredItem *mon) {
     UA_Job job = {.type = UA_JOBTYPE_METHODCALL,
-                  .job.methodCall = {.method = (UA_ServerCallback)SampleCallback, .data = mon} };
+                  .job.methodCall = {.method = (UA_ServerCallback)UA_MoniteredItem_SampleCallback, .data = mon} };
     UA_StatusCode retval = UA_Server_addRepeatedJob(server, job, (UA_UInt32)mon->samplingInterval,
                                                     &mon->sampleJobGuid);
     if(retval == UA_STATUSCODE_GOOD)

src/server/ua_subscription.h

@@ -53,6 +53,7 @@ typedef struct UA_MonitoredItem {
 
 UA_MonitoredItem *UA_MonitoredItem_new(void);
 void MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem);
+void UA_MoniteredItem_SampleCallback(UA_Server *server, UA_MonitoredItem *monitoredItem);
 UA_StatusCode MonitoredItem_registerSampleJob(UA_Server *server, UA_MonitoredItem *mon);
 UA_StatusCode MonitoredItem_unregisterSampleJob(UA_Server *server, UA_MonitoredItem *mon);
 

tests/CMakeLists.txt

@@ -113,3 +113,7 @@ add_test(check_server_binary_messages_write ${CMAKE_CURRENT_BINARY_DIR}/check_se
 add_executable(check_client check_client.c $<TARGET_OBJECTS:open62541-object>)
 target_link_libraries(check_client ${LIBS})
 add_test(check_client ${CMAKE_CURRENT_BINARY_DIR}/check_client)
+
+add_executable(check_client_subscriptions check_client_subscriptions.c $<TARGET_OBJECTS:open62541-object>)
+target_link_libraries(check_client_subscriptions ${LIBS})
+add_test(check_client_subscriptions ${CMAKE_CURRENT_BINARY_DIR}/check_client_subscriptions)

tests/check_client.c

@@ -64,6 +64,7 @@ static Suite* testSuite_Client(void) {
 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);

tests/check_client_subscriptions.c

@@ -0,0 +1,93 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <pthread.h>
+
+#include "ua_types.h"
+#include "ua_server.h"
+#include "ua_client.h"
+#include "ua_client_highlevel.h"
+#include "ua_config_standard.h"
+#include "ua_network_tcp.h"
+#include "check.h"
+
+UA_Server *server;
+UA_Boolean *running;
+UA_ServerNetworkLayer nl;
+pthread_t server_thread;
+
+static void * serverloop(void *_) {
+    while(*running)
+        UA_Server_run_iterate(server, true);
+    return NULL;
+}
+
+static void setup(void) {
+    running = UA_Boolean_new();
+    *running = true;
+    UA_ServerConfig config = UA_ServerConfig_standard;
+    nl = UA_ServerNetworkLayerTCP(UA_ConnectionConfig_standard, 16664);
+    config.networkLayers = &nl;
+    config.networkLayersSize = 1;
+    server = UA_Server_new(config);
+    UA_Server_run_startup(server);
+    pthread_create(&server_thread, NULL, serverloop, NULL);
+}
+
+static void teardown(void) {
+    *running = false;
+    pthread_join(server_thread, NULL);
+    UA_Server_run_shutdown(server);
+    UA_Boolean_delete(running);
+    UA_Server_delete(server);
+    nl.deleteMembers(&nl);
+}
+
+UA_Boolean notificationReceived;
+
+static void monitoredItemHandler(UA_UInt32 monId, UA_DataValue *value, void *context) {
+    notificationReceived = true;
+}
+
+START_TEST(Client_subscription) {
+    UA_Client *client = UA_Client_new(UA_ClientConfig_standard);
+    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:16664");
+    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
+
+    UA_UInt32 subId;
+    retval = UA_Client_Subscriptions_new(client, UA_SubscriptionSettings_standard, &subId);
+    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
+
+    /* monitor the server state */
+    UA_UInt32 monId;
+    retval = UA_Client_Subscriptions_addMonitoredItem(client, subId, UA_NODEID_NUMERIC(0, 2259),
+                                                      UA_ATTRIBUTEID_VALUE, monitoredItemHandler, NULL, &monId);
+    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
+
+    notificationReceived = false;
+    retval = UA_Client_Subscriptions_manuallySendPublishRequest(client);
+    ck_assert_uint_eq(retval, UA_STATUSCODE_GOOD);
+    ck_assert_uint_eq(notificationReceived, true);
+
+    UA_Client_disconnect(client);
+    UA_Client_delete(client);
+}
+END_TEST
+
+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);
+    suite_add_tcase(s,tc_client);
+    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;
+}