ua_subscription.c 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602
  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. sub->monitoredItemsSize = 0;
  53. /* Delete Retransmission Queue */
  54. UA_NotificationMessageEntry *nme, *nme_tmp;
  55. TAILQ_FOREACH_SAFE(nme, &sub->retransmissionQueue, listEntry, nme_tmp) {
  56. TAILQ_REMOVE(&sub->retransmissionQueue, nme, listEntry);
  57. UA_NotificationMessage_deleteMembers(&nme->message);
  58. UA_free(nme);
  59. --sub->session->totalRetransmissionQueueSize;
  60. --sub->retransmissionQueueSize;
  61. }
  62. UA_assert(sub->retransmissionQueueSize == 0);
  63. UA_LOG_INFO_SESSION(&server->config.logger, sub->session,
  64. "Subscription %u | Deleted the Subscription",
  65. sub->subscriptionId);
  66. }
  67. UA_MonitoredItem *
  68. UA_Subscription_getMonitoredItem(UA_Subscription *sub, UA_UInt32 monitoredItemId) {
  69. UA_MonitoredItem *mon;
  70. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  71. if(mon->monitoredItemId == monitoredItemId)
  72. break;
  73. }
  74. return mon;
  75. }
  76. UA_StatusCode
  77. UA_Subscription_deleteMonitoredItem(UA_Server *server, UA_Subscription *sub,
  78. UA_UInt32 monitoredItemId) {
  79. /* Find the MonitoredItem */
  80. UA_MonitoredItem *mon;
  81. LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
  82. if(mon->monitoredItemId == monitoredItemId)
  83. break;
  84. }
  85. if(!mon)
  86. return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  87. UA_LOG_INFO_SESSION(&server->config.logger, sub->session,
  88. "Subscription %u | MonitoredItem %i | "
  89. "Delete the MonitoredItem", sub->subscriptionId,
  90. mon->monitoredItemId);
  91. /* Remove the MonitoredItem */
  92. LIST_REMOVE(mon, listEntry);
  93. sub->monitoredItemsSize--;
  94. /* Remove content and delayed free */
  95. UA_MonitoredItem_delete(server, mon);
  96. return UA_STATUSCODE_GOOD;
  97. }
  98. void
  99. UA_Subscription_addMonitoredItem(UA_Subscription *sub, UA_MonitoredItem *newMon) {
  100. sub->monitoredItemsSize++;
  101. LIST_INSERT_HEAD(&sub->monitoredItems, newMon, listEntry);
  102. }
  103. static void
  104. removeOldestRetransmissionMessage(UA_Session *session) {
  105. UA_NotificationMessageEntry *oldestEntry = NULL;
  106. UA_Subscription *oldestSub = NULL;
  107. UA_Subscription *sub;
  108. LIST_FOREACH(sub, &session->serverSubscriptions, listEntry) {
  109. UA_NotificationMessageEntry *first =
  110. TAILQ_LAST(&sub->retransmissionQueue, ListOfNotificationMessages);
  111. if(!first)
  112. continue;
  113. if(!oldestEntry || oldestEntry->message.publishTime > first->message.publishTime) {
  114. oldestEntry = first;
  115. oldestSub = sub;
  116. }
  117. }
  118. UA_assert(oldestEntry);
  119. UA_assert(oldestSub);
  120. TAILQ_REMOVE(&oldestSub->retransmissionQueue, oldestEntry, listEntry);
  121. UA_NotificationMessage_deleteMembers(&oldestEntry->message);
  122. UA_free(oldestEntry);
  123. --session->totalRetransmissionQueueSize;
  124. --oldestSub->retransmissionQueueSize;
  125. }
  126. static void
  127. UA_Subscription_addRetransmissionMessage(UA_Server *server, UA_Subscription *sub,
  128. UA_NotificationMessageEntry *entry) {
  129. /* Release the oldest entry if there is not enough space */
  130. if(server->config.maxRetransmissionQueueSize > 0 &&
  131. sub->session->totalRetransmissionQueueSize >= server->config.maxRetransmissionQueueSize) {
  132. UA_LOG_WARNING_SESSION(&server->config.logger, sub->session, "Subscription %u | "
  133. "Retransmission queue overflow", sub->subscriptionId);
  134. removeOldestRetransmissionMessage(sub->session);
  135. }
  136. /* Add entry */
  137. TAILQ_INSERT_TAIL(&sub->retransmissionQueue, entry, listEntry);
  138. ++sub->session->totalRetransmissionQueueSize;
  139. ++sub->retransmissionQueueSize;
  140. }
  141. UA_StatusCode
  142. UA_Subscription_removeRetransmissionMessage(UA_Subscription *sub, UA_UInt32 sequenceNumber) {
  143. /* Find the retransmission message */
  144. UA_NotificationMessageEntry *entry;
  145. TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
  146. if(entry->message.sequenceNumber == sequenceNumber)
  147. break;
  148. }
  149. if(!entry)
  150. return UA_STATUSCODE_BADSEQUENCENUMBERUNKNOWN;
  151. /* Remove the retransmission message */
  152. TAILQ_REMOVE(&sub->retransmissionQueue, entry, listEntry);
  153. --sub->session->totalRetransmissionQueueSize;
  154. --sub->retransmissionQueueSize;
  155. UA_NotificationMessage_deleteMembers(&entry->message);
  156. UA_free(entry);
  157. return UA_STATUSCODE_GOOD;
  158. }
  159. static UA_StatusCode
  160. prepareNotificationMessage(UA_Server *server, UA_Subscription *sub,
  161. UA_NotificationMessage *message, size_t notifications) {
  162. UA_assert(notifications > 0);
  163. /* Allocate an ExtensionObject for events and data */
  164. message->notificationData = (UA_ExtensionObject*)
  165. UA_Array_new(2, &UA_TYPES[UA_TYPES_EXTENSIONOBJECT]);
  166. if(!message->notificationData)
  167. return UA_STATUSCODE_BADOUTOFMEMORY;
  168. message->notificationDataSize = 2;
  169. /* Pre-allocate DataChangeNotifications */
  170. size_t notificationDataIdx = 0;
  171. UA_DataChangeNotification *dcn = NULL;
  172. if(sub->dataChangeNotifications > 0) {
  173. dcn = UA_DataChangeNotification_new();
  174. if(!dcn) {
  175. UA_NotificationMessage_deleteMembers(message);
  176. return UA_STATUSCODE_BADOUTOFMEMORY;
  177. }
  178. message->notificationData->encoding = UA_EXTENSIONOBJECT_DECODED;
  179. message->notificationData->content.decoded.data = dcn;
  180. message->notificationData->content.decoded.type = &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION];
  181. size_t dcnSize = sub->dataChangeNotifications;
  182. if(dcnSize > notifications)
  183. dcnSize = notifications;
  184. dcn->monitoredItems = (UA_MonitoredItemNotification*)
  185. UA_Array_new(dcnSize, &UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION]);
  186. if(!dcn->monitoredItems) {
  187. UA_NotificationMessage_deleteMembers(message);
  188. return UA_STATUSCODE_BADOUTOFMEMORY;
  189. }
  190. dcn->monitoredItemsSize = dcnSize;
  191. notificationDataIdx++;
  192. }
  193. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  194. UA_EventNotificationList *enl = NULL;
  195. UA_StatusChangeNotification *scn = NULL;
  196. /* Pre-allocate either StatusChange or EventNotifications. Sending a
  197. * (single) StatusChangeNotification has priority. */
  198. if(sub->statusChangeNotifications > 0) {
  199. scn = UA_StatusChangeNotification_new();
  200. if(!scn) {
  201. UA_NotificationMessage_deleteMembers(message);
  202. return UA_STATUSCODE_BADOUTOFMEMORY;
  203. }
  204. message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
  205. message->notificationData[notificationDataIdx].content.decoded.data = scn;
  206. message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION];
  207. notificationDataIdx++;
  208. } else if(sub->eventNotifications > 0) {
  209. enl = UA_EventNotificationList_new();
  210. if(!enl) {
  211. UA_NotificationMessage_deleteMembers(message);
  212. return UA_STATUSCODE_BADOUTOFMEMORY;
  213. }
  214. message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
  215. message->notificationData[notificationDataIdx].content.decoded.data = enl;
  216. message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST];
  217. size_t enlSize = sub->eventNotifications;
  218. if(enlSize > notifications)
  219. enlSize = notifications;
  220. enl->events = (UA_EventFieldList*) UA_Array_new(enlSize, &UA_TYPES[UA_TYPES_EVENTFIELDLIST]);
  221. if(!enl->events) {
  222. UA_NotificationMessage_deleteMembers(message);
  223. return UA_STATUSCODE_BADOUTOFMEMORY;
  224. }
  225. enl->eventsSize = enlSize;
  226. notificationDataIdx++;
  227. }
  228. #endif
  229. UA_assert(notificationDataIdx > 0);
  230. message->notificationDataSize = notificationDataIdx;
  231. /* <-- The point of no return --> */
  232. size_t totalNotifications = 0; /* How many notifications were moved to the response overall? */
  233. size_t dcnPos = 0; /* How many DataChangeNotifications were put into the list? */
  234. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  235. size_t enlPos = 0; /* How many EventNotifications were moved into the list */
  236. #endif
  237. UA_Notification *notification, *notification_tmp;
  238. TAILQ_FOREACH_SAFE(notification, &sub->notificationQueue, globalEntry, notification_tmp) {
  239. if(totalNotifications >= notifications)
  240. break;
  241. UA_MonitoredItem *mon = notification->mon;
  242. /* Remove from the queues and decrease the counters */
  243. UA_Notification_dequeue(server, notification);
  244. /* Move the content to the response */
  245. if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
  246. UA_assert(dcn != NULL); /* Have at least one change notification */
  247. /* Move the content to the response */
  248. UA_MonitoredItemNotification *min = &dcn->monitoredItems[dcnPos];
  249. min->clientHandle = mon->clientHandle;
  250. min->value = notification->data.value;
  251. UA_DataValue_init(&notification->data.value); /* Reset after the value has been moved */
  252. dcnPos++;
  253. }
  254. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  255. else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY && scn) {
  256. // TODO: Handling of StatusChangeNotifications
  257. scn = NULL; /* At most one per PublishReponse */
  258. } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY && enl) {
  259. UA_assert(enl != NULL); /* Have at least one event notification */
  260. /* Move the content to the response */
  261. UA_EventFieldList *efl = &enl->events[enlPos];
  262. *efl = notification->data.event.fields;
  263. UA_EventFieldList_init(&notification->data.event.fields);
  264. efl->clientHandle = mon->clientHandle;
  265. enlPos++;
  266. }
  267. #endif
  268. else {
  269. UA_Notification_delete(notification);
  270. continue; /* Unknown type. Nothing to do */
  271. }
  272. UA_Notification_delete(notification);
  273. totalNotifications++;
  274. }
  275. /* Set sizes */
  276. if(dcn) {
  277. dcn->monitoredItemsSize = dcnPos;
  278. if(dcnPos == 0) {
  279. UA_free(dcn->monitoredItems);
  280. dcn->monitoredItems = NULL;
  281. }
  282. }
  283. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  284. if(enl) {
  285. enl->eventsSize = enlPos;
  286. if(enlPos == 0) {
  287. UA_free(enl->events);
  288. enl->events = NULL;
  289. }
  290. }
  291. #endif
  292. return UA_STATUSCODE_GOOD;
  293. }
  294. /* According to OPC Unified Architecture, Part 4 5.13.1.1 i) The value 0 is
  295. * never used for the sequence number */
  296. static UA_UInt32
  297. UA_Subscription_nextSequenceNumber(UA_UInt32 sequenceNumber) {
  298. UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
  299. if(nextSequenceNumber == 0)
  300. nextSequenceNumber = 1;
  301. return nextSequenceNumber;
  302. }
  303. static void
  304. publishCallback(UA_Server *server, UA_Subscription *sub) {
  305. sub->readyNotifications = sub->notificationQueueSize;
  306. UA_Subscription_publish(server, sub);
  307. }
  308. void
  309. UA_Subscription_publish(UA_Server *server, UA_Subscription *sub) {
  310. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session, "Subscription %u | "
  311. "Publish Callback", sub->subscriptionId);
  312. /* Dequeue a response */
  313. UA_PublishResponseEntry *pre = UA_Session_dequeuePublishReq(sub->session);
  314. if(pre) {
  315. sub->currentLifetimeCount = 0; /* Reset the LifetimeCounter */
  316. } else {
  317. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
  318. "Subscription %u | The publish queue is empty",
  319. sub->subscriptionId);
  320. ++sub->currentLifetimeCount;
  321. if(sub->currentLifetimeCount > sub->lifeTimeCount) {
  322. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
  323. "Subscription %u | End of lifetime "
  324. "for subscription", sub->subscriptionId);
  325. UA_Session_deleteSubscription(server, sub->session, sub->subscriptionId);
  326. /* TODO: send a StatusChangeNotification with Bad_Timeout */
  327. return;
  328. }
  329. }
  330. /* If there are several late publish responses... */
  331. if(sub->readyNotifications > sub->notificationQueueSize)
  332. sub->readyNotifications = sub->notificationQueueSize;
  333. /* Count the available notifications */
  334. UA_UInt32 notifications = sub->readyNotifications;
  335. if(!sub->publishingEnabled)
  336. notifications = 0;
  337. UA_Boolean moreNotifications = false;
  338. if(notifications > sub->notificationsPerPublish) {
  339. notifications = sub->notificationsPerPublish;
  340. moreNotifications = true;
  341. }
  342. /* Return if no notifications and no keepalive */
  343. if(notifications == 0) {
  344. ++sub->currentKeepAliveCount;
  345. if(sub->currentKeepAliveCount < sub->maxKeepAliveCount) {
  346. if(pre)
  347. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  348. return;
  349. }
  350. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
  351. "Subscription %u | Sending a KeepAlive",
  352. sub->subscriptionId);
  353. }
  354. /* We want to send a response. Is the channel open? */
  355. UA_SecureChannel *channel = sub->session->header.channel;
  356. if(!channel || !pre) {
  357. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
  358. "Subscription %u | Want to send a publish response but can't. "
  359. "The subscription is late.", sub->subscriptionId);
  360. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  361. if(pre)
  362. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  363. return;
  364. }
  365. /* Prepare the response */
  366. UA_PublishResponse *response = &pre->response;
  367. UA_NotificationMessage *message = &response->notificationMessage;
  368. UA_NotificationMessageEntry *retransmission = NULL;
  369. if(notifications > 0) {
  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. /* Prepare the response */
  381. UA_StatusCode retval = prepareNotificationMessage(server, sub, message, notifications);
  382. if(retval != UA_STATUSCODE_GOOD) {
  383. UA_LOG_WARNING_SESSION(&server->config.logger, sub->session,
  384. "Subscription %u | Could not prepare the notification message. "
  385. "The subscription is late.", sub->subscriptionId);
  386. UA_free(retransmission);
  387. sub->state = UA_SUBSCRIPTIONSTATE_LATE;
  388. UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
  389. return;
  390. }
  391. }
  392. /* <-- The point of no return --> */
  393. /* Adjust the number of ready notifications */
  394. UA_assert(sub->readyNotifications >= notifications);
  395. sub->readyNotifications -= notifications;
  396. /* Set up the response */
  397. response->responseHeader.timestamp = UA_DateTime_now();
  398. response->subscriptionId = sub->subscriptionId;
  399. response->moreNotifications = moreNotifications;
  400. message->publishTime = response->responseHeader.timestamp;
  401. /* Set sequence number to message. Started at 1 which is given
  402. * during creating a new subscription. The 1 is required for
  403. * initial publish response with or without an monitored item. */
  404. message->sequenceNumber = sub->nextSequenceNumber;
  405. if(notifications > 0) {
  406. /* Put the notification message into the retransmission queue. This
  407. * needs to be done here, so that the message itself is included in the
  408. * available sequence numbers for acknowledgement. */
  409. retransmission->message = response->notificationMessage;
  410. UA_Subscription_addRetransmissionMessage(server, sub, retransmission);
  411. /* Only if a notification was created, the sequence number must be increased.
  412. * For a keepalive the sequence number can be reused. */
  413. sub->nextSequenceNumber = UA_Subscription_nextSequenceNumber(sub->nextSequenceNumber);
  414. }
  415. /* Get the available sequence numbers from the retransmission queue */
  416. size_t available = sub->retransmissionQueueSize;
  417. UA_STACKARRAY(UA_UInt32, seqNumbers, available);
  418. if(available > 0) {
  419. response->availableSequenceNumbers = seqNumbers;
  420. response->availableSequenceNumbersSize = available;
  421. size_t i = 0;
  422. UA_NotificationMessageEntry *nme;
  423. TAILQ_FOREACH(nme, &sub->retransmissionQueue, listEntry) {
  424. response->availableSequenceNumbers[i] = nme->message.sequenceNumber;
  425. ++i;
  426. }
  427. }
  428. /* Send the response */
  429. UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
  430. "Subscription %u | Sending out a publish response "
  431. "with %u notifications", sub->subscriptionId,
  432. (UA_UInt32)notifications);
  433. UA_SecureChannel_sendSymmetricMessage(sub->session->header.channel, pre->requestId,
  434. UA_MESSAGETYPE_MSG, response,
  435. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  436. /* Reset subscription state to normal */
  437. sub->state = UA_SUBSCRIPTIONSTATE_NORMAL;
  438. sub->currentKeepAliveCount = 0;
  439. /* Free the response */
  440. UA_Array_delete(response->results, response->resultsSize, &UA_TYPES[UA_TYPES_UINT32]);
  441. UA_free(pre); /* No need for UA_PublishResponse_deleteMembers */
  442. /* Repeat sending responses if there are more notifications to send */
  443. if(moreNotifications)
  444. UA_Subscription_publish(server, sub);
  445. }
  446. UA_Boolean
  447. UA_Subscription_reachedPublishReqLimit(UA_Server *server, UA_Session *session) {
  448. UA_LOG_DEBUG_SESSION(&server->config.logger, session,
  449. "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. void
  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;
  499. UA_Server_removeRepeatedCallback(server, sub->publishCallbackId);
  500. sub->publishCallbackIsRegistered = false;
  501. }
  502. /* When the session has publish requests stored but the last subscription is
  503. * deleted... Send out empty responses */
  504. void
  505. UA_Subscription_answerPublishRequestsNoSubscription(UA_Server *server, UA_Session *session) {
  506. /* No session or there are remaining subscriptions */
  507. if(!session || LIST_FIRST(&session->serverSubscriptions))
  508. return;
  509. /* Send a response for every queued request */
  510. UA_PublishResponseEntry *pre;
  511. while((pre = UA_Session_dequeuePublishReq(session))) {
  512. UA_PublishResponse *response = &pre->response;
  513. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
  514. response->responseHeader.timestamp = UA_DateTime_now();
  515. UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId, UA_MESSAGETYPE_MSG,
  516. response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  517. UA_PublishResponse_deleteMembers(response);
  518. UA_free(pre);
  519. }
  520. }
  521. #endif /* UA_ENABLE_SUBSCRIPTIONS */