ua_services_subscription.c 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585
  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. #include "ua_server_internal.h"
  5. #include "ua_services.h"
  6. #include "ua_subscription.h"
  7. #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
  8. #define UA_BOUNDEDVALUE_SETWBOUNDS(BOUNDS, SRC, DST) { \
  9. if(SRC > BOUNDS.max) DST = BOUNDS.max; \
  10. else if(SRC < BOUNDS.min) DST = BOUNDS.min; \
  11. else DST = SRC; \
  12. }
  13. static void
  14. setSubscriptionSettings(UA_Server *server, UA_Subscription *subscription,
  15. UA_Double requestedPublishingInterval,
  16. UA_UInt32 requestedLifetimeCount,
  17. UA_UInt32 requestedMaxKeepAliveCount,
  18. UA_UInt32 maxNotificationsPerPublish, UA_Byte priority) {
  19. /* deregister the callback if required */
  20. UA_StatusCode retval = Subscription_unregisterPublishCallback(server, subscription);
  21. if(retval != UA_STATUSCODE_GOOD)
  22. UA_LOG_DEBUG_SESSION(server->config.logger, subscription->session, "Subscription %u | "
  23. "Could not unregister publish callback with error code %s",
  24. subscription->subscriptionID, UA_StatusCode_name(retval));
  25. /* re-parameterize the subscription */
  26. subscription->publishingInterval = requestedPublishingInterval;
  27. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.publishingIntervalLimits,
  28. requestedPublishingInterval, subscription->publishingInterval);
  29. /* check for nan*/
  30. if(requestedPublishingInterval != requestedPublishingInterval)
  31. subscription->publishingInterval = server->config.publishingIntervalLimits.min;
  32. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.keepAliveCountLimits,
  33. requestedMaxKeepAliveCount, subscription->maxKeepAliveCount);
  34. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.lifeTimeCountLimits,
  35. requestedLifetimeCount, subscription->lifeTimeCount);
  36. if(subscription->lifeTimeCount < 3 * subscription->maxKeepAliveCount)
  37. subscription->lifeTimeCount = 3 * subscription->maxKeepAliveCount;
  38. subscription->notificationsPerPublish = maxNotificationsPerPublish;
  39. if(maxNotificationsPerPublish == 0 ||
  40. maxNotificationsPerPublish > server->config.maxNotificationsPerPublish)
  41. subscription->notificationsPerPublish = server->config.maxNotificationsPerPublish;
  42. subscription->priority = priority;
  43. retval = Subscription_registerPublishCallback(server, subscription);
  44. if(retval != UA_STATUSCODE_GOOD)
  45. UA_LOG_DEBUG_SESSION(server->config.logger, subscription->session, "Subscription %u | "
  46. "Could not register publish callback with error code %s",
  47. subscription->subscriptionID, UA_StatusCode_name(retval));
  48. }
  49. void
  50. Service_CreateSubscription(UA_Server *server, UA_Session *session,
  51. const UA_CreateSubscriptionRequest *request,
  52. UA_CreateSubscriptionResponse *response) {
  53. /* Create the subscription */
  54. UA_Subscription *newSubscription = UA_Subscription_new(session, response->subscriptionId);
  55. if(!newSubscription) {
  56. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  57. "Processing CreateSubscriptionRequest failed");
  58. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  59. return;
  60. }
  61. newSubscription->subscriptionID = UA_Session_getUniqueSubscriptionID(session);
  62. UA_Session_addSubscription(session, newSubscription);
  63. /* Set the subscription parameters */
  64. newSubscription->publishingEnabled = request->publishingEnabled;
  65. setSubscriptionSettings(server, newSubscription, request->requestedPublishingInterval,
  66. request->requestedLifetimeCount, request->requestedMaxKeepAliveCount,
  67. request->maxNotificationsPerPublish, request->priority);
  68. newSubscription->currentKeepAliveCount = newSubscription->maxKeepAliveCount; /* set settings first */
  69. /* Prepare the response */
  70. response->subscriptionId = newSubscription->subscriptionID;
  71. response->revisedPublishingInterval = newSubscription->publishingInterval;
  72. response->revisedLifetimeCount = newSubscription->lifeTimeCount;
  73. response->revisedMaxKeepAliveCount = newSubscription->maxKeepAliveCount;
  74. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  75. "CreateSubscriptionRequest: Created Subscription %u "
  76. "with a publishing interval of %f ms", response->subscriptionId,
  77. newSubscription->publishingInterval);
  78. }
  79. void
  80. Service_ModifySubscription(UA_Server *server, UA_Session *session,
  81. const UA_ModifySubscriptionRequest *request,
  82. UA_ModifySubscriptionResponse *response) {
  83. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  84. "Processing ModifySubscriptionRequest");
  85. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  86. if(!sub) {
  87. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  88. return;
  89. }
  90. setSubscriptionSettings(server, sub, request->requestedPublishingInterval,
  91. request->requestedLifetimeCount, request->requestedMaxKeepAliveCount,
  92. request->maxNotificationsPerPublish, request->priority);
  93. sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  94. response->revisedPublishingInterval = sub->publishingInterval;
  95. response->revisedLifetimeCount = sub->lifeTimeCount;
  96. response->revisedMaxKeepAliveCount = sub->maxKeepAliveCount;
  97. }
  98. void
  99. Service_SetPublishingMode(UA_Server *server, UA_Session *session,
  100. const UA_SetPublishingModeRequest *request,
  101. UA_SetPublishingModeResponse *response) {
  102. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  103. "Processing SetPublishingModeRequest");
  104. if(request->subscriptionIdsSize <= 0) {
  105. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  106. return;
  107. }
  108. size_t size = request->subscriptionIdsSize;
  109. response->results = (UA_StatusCode *)UA_Array_new(size, &UA_TYPES[UA_TYPES_STATUSCODE]);
  110. if(!response->results) {
  111. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  112. return;
  113. }
  114. response->resultsSize = size;
  115. for(size_t i = 0; i < size; ++i) {
  116. UA_Subscription *sub =
  117. UA_Session_getSubscriptionByID(session, request->subscriptionIds[i]);
  118. if(!sub) {
  119. response->results[i] = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  120. continue;
  121. }
  122. if(sub->publishingEnabled != request->publishingEnabled) {
  123. sub->publishingEnabled = request->publishingEnabled;
  124. sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  125. }
  126. }
  127. }
  128. static void
  129. setMonitoredItemSettings(UA_Server *server, UA_MonitoredItem *mon,
  130. UA_MonitoringMode monitoringMode,
  131. const UA_MonitoringParameters *params) {
  132. MonitoredItem_unregisterSampleCallback(server, mon);
  133. mon->monitoringMode = monitoringMode;
  134. /* ClientHandle */
  135. mon->clientHandle = params->clientHandle;
  136. /* SamplingInterval */
  137. UA_Double samplingInterval = params->samplingInterval;
  138. if(mon->attributeID == UA_ATTRIBUTEID_VALUE) {
  139. const UA_VariableNode *vn = (const UA_VariableNode*)
  140. UA_NodeStore_get(server->nodestore, &mon->monitoredNodeId);
  141. if(vn && vn->nodeClass == UA_NODECLASS_VARIABLE &&
  142. samplingInterval < vn->minimumSamplingInterval)
  143. samplingInterval = vn->minimumSamplingInterval;
  144. } else if(mon->attributeID == UA_ATTRIBUTEID_EVENTNOTIFIER) {
  145. /* TODO: events should not need a samplinginterval */
  146. samplingInterval = 10000.0f; // 10 seconds to reduce the load
  147. }
  148. mon->samplingInterval = samplingInterval;
  149. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.samplingIntervalLimits,
  150. samplingInterval, mon->samplingInterval);
  151. if(samplingInterval != samplingInterval) /* Check for nan */
  152. mon->samplingInterval = server->config.samplingIntervalLimits.min;
  153. /* Filter */
  154. if(params->filter.encoding != UA_EXTENSIONOBJECT_DECODED ||
  155. params->filter.content.decoded.type != &UA_TYPES[UA_TYPES_DATACHANGEFILTER]) {
  156. /* Default: Trigger only on the value and the statuscode */
  157. mon->trigger = UA_DATACHANGETRIGGER_STATUSVALUE;
  158. } else {
  159. UA_DataChangeFilter *filter = (UA_DataChangeFilter *)params->filter.content.decoded.data;
  160. mon->trigger = filter->trigger;
  161. }
  162. /* QueueSize */
  163. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.queueSizeLimits,
  164. params->queueSize, mon->maxQueueSize);
  165. /* DiscardOldest */
  166. mon->discardOldest = params->discardOldest;
  167. /* Register sample callback if reporting is enabled */
  168. if(monitoringMode == UA_MONITORINGMODE_REPORTING)
  169. MonitoredItem_registerSampleCallback(server, mon);
  170. }
  171. static const UA_String binaryEncoding = {sizeof("Default Binary")-1, (UA_Byte*)"Default Binary"};
  172. /* static const UA_String xmlEncoding = {sizeof("Default Xml")-1, (UA_Byte*)"Default Xml"}; */
  173. static void
  174. Service_CreateMonitoredItems_single(UA_Server *server, UA_Session *session,
  175. UA_Subscription *sub,
  176. const UA_TimestampsToReturn timestampsToReturn,
  177. const UA_MonitoredItemCreateRequest *request,
  178. UA_MonitoredItemCreateResult *result) {
  179. /* Make an example read to get errors in the itemToMonitor. Allow return
  180. * codes "good" and "uncertain", as well as a list of statuscodes that might
  181. * be repaired inside the data source. */
  182. UA_DataValue v;
  183. UA_DataValue_init(&v);
  184. Service_Read_single(server, session, timestampsToReturn, &request->itemToMonitor, &v);
  185. if(v.hasStatus && (v.status >> 30) > 1 &&
  186. v.status != UA_STATUSCODE_BADRESOURCEUNAVAILABLE &&
  187. v.status != UA_STATUSCODE_BADCOMMUNICATIONERROR &&
  188. v.status != UA_STATUSCODE_BADWAITINGFORINITIALDATA) {
  189. result->statusCode = v.status;
  190. UA_DataValue_deleteMembers(&v);
  191. return;
  192. }
  193. UA_DataValue_deleteMembers(&v);
  194. /* Check if the encoding is supported */
  195. if(request->itemToMonitor.dataEncoding.name.length > 0 &&
  196. (!UA_String_equal(&binaryEncoding, &request->itemToMonitor.dataEncoding.name) ||
  197. request->itemToMonitor.dataEncoding.namespaceIndex != 0)) {
  198. result->statusCode = UA_STATUSCODE_BADDATAENCODINGUNSUPPORTED;
  199. return;
  200. }
  201. /* Check if the encoding is set for a value */
  202. if(request->itemToMonitor.attributeId != UA_ATTRIBUTEID_VALUE &&
  203. request->itemToMonitor.dataEncoding.name.length > 0) {
  204. result->statusCode = UA_STATUSCODE_BADDATAENCODINGINVALID;
  205. return;
  206. }
  207. /* Create the monitoreditem */
  208. UA_MonitoredItem *newMon = UA_MonitoredItem_new();
  209. if(!newMon) {
  210. result->statusCode = UA_STATUSCODE_BADOUTOFMEMORY;
  211. return;
  212. }
  213. UA_StatusCode retval = UA_NodeId_copy(&request->itemToMonitor.nodeId,
  214. &newMon->monitoredNodeId);
  215. if(retval != UA_STATUSCODE_GOOD) {
  216. result->statusCode = retval;
  217. MonitoredItem_delete(server, newMon);
  218. return;
  219. }
  220. newMon->subscription = sub;
  221. newMon->attributeID = request->itemToMonitor.attributeId;
  222. newMon->itemId = ++(sub->lastMonitoredItemId);
  223. newMon->timestampsToReturn = timestampsToReturn;
  224. setMonitoredItemSettings(server, newMon, request->monitoringMode,
  225. &request->requestedParameters);
  226. LIST_INSERT_HEAD(&sub->monitoredItems, newMon, listEntry);
  227. /* Create the first sample */
  228. if(request->monitoringMode == UA_MONITORINGMODE_REPORTING)
  229. UA_MoniteredItem_SampleCallback(server, newMon);
  230. /* Prepare the response */
  231. UA_String_copy(&request->itemToMonitor.indexRange, &newMon->indexRange);
  232. result->revisedSamplingInterval = newMon->samplingInterval;
  233. result->revisedQueueSize = newMon->maxQueueSize;
  234. result->monitoredItemId = newMon->itemId;
  235. }
  236. void
  237. Service_CreateMonitoredItems(UA_Server *server, UA_Session *session,
  238. const UA_CreateMonitoredItemsRequest *request,
  239. UA_CreateMonitoredItemsResponse *response) {
  240. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  241. "Processing CreateMonitoredItemsRequest");
  242. /* Check if the timestampstoreturn is valid */
  243. if(request->timestampsToReturn > UA_TIMESTAMPSTORETURN_NEITHER) {
  244. response->responseHeader.serviceResult = UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID;
  245. return;
  246. }
  247. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  248. if(!sub) {
  249. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  250. return;
  251. }
  252. /* Reset the subscription lifetime */
  253. sub->currentLifetimeCount = 0;
  254. if(request->itemsToCreateSize <= 0) {
  255. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  256. return;
  257. }
  258. response->results = (UA_MonitoredItemCreateResult *)UA_Array_new(request->itemsToCreateSize,
  259. &UA_TYPES[UA_TYPES_MONITOREDITEMCREATERESULT]);
  260. if(!response->results) {
  261. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  262. return;
  263. }
  264. response->resultsSize = request->itemsToCreateSize;
  265. for(size_t i = 0; i < request->itemsToCreateSize; ++i)
  266. Service_CreateMonitoredItems_single(server, session, sub, request->timestampsToReturn,
  267. &request->itemsToCreate[i], &response->results[i]);
  268. }
  269. static void
  270. Service_ModifyMonitoredItems_single(UA_Server *server, UA_Session *session, UA_Subscription *sub,
  271. const UA_MonitoredItemModifyRequest *request,
  272. UA_MonitoredItemModifyResult *result) {
  273. UA_MonitoredItem *mon = UA_Subscription_getMonitoredItem(sub, request->monitoredItemId);
  274. if(!mon) {
  275. result->statusCode = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  276. return;
  277. }
  278. setMonitoredItemSettings(server, mon, mon->monitoringMode,
  279. &request->requestedParameters);
  280. result->revisedSamplingInterval = mon->samplingInterval;
  281. result->revisedQueueSize = mon->maxQueueSize;
  282. }
  283. void Service_ModifyMonitoredItems(UA_Server *server, UA_Session *session,
  284. const UA_ModifyMonitoredItemsRequest *request,
  285. UA_ModifyMonitoredItemsResponse *response) {
  286. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  287. "Processing ModifyMonitoredItemsRequest");
  288. /* check if the timestampstoreturn is valid */
  289. if(request->timestampsToReturn > UA_TIMESTAMPSTORETURN_NEITHER) {
  290. response->responseHeader.serviceResult = UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID;
  291. return;
  292. }
  293. /* Get the subscription */
  294. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  295. if(!sub) {
  296. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  297. return;
  298. }
  299. /* Reset the subscription lifetime */
  300. sub->currentLifetimeCount = 0;
  301. if(request->itemsToModifySize <= 0) {
  302. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  303. return;
  304. }
  305. response->results = (UA_MonitoredItemModifyResult *)UA_Array_new(request->itemsToModifySize,
  306. &UA_TYPES[UA_TYPES_MONITOREDITEMMODIFYRESULT]);
  307. if(!response->results) {
  308. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  309. return;
  310. }
  311. response->resultsSize = request->itemsToModifySize;
  312. for(size_t i = 0; i < request->itemsToModifySize; ++i)
  313. Service_ModifyMonitoredItems_single(server, session, sub, &request->itemsToModify[i],
  314. &response->results[i]);
  315. }
  316. void Service_SetMonitoringMode(UA_Server *server, UA_Session *session,
  317. const UA_SetMonitoringModeRequest *request,
  318. UA_SetMonitoringModeResponse *response) {
  319. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing SetMonitoringMode");
  320. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  321. if(!sub) {
  322. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  323. return;
  324. }
  325. if(request->monitoredItemIdsSize == 0) {
  326. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  327. return;
  328. }
  329. response->results = (UA_StatusCode *)UA_Array_new(request->monitoredItemIdsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
  330. if(!response->results) {
  331. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  332. return;
  333. }
  334. response->resultsSize = request->monitoredItemIdsSize;
  335. for(size_t i = 0; i < response->resultsSize; ++i) {
  336. UA_MonitoredItem *mon =
  337. UA_Subscription_getMonitoredItem(sub, request->monitoredItemIds[i]);
  338. if(!mon) {
  339. response->results[i] = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  340. continue;
  341. }
  342. if(request->monitoringMode == mon->monitoringMode)
  343. continue;
  344. mon->monitoringMode = request->monitoringMode;
  345. if(mon->monitoringMode == UA_MONITORINGMODE_REPORTING)
  346. MonitoredItem_registerSampleCallback(server, mon);
  347. else
  348. MonitoredItem_unregisterSampleCallback(server, mon);
  349. }
  350. }
  351. /* TODO: Unify with senderror in ua_server_binary.c */
  352. static void
  353. subscriptionSendError(UA_SecureChannel *channel, UA_UInt32 requestHandle,
  354. UA_UInt32 requestId, UA_StatusCode error) {
  355. UA_PublishResponse err_response;
  356. UA_PublishResponse_init(&err_response);
  357. err_response.responseHeader.requestHandle = requestHandle;
  358. err_response.responseHeader.timestamp = UA_DateTime_now();
  359. err_response.responseHeader.serviceResult = error;
  360. UA_SecureChannel_sendBinaryMessage(channel, requestId, &err_response,
  361. &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  362. }
  363. void
  364. Service_Publish(UA_Server *server, UA_Session *session,
  365. const UA_PublishRequest *request, UA_UInt32 requestId) {
  366. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing PublishRequest");
  367. /* Return an error if the session has no subscription */
  368. if(LIST_EMPTY(&session->serverSubscriptions)) {
  369. subscriptionSendError(session->channel, request->requestHeader.requestHandle,
  370. requestId, UA_STATUSCODE_BADNOSUBSCRIPTION);
  371. return;
  372. }
  373. UA_PublishResponseEntry *entry = (UA_PublishResponseEntry *)UA_malloc(sizeof(UA_PublishResponseEntry));
  374. if(!entry) {
  375. subscriptionSendError(session->channel, requestId,
  376. request->requestHeader.requestHandle, UA_STATUSCODE_BADOUTOFMEMORY);
  377. return;
  378. }
  379. entry->requestId = requestId;
  380. /* Build the response */
  381. UA_PublishResponse *response = &entry->response;
  382. UA_PublishResponse_init(response);
  383. response->responseHeader.requestHandle = request->requestHeader.requestHandle;
  384. if(request->subscriptionAcknowledgementsSize > 0) {
  385. response->results = (UA_StatusCode *)UA_Array_new(request->subscriptionAcknowledgementsSize,
  386. &UA_TYPES[UA_TYPES_STATUSCODE]);
  387. if(!response->results) {
  388. UA_free(entry);
  389. subscriptionSendError(session->channel, requestId,
  390. request->requestHeader.requestHandle,
  391. UA_STATUSCODE_BADOUTOFMEMORY);
  392. return;
  393. }
  394. response->resultsSize = request->subscriptionAcknowledgementsSize;
  395. }
  396. /* Delete Acknowledged Subscription Messages */
  397. for(size_t i = 0; i < request->subscriptionAcknowledgementsSize; ++i) {
  398. UA_SubscriptionAcknowledgement *ack = &request->subscriptionAcknowledgements[i];
  399. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, ack->subscriptionId);
  400. if(!sub) {
  401. response->results[i] = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  402. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  403. "Cannot process acknowledgements subscription %u",
  404. ack->subscriptionId);
  405. continue;
  406. }
  407. /* Remove the acked transmission from the retransmission queue */
  408. response->results[i] = UA_Subscription_removeRetransmissionMessage(sub, ack->sequenceNumber);
  409. }
  410. /* Queue the publish response */
  411. SIMPLEQ_INSERT_TAIL(&session->responseQueue, entry, listEntry);
  412. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Queued a publication message",
  413. session->authenticationToken.identifier.numeric);
  414. /* Answer immediately to a late subscription */
  415. UA_Subscription *immediate;
  416. LIST_FOREACH(immediate, &session->serverSubscriptions, listEntry) {
  417. if(immediate->state == UA_SUBSCRIPTIONSTATE_LATE) {
  418. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Subscription %u | "
  419. "Response on a late subscription", immediate->subscriptionID,
  420. session->authenticationToken.identifier.numeric);
  421. UA_Subscription_publishCallback(server, immediate);
  422. break;
  423. }
  424. }
  425. }
  426. void
  427. Service_DeleteSubscriptions(UA_Server *server, UA_Session *session,
  428. const UA_DeleteSubscriptionsRequest *request,
  429. UA_DeleteSubscriptionsResponse *response) {
  430. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  431. "Processing DeleteSubscriptionsRequest");
  432. if(request->subscriptionIdsSize == 0) {
  433. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  434. return;
  435. }
  436. response->results = (UA_StatusCode *)UA_malloc(sizeof(UA_StatusCode) * request->subscriptionIdsSize);
  437. if(!response->results) {
  438. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  439. return;
  440. }
  441. response->resultsSize = request->subscriptionIdsSize;
  442. for(size_t i = 0; i < request->subscriptionIdsSize; ++i) {
  443. response->results[i] = UA_Session_deleteSubscription(server, session, request->subscriptionIds[i]);
  444. if(response->results[i] == UA_STATUSCODE_GOOD) {
  445. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Subscription %u | "
  446. "Subscription deleted", request->subscriptionIds[i]);
  447. } else {
  448. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Deleting Subscription with Id "
  449. "%u failed with error code 0x%08x", request->subscriptionIds[i],
  450. response->results[i]);
  451. }
  452. }
  453. /* Send dangling publish responses in a delayed callback if the last
  454. * subscription was removed */
  455. if(LIST_FIRST(&session->serverSubscriptions))
  456. return;
  457. UA_NodeId *sessionToken = UA_NodeId_new();
  458. if(!sessionToken)
  459. return;
  460. UA_NodeId_copy(&session->authenticationToken, sessionToken);
  461. UA_Server_delayedCallback(server,
  462. (UA_ServerCallback)UA_Subscription_answerPublishRequestsNoSubscription,
  463. sessionToken);
  464. }
  465. void Service_DeleteMonitoredItems(UA_Server *server, UA_Session *session,
  466. const UA_DeleteMonitoredItemsRequest *request,
  467. UA_DeleteMonitoredItemsResponse *response) {
  468. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  469. "Processing DeleteMonitoredItemsRequest");
  470. if(request->monitoredItemIdsSize == 0) {
  471. response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
  472. return;
  473. }
  474. /* Get the subscription */
  475. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  476. if(!sub) {
  477. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  478. return;
  479. }
  480. /* Reset the subscription lifetime */
  481. sub->currentLifetimeCount = 0;
  482. response->results = (UA_StatusCode *)UA_malloc(sizeof(UA_StatusCode) * request->monitoredItemIdsSize);
  483. if(!response->results) {
  484. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  485. return;
  486. }
  487. response->resultsSize = request->monitoredItemIdsSize;
  488. for(size_t i = 0; i < request->monitoredItemIdsSize; ++i)
  489. response->results[i] = UA_Subscription_deleteMonitoredItem(server, sub, request->monitoredItemIds[i]);
  490. }
  491. void Service_Republish(UA_Server *server, UA_Session *session, const UA_RepublishRequest *request,
  492. UA_RepublishResponse *response) {
  493. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing RepublishRequest");
  494. /* get the subscription */
  495. UA_Subscription *sub = UA_Session_getSubscriptionByID(session, request->subscriptionId);
  496. if (!sub) {
  497. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  498. return;
  499. }
  500. /* Reset the subscription lifetime */
  501. sub->currentLifetimeCount = 0;
  502. /* Find the notification in the retransmission queue */
  503. UA_NotificationMessageEntry *entry;
  504. TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
  505. if(entry->message.sequenceNumber == request->retransmitSequenceNumber)
  506. break;
  507. }
  508. if(entry)
  509. response->responseHeader.serviceResult =
  510. UA_NotificationMessage_copy(&entry->message, &response->notificationMessage);
  511. else
  512. response->responseHeader.serviceResult = UA_STATUSCODE_BADMESSAGENOTAVAILABLE;
  513. }
  514. #endif /* UA_ENABLE_SUBSCRIPTIONS */