ua_subscription.c 23 KB

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