ua_services_subscription.c 25 KB

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