ua_subscription.c 24 KB

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