ua_subscription.c 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525
  1. /* This Source Code Form is subject to the terms of the Mozilla Public
  2. * License, v. 2.0. If a copy of the MPL was not distributed with this
  3. * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  4. *
  5. * Copyright 2015-2017 (c) Julius Pfrommer, Fraunhofer IOSB
  6. * Copyright 2015 (c) Chris Iatrou
  7. * Copyright 2015-2016 (c) Sten Grüner
  8. * Copyright 2017-2018 (c) Thomas Stalder, Blue Time Concept SA
  9. * Copyright 2015 (c) Joakim L. Gilje
  10. * Copyright 2016-2017 (c) Florian Palm
  11. * Copyright 2015-2016 (c) Oleksiy Vasylyev
  12. * Copyright 2017 (c) frax2222
  13. * Copyright 2017 (c) Stefan Profanter, fortiss GmbH
  14. * Copyright 2017 (c) Mattias Bornhager
  15. */
  16. #include "ua_subscription.h"
  17. #include "ua_server_internal.h"
  18. #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
  19. UA_Subscription *
  20. UA_Subscription_new(UA_Session *session, UA_UInt32 subscriptionId) {
  21. /* Allocate the memory */
  22. UA_Subscription *newItem =
  23. (UA_Subscription*)UA_calloc(1, sizeof(UA_Subscription));
  24. if(!newItem)
  25. return NULL;
  26. /* Remaining members are covered by calloc zeroing out the memory */
  27. newItem->session = session;
  28. newItem->subscriptionId = subscriptionId;
  29. newItem->numMonitoredItems = 0;
  30. newItem->state = UA_SUBSCRIPTIONSTATE_NORMAL; /* The first publish response is sent immediately */
  31. TAILQ_INIT(&newItem->retransmissionQueue);
  32. return newItem;
  33. }
  34. void
  35. UA_Subscription_deleteMembers(UA_Server *server, UA_Subscription *sub) {
  36. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  37. "Subscription %u | Delete the subscription",
  38. sub->subscriptionId);
  39. Subscription_unregisterPublishCallback(server, sub);
  40. /* Delete monitored Items */
  41. UA_MonitoredItem *mon, *tmp_mon;
  42. LIST_FOREACH_SAFE(mon, &sub->monitoredItems, listEntry, tmp_mon) {
  43. LIST_REMOVE(mon, listEntry);
  44. MonitoredItem_delete(server, mon);
  45. }
  46. /* Delete Retransmission Queue */
  47. UA_NotificationMessageEntry *nme, *nme_tmp;
  48. TAILQ_FOREACH_SAFE(nme, &sub->retransmissionQueue, listEntry, nme_tmp) {
  49. TAILQ_REMOVE(&sub->retransmissionQueue, nme, listEntry);
  50. UA_NotificationMessage_deleteMembers(&nme->message);
  51. UA_free(nme);
  52. }
  53. sub->retransmissionQueueSize = 0;
  54. }
  55. UA_MonitoredItem *
  56. UA_Subscription_getMonitoredItem(UA_Subscription *sub,
  57. UA_UInt32 monitoredItemId) {
  58. UA_MonitoredItem *mon;
  59. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  60. if(mon->itemId == monitoredItemId)
  61. break;
  62. }
  63. return mon;
  64. }
  65. UA_StatusCode
  66. UA_Subscription_deleteMonitoredItem(UA_Server *server, UA_Subscription *sub,
  67. UA_UInt32 monitoredItemId) {
  68. /* Find the MonitoredItem */
  69. UA_MonitoredItem *mon;
  70. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  71. if(mon->itemId == monitoredItemId)
  72. break;
  73. }
  74. if(!mon)
  75. return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  76. /* Remove the MonitoredItem */
  77. LIST_REMOVE(mon, listEntry);
  78. MonitoredItem_delete(server, mon);
  79. sub->numMonitoredItems--;
  80. return UA_STATUSCODE_GOOD;
  81. }
  82. void
  83. UA_Subscription_addMonitoredItem(UA_Subscription *sub, UA_MonitoredItem *newMon) {
  84. sub->numMonitoredItems++;
  85. LIST_INSERT_HEAD(&sub->monitoredItems, newMon, listEntry);
  86. }
  87. UA_UInt32
  88. UA_Subscription_getNumMonitoredItems(UA_Subscription *sub) {
  89. return sub->numMonitoredItems;
  90. }
  91. static size_t
  92. countQueuedNotifications(UA_Subscription *sub,
  93. UA_Boolean *moreNotifications) {
  94. if(!sub->publishingEnabled)
  95. return 0;
  96. size_t notifications = 0;
  97. UA_MonitoredItem *mon;
  98. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  99. MonitoredItem_queuedValue *qv;
  100. TAILQ_FOREACH(qv, &mon->queue, listEntry) {
  101. if(notifications >= sub->notificationsPerPublish) {
  102. *moreNotifications = true;
  103. break;
  104. }
  105. ++notifications;
  106. }
  107. }
  108. return notifications;
  109. }
  110. static void
  111. UA_Subscription_addRetransmissionMessage(UA_Server *server, UA_Subscription *sub,
  112. UA_NotificationMessageEntry *entry) {
  113. /* Release the oldest entry if there is not enough space */
  114. if(server->config.maxRetransmissionQueueSize > 0 &&
  115. sub->retransmissionQueueSize >= server->config.maxRetransmissionQueueSize) {
  116. UA_NotificationMessageEntry *lastentry =
  117. TAILQ_LAST(&sub->retransmissionQueue, ListOfNotificationMessages);
  118. TAILQ_REMOVE(&sub->retransmissionQueue, lastentry, listEntry);
  119. --sub->retransmissionQueueSize;
  120. UA_NotificationMessage_deleteMembers(&lastentry->message);
  121. UA_free(lastentry);
  122. }
  123. /* Add entry */
  124. TAILQ_INSERT_HEAD(&sub->retransmissionQueue, entry, listEntry);
  125. ++sub->retransmissionQueueSize;
  126. }
  127. UA_StatusCode
  128. UA_Subscription_removeRetransmissionMessage(UA_Subscription *sub,
  129. UA_UInt32 sequenceNumber) {
  130. /* Find the retransmission message */
  131. UA_NotificationMessageEntry *entry;
  132. TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
  133. if(entry->message.sequenceNumber == sequenceNumber)
  134. break;
  135. }
  136. if(!entry)
  137. return UA_STATUSCODE_BADSEQUENCENUMBERUNKNOWN;
  138. /* Remove the retransmission message */
  139. TAILQ_REMOVE(&sub->retransmissionQueue, entry, listEntry);
  140. --sub->retransmissionQueueSize;
  141. UA_NotificationMessage_deleteMembers(&entry->message);
  142. UA_free(entry);
  143. return UA_STATUSCODE_GOOD;
  144. }
  145. static UA_MonitoredItem *
  146. selectFirstMonToIterate(UA_Subscription *sub) {
  147. UA_MonitoredItem *mon = LIST_FIRST(&sub->monitoredItems);
  148. if(sub->lastSendMonitoredItemId > 0) {
  149. while(mon) {
  150. if(mon->itemId == sub->lastSendMonitoredItemId)
  151. break;
  152. mon = LIST_NEXT(mon, listEntry);
  153. }
  154. if(!mon)
  155. mon = LIST_FIRST(&sub->monitoredItems);
  156. }
  157. return mon;
  158. }
  159. /* Iterate over the monitoreditems of the subscription, starting at mon, and
  160. * move notifications into the response. */
  161. static void
  162. moveNotificationsFromMonitoredItems(UA_Subscription *sub, UA_MonitoredItem *mon,
  163. UA_MonitoredItemNotification *mins, size_t minsSize,
  164. size_t *pos) {
  165. MonitoredItem_queuedValue *qv, *qv_tmp;
  166. while(mon) {
  167. sub->lastSendMonitoredItemId = mon->itemId;
  168. TAILQ_FOREACH_SAFE(qv, &mon->queue, listEntry, qv_tmp) {
  169. if(*pos >= minsSize)
  170. return;
  171. UA_MonitoredItemNotification *min = &mins[*pos];
  172. min->clientHandle = qv->clientHandle;
  173. if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  174. min->value = qv->data.value;
  175. } else {
  176. /* TODO implementation for events */
  177. }
  178. TAILQ_REMOVE(&mon->queue, qv, listEntry);
  179. UA_free(qv);
  180. --mon->currentQueueSize;
  181. ++(*pos);
  182. }
  183. mon = LIST_NEXT(mon, listEntry);
  184. }
  185. }
  186. static UA_StatusCode
  187. prepareNotificationMessage(UA_Subscription *sub,
  188. UA_NotificationMessage *message,
  189. size_t notifications) {
  190. /* Array of ExtensionObject to hold different kinds of notifications
  191. * (currently only DataChangeNotifications) */
  192. message->notificationData = UA_ExtensionObject_new();
  193. if(!message->notificationData)
  194. return UA_STATUSCODE_BADOUTOFMEMORY;
  195. message->notificationDataSize = 1;
  196. /* Allocate Notification */
  197. UA_DataChangeNotification *dcn = UA_DataChangeNotification_new();
  198. if(!dcn) {
  199. UA_NotificationMessage_deleteMembers(message);
  200. return UA_STATUSCODE_BADOUTOFMEMORY;
  201. }
  202. UA_ExtensionObject *data = message->notificationData;
  203. data->encoding = UA_EXTENSIONOBJECT_DECODED;
  204. data->content.decoded.data = dcn;
  205. data->content.decoded.type = &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION];
  206. /* Allocate array of notifications */
  207. dcn->monitoredItems = (UA_MonitoredItemNotification *)
  208. UA_Array_new(notifications,
  209. &UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION]);
  210. if(!dcn->monitoredItems) {
  211. UA_NotificationMessage_deleteMembers(message);
  212. return UA_STATUSCODE_BADOUTOFMEMORY;
  213. }
  214. dcn->monitoredItemsSize = notifications;
  215. /* Move notifications into the response .. the point of no return */
  216. /* Select the first monitoredItem or the first monitoreditem after the last
  217. * that was processed. */
  218. UA_MonitoredItem *mon = selectFirstMonToIterate(sub);
  219. /* Move notifications into the response */
  220. size_t l = 0;
  221. moveNotificationsFromMonitoredItems(sub, mon, dcn->monitoredItems, notifications, &l);
  222. if(l < notifications) {
  223. /* Not done. We skipped MonitoredItems. Restart at the beginning. */
  224. moveNotificationsFromMonitoredItems(sub, LIST_FIRST(&sub->monitoredItems),
  225. dcn->monitoredItems, notifications, &l);
  226. }
  227. return UA_STATUSCODE_GOOD;
  228. }
  229. /* According to OPC Unified Architecture, Part 4 5.13.1.1 i) The value 0 is
  230. * never used for the sequence number */
  231. static UA_UInt32
  232. UA_Subscription_nextSequenceNumber(UA_UInt32 sequenceNumber) {
  233. UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
  234. if(nextSequenceNumber == 0)
  235. nextSequenceNumber = 1;
  236. return nextSequenceNumber;
  237. }
  238. void
  239. UA_Subscription_publishCallback(UA_Server *server, UA_Subscription *sub) {
  240. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  241. "Subscription %u | Publish Callback",
  242. sub->subscriptionId);
  243. /* Dequeue a response */
  244. UA_PublishResponseEntry *pre = UA_Session_getPublishReq(sub->session);
  245. if(pre) {
  246. sub->currentLifetimeCount = 0; /* Reset the lifetimecounter */
  247. } else {
  248. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  249. "Subscription %u | The publish queue is empty",
  250. sub->subscriptionId);
  251. ++sub->currentLifetimeCount;
  252. if(sub->currentLifetimeCount > sub->lifeTimeCount) {
  253. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  254. "Subscription %u | End of lifetime "
  255. "for subscription", sub->subscriptionId);
  256. UA_Session_deleteSubscription(server, sub->session, sub->subscriptionId);
  257. /* TODO: send a StatusChangeNotification with Bad_Timeout */
  258. return;
  259. }
  260. }
  261. /* Count the available notifications */
  262. UA_Boolean moreNotifications = false;
  263. size_t notifications = countQueuedNotifications(sub, &moreNotifications);
  264. /* Return if no notifications and no keepalive */
  265. if(notifications == 0) {
  266. ++sub->currentKeepAliveCount;
  267. if(sub->currentKeepAliveCount < sub->maxKeepAliveCount)
  268. return;
  269. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  270. "Subscription %u | Sending a KeepAlive",
  271. sub->subscriptionId);
  272. }
  273. /* Can we send a response? */
  274. UA_SecureChannel *channel = sub->session->header.channel;
  275. if(!channel || !pre) {
  276. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  277. "Subscription %u | Want to send a publish response;"
  278. "but not possible", sub->subscriptionId);
  279. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  280. return;
  281. }
  282. /* Prepare the response */
  283. UA_PublishResponse *response = &pre->response;
  284. UA_NotificationMessage *message = &response->notificationMessage;
  285. UA_NotificationMessageEntry *retransmission = NULL;
  286. if(notifications > 0) {
  287. /* Allocate the retransmission entry */
  288. retransmission = (UA_NotificationMessageEntry*)
  289. UA_malloc(sizeof(UA_NotificationMessageEntry));
  290. if(!retransmission) {
  291. UA_LOG_WARNING_SESSION(server->config.logger, sub->session,
  292. "Subscription %u | Could not allocate memory "
  293. "for retransmission", sub->subscriptionId);
  294. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  295. return;
  296. }
  297. /* Prepare the response */
  298. UA_StatusCode retval = prepareNotificationMessage(sub, message, notifications);
  299. if(retval != UA_STATUSCODE_GOOD) {
  300. UA_LOG_WARNING_SESSION(server->config.logger, sub->session,
  301. "Subscription %u | Could not prepare the "
  302. "notification message", sub->subscriptionId);
  303. UA_free(retransmission);
  304. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  305. return;
  306. }
  307. }
  308. /* <-- The point of no return --> */
  309. /* Remove the response from the response queue */
  310. UA_Session_removePublishReq(sub->session, pre);
  311. /* Set up the response */
  312. response->responseHeader.timestamp = UA_DateTime_now();
  313. response->subscriptionId = sub->subscriptionId;
  314. response->moreNotifications = moreNotifications;
  315. message->publishTime = response->responseHeader.timestamp;
  316. /* Set the sequence number. The sequence number will be reused if there are
  317. * no notifications (and this is a keepalive message). */
  318. message->sequenceNumber = UA_Subscription_nextSequenceNumber(sub->sequenceNumber);
  319. if(notifications != 0) {
  320. /* There are notifications. So we can't reuse the sequence number. */
  321. sub->sequenceNumber = message->sequenceNumber;
  322. /* Put the notification message into the retransmission queue. This
  323. * needs to be done here, so that the message itself is included in the
  324. * available sequence numbers for acknowledgement. */
  325. retransmission->message = response->notificationMessage;
  326. UA_Subscription_addRetransmissionMessage(server, sub, retransmission);
  327. }
  328. /* Get the available sequence numbers from the retransmission queue */
  329. size_t available = sub->retransmissionQueueSize;
  330. if(available > 0) {
  331. response->availableSequenceNumbers =
  332. (UA_UInt32*)UA_alloca(available * sizeof(UA_UInt32));
  333. response->availableSequenceNumbersSize = available;
  334. size_t i = 0;
  335. UA_NotificationMessageEntry *nme;
  336. TAILQ_FOREACH(nme, &sub->retransmissionQueue, listEntry) {
  337. response->availableSequenceNumbers[i] = nme->message.sequenceNumber;
  338. ++i;
  339. }
  340. }
  341. /* Send the response */
  342. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  343. "Subscription %u | Sending out a publish response "
  344. "with %u notifications", sub->subscriptionId,
  345. (UA_UInt32)notifications);
  346. UA_SecureChannel_sendSymmetricMessage(sub->session->header.channel, pre->requestId,
  347. UA_MESSAGETYPE_MSG, response,
  348. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  349. /* Reset subscription state to normal. */
  350. sub->state = UA_SUBSCRIPTIONSTATE_NORMAL;
  351. sub->currentKeepAliveCount = 0;
  352. /* Free the response */
  353. UA_Array_delete(response->results, response->resultsSize,
  354. &UA_TYPES[UA_TYPES_UINT32]);
  355. UA_free(pre); /* no need for UA_PublishResponse_deleteMembers */
  356. if(!moreNotifications) {
  357. /* All notifications were sent. The next time, just start at the first
  358. * monitoreditem. */
  359. sub->lastSendMonitoredItemId = 0;
  360. } else {
  361. /* Repeat sending responses right away if there are more notifications
  362. * to send */
  363. UA_Subscription_publishCallback(server, sub);
  364. }
  365. }
  366. UA_Boolean
  367. UA_Subscription_reachedPublishReqLimit(UA_Server *server, UA_Session *session) {
  368. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  369. "Reached number of publish request limit");
  370. /* Dequeue a response */
  371. UA_PublishResponseEntry *pre = UA_Session_getPublishReq(session);
  372. /* Cannot publish without a response */
  373. if(!pre) {
  374. UA_LOG_FATAL_SESSION(server->config.logger, session, "No publish requests available");
  375. return false;
  376. }
  377. UA_PublishResponse *response = &pre->response;
  378. UA_NotificationMessage *message = &response->notificationMessage;
  379. /* <-- The point of no return --> */
  380. /* Remove the response from the response queue */
  381. UA_Session_removePublishReq(session, pre);
  382. /* Set up the response. Note that this response has no related subscription id */
  383. response->responseHeader.timestamp = UA_DateTime_now();
  384. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS;
  385. response->subscriptionId = 0;
  386. response->moreNotifications = false;
  387. message->publishTime = response->responseHeader.timestamp;
  388. message->sequenceNumber = 0;
  389. response->availableSequenceNumbersSize = 0;
  390. /* Send the response */
  391. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  392. "Sending out a publish response triggered by too many publish requests");
  393. UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId,
  394. UA_MESSAGETYPE_MSG, response,
  395. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  396. /* Free the response */
  397. UA_Array_delete(response->results, response->resultsSize,
  398. &UA_TYPES[UA_TYPES_UINT32]);
  399. UA_free(pre); /* no need for UA_PublishResponse_deleteMembers */
  400. return true;
  401. }
  402. UA_StatusCode
  403. Subscription_registerPublishCallback(UA_Server *server, UA_Subscription *sub) {
  404. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  405. "Subscription %u | Register subscription "
  406. "publishing callback", sub->subscriptionId);
  407. if(sub->publishCallbackIsRegistered)
  408. return UA_STATUSCODE_GOOD;
  409. UA_StatusCode retval =
  410. UA_Server_addRepeatedCallback(server,
  411. (UA_ServerCallback)UA_Subscription_publishCallback,
  412. sub, (UA_UInt32)sub->publishingInterval,
  413. &sub->publishCallbackId);
  414. if(retval != UA_STATUSCODE_GOOD)
  415. return retval;
  416. sub->publishCallbackIsRegistered = true;
  417. return UA_STATUSCODE_GOOD;
  418. }
  419. UA_StatusCode
  420. Subscription_unregisterPublishCallback(UA_Server *server, UA_Subscription *sub) {
  421. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  422. "Subscription %u | Unregister subscription "
  423. "publishing callback", sub->subscriptionId);
  424. if(!sub->publishCallbackIsRegistered)
  425. return UA_STATUSCODE_GOOD;
  426. UA_StatusCode retval =
  427. UA_Server_removeRepeatedCallback(server, sub->publishCallbackId);
  428. if(retval != UA_STATUSCODE_GOOD)
  429. return retval;
  430. sub->publishCallbackIsRegistered = false;
  431. return UA_STATUSCODE_GOOD;
  432. }
  433. /* When the session has publish requests stored but the last subscription is
  434. * deleted... Send out empty responses */
  435. void
  436. UA_Subscription_answerPublishRequestsNoSubscription(UA_Server *server,
  437. UA_Session *session) {
  438. /* No session or there are remaining subscriptions */
  439. if(!session || LIST_FIRST(&session->serverSubscriptions))
  440. return;
  441. /* Send a response for every queued request */
  442. UA_PublishResponseEntry *pre;
  443. while((pre = UA_Session_getPublishReq(session))) {
  444. UA_Session_removePublishReq(session, pre);
  445. UA_PublishResponse *response = &pre->response;
  446. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
  447. response->responseHeader.timestamp = UA_DateTime_now();
  448. UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId,
  449. UA_MESSAGETYPE_MSG, response,
  450. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  451. UA_PublishResponse_deleteMembers(response);
  452. UA_free(pre);
  453. }
  454. }
  455. #endif /* UA_ENABLE_SUBSCRIPTIONS */