ua_subscription.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552
  1. #include "ua_subscription.h"
  2. #include "ua_server_internal.h"
  3. #include "ua_nodestore.h"
  4. /****************/
  5. /* Subscription */
  6. /****************/
  7. UA_Subscription *UA_Subscription_new(UA_Int32 subscriptionID) {
  8. UA_Subscription *new = UA_malloc(sizeof(UA_Subscription));
  9. if(!new)
  10. return UA_NULL;
  11. new->subscriptionID = subscriptionID;
  12. new->lastPublished = 0;
  13. new->sequenceNumber = 1;
  14. memset(&new->timedUpdateJobGuid, 0, sizeof(UA_Guid));
  15. new->timedUpdateJob = UA_NULL;
  16. new->timedUpdateIsRegistered = UA_FALSE;
  17. LIST_INIT(&new->MonitoredItems);
  18. LIST_INIT(&new->unpublishedNotifications);
  19. return new;
  20. }
  21. void UA_Subscription_deleteMembers(UA_Subscription *subscription, UA_Server *server) {
  22. UA_unpublishedNotification *not, *tmp_not;
  23. UA_MonitoredItem *mon, *tmp_mon;
  24. // Just in case any parallel process attempts to access this subscription
  25. // while we are deleting it... make it vanish.
  26. subscription->subscriptionID = 0;
  27. // Delete monitored Items
  28. LIST_FOREACH_SAFE(mon, &subscription->MonitoredItems, listEntry, tmp_mon) {
  29. LIST_REMOVE(mon, listEntry);
  30. MonitoredItem_delete(mon);
  31. }
  32. // Delete unpublished Notifications
  33. LIST_FOREACH_SAFE(not, &subscription->unpublishedNotifications, listEntry, tmp_not) {
  34. Subscription_deleteUnpublishedNotification(not->notification->sequenceNumber, subscription);
  35. }
  36. // Unhook/Unregister any timed work assiociated with this subscription
  37. if(subscription->timedUpdateJob != UA_NULL){
  38. Subscription_unregisterUpdateJob(server, subscription);
  39. UA_free(subscription->timedUpdateJob);
  40. }
  41. }
  42. UA_UInt32 Subscription_queuedNotifications(UA_Subscription *subscription) {
  43. if(!subscription)
  44. return 0;
  45. UA_UInt32 j = 0;
  46. UA_unpublishedNotification *i;
  47. LIST_FOREACH(i, &subscription->unpublishedNotifications, listEntry)
  48. j++;
  49. return j;
  50. }
  51. void Subscription_generateKeepAlive(UA_Subscription *subscription) {
  52. if(subscription->keepAliveCount.currentValue > subscription->keepAliveCount.minValue &&
  53. subscription->keepAliveCount.currentValue <= subscription->keepAliveCount.maxValue)
  54. return;
  55. UA_unpublishedNotification *msg = UA_malloc(sizeof(UA_unpublishedNotification));
  56. if(!msg)
  57. return;
  58. msg->notification = UA_NULL;
  59. msg->notification = UA_malloc(sizeof(UA_NotificationMessage));
  60. msg->notification->notificationData = UA_NULL;
  61. // KeepAlive uses next message, but does not increment counter
  62. msg->notification->sequenceNumber = subscription->sequenceNumber + 1;
  63. msg->notification->publishTime = UA_DateTime_now();
  64. msg->notification->notificationDataSize = 0;
  65. LIST_INSERT_HEAD(&subscription->unpublishedNotifications, msg, listEntry);
  66. subscription->keepAliveCount.currentValue = subscription->keepAliveCount.maxValue;
  67. }
  68. void Subscription_updateNotifications(UA_Subscription *subscription) {
  69. UA_MonitoredItem *mon;
  70. //MonitoredItem_queuedValue *queuedValue;
  71. UA_unpublishedNotification *msg = NULL, *tempmsg;
  72. UA_UInt32 monItemsChangeT = 0, monItemsStatusT = 0, monItemsEventT = 0;
  73. UA_DataChangeNotification *changeNotification;
  74. size_t notificationOffset;
  75. if(!subscription || subscription->lastPublished + subscription->publishingInterval > UA_DateTime_now())
  76. return;
  77. // Make sure there is data to be published and establish which message types
  78. // will need to be generated
  79. LIST_FOREACH(mon, &subscription->MonitoredItems, listEntry) {
  80. // Check if this MonitoredItems Queue holds data and how much data is held in total
  81. if(!TAILQ_FIRST(&mon->queue))
  82. continue;
  83. if((mon->monitoredItemType & MONITOREDITEM_TYPE_CHANGENOTIFY) != 0)
  84. monItemsChangeT+=mon->queueSize.currentValue;
  85. else if((mon->monitoredItemType & MONITOREDITEM_TYPE_STATUSNOTIFY) != 0)
  86. monItemsStatusT+=mon->queueSize.currentValue;
  87. else if((mon->monitoredItemType & MONITOREDITEM_TYPE_EVENTNOTIFY) != 0)
  88. monItemsEventT+=mon->queueSize.currentValue;
  89. }
  90. // FIXME: This is hardcoded to 100 because it is not covered by the spec but we need to protect the server!
  91. if(Subscription_queuedNotifications(subscription) >= 10) {
  92. // Remove last entry
  93. LIST_FOREACH_SAFE(msg, &subscription->unpublishedNotifications, listEntry, tempmsg)
  94. Subscription_deleteUnpublishedNotification(msg->notification->sequenceNumber, subscription);
  95. }
  96. if(monItemsChangeT == 0 && monItemsEventT == 0 && monItemsStatusT == 0) {
  97. // Decrement KeepAlive
  98. subscription->keepAliveCount.currentValue--;
  99. // +- Generate KeepAlive msg if counter overruns
  100. if (subscription->keepAliveCount.currentValue < subscription->keepAliveCount.minValue)
  101. Subscription_generateKeepAlive(subscription);
  102. return;
  103. }
  104. msg = (UA_unpublishedNotification *) UA_malloc(sizeof(UA_unpublishedNotification));
  105. msg->notification = UA_malloc(sizeof(UA_NotificationMessage));
  106. INITPOINTER(msg->notification->notificationData);
  107. msg->notification->sequenceNumber = subscription->sequenceNumber++;
  108. msg->notification->publishTime = UA_DateTime_now();
  109. // NotificationData is an array of Change, Status and Event messages, each containing the appropriate
  110. // list of Queued values from all monitoredItems of that type
  111. msg->notification->notificationDataSize = ISNOTZERO(monItemsChangeT);
  112. // + ISNOTZERO(monItemsEventT) + ISNOTZERO(monItemsStatusT);
  113. msg->notification->notificationData = UA_Array_new(&UA_TYPES[UA_TYPES_EXTENSIONOBJECT],
  114. msg->notification->notificationDataSize);
  115. for(int notmsgn=0; notmsgn < msg->notification->notificationDataSize; notmsgn++) {
  116. // Set the notification message type and encoding for each of
  117. // the three possible NotificationData Types
  118. msg->notification->notificationData[notmsgn].encoding = 1; // Encoding is always binary
  119. msg->notification->notificationData[notmsgn].typeId = UA_NODEID_NUMERIC(0, 811);
  120. if(notmsgn == 0) {
  121. // Construct a DataChangeNotification
  122. changeNotification = UA_malloc(sizeof(UA_DataChangeNotification));
  123. // Create one DataChangeNotification for each queue item held in each monitoredItems queue:
  124. changeNotification->monitoredItems = UA_Array_new(&UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION],
  125. monItemsChangeT);
  126. // Scan all monitoredItems in this subscription and have their queue transformed into an Array of
  127. // the propper NotificationMessageType (Status, Change, Event)
  128. monItemsChangeT = 0;
  129. LIST_FOREACH(mon, &subscription->MonitoredItems, listEntry) {
  130. if(mon->monitoredItemType != MONITOREDITEM_TYPE_CHANGENOTIFY || !TAILQ_FIRST(&mon->queue))
  131. continue;
  132. // Note: Monitored Items might not return a queuedValue if there is a problem encoding it.
  133. monItemsChangeT += MonitoredItem_QueueToDataChangeNotifications(&changeNotification->monitoredItems[monItemsChangeT], mon);
  134. MonitoredItem_ClearQueue(mon);
  135. }
  136. changeNotification->monitoredItemsSize = monItemsChangeT;
  137. changeNotification->diagnosticInfosSize = 0;
  138. changeNotification->diagnosticInfos = UA_NULL;
  139. msg->notification->notificationData[notmsgn].body.length =
  140. UA_calcSizeBinary(changeNotification, &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION]);
  141. msg->notification->notificationData[notmsgn].body.data =
  142. UA_calloc(msg->notification->notificationData[notmsgn].body.length, sizeof(UA_Byte));
  143. notificationOffset = 0;
  144. UA_encodeBinary(changeNotification, &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION],
  145. &msg->notification->notificationData[notmsgn].body, &notificationOffset);
  146. // FIXME: Not properly freed!
  147. for(unsigned int i=0; i<monItemsChangeT; i++) {
  148. UA_MonitoredItemNotification *thisNotification = &(changeNotification->monitoredItems[i]);
  149. UA_DataValue_deleteMembers(&(thisNotification->value));
  150. }
  151. UA_free(changeNotification->monitoredItems);
  152. UA_free(changeNotification);
  153. } else if(notmsgn == 1) {
  154. // FIXME: Constructing a StatusChangeNotification is not implemented
  155. } else if(notmsgn == 2) {
  156. // FIXME: Constructing a EventListNotification is not implemented
  157. }
  158. }
  159. LIST_INSERT_HEAD(&subscription->unpublishedNotifications, msg, listEntry);
  160. }
  161. UA_UInt32 *Subscription_getAvailableSequenceNumbers(UA_Subscription *sub) {
  162. UA_UInt32 *seqArray = UA_malloc(sizeof(UA_UInt32) * Subscription_queuedNotifications(sub));
  163. if(!seqArray)
  164. return UA_NULL;
  165. int i = 0;
  166. UA_unpublishedNotification *not;
  167. LIST_FOREACH(not, &sub->unpublishedNotifications, listEntry) {
  168. seqArray[i] = not->notification->sequenceNumber;
  169. i++;
  170. }
  171. return seqArray;
  172. }
  173. void Subscription_copyTopNotificationMessage(UA_NotificationMessage *dst, UA_Subscription *sub) {
  174. if(!dst)
  175. return;
  176. if(Subscription_queuedNotifications(sub) == 0) {
  177. dst->notificationDataSize = 0;
  178. dst->publishTime = UA_DateTime_now();
  179. dst->sequenceNumber = 0;
  180. return;
  181. }
  182. UA_NotificationMessage *latest = LIST_FIRST(&sub->unpublishedNotifications)->notification;
  183. dst->notificationDataSize = latest->notificationDataSize;
  184. dst->publishTime = latest->publishTime;
  185. dst->sequenceNumber = latest->sequenceNumber;
  186. if(latest->notificationDataSize == 0)
  187. return;
  188. dst->notificationData = (UA_ExtensionObject *) UA_malloc(sizeof(UA_ExtensionObject));
  189. dst->notificationData->encoding = latest->notificationData->encoding;
  190. dst->notificationData->typeId = latest->notificationData->typeId;
  191. UA_ByteString_copy(&latest->notificationData->body,
  192. &dst->notificationData->body);
  193. }
  194. UA_UInt32 Subscription_deleteUnpublishedNotification(UA_UInt32 seqNo, UA_Subscription *sub) {
  195. UA_UInt32 deletedItems = 0;
  196. UA_unpublishedNotification *not, *tmp;
  197. LIST_FOREACH_SAFE(not, &sub->unpublishedNotifications, listEntry, tmp) {
  198. if(not->notification->sequenceNumber != seqNo)
  199. continue;
  200. LIST_REMOVE(not, listEntry);
  201. if(not->notification) {
  202. if(not->notification->notificationData) {
  203. if(not->notification->notificationData->body.data)
  204. UA_free(not->notification->notificationData->body.data);
  205. UA_free(not->notification->notificationData);
  206. }
  207. UA_free(not->notification);
  208. }
  209. UA_free(not);
  210. deletedItems++;
  211. }
  212. return deletedItems;
  213. }
  214. static void Subscription_timedUpdateNotificationsJob(UA_Server *server, void *data) {
  215. // Timed-Worker/Job Version of updateNotifications
  216. UA_Subscription *sub = (UA_Subscription *) data;
  217. UA_MonitoredItem *mon;
  218. if(!data || !server)
  219. return;
  220. // This is set by the Subscription_delete function to detere us from fiddling with
  221. // this subscription if it is being deleted (not technically thread save, but better
  222. // then nothing at all)
  223. if(sub->subscriptionID == 0)
  224. return;
  225. // FIXME: This should be done by the event system
  226. LIST_FOREACH(mon, &sub->MonitoredItems, listEntry)
  227. MonitoredItem_QueuePushDataValue(server, mon);
  228. Subscription_updateNotifications(sub);
  229. }
  230. UA_StatusCode Subscription_createdUpdateJob(UA_Server *server, UA_Guid jobId, UA_Subscription *sub) {
  231. if(server == UA_NULL || sub == UA_NULL)
  232. return UA_STATUSCODE_BADSERVERINDEXINVALID;
  233. UA_Job *theWork;
  234. theWork = (UA_Job *) UA_malloc(sizeof(UA_Job));
  235. if(!theWork)
  236. return UA_STATUSCODE_BADOUTOFMEMORY;
  237. *theWork = (UA_Job) {.type = UA_JOBTYPE_METHODCALL,
  238. .job.methodCall = {.method = Subscription_timedUpdateNotificationsJob, .data = sub} };
  239. sub->timedUpdateJobGuid = jobId;
  240. sub->timedUpdateJob = theWork;
  241. return UA_STATUSCODE_GOOD;
  242. }
  243. UA_StatusCode Subscription_registerUpdateJob(UA_Server *server, UA_Subscription *sub) {
  244. if(server == UA_NULL || sub == UA_NULL)
  245. return UA_STATUSCODE_BADSERVERINDEXINVALID;
  246. if(sub->publishingInterval <= 5 )
  247. return UA_STATUSCODE_BADNOTSUPPORTED;
  248. // Practically enough, the client sends a uint32 in ms, which we store as datetime, which here is required in as uint32 in ms as the interval
  249. UA_StatusCode retval = UA_Server_addRepeatedJob(server, *sub->timedUpdateJob, sub->publishingInterval,
  250. &sub->timedUpdateJobGuid);
  251. if(!retval)
  252. sub->timedUpdateIsRegistered = UA_TRUE;
  253. return retval;
  254. }
  255. UA_StatusCode Subscription_unregisterUpdateJob(UA_Server *server, UA_Subscription *sub) {
  256. if(server == UA_NULL || sub == UA_NULL)
  257. return UA_STATUSCODE_BADSERVERINDEXINVALID;
  258. UA_Int32 retval = UA_Server_removeRepeatedJob(server, sub->timedUpdateJobGuid);
  259. sub->timedUpdateIsRegistered = UA_FALSE;
  260. return retval;
  261. }
  262. /*****************/
  263. /* MonitoredItem */
  264. /*****************/
  265. UA_MonitoredItem *UA_MonitoredItem_new() {
  266. UA_MonitoredItem *new = (UA_MonitoredItem *) UA_malloc(sizeof(UA_MonitoredItem));
  267. new->queueSize = (UA_UInt32_BoundedValue) { .minValue = 0, .maxValue = 0, .currentValue = 0};
  268. new->lastSampled = 0;
  269. // FIXME: This is currently hardcoded;
  270. new->monitoredItemType = MONITOREDITEM_TYPE_CHANGENOTIFY;
  271. TAILQ_INIT(&new->queue);
  272. UA_NodeId_init(&new->monitoredNodeId);
  273. INITPOINTER(new->lastSampledValue.data );
  274. return new;
  275. }
  276. void MonitoredItem_delete(UA_MonitoredItem *monitoredItem) {
  277. // Delete Queued Data
  278. MonitoredItem_ClearQueue(monitoredItem);
  279. // Remove from subscription list
  280. LIST_REMOVE(monitoredItem, listEntry);
  281. // Release comparison sample
  282. if(monitoredItem->lastSampledValue.data != NULL) {
  283. UA_free(monitoredItem->lastSampledValue.data);
  284. }
  285. UA_NodeId_deleteMembers(&(monitoredItem->monitoredNodeId));
  286. UA_free(monitoredItem);
  287. }
  288. int MonitoredItem_QueueToDataChangeNotifications(UA_MonitoredItemNotification *dst,
  289. UA_MonitoredItem *monitoredItem) {
  290. int queueSize = 0;
  291. MonitoredItem_queuedValue *queueItem;
  292. // Count instead of relying on the items currentValue
  293. TAILQ_FOREACH(queueItem, &monitoredItem->queue, listEntry) {
  294. dst[queueSize].clientHandle = monitoredItem->clientHandle;
  295. dst[queueSize].value.hasServerPicoseconds = UA_FALSE;
  296. dst[queueSize].value.hasServerTimestamp = UA_TRUE;
  297. dst[queueSize].value.serverTimestamp = UA_DateTime_now();
  298. dst[queueSize].value.hasSourcePicoseconds = UA_FALSE;
  299. dst[queueSize].value.hasSourceTimestamp = UA_TRUE;
  300. dst[queueSize].value.sourceTimestamp = UA_DateTime_now();
  301. dst[queueSize].value.hasValue = UA_TRUE;
  302. dst[queueSize].value.status = UA_STATUSCODE_GOOD;
  303. UA_Variant_copy(&(queueItem->value), &(dst[queueSize].value.value));
  304. // Do not create variants with no type -> will make calcSizeBinary() segfault.
  305. if(dst[queueSize].value.value.type)
  306. queueSize++;
  307. }
  308. return queueSize;
  309. }
  310. void MonitoredItem_ClearQueue(UA_MonitoredItem *monitoredItem) {
  311. MonitoredItem_queuedValue *val, *val_tmp;
  312. TAILQ_FOREACH_SAFE(val, &monitoredItem->queue, listEntry, val_tmp) {
  313. TAILQ_REMOVE(&monitoredItem->queue, val, listEntry);
  314. UA_Variant_deleteMembers(&val->value);
  315. UA_free(val);
  316. }
  317. monitoredItem->queueSize.currentValue = 0;
  318. }
  319. UA_Boolean MonitoredItem_CopyMonitoredValueToVariant(UA_UInt32 attributeID, const UA_Node *src,
  320. UA_Variant *dst) {
  321. UA_Boolean samplingError = UA_TRUE;
  322. UA_DataValue sourceDataValue;
  323. UA_DataValue_init(&sourceDataValue);
  324. const UA_VariableNode *srcAsVariableNode = (const UA_VariableNode*)src;
  325. // FIXME: Not all attributeIDs can be monitored yet
  326. switch(attributeID) {
  327. case UA_ATTRIBUTEID_NODEID:
  328. UA_Variant_setScalarCopy(dst, (const UA_NodeId*)&src->nodeId, &UA_TYPES[UA_TYPES_NODEID]);
  329. samplingError = UA_FALSE;
  330. break;
  331. case UA_ATTRIBUTEID_NODECLASS:
  332. UA_Variant_setScalarCopy(dst, (const UA_Int32*)&src->nodeClass, &UA_TYPES[UA_TYPES_INT32]);
  333. samplingError = UA_FALSE;
  334. break;
  335. case UA_ATTRIBUTEID_BROWSENAME:
  336. UA_Variant_setScalarCopy(dst, (const UA_String*)&src->browseName, &UA_TYPES[UA_TYPES_QUALIFIEDNAME]);
  337. samplingError = UA_FALSE;
  338. break;
  339. case UA_ATTRIBUTEID_DISPLAYNAME:
  340. UA_Variant_setScalarCopy(dst, (const UA_String*)&src->displayName, &UA_TYPES[UA_TYPES_LOCALIZEDTEXT]);
  341. samplingError = UA_FALSE;
  342. break;
  343. case UA_ATTRIBUTEID_DESCRIPTION:
  344. UA_Variant_setScalarCopy(dst, (const UA_String*)&src->displayName, &UA_TYPES[UA_TYPES_LOCALIZEDTEXT]);
  345. samplingError = UA_FALSE;
  346. break;
  347. case UA_ATTRIBUTEID_WRITEMASK:
  348. UA_Variant_setScalarCopy(dst, (const UA_String*)&src->writeMask, &UA_TYPES[UA_TYPES_UINT32]);
  349. samplingError = UA_FALSE;
  350. break;
  351. case UA_ATTRIBUTEID_USERWRITEMASK:
  352. UA_Variant_setScalarCopy(dst, (const UA_String*)&src->writeMask, &UA_TYPES[UA_TYPES_UINT32]);
  353. samplingError = UA_FALSE;
  354. break;
  355. case UA_ATTRIBUTEID_ISABSTRACT:
  356. break;
  357. case UA_ATTRIBUTEID_SYMMETRIC:
  358. break;
  359. case UA_ATTRIBUTEID_INVERSENAME:
  360. break;
  361. case UA_ATTRIBUTEID_CONTAINSNOLOOPS:
  362. break;
  363. case UA_ATTRIBUTEID_EVENTNOTIFIER:
  364. break;
  365. case UA_ATTRIBUTEID_VALUE:
  366. if(src->nodeClass == UA_NODECLASS_VARIABLE) {
  367. const UA_VariableNode *vsrc = (const UA_VariableNode*)src;
  368. if(srcAsVariableNode->valueSource == UA_VALUESOURCE_VARIANT) {
  369. UA_Variant_copy(&vsrc->value.variant.value, dst);
  370. //no onRead callback here since triggered by a subscription
  371. samplingError = UA_FALSE;
  372. } else {
  373. if(srcAsVariableNode->valueSource != UA_VALUESOURCE_DATASOURCE)
  374. break;
  375. // todo: handle numeric ranges
  376. if(srcAsVariableNode->value.dataSource.read(vsrc->value.dataSource.handle, vsrc->nodeId, UA_TRUE, UA_NULL,
  377. &sourceDataValue) != UA_STATUSCODE_GOOD)
  378. break;
  379. UA_Variant_copy(&sourceDataValue.value, dst);
  380. if(sourceDataValue.value.data) {
  381. UA_deleteMembers(sourceDataValue.value.data, sourceDataValue.value.type);
  382. UA_free(sourceDataValue.value.data);
  383. sourceDataValue.value.data = UA_NULL;
  384. }
  385. UA_DataValue_deleteMembers(&sourceDataValue);
  386. samplingError = UA_FALSE;
  387. }
  388. }
  389. break;
  390. case UA_ATTRIBUTEID_DATATYPE:
  391. break;
  392. case UA_ATTRIBUTEID_VALUERANK:
  393. break;
  394. case UA_ATTRIBUTEID_ARRAYDIMENSIONS:
  395. break;
  396. case UA_ATTRIBUTEID_ACCESSLEVEL:
  397. break;
  398. case UA_ATTRIBUTEID_USERACCESSLEVEL:
  399. break;
  400. case UA_ATTRIBUTEID_MINIMUMSAMPLINGINTERVAL:
  401. break;
  402. case UA_ATTRIBUTEID_HISTORIZING:
  403. break;
  404. case UA_ATTRIBUTEID_EXECUTABLE:
  405. break;
  406. case UA_ATTRIBUTEID_USEREXECUTABLE:
  407. break;
  408. default:
  409. break;
  410. }
  411. return samplingError;
  412. }
  413. void MonitoredItem_QueuePushDataValue(UA_Server *server, UA_MonitoredItem *monitoredItem) {
  414. UA_ByteString newValueAsByteString = { .length=0, .data=NULL };
  415. size_t encodingOffset = 0;
  416. if(!monitoredItem || monitoredItem->lastSampled + monitoredItem->samplingInterval > UA_DateTime_now())
  417. return;
  418. // FIXME: Actively suppress non change value based monitoring. There should be
  419. // another function to handle status and events.
  420. if(monitoredItem->monitoredItemType != MONITOREDITEM_TYPE_CHANGENOTIFY)
  421. return;
  422. MonitoredItem_queuedValue *newvalue = UA_malloc(sizeof(MonitoredItem_queuedValue));
  423. if(!newvalue)
  424. return;
  425. newvalue->listEntry.tqe_next = UA_NULL;
  426. newvalue->listEntry.tqe_prev = UA_NULL;
  427. UA_Variant_init(&newvalue->value);
  428. // Verify that the *Node being monitored is still valid
  429. // Looking up the in the nodestore is only necessary if we suspect that it is changed during writes
  430. // e.g. in multithreaded applications
  431. const UA_Node *target = UA_NodeStore_get(server->nodestore, &monitoredItem->monitoredNodeId);
  432. if(!target) {
  433. UA_free(newvalue);
  434. return;
  435. }
  436. UA_Boolean samplingError = MonitoredItem_CopyMonitoredValueToVariant(monitoredItem->attributeID, target,
  437. &newvalue->value);
  438. UA_NodeStore_release(target);
  439. if(samplingError != UA_FALSE || !newvalue->value.type) {
  440. UA_Variant_deleteMembers(&newvalue->value);
  441. UA_free(newvalue);
  442. return;
  443. }
  444. if(monitoredItem->queueSize.currentValue >= monitoredItem->queueSize.maxValue) {
  445. if(monitoredItem->discardOldest != UA_TRUE) {
  446. // We cannot remove the oldest value and theres no queue space left. We're done here.
  447. UA_free(newvalue);
  448. return;
  449. }
  450. MonitoredItem_queuedValue *queueItem = TAILQ_LAST(&monitoredItem->queue, QueueOfQueueDataValues);
  451. TAILQ_REMOVE(&monitoredItem->queue, queueItem, listEntry);
  452. UA_free(queueItem);
  453. monitoredItem->queueSize.currentValue--;
  454. }
  455. // encode the data to find if its different to the previous
  456. newValueAsByteString.length = UA_calcSizeBinary(&newvalue->value, &UA_TYPES[UA_TYPES_VARIANT]);
  457. newValueAsByteString.data = UA_malloc(newValueAsByteString.length);
  458. UA_encodeBinary(&newvalue->value, &UA_TYPES[UA_TYPES_VARIANT], &newValueAsByteString, &encodingOffset);
  459. if(!monitoredItem->lastSampledValue.data) {
  460. UA_ByteString_copy(&newValueAsByteString, &monitoredItem->lastSampledValue);
  461. TAILQ_INSERT_HEAD(&monitoredItem->queue, newvalue, listEntry);
  462. monitoredItem->queueSize.currentValue++;
  463. monitoredItem->lastSampled = UA_DateTime_now();
  464. UA_free(newValueAsByteString.data);
  465. } else {
  466. if(UA_String_equal(&newValueAsByteString, &monitoredItem->lastSampledValue) == UA_TRUE) {
  467. UA_Variant_deleteMembers(&newvalue->value);
  468. UA_free(newvalue);
  469. UA_String_deleteMembers(&newValueAsByteString);
  470. return;
  471. }
  472. UA_ByteString_deleteMembers(&monitoredItem->lastSampledValue);
  473. monitoredItem->lastSampledValue = newValueAsByteString;
  474. TAILQ_INSERT_HEAD(&monitoredItem->queue, newvalue, listEntry);
  475. monitoredItem->queueSize.currentValue++;
  476. monitoredItem->lastSampled = UA_DateTime_now();
  477. }
  478. }