ua_services_subscription.c 25 KB

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