ua_subscription.c 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367
  1. #include "ua_subscription.h"
  2. #include "ua_server_internal.h"
  3. #include "ua_services.h"
  4. #include "ua_nodestore.h"
  5. #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
  6. /*****************/
  7. /* MonitoredItem */
  8. /*****************/
  9. UA_MonitoredItem * UA_MonitoredItem_new() {
  10. UA_MonitoredItem *new = UA_malloc(sizeof(UA_MonitoredItem));
  11. new->subscription = NULL;
  12. new->currentQueueSize = 0;
  13. new->maxQueueSize = 0;
  14. new->monitoredItemType = UA_MONITOREDITEMTYPE_CHANGENOTIFY; /* currently hardcoded */
  15. new->timestampsToReturn = UA_TIMESTAMPSTORETURN_SOURCE;
  16. UA_String_init(&new->indexRange);
  17. TAILQ_INIT(&new->queue);
  18. UA_NodeId_init(&new->monitoredNodeId);
  19. new->lastSampledValue = UA_BYTESTRING_NULL;
  20. memset(&new->sampleJobGuid, 0, sizeof(UA_Guid));
  21. new->sampleJobIsRegistered = false;
  22. new->itemId = 0;
  23. return new;
  24. }
  25. void MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem) {
  26. MonitoredItem_unregisterSampleJob(server, monitoredItem);
  27. /* clear the queued samples */
  28. MonitoredItem_queuedValue *val, *val_tmp;
  29. TAILQ_FOREACH_SAFE(val, &monitoredItem->queue, listEntry, val_tmp) {
  30. TAILQ_REMOVE(&monitoredItem->queue, val, listEntry);
  31. UA_DataValue_deleteMembers(&val->value);
  32. UA_free(val);
  33. }
  34. monitoredItem->currentQueueSize = 0;
  35. LIST_REMOVE(monitoredItem, listEntry);
  36. UA_String_deleteMembers(&monitoredItem->indexRange);
  37. UA_ByteString_deleteMembers(&monitoredItem->lastSampledValue);
  38. UA_NodeId_deleteMembers(&monitoredItem->monitoredNodeId);
  39. UA_free(monitoredItem);
  40. }
  41. static void SampleCallback(UA_Server *server, UA_MonitoredItem *monitoredItem) {
  42. if(monitoredItem->monitoredItemType != UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  43. UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER, "Session " PRINTF_GUID_FORMAT " | MonitoredItem %i | "
  44. "Cannot process a monitoreditem that is not a data change notification",
  45. PRINTF_GUID_DATA(monitoredItem->subscription->session->sessionId), monitoredItem->itemId);
  46. return;
  47. }
  48. MonitoredItem_queuedValue *newvalue = UA_malloc(sizeof(MonitoredItem_queuedValue));
  49. if(!newvalue) {
  50. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  51. "Session " PRINTF_GUID_FORMAT " | MonitoredItem %i | Skipped a sample due to lack of memory",
  52. PRINTF_GUID_DATA(monitoredItem->subscription->session->sessionId), monitoredItem->itemId);
  53. return;
  54. }
  55. UA_DataValue_init(&newvalue->value);
  56. newvalue->clientHandle = monitoredItem->clientHandle;
  57. /* Read the value */
  58. UA_ReadValueId rvid;
  59. UA_ReadValueId_init(&rvid);
  60. rvid.nodeId = monitoredItem->monitoredNodeId;
  61. rvid.attributeId = monitoredItem->attributeID;
  62. rvid.indexRange = monitoredItem->indexRange;
  63. UA_Subscription *sub = monitoredItem->subscription;
  64. Service_Read_single(server, sub->session, monitoredItem->timestampsToReturn, &rvid, &newvalue->value);
  65. /* encode to see if the data has changed */
  66. size_t binsize = UA_calcSizeBinary(&newvalue->value.value, &UA_TYPES[UA_TYPES_VARIANT]);
  67. UA_ByteString newValueAsByteString;
  68. UA_StatusCode retval = UA_ByteString_allocBuffer(&newValueAsByteString, binsize);
  69. if(retval != UA_STATUSCODE_GOOD) {
  70. UA_DataValue_deleteMembers(&newvalue->value);
  71. UA_free(newvalue);
  72. return;
  73. }
  74. size_t encodingOffset = 0;
  75. retval = UA_encodeBinary(&newvalue->value.value, &UA_TYPES[UA_TYPES_VARIANT],
  76. NULL, NULL, &newValueAsByteString, &encodingOffset);
  77. /* error or the content has not changed */
  78. if(retval != UA_STATUSCODE_GOOD ||
  79. (monitoredItem->lastSampledValue.data &&
  80. UA_String_equal(&newValueAsByteString, &monitoredItem->lastSampledValue))) {
  81. UA_ByteString_deleteMembers(&newValueAsByteString);
  82. UA_DataValue_deleteMembers(&newvalue->value);
  83. UA_free(newvalue);
  84. UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER, "Session " PRINTF_GUID_FORMAT " | Subscription %u | "
  85. "MonitoredItem %u | Do not sample an unchanged value",
  86. PRINTF_GUID_DATA(monitoredItem->subscription->session->sessionId), monitoredItem->subscription->subscriptionID, monitoredItem->itemId);
  87. return;
  88. }
  89. UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER, "Session " PRINTF_GUID_FORMAT " | Subscription %u | "
  90. "MonitoredItem %u | Sampling the value",
  91. PRINTF_GUID_DATA(monitoredItem->subscription->session->sessionId),
  92. monitoredItem->subscription->subscriptionID, monitoredItem->itemId);
  93. /* do we have space in the queue? */
  94. if(monitoredItem->currentQueueSize >= monitoredItem->maxQueueSize) {
  95. if(!monitoredItem->discardOldest) {
  96. // We cannot remove the oldest value and theres no queue space left. We're done here.
  97. UA_ByteString_deleteMembers(&newValueAsByteString);
  98. UA_DataValue_deleteMembers(&newvalue->value);
  99. UA_free(newvalue);
  100. return;
  101. }
  102. MonitoredItem_queuedValue *queueItem = TAILQ_LAST(&monitoredItem->queue, QueueOfQueueDataValues);
  103. TAILQ_REMOVE(&monitoredItem->queue, queueItem, listEntry);
  104. UA_DataValue_deleteMembers(&queueItem->value);
  105. UA_free(queueItem);
  106. monitoredItem->currentQueueSize--;
  107. }
  108. /* add the sample */
  109. UA_ByteString_deleteMembers(&monitoredItem->lastSampledValue);
  110. monitoredItem->lastSampledValue = newValueAsByteString;
  111. TAILQ_INSERT_TAIL(&monitoredItem->queue, newvalue, listEntry);
  112. monitoredItem->currentQueueSize++;
  113. }
  114. UA_StatusCode MonitoredItem_registerSampleJob(UA_Server *server, UA_MonitoredItem *mon) {
  115. UA_Job job = {.type = UA_JOBTYPE_METHODCALL,
  116. .job.methodCall = {.method = (UA_ServerCallback)SampleCallback, .data = mon} };
  117. UA_StatusCode retval = UA_Server_addRepeatedJob(server, job, (UA_UInt32)mon->samplingInterval,
  118. &mon->sampleJobGuid);
  119. if(retval == UA_STATUSCODE_GOOD)
  120. mon->sampleJobIsRegistered = true;
  121. return retval;
  122. }
  123. UA_StatusCode MonitoredItem_unregisterSampleJob(UA_Server *server, UA_MonitoredItem *mon) {
  124. if(!mon->sampleJobIsRegistered)
  125. return UA_STATUSCODE_GOOD;
  126. mon->sampleJobIsRegistered = false;
  127. return UA_Server_removeRepeatedJob(server, mon->sampleJobGuid);
  128. }
  129. /****************/
  130. /* Subscription */
  131. /****************/
  132. UA_Subscription * UA_Subscription_new(UA_Session *session, UA_UInt32 subscriptionID) {
  133. UA_Subscription *new = UA_malloc(sizeof(UA_Subscription));
  134. if(!new)
  135. return NULL;
  136. new->session = session;
  137. new->subscriptionID = subscriptionID;
  138. new->sequenceNumber = 0;
  139. new->maxKeepAliveCount = 0;
  140. new->publishingEnabled = false;
  141. memset(&new->publishJobGuid, 0, sizeof(UA_Guid));
  142. new->publishJobIsRegistered = false;
  143. new->currentKeepAliveCount = 0;
  144. new->currentLifetimeCount = 0;
  145. new->lastMonitoredItemId = 0;
  146. new->state = UA_SUBSCRIPTIONSTATE_LATE; /* The first publish response is sent immediately */
  147. LIST_INIT(&new->retransmissionQueue);
  148. LIST_INIT(&new->MonitoredItems);
  149. return new;
  150. }
  151. void UA_Subscription_deleteMembers(UA_Subscription *subscription, UA_Server *server) {
  152. Subscription_unregisterPublishJob(server, subscription);
  153. /* Delete monitored Items */
  154. UA_MonitoredItem *mon, *tmp_mon;
  155. LIST_FOREACH_SAFE(mon, &subscription->MonitoredItems, listEntry, tmp_mon) {
  156. LIST_REMOVE(mon, listEntry);
  157. MonitoredItem_delete(server, mon);
  158. }
  159. /* Delete Retransmission Queue */
  160. UA_NotificationMessageEntry *nme, *nme_tmp;
  161. LIST_FOREACH_SAFE(nme, &subscription->retransmissionQueue, listEntry, nme_tmp) {
  162. LIST_REMOVE(nme, listEntry);
  163. UA_NotificationMessage_deleteMembers(&nme->message);
  164. UA_free(nme);
  165. }
  166. }
  167. UA_MonitoredItem *
  168. UA_Subscription_getMonitoredItem(UA_Subscription *sub, UA_UInt32 monitoredItemID) {
  169. UA_MonitoredItem *mon;
  170. LIST_FOREACH(mon, &sub->MonitoredItems, listEntry) {
  171. if(mon->itemId == monitoredItemID)
  172. break;
  173. }
  174. return mon;
  175. }
  176. UA_StatusCode
  177. UA_Subscription_deleteMonitoredItem(UA_Server *server, UA_Subscription *sub,
  178. UA_UInt32 monitoredItemID) {
  179. UA_MonitoredItem *mon;
  180. LIST_FOREACH(mon, &sub->MonitoredItems, listEntry) {
  181. if(mon->itemId == monitoredItemID) {
  182. LIST_REMOVE(mon, listEntry);
  183. MonitoredItem_delete(server, mon);
  184. return UA_STATUSCODE_GOOD;
  185. }
  186. }
  187. return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  188. }
  189. void UA_Subscription_publishCallback(UA_Server *server, UA_Subscription *sub) {
  190. /* Count the available notifications */
  191. size_t notifications = 0;
  192. UA_Boolean moreNotifications = false;
  193. if(sub->publishingEnabled) {
  194. UA_MonitoredItem *mon;
  195. LIST_FOREACH(mon, &sub->MonitoredItems, listEntry) {
  196. MonitoredItem_queuedValue *qv;
  197. TAILQ_FOREACH(qv, &mon->queue, listEntry) {
  198. if(notifications >= sub->notificationsPerPublish) {
  199. moreNotifications = true;
  200. break;
  201. }
  202. notifications++;
  203. }
  204. }
  205. }
  206. /* Return if nothing to do */
  207. if(notifications == 0) {
  208. sub->currentKeepAliveCount++;
  209. if(sub->currentKeepAliveCount < sub->maxKeepAliveCount)
  210. return;
  211. }
  212. /* Check if the securechannel is valid */
  213. UA_SecureChannel *channel = sub->session->channel;
  214. if(!channel)
  215. return;
  216. /* Dequeue a response */
  217. UA_PublishResponseEntry *pre = SIMPLEQ_FIRST(&sub->session->responseQueue);
  218. if(!pre) {
  219. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  220. "Cannot send a publish response on subscription %u, "
  221. "since the publish queue is empty", sub->subscriptionID)
  222. if(sub->state != UA_SUBSCRIPTIONSTATE_LATE) {
  223. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  224. } else {
  225. sub->currentLifetimeCount++;
  226. if(sub->currentLifetimeCount >= sub->lifeTimeCount) {
  227. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  228. "Session %u | Subscription %u | End of lifetime for subscription",
  229. sub->session->authenticationToken.identifier.numeric, sub->subscriptionID);
  230. UA_Session_deleteSubscription(server, sub->session, sub->subscriptionID);
  231. }
  232. }
  233. return;
  234. }
  235. SIMPLEQ_REMOVE_HEAD(&sub->session->responseQueue, listEntry);
  236. UA_PublishResponse *response = &pre->response;
  237. UA_UInt32 requestId = pre->requestId;
  238. /* We have a request. Reset state to normal. */
  239. sub->state = UA_SUBSCRIPTIONSTATE_NORMAL;
  240. sub->currentKeepAliveCount = 0;
  241. sub->currentLifetimeCount = 0;
  242. /* Prepare the response */
  243. response->responseHeader.timestamp = UA_DateTime_now();
  244. response->subscriptionId = sub->subscriptionID;
  245. response->moreNotifications = moreNotifications;
  246. UA_NotificationMessage *message = &response->notificationMessage;
  247. message->publishTime = response->responseHeader.timestamp;
  248. if(notifications == 0) {
  249. /* Send sequence number for the next notification */
  250. message->sequenceNumber = sub->sequenceNumber + 1;
  251. } else {
  252. /* Increase the sequence number */
  253. message->sequenceNumber = ++sub->sequenceNumber;
  254. /* Collect the notification messages */
  255. message->notificationData = UA_ExtensionObject_new();
  256. message->notificationDataSize = 1;
  257. UA_ExtensionObject *data = message->notificationData;
  258. UA_DataChangeNotification *dcn = UA_DataChangeNotification_new();
  259. dcn->monitoredItems = UA_Array_new(notifications, &UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION]);
  260. dcn->monitoredItemsSize = notifications;
  261. size_t l = 0;
  262. UA_MonitoredItem *mon;
  263. LIST_FOREACH(mon, &sub->MonitoredItems, listEntry) {
  264. MonitoredItem_queuedValue *qv, *qv_tmp;
  265. TAILQ_FOREACH_SAFE(qv, &mon->queue, listEntry, qv_tmp) {
  266. if(notifications <= l)
  267. break;
  268. UA_MonitoredItemNotification *min = &dcn->monitoredItems[l];
  269. min->clientHandle = qv->clientHandle;
  270. min->value = qv->value;
  271. TAILQ_REMOVE(&mon->queue, qv, listEntry);
  272. UA_free(qv);
  273. mon->currentQueueSize--;
  274. l++;
  275. }
  276. }
  277. data->encoding = UA_EXTENSIONOBJECT_DECODED;
  278. data->content.decoded.data = dcn;
  279. data->content.decoded.type = &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION];
  280. /* Put the notification message into the retransmission queue */
  281. UA_NotificationMessageEntry *retransmission = malloc(sizeof(UA_NotificationMessageEntry));
  282. retransmission->message = response->notificationMessage;
  283. LIST_INSERT_HEAD(&sub->retransmissionQueue, retransmission, listEntry);
  284. }
  285. /* Get the available sequence numbers from the retransmission queue */
  286. size_t available = 0, i = 0;
  287. UA_NotificationMessageEntry *nme;
  288. LIST_FOREACH(nme, &sub->retransmissionQueue, listEntry)
  289. available++;
  290. //cppcheck-suppress knownConditionTrueFalse
  291. if(available > 0) {
  292. response->availableSequenceNumbers = UA_alloca(available * sizeof(UA_UInt32));
  293. response->availableSequenceNumbersSize = available;
  294. }
  295. LIST_FOREACH(nme, &sub->retransmissionQueue, listEntry) {
  296. response->availableSequenceNumbers[i] = nme->message.sequenceNumber;
  297. i++;
  298. }
  299. /* Send the response */
  300. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  301. "Sending out a publish response with %u notifications", (UA_UInt32)notifications);
  302. UA_SecureChannel_sendBinaryMessage(sub->session->channel, requestId, response,
  303. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  304. /* Remove the queued request */
  305. UA_NotificationMessage_init(&response->notificationMessage); /* message was copied to the queue */
  306. response->availableSequenceNumbers = NULL; /* stack-allocated */
  307. response->availableSequenceNumbersSize = 0;
  308. UA_PublishResponse_deleteMembers(&pre->response);
  309. UA_free(pre);
  310. /* Repeat if there are more notifications to send */
  311. if(moreNotifications)
  312. UA_Subscription_publishCallback(server, sub);
  313. }
  314. UA_StatusCode Subscription_registerPublishJob(UA_Server *server, UA_Subscription *sub) {
  315. UA_Job job = (UA_Job) {.type = UA_JOBTYPE_METHODCALL,
  316. .job.methodCall = {.method = (UA_ServerCallback)UA_Subscription_publishCallback,
  317. .data = sub} };
  318. UA_StatusCode retval = UA_Server_addRepeatedJob(server, job, (UA_UInt32)sub->publishingInterval,
  319. &sub->publishJobGuid);
  320. if(retval == UA_STATUSCODE_GOOD)
  321. sub->publishJobIsRegistered = true;
  322. return retval;
  323. }
  324. UA_StatusCode Subscription_unregisterPublishJob(UA_Server *server, UA_Subscription *sub) {
  325. if(!sub->publishJobIsRegistered)
  326. return UA_STATUSCODE_GOOD;
  327. sub->publishJobIsRegistered = false;
  328. return UA_Server_removeRepeatedJob(server, sub->publishJobGuid);
  329. }
  330. #endif /* UA_ENABLE_SUBSCRIPTIONS */