ua_subscription.c 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608
  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-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
  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) Ari Breitkreuz, fortiss GmbH
  15. * Copyright 2017 (c) Mattias Bornhager
  16. */
  17. #include "ua_server_internal.h"
  18. #include "ua_subscription.h"
  19. #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
  20. void
  21. UA_Notification_enqueue(UA_Server *server, UA_Subscription *sub,
  22. UA_MonitoredItem *mon, UA_Notification *n) {
  23. /* Add to the MonitoredItem */
  24. TAILQ_INSERT_TAIL(&mon->queue, n, listEntry);
  25. ++mon->queueSize;
  26. /* Add to the subscription */
  27. TAILQ_INSERT_TAIL(&sub->notificationQueue, n, globalEntry);
  28. ++sub->notificationQueueSize;
  29. if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  30. ++sub->dataChangeNotifications;
  31. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
  32. ++sub->eventNotifications;
  33. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
  34. ++sub->statusChangeNotifications;
  35. }
  36. /* Ensure enough space is available in the MonitoredItem. Do this only after
  37. * adding the new Notification. */
  38. UA_MonitoredItem_ensureQueueSpace(server, mon);
  39. }
  40. void
  41. UA_Notification_delete(UA_Subscription *sub, UA_MonitoredItem *mon,
  42. UA_Notification *n) {
  43. if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  44. UA_DataValue_deleteMembers(&n->data.value);
  45. --sub->dataChangeNotifications;
  46. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  47. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
  48. UA_EventFieldList_deleteMembers(&n->data.event.fields);
  49. /* EventFilterResult currently isn't being used
  50. * UA_EventFilterResult_delete(notification->data.event->result); */
  51. --sub->eventNotifications;
  52. #endif
  53. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY) {
  54. --sub->statusChangeNotifications;
  55. }
  56. TAILQ_REMOVE(&mon->queue, n, listEntry);
  57. --mon->queueSize;
  58. TAILQ_REMOVE(&sub->notificationQueue, n, globalEntry);
  59. --sub->notificationQueueSize;
  60. UA_free(n);
  61. }
  62. UA_Subscription *
  63. UA_Subscription_new(UA_Session *session, UA_UInt32 subscriptionId) {
  64. /* Allocate the memory */
  65. UA_Subscription *newSub =
  66. (UA_Subscription*)UA_calloc(1, sizeof(UA_Subscription));
  67. if(!newSub)
  68. return NULL;
  69. /* Remaining members are covered by calloc zeroing out the memory */
  70. newSub->session = session;
  71. newSub->subscriptionId = subscriptionId;
  72. newSub->state = UA_SUBSCRIPTIONSTATE_NORMAL; /* The first publish response is sent immediately */
  73. TAILQ_INIT(&newSub->retransmissionQueue);
  74. TAILQ_INIT(&newSub->notificationQueue);
  75. return newSub;
  76. }
  77. void
  78. UA_Subscription_deleteMembers(UA_Server *server, UA_Subscription *sub) {
  79. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session, "Subscription %u | "
  80. "Delete the subscription", sub->subscriptionId);
  81. Subscription_unregisterPublishCallback(server, sub);
  82. /* Delete monitored Items */
  83. UA_MonitoredItem *mon, *tmp_mon;
  84. LIST_FOREACH_SAFE(mon, &sub->monitoredItems, listEntry, tmp_mon) {
  85. LIST_REMOVE(mon, listEntry);
  86. UA_MonitoredItem_delete(server, mon);
  87. }
  88. sub->monitoredItemsSize = 0;
  89. /* Delete Retransmission Queue */
  90. UA_NotificationMessageEntry *nme, *nme_tmp;
  91. TAILQ_FOREACH_SAFE(nme, &sub->retransmissionQueue, listEntry, nme_tmp) {
  92. TAILQ_REMOVE(&sub->retransmissionQueue, nme, listEntry);
  93. UA_NotificationMessage_deleteMembers(&nme->message);
  94. UA_free(nme);
  95. }
  96. sub->retransmissionQueueSize = 0;
  97. }
  98. UA_MonitoredItem *
  99. UA_Subscription_getMonitoredItem(UA_Subscription *sub, UA_UInt32 monitoredItemId) {
  100. UA_MonitoredItem *mon;
  101. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  102. if(mon->monitoredItemId == monitoredItemId)
  103. break;
  104. }
  105. return mon;
  106. }
  107. UA_StatusCode
  108. UA_Subscription_deleteMonitoredItem(UA_Server *server, UA_Subscription *sub,
  109. UA_UInt32 monitoredItemId) {
  110. /* Find the MonitoredItem */
  111. UA_MonitoredItem *mon;
  112. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  113. if(mon->monitoredItemId == monitoredItemId)
  114. break;
  115. }
  116. if(!mon)
  117. return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  118. /* Remove the MonitoredItem */
  119. LIST_REMOVE(mon, listEntry);
  120. sub->monitoredItemsSize--;
  121. /* Remove content and delayed free */
  122. UA_MonitoredItem_delete(server, mon);
  123. return UA_STATUSCODE_GOOD;
  124. }
  125. void
  126. UA_Subscription_addMonitoredItem(UA_Subscription *sub, UA_MonitoredItem *newMon) {
  127. sub->monitoredItemsSize++;
  128. LIST_INSERT_HEAD(&sub->monitoredItems, newMon, listEntry);
  129. }
  130. static void
  131. UA_Subscription_addRetransmissionMessage(UA_Server *server, UA_Subscription *sub,
  132. UA_NotificationMessageEntry *entry) {
  133. /* Release the oldest entry if there is not enough space */
  134. if(server->config.maxRetransmissionQueueSize > 0 &&
  135. sub->retransmissionQueueSize >= server->config.maxRetransmissionQueueSize) {
  136. UA_NotificationMessageEntry *lastentry =
  137. TAILQ_LAST(&sub->retransmissionQueue, ListOfNotificationMessages);
  138. TAILQ_REMOVE(&sub->retransmissionQueue, lastentry, listEntry);
  139. --sub->retransmissionQueueSize;
  140. UA_NotificationMessage_deleteMembers(&lastentry->message);
  141. UA_free(lastentry);
  142. }
  143. /* Add entry */
  144. TAILQ_INSERT_HEAD(&sub->retransmissionQueue, entry, listEntry);
  145. ++sub->retransmissionQueueSize;
  146. }
  147. UA_StatusCode
  148. UA_Subscription_removeRetransmissionMessage(UA_Subscription *sub, UA_UInt32 sequenceNumber) {
  149. /* Find the retransmission message */
  150. UA_NotificationMessageEntry *entry;
  151. TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
  152. if(entry->message.sequenceNumber == sequenceNumber)
  153. break;
  154. }
  155. if(!entry)
  156. return UA_STATUSCODE_BADSEQUENCENUMBERUNKNOWN;
  157. /* Remove the retransmission message */
  158. TAILQ_REMOVE(&sub->retransmissionQueue, entry, listEntry);
  159. --sub->retransmissionQueueSize;
  160. UA_NotificationMessage_deleteMembers(&entry->message);
  161. UA_free(entry);
  162. return UA_STATUSCODE_GOOD;
  163. }
  164. static UA_StatusCode
  165. prepareNotificationMessage(UA_Server *server, UA_Subscription *sub,
  166. UA_NotificationMessage *message, size_t notifications) {
  167. UA_assert(notifications > 0);
  168. /* Allocate an ExtensionObject for events and data */
  169. message->notificationData = (UA_ExtensionObject*)
  170. UA_Array_new(2, &UA_TYPES[UA_TYPES_EXTENSIONOBJECT]);
  171. if(!message->notificationData)
  172. return UA_STATUSCODE_BADOUTOFMEMORY;
  173. message->notificationDataSize = 2;
  174. /* Pre-allocate DataChangeNotifications */
  175. size_t notificationDataIdx = 0;
  176. UA_DataChangeNotification *dcn = NULL;
  177. if(sub->dataChangeNotifications > 0) {
  178. dcn = UA_DataChangeNotification_new();
  179. if(!dcn) {
  180. UA_NotificationMessage_deleteMembers(message);
  181. return UA_STATUSCODE_BADOUTOFMEMORY;
  182. }
  183. message->notificationData->encoding = UA_EXTENSIONOBJECT_DECODED;
  184. message->notificationData->content.decoded.data = dcn;
  185. message->notificationData->content.decoded.type = &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION];
  186. size_t dcnSize = sub->dataChangeNotifications;
  187. if(dcnSize > notifications)
  188. dcnSize = notifications;
  189. dcn->monitoredItems = (UA_MonitoredItemNotification*)
  190. UA_Array_new(dcnSize, &UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION]);
  191. if(!dcn->monitoredItems) {
  192. UA_NotificationMessage_deleteMembers(message);
  193. return UA_STATUSCODE_BADOUTOFMEMORY;
  194. }
  195. dcn->monitoredItemsSize = dcnSize;
  196. notificationDataIdx++;
  197. }
  198. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  199. UA_EventNotificationList *enl = NULL;
  200. UA_StatusChangeNotification *scn = NULL;
  201. /* Pre-allocate either StatusChange or EventNotifications. Sending a
  202. * (single) StatusChangeNotification has priority. */
  203. if(sub->statusChangeNotifications > 0) {
  204. scn = UA_StatusChangeNotification_new();
  205. if(!scn) {
  206. UA_NotificationMessage_deleteMembers(message);
  207. return UA_STATUSCODE_BADOUTOFMEMORY;
  208. }
  209. message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
  210. message->notificationData[notificationDataIdx].content.decoded.data = scn;
  211. message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION];
  212. notificationDataIdx++;
  213. } else if(sub->eventNotifications > 0) {
  214. enl = UA_EventNotificationList_new();
  215. if(!enl) {
  216. UA_NotificationMessage_deleteMembers(message);
  217. return UA_STATUSCODE_BADOUTOFMEMORY;
  218. }
  219. message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
  220. message->notificationData[notificationDataIdx].content.decoded.data = enl;
  221. message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST];
  222. size_t enlSize = sub->eventNotifications;
  223. if(enlSize > notifications)
  224. enlSize = notifications;
  225. enl->events = (UA_EventFieldList*) UA_Array_new(enlSize, &UA_TYPES[UA_TYPES_EVENTFIELDLIST]);
  226. if(!enl->events) {
  227. UA_NotificationMessage_deleteMembers(message);
  228. return UA_STATUSCODE_BADOUTOFMEMORY;
  229. }
  230. enl->eventsSize = enlSize;
  231. notificationDataIdx++;
  232. }
  233. #endif
  234. UA_assert(notificationDataIdx > 0);
  235. message->notificationDataSize = notificationDataIdx;
  236. /* <-- The point of no return --> */
  237. size_t totalNotifications = 0; /* How many notifications were moved to the response overall? */
  238. size_t dcnPos = 0; /* How many DataChangeNotifications were put into the list? */
  239. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  240. size_t enlPos = 0; /* How many EventNotifications were moved into the list */
  241. #endif
  242. UA_Notification *notification, *notification_tmp;
  243. TAILQ_FOREACH_SAFE(notification, &sub->notificationQueue, globalEntry, notification_tmp) {
  244. if(totalNotifications >= notifications)
  245. break;
  246. UA_MonitoredItem *mon = notification->mon;
  247. /* Move the content to the response */
  248. if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  249. UA_assert(dcn != NULL); /* Have at least one change notification */
  250. /* Move the content to the response */
  251. UA_MonitoredItemNotification *min = &dcn->monitoredItems[dcnPos];
  252. min->clientHandle = mon->clientHandle;
  253. min->value = notification->data.value;
  254. UA_DataValue_init(&notification->data.value); /* Reset after the value has been moved */
  255. dcnPos++;
  256. }
  257. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  258. else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY && scn) {
  259. // TODO: Handling of StatusChangeNotifications
  260. scn = NULL; /* At most one per PublishReponse */
  261. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY && enl) {
  262. UA_assert(enl != NULL); /* Have at least one event notification */
  263. /* Move the content to the response */
  264. UA_EventFieldList *efl = &enl->events[enlPos];
  265. *efl = notification->data.event.fields;
  266. UA_EventFieldList_init(&notification->data.event.fields);
  267. efl->clientHandle = mon->clientHandle;
  268. enlPos++;
  269. }
  270. #endif
  271. else {
  272. continue; /* Nothing to do */
  273. }
  274. /* Remove the notification from the queues */
  275. UA_Notification_delete(sub, mon, notification);
  276. totalNotifications++;
  277. }
  278. /* Set sizes */
  279. if(dcn) {
  280. dcn->monitoredItemsSize = dcnPos;
  281. if(dcnPos == 0) {
  282. UA_free(dcn->monitoredItems);
  283. dcn->monitoredItems = NULL;
  284. }
  285. }
  286. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  287. if(enl) {
  288. enl->eventsSize = enlPos;
  289. if(enlPos == 0) {
  290. UA_free(enl->events);
  291. enl->events = NULL;
  292. }
  293. }
  294. #endif
  295. return UA_STATUSCODE_GOOD;
  296. }
  297. /* According to OPC Unified Architecture, Part 4 5.13.1.1 i) The value 0 is
  298. * never used for the sequence number */
  299. static UA_UInt32
  300. UA_Subscription_nextSequenceNumber(UA_UInt32 sequenceNumber) {
  301. UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
  302. if(nextSequenceNumber == 0)
  303. nextSequenceNumber = 1;
  304. return nextSequenceNumber;
  305. }
  306. static void
  307. publishCallback(UA_Server *server, UA_Subscription *sub) {
  308. sub->readyNotifications = sub->notificationQueueSize;
  309. UA_Subscription_publish(server, sub);
  310. }
  311. void
  312. UA_Subscription_publish(UA_Server *server, UA_Subscription *sub) {
  313. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session, "Subscription %u | "
  314. "Publish Callback", sub->subscriptionId);
  315. /* Dequeue a response */
  316. UA_PublishResponseEntry *pre = UA_Session_dequeuePublishReq(sub->session);
  317. if(pre) {
  318. sub->currentLifetimeCount = 0; /* Reset the LifetimeCounter */
  319. } else {
  320. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  321. "Subscription %u | The publish queue is empty",
  322. sub->subscriptionId);
  323. ++sub->currentLifetimeCount;
  324. if(sub->currentLifetimeCount > sub->lifeTimeCount) {
  325. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  326. "Subscription %u | End of lifetime "
  327. "for subscription", sub->subscriptionId);
  328. UA_Session_deleteSubscription(server, sub->session, sub->subscriptionId);
  329. /* TODO: send a StatusChangeNotification with Bad_Timeout */
  330. return;
  331. }
  332. }
  333. /* If there are several late publish responses... */
  334. if(sub->readyNotifications > sub->notificationQueueSize)
  335. sub->readyNotifications = sub->notificationQueueSize;
  336. /* Count the available notifications */
  337. UA_UInt32 notifications = sub->readyNotifications;
  338. if(!sub->publishingEnabled)
  339. notifications = 0;
  340. UA_Boolean moreNotifications = false;
  341. if(notifications > sub->notificationsPerPublish) {
  342. notifications = sub->notificationsPerPublish;
  343. moreNotifications = true;
  344. }
  345. /* Return if no notifications and no keepalive */
  346. if(notifications == 0) {
  347. ++sub->currentKeepAliveCount;
  348. if(sub->currentKeepAliveCount < sub->maxKeepAliveCount) {
  349. if(pre)
  350. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  351. return;
  352. }
  353. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  354. "Subscription %u | Sending a KeepAlive",
  355. sub->subscriptionId);
  356. }
  357. /* We want to send a response. Is the channel open? */
  358. UA_SecureChannel *channel = sub->session->header.channel;
  359. if(!channel || !pre) {
  360. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  361. "Subscription %u | Want to send a publish response but can't. "
  362. "The subscription is late.", sub->subscriptionId);
  363. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  364. if(pre)
  365. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  366. return;
  367. }
  368. /* Prepare the response */
  369. UA_PublishResponse *response = &pre->response;
  370. UA_NotificationMessage *message = &response->notificationMessage;
  371. UA_NotificationMessageEntry *retransmission = NULL;
  372. if(notifications > 0) {
  373. /* Allocate the retransmission entry */
  374. retransmission = (UA_NotificationMessageEntry*)UA_malloc(sizeof(UA_NotificationMessageEntry));
  375. if(!retransmission) {
  376. UA_LOG_WARNING_SESSION(server->config.logger, sub->session,
  377. "Subscription %u | Could not allocate memory for retransmission. "
  378. "The subscription is late.", sub->subscriptionId);
  379. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  380. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  381. return;
  382. }
  383. /* Prepare the response */
  384. UA_StatusCode retval = prepareNotificationMessage(server, sub, message, notifications);
  385. if(retval != UA_STATUSCODE_GOOD) {
  386. UA_LOG_WARNING_SESSION(server->config.logger, sub->session,
  387. "Subscription %u | Could not prepare the notification message. "
  388. "The subscription is late.", sub->subscriptionId);
  389. UA_free(retransmission);
  390. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  391. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  392. return;
  393. }
  394. }
  395. /* <-- The point of no return --> */
  396. /* Adjust the number of ready notifications */
  397. UA_assert(sub->readyNotifications >= notifications);
  398. sub->readyNotifications -= notifications;
  399. /* Set up the response */
  400. response->responseHeader.timestamp = UA_DateTime_now();
  401. response->subscriptionId = sub->subscriptionId;
  402. response->moreNotifications = moreNotifications;
  403. message->publishTime = response->responseHeader.timestamp;
  404. /* Set the sequence number. The sequence number will be reused if there are
  405. * no notifications (and this is a keepalive message). */
  406. message->sequenceNumber = UA_Subscription_nextSequenceNumber(sub->sequenceNumber);
  407. if(notifications > 0) {
  408. /* There are notifications. So we can't reuse the sequence number. */
  409. sub->sequenceNumber = message->sequenceNumber;
  410. /* Put the notification message into the retransmission queue. This
  411. * needs to be done here, so that the message itself is included in the
  412. * available sequence numbers for acknowledgement. */
  413. retransmission->message = response->notificationMessage;
  414. UA_Subscription_addRetransmissionMessage(server, sub, retransmission);
  415. }
  416. /* Get the available sequence numbers from the retransmission queue */
  417. size_t available = sub->retransmissionQueueSize;
  418. UA_STACKARRAY(UA_UInt32, seqNumbers, available);
  419. if(available > 0) {
  420. response->availableSequenceNumbers = seqNumbers;
  421. response->availableSequenceNumbersSize = available;
  422. size_t i = 0;
  423. UA_NotificationMessageEntry *nme;
  424. TAILQ_FOREACH(nme, &sub->retransmissionQueue, listEntry) {
  425. response->availableSequenceNumbers[i] = nme->message.sequenceNumber;
  426. ++i;
  427. }
  428. }
  429. /* Send the response */
  430. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  431. "Subscription %u | Sending out a publish response "
  432. "with %u notifications", sub->subscriptionId,
  433. (UA_UInt32)notifications);
  434. UA_SecureChannel_sendSymmetricMessage(sub->session->header.channel, pre->requestId,
  435. UA_MESSAGETYPE_MSG, response,
  436. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  437. /* Reset subscription state to normal */
  438. sub->state = UA_SUBSCRIPTIONSTATE_NORMAL;
  439. sub->currentKeepAliveCount = 0;
  440. /* Free the response */
  441. UA_Array_delete(response->results, response->resultsSize, &UA_TYPES[UA_TYPES_UINT32]);
  442. UA_free(pre); /* No need for UA_PublishResponse_deleteMembers */
  443. /* Repeat sending responses if there are more notifications to send */
  444. if(moreNotifications)
  445. UA_Subscription_publish(server, sub);
  446. }
  447. UA_Boolean
  448. UA_Subscription_reachedPublishReqLimit(UA_Server *server, UA_Session *session) {
  449. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Reached number of publish request limit");
  450. /* Dequeue a response */
  451. UA_PublishResponseEntry *pre = UA_Session_dequeuePublishReq(session);
  452. /* Cannot publish without a response */
  453. if(!pre) {
  454. UA_LOG_FATAL_SESSION(server->config.logger, session, "No publish requests available");
  455. return false;
  456. }
  457. /* <-- The point of no return --> */
  458. UA_PublishResponse *response = &pre->response;
  459. UA_NotificationMessage *message = &response->notificationMessage;
  460. /* Set up the response. Note that this response has no related subscription id */
  461. response->responseHeader.timestamp = UA_DateTime_now();
  462. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS;
  463. response->subscriptionId = 0;
  464. response->moreNotifications = false;
  465. message->publishTime = response->responseHeader.timestamp;
  466. message->sequenceNumber = 0;
  467. response->availableSequenceNumbersSize = 0;
  468. /* Send the response */
  469. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  470. "Sending out a publish response triggered by too many publish requests");
  471. UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId,
  472. UA_MESSAGETYPE_MSG, response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  473. /* Free the response */
  474. UA_Array_delete(response->results, response->resultsSize, &UA_TYPES[UA_TYPES_UINT32]);
  475. UA_free(pre); /* no need for UA_PublishResponse_deleteMembers */
  476. return true;
  477. }
  478. UA_StatusCode
  479. Subscription_registerPublishCallback(UA_Server *server, UA_Subscription *sub) {
  480. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session,
  481. "Subscription %u | Register subscription "
  482. "publishing callback", sub->subscriptionId);
  483. if(sub->publishCallbackIsRegistered)
  484. return UA_STATUSCODE_GOOD;
  485. UA_StatusCode retval =
  486. UA_Server_addRepeatedCallback(server, (UA_ServerCallback)publishCallback,
  487. sub, (UA_UInt32)sub->publishingInterval, &sub->publishCallbackId);
  488. if(retval != UA_STATUSCODE_GOOD)
  489. return retval;
  490. sub->publishCallbackIsRegistered = true;
  491. return UA_STATUSCODE_GOOD;
  492. }
  493. UA_StatusCode
  494. Subscription_unregisterPublishCallback(UA_Server *server, UA_Subscription *sub) {
  495. UA_LOG_DEBUG_SESSION(server->config.logger, sub->session, "Subscription %u | "
  496. "Unregister subscription publishing callback", sub->subscriptionId);
  497. if(!sub->publishCallbackIsRegistered)
  498. return UA_STATUSCODE_GOOD;
  499. UA_StatusCode retval = UA_Server_removeRepeatedCallback(server, sub->publishCallbackId);
  500. if(retval != UA_STATUSCODE_GOOD)
  501. return retval;
  502. sub->publishCallbackIsRegistered = false;
  503. return UA_STATUSCODE_GOOD;
  504. }
  505. /* When the session has publish requests stored but the last subscription is
  506. * deleted... Send out empty responses */
  507. void
  508. UA_Subscription_answerPublishRequestsNoSubscription(UA_Server *server, UA_Session *session) {
  509. /* No session or there are remaining subscriptions */
  510. if(!session || LIST_FIRST(&session->serverSubscriptions))
  511. return;
  512. /* Send a response for every queued request */
  513. UA_PublishResponseEntry *pre;
  514. while((pre = UA_Session_dequeuePublishReq(session))) {
  515. UA_PublishResponse *response = &pre->response;
  516. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
  517. response->responseHeader.timestamp = UA_DateTime_now();
  518. UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId, UA_MESSAGETYPE_MSG,
  519. response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  520. UA_PublishResponse_deleteMembers(response);
  521. UA_free(pre);
  522. }
  523. }
  524. #endif /* UA_ENABLE_SUBSCRIPTIONS */