ua_services_subscription.c 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842
  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 2014-2017 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
  6. * Copyright 2016-2017 (c) Florian Palm
  7. * Copyright 2015 (c) Chris Iatrou
  8. * Copyright 2015-2016 (c) Sten Grüner
  9. * Copyright 2015-2016 (c) Oleksiy Vasylyev
  10. * Copyright 2017 (c) Stefan Profanter, fortiss GmbH
  11. * Copyright 2018 (c) Ari Breitkreuz, fortiss GmbH
  12. * Copyright 2017 (c) Mattias Bornhager
  13. * Copyright 2017 (c) Henrik Norrman
  14. * Copyright 2017-2018 (c) Thomas Stalder, Blue Time Concept SA
  15. * Copyright 2018 (c) Fabian Arndt, Root-Core
  16. */
  17. #include "ua_server_internal.h"
  18. #include "ua_services.h"
  19. #include "ua_subscription.h"
  20. #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
  21. #define UA_BOUNDEDVALUE_SETWBOUNDS(BOUNDS, SRC, DST) { \
  22. if(SRC > BOUNDS.max) DST = BOUNDS.max; \
  23. else if(SRC < BOUNDS.min) DST = BOUNDS.min; \
  24. else DST = SRC; \
  25. }
  26. static void
  27. setSubscriptionSettings(UA_Server *server, UA_Subscription *subscription,
  28. UA_Double requestedPublishingInterval,
  29. UA_UInt32 requestedLifetimeCount,
  30. UA_UInt32 requestedMaxKeepAliveCount,
  31. UA_UInt32 maxNotificationsPerPublish, UA_Byte priority) {
  32. /* deregister the callback if required */
  33. UA_StatusCode retval = Subscription_unregisterPublishCallback(server, subscription);
  34. if(retval != UA_STATUSCODE_GOOD)
  35. UA_LOG_DEBUG_SESSION(server->config.logger, subscription->session,
  36. "Subscription %u | Could not unregister publish callback with error code %s",
  37. subscription->subscriptionId, UA_StatusCode_name(retval));
  38. /* re-parameterize the subscription */
  39. subscription->publishingInterval = requestedPublishingInterval;
  40. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.publishingIntervalLimits,
  41. requestedPublishingInterval, subscription->publishingInterval);
  42. /* check for nan*/
  43. if(requestedPublishingInterval != requestedPublishingInterval)
  44. subscription->publishingInterval = server->config.publishingIntervalLimits.min;
  45. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.keepAliveCountLimits,
  46. requestedMaxKeepAliveCount, subscription->maxKeepAliveCount);
  47. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.lifeTimeCountLimits,
  48. requestedLifetimeCount, subscription->lifeTimeCount);
  49. if(subscription->lifeTimeCount < 3 * subscription->maxKeepAliveCount)
  50. subscription->lifeTimeCount = 3 * subscription->maxKeepAliveCount;
  51. subscription->notificationsPerPublish = maxNotificationsPerPublish;
  52. if(maxNotificationsPerPublish == 0 ||
  53. maxNotificationsPerPublish > server->config.maxNotificationsPerPublish)
  54. subscription->notificationsPerPublish = server->config.maxNotificationsPerPublish;
  55. subscription->priority = priority;
  56. retval = Subscription_registerPublishCallback(server, subscription);
  57. if(retval != UA_STATUSCODE_GOOD)
  58. UA_LOG_DEBUG_SESSION(server->config.logger, subscription->session,
  59. "Subscription %u | Could not register publish callback with error code %s",
  60. subscription->subscriptionId, UA_StatusCode_name(retval));
  61. }
  62. void
  63. Service_CreateSubscription(UA_Server *server, UA_Session *session,
  64. const UA_CreateSubscriptionRequest *request,
  65. UA_CreateSubscriptionResponse *response) {
  66. /* Check limits for the number of subscriptions */
  67. if((server->config.maxSubscriptionsPerSession != 0) &&
  68. (session->numSubscriptions >= server->config.maxSubscriptionsPerSession)) {
  69. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYSUBSCRIPTIONS;
  70. return;
  71. }
  72. /* Create the subscription */
  73. UA_Subscription *newSubscription = UA_Subscription_new(session, response->subscriptionId);
  74. if(!newSubscription) {
  75. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  76. "Processing CreateSubscriptionRequest failed");
  77. response->responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
  78. return;
  79. }
  80. UA_Session_addSubscription(session, newSubscription); /* Also assigns the subscription id */
  81. /* Set the subscription parameters */
  82. newSubscription->publishingEnabled = request->publishingEnabled;
  83. setSubscriptionSettings(server, newSubscription, request->requestedPublishingInterval,
  84. request->requestedLifetimeCount, request->requestedMaxKeepAliveCount,
  85. request->maxNotificationsPerPublish, request->priority);
  86. newSubscription->currentKeepAliveCount = newSubscription->maxKeepAliveCount; /* set settings first */
  87. /* Prepare the response */
  88. response->subscriptionId = newSubscription->subscriptionId;
  89. response->revisedPublishingInterval = newSubscription->publishingInterval;
  90. response->revisedLifetimeCount = newSubscription->lifeTimeCount;
  91. response->revisedMaxKeepAliveCount = newSubscription->maxKeepAliveCount;
  92. UA_LOG_INFO_SESSION(server->config.logger, session, "Subscription %u | "
  93. "Created the Subscription with a publishing interval of %.2f ms",
  94. response->subscriptionId, newSubscription->publishingInterval);
  95. }
  96. void
  97. Service_ModifySubscription(UA_Server *server, UA_Session *session,
  98. const UA_ModifySubscriptionRequest *request,
  99. UA_ModifySubscriptionResponse *response) {
  100. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing ModifySubscriptionRequest");
  101. UA_Subscription *sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  102. if(!sub) {
  103. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  104. return;
  105. }
  106. setSubscriptionSettings(server, sub, request->requestedPublishingInterval,
  107. request->requestedLifetimeCount, request->requestedMaxKeepAliveCount,
  108. request->maxNotificationsPerPublish, request->priority);
  109. sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  110. response->revisedPublishingInterval = sub->publishingInterval;
  111. response->revisedLifetimeCount = sub->lifeTimeCount;
  112. response->revisedMaxKeepAliveCount = sub->maxKeepAliveCount;
  113. }
  114. static void
  115. Operation_SetPublishingMode(UA_Server *Server, UA_Session *session,
  116. UA_Boolean *publishingEnabled, UA_UInt32 *subscriptionId,
  117. UA_StatusCode *result) {
  118. UA_Subscription *sub = UA_Session_getSubscriptionById(session, *subscriptionId);
  119. if(!sub) {
  120. *result = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  121. return;
  122. }
  123. sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  124. sub->publishingEnabled = *publishingEnabled; /* Set the publishing mode */
  125. }
  126. void
  127. Service_SetPublishingMode(UA_Server *server, UA_Session *session,
  128. const UA_SetPublishingModeRequest *request,
  129. UA_SetPublishingModeResponse *response) {
  130. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing SetPublishingModeRequest");
  131. UA_Boolean publishingEnabled = request->publishingEnabled; /* request is const */
  132. response->responseHeader.serviceResult =
  133. UA_Server_processServiceOperations(server, session, (UA_ServiceOperation)Operation_SetPublishingMode,
  134. &publishingEnabled,
  135. &request->subscriptionIdsSize, &UA_TYPES[UA_TYPES_UINT32],
  136. &response->resultsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
  137. }
  138. static UA_StatusCode
  139. setMonitoredItemSettings(UA_Server *server, UA_MonitoredItem *mon,
  140. UA_MonitoringMode monitoringMode,
  141. const UA_MonitoringParameters *params,
  142. const UA_DataType* dataType) {
  143. /* Filter */
  144. if(params->filter.encoding != UA_EXTENSIONOBJECT_DECODED) {
  145. UA_DataChangeFilter_init(&(mon->filter.dataChangeFilter));
  146. mon->filter.dataChangeFilter.trigger = UA_DATACHANGETRIGGER_STATUSVALUE;
  147. } else if(params->filter.content.decoded.type == &UA_TYPES[UA_TYPES_DATACHANGEFILTER]) {
  148. UA_DataChangeFilter *filter = (UA_DataChangeFilter *)params->filter.content.decoded.data;
  149. // TODO implement EURange to support UA_DEADBANDTYPE_PERCENT
  150. switch(filter->deadbandType) {
  151. case UA_DEADBANDTYPE_NONE:
  152. break;
  153. case UA_DEADBANDTYPE_ABSOLUTE:
  154. if(!dataType || !isDataTypeNumeric(dataType))
  155. return UA_STATUSCODE_BADFILTERNOTALLOWED;
  156. break;
  157. case UA_DEADBANDTYPE_PERCENT:
  158. return UA_STATUSCODE_BADMONITOREDITEMFILTERUNSUPPORTED;
  159. default:
  160. return UA_STATUSCODE_BADMONITOREDITEMFILTERUNSUPPORTED;
  161. }
  162. UA_DataChangeFilter_copy(filter, &mon->filter.dataChangeFilter);
  163. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  164. } else if (params->filter.content.decoded.type == &UA_TYPES[UA_TYPES_EVENTFILTER]) {
  165. UA_EventFilter_copy((UA_EventFilter *)params->filter.content.decoded.data,
  166. &mon->filter.eventFilter);
  167. #endif
  168. } else {
  169. return UA_STATUSCODE_BADMONITOREDITEMFILTERINVALID;
  170. }
  171. /* <-- The point of no return --> */
  172. /* Unregister the callback */
  173. UA_MonitoredItem_unregisterSampleCallback(server, mon);
  174. /* Remove the old samples */
  175. UA_ByteString_deleteMembers(&mon->lastSampledValue);
  176. UA_Variant_deleteMembers(&mon->lastValue);
  177. /* ClientHandle */
  178. mon->clientHandle = params->clientHandle;
  179. /* SamplingInterval */
  180. UA_Double samplingInterval = params->samplingInterval;
  181. if(mon->attributeId == UA_ATTRIBUTEID_VALUE) {
  182. mon->monitoredItemType = UA_MONITOREDITEMTYPE_CHANGENOTIFY;
  183. const UA_VariableNode *vn = (const UA_VariableNode *)
  184. UA_Nodestore_get(server, &mon->monitoredNodeId);
  185. if(vn) {
  186. if(vn->nodeClass == UA_NODECLASS_VARIABLE &&
  187. samplingInterval < vn->minimumSamplingInterval)
  188. samplingInterval = vn->minimumSamplingInterval;
  189. UA_Nodestore_release(server, (const UA_Node *)vn);
  190. }
  191. } else if(mon->attributeId == UA_ATTRIBUTEID_EVENTNOTIFIER) {
  192. /* TODO: events should not need a samplinginterval */
  193. samplingInterval = 10000.0f; // 10 seconds to reduce the load
  194. mon->monitoredItemType = UA_MONITOREDITEMTYPE_EVENTNOTIFY;
  195. } else {
  196. mon->monitoredItemType = UA_MONITOREDITEMTYPE_CHANGENOTIFY;
  197. }
  198. mon->samplingInterval = samplingInterval;
  199. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.samplingIntervalLimits,
  200. samplingInterval, mon->samplingInterval);
  201. if(samplingInterval != samplingInterval) /* Check for nan */
  202. mon->samplingInterval = server->config.samplingIntervalLimits.min;
  203. /* QueueSize */
  204. UA_BOUNDEDVALUE_SETWBOUNDS(server->config.queueSizeLimits,
  205. params->queueSize, mon->maxQueueSize);
  206. /* DiscardOldest */
  207. mon->discardOldest = params->discardOldest;
  208. /* Register sample callback if reporting is enabled */
  209. mon->monitoringMode = monitoringMode;
  210. if(monitoringMode == UA_MONITORINGMODE_REPORTING)
  211. UA_MonitoredItem_registerSampleCallback(server, mon);
  212. return UA_STATUSCODE_GOOD;
  213. }
  214. static const UA_String binaryEncoding = {sizeof("Default Binary") - 1, (UA_Byte *)"Default Binary"};
  215. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  216. static UA_StatusCode UA_Server_addMonitoredItemToNodeEditNodeCallback(UA_Server *server, UA_Session *session,
  217. UA_Node *node, void *data) {
  218. /* data is the MonitoredItem */
  219. /* SLIST_INSERT_HEAD */
  220. ((UA_MonitoredItem *)data)->next = ((UA_ObjectNode *)node)->monitoredItemQueue;
  221. ((UA_ObjectNode *)node)->monitoredItemQueue = (UA_MonitoredItem *)data;
  222. return UA_STATUSCODE_GOOD;
  223. }
  224. #endif
  225. /* Thread-local variables to pass additional arguments into the operation */
  226. struct createMonContext {
  227. UA_Subscription *sub;
  228. UA_TimestampsToReturn timestampsToReturn;
  229. /* If sub is NULL, use local callbacks */
  230. UA_Server_DataChangeNotificationCallback dataChangeCallback;
  231. void *context;
  232. };
  233. static void
  234. Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct createMonContext *cmc,
  235. const UA_MonitoredItemCreateRequest *request,
  236. UA_MonitoredItemCreateResult *result) {
  237. /* Check available capacity */
  238. if(server->config.maxMonitoredItemsPerSubscription != 0 && cmc->sub &&
  239. cmc->sub->monitoredItemsSize >= server->config.maxMonitoredItemsPerSubscription) {
  240. result->statusCode = UA_STATUSCODE_BADTOOMANYMONITOREDITEMS;
  241. return;
  242. }
  243. /* Make an example read to get errors in the itemToMonitor. Allow return
  244. * codes "good" and "uncertain", as well as a list of statuscodes that might
  245. * be repaired inside the data source. */
  246. UA_DataValue v = UA_Server_readWithSession(server, session, &request->itemToMonitor,
  247. cmc->timestampsToReturn);
  248. if(v.hasStatus && (v.status >> 30) > 1 &&
  249. v.status != UA_STATUSCODE_BADRESOURCEUNAVAILABLE &&
  250. v.status != UA_STATUSCODE_BADCOMMUNICATIONERROR &&
  251. v.status != UA_STATUSCODE_BADWAITINGFORINITIALDATA &&
  252. v.status != UA_STATUSCODE_BADUSERACCESSDENIED &&
  253. v.status != UA_STATUSCODE_BADNOTREADABLE &&
  254. v.status != UA_STATUSCODE_BADINDEXRANGENODATA) {
  255. result->statusCode = v.status;
  256. UA_DataValue_deleteMembers(&v);
  257. return;
  258. }
  259. /* Check if the encoding is supported */
  260. if(request->itemToMonitor.dataEncoding.name.length > 0 &&
  261. (!UA_String_equal(&binaryEncoding, &request->itemToMonitor.dataEncoding.name) ||
  262. request->itemToMonitor.dataEncoding.namespaceIndex != 0)) {
  263. result->statusCode = UA_STATUSCODE_BADDATAENCODINGUNSUPPORTED;
  264. UA_DataValue_deleteMembers(&v);
  265. return;
  266. }
  267. /* Check if the encoding is set for a value */
  268. if(request->itemToMonitor.attributeId != UA_ATTRIBUTEID_VALUE &&
  269. request->itemToMonitor.dataEncoding.name.length > 0) {
  270. result->statusCode = UA_STATUSCODE_BADDATAENCODINGINVALID;
  271. UA_DataValue_deleteMembers(&v);
  272. return;
  273. }
  274. /* Allocate the MonitoredItem */
  275. size_t nmsize = sizeof(UA_MonitoredItem);
  276. if(!cmc->sub)
  277. nmsize = sizeof(UA_LocalMonitoredItem);
  278. UA_MonitoredItem *newMon = (UA_MonitoredItem*)UA_malloc(nmsize);
  279. if(!newMon) {
  280. result->statusCode = UA_STATUSCODE_BADOUTOFMEMORY;
  281. UA_DataValue_deleteMembers(&v);
  282. return;
  283. }
  284. /* Initialize the MonitoredItem */
  285. UA_MonitoredItem_init(newMon, cmc->sub);
  286. newMon->monitoredItemType = UA_MONITOREDITEMTYPE_CHANGENOTIFY;
  287. newMon->attributeId = request->itemToMonitor.attributeId;
  288. newMon->timestampsToReturn = cmc->timestampsToReturn;
  289. UA_StatusCode retval = UA_STATUSCODE_GOOD;
  290. retval |= UA_NodeId_copy(&request->itemToMonitor.nodeId, &newMon->monitoredNodeId);
  291. retval |= UA_String_copy(&request->itemToMonitor.indexRange, &newMon->indexRange);
  292. retval |= setMonitoredItemSettings(server, newMon, request->monitoringMode,
  293. &request->requestedParameters, v.value.type);
  294. UA_DataValue_deleteMembers(&v);
  295. if(retval != UA_STATUSCODE_GOOD) {
  296. UA_LOG_INFO_SESSION(server->config.logger, session,
  297. "Subscription %u | Could not create a MonitoredItem "
  298. "with StatusCode %s", cmc->sub ? cmc->sub->subscriptionId : 0,
  299. UA_StatusCode_name(retval));
  300. result->statusCode = retval;
  301. UA_MonitoredItem_delete(server, newMon);
  302. return;
  303. }
  304. /* Add to the subscriptions or the local MonitoredItems */
  305. if(cmc->sub) {
  306. newMon->monitoredItemId = ++cmc->sub->lastMonitoredItemId;
  307. UA_Subscription_addMonitoredItem(cmc->sub, newMon);
  308. #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
  309. if(newMon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
  310. /* Insert the monitored item into the node's queue */
  311. UA_Server_editNode(server, NULL, &newMon->monitoredNodeId,
  312. UA_Server_addMonitoredItemToNodeEditNodeCallback, newMon);
  313. }
  314. #endif
  315. } else {
  316. //TODO support events for local monitored items
  317. UA_LocalMonitoredItem *localMon = (UA_LocalMonitoredItem*)newMon;
  318. localMon->context = cmc->context;
  319. localMon->callback.dataChangeCallback = cmc->dataChangeCallback;
  320. newMon->monitoredItemId = ++server->lastLocalMonitoredItemId;
  321. LIST_INSERT_HEAD(&server->localMonitoredItems, newMon, listEntry);
  322. }
  323. /* Register MonitoredItem in userland */
  324. if(server->config.monitoredItemRegisterCallback) {
  325. void *targetContext = NULL;
  326. UA_Server_getNodeContext(server, request->itemToMonitor.nodeId, &targetContext);
  327. server->config.monitoredItemRegisterCallback(server, &session->sessionId,
  328. session->sessionHandle,
  329. &request->itemToMonitor.nodeId,
  330. targetContext, newMon->attributeId, false);
  331. newMon->registered = true;
  332. }
  333. UA_LOG_INFO_SESSION(server->config.logger, session,
  334. "Subscription %u | MonitoredItem %i | "
  335. "Created the MonitoredItem",
  336. cmc->sub ? cmc->sub->subscriptionId : 0,
  337. newMon->monitoredItemId);
  338. /* Create the first sample */
  339. if(request->monitoringMode == UA_MONITORINGMODE_REPORTING)
  340. UA_MonitoredItem_sampleCallback(server, newMon);
  341. /* Prepare the response */
  342. result->revisedSamplingInterval = newMon->samplingInterval;
  343. result->revisedQueueSize = newMon->maxQueueSize;
  344. result->monitoredItemId = newMon->monitoredItemId;
  345. }
  346. void
  347. Service_CreateMonitoredItems(UA_Server *server, UA_Session *session,
  348. const UA_CreateMonitoredItemsRequest *request,
  349. UA_CreateMonitoredItemsResponse *response) {
  350. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing CreateMonitoredItemsRequest");
  351. if(server->config.maxMonitoredItemsPerCall != 0 &&
  352. request->itemsToCreateSize > server->config.maxMonitoredItemsPerCall) {
  353. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYOPERATIONS;
  354. return;
  355. }
  356. /* Check if the timestampstoreturn is valid */
  357. struct createMonContext cmc;
  358. cmc.timestampsToReturn = request->timestampsToReturn;
  359. if(cmc.timestampsToReturn > UA_TIMESTAMPSTORETURN_NEITHER) {
  360. response->responseHeader.serviceResult = UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID;
  361. return;
  362. }
  363. /* Find the subscription */
  364. cmc.sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  365. if(!cmc.sub) {
  366. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  367. return;
  368. }
  369. /* Reset the subscription lifetime */
  370. cmc.sub->currentLifetimeCount = 0;
  371. response->responseHeader.serviceResult =
  372. UA_Server_processServiceOperations(server, session, (UA_ServiceOperation)Operation_CreateMonitoredItem, &cmc,
  373. &request->itemsToCreateSize, &UA_TYPES[UA_TYPES_MONITOREDITEMCREATEREQUEST],
  374. &response->resultsSize, &UA_TYPES[UA_TYPES_MONITOREDITEMCREATERESULT]);
  375. }
  376. UA_MonitoredItemCreateResult
  377. UA_Server_createDataChangeMonitoredItem(UA_Server *server,
  378. UA_TimestampsToReturn timestampsToReturn,
  379. const UA_MonitoredItemCreateRequest item,
  380. void *monitoredItemContext,
  381. UA_Server_DataChangeNotificationCallback callback) {
  382. struct createMonContext cmc;
  383. cmc.sub = NULL;
  384. cmc.context = monitoredItemContext;
  385. cmc.dataChangeCallback = callback;
  386. cmc.timestampsToReturn = timestampsToReturn;
  387. UA_MonitoredItemCreateResult result;
  388. UA_MonitoredItemCreateResult_init(&result);
  389. Operation_CreateMonitoredItem(server, &server->adminSession, &cmc, &item, &result);
  390. return result;
  391. }
  392. static void
  393. Operation_ModifyMonitoredItem(UA_Server *server, UA_Session *session, UA_Subscription *sub,
  394. const UA_MonitoredItemModifyRequest *request,
  395. UA_MonitoredItemModifyResult *result) {
  396. /* Get the MonitoredItem */
  397. UA_MonitoredItem *mon = UA_Subscription_getMonitoredItem(sub, request->monitoredItemId);
  398. if(!mon) {
  399. result->statusCode = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  400. return;
  401. }
  402. /* Read the current value to test if filters are possible.
  403. * Can return an empty value (v.value.type == NULL). */
  404. UA_ReadValueId rvid;
  405. UA_ReadValueId_init(&rvid);
  406. rvid.nodeId = mon->monitoredNodeId;
  407. rvid.attributeId = mon->attributeId;
  408. rvid.indexRange = mon->indexRange;
  409. UA_DataValue v = UA_Server_readWithSession(server, session, &rvid, mon->timestampsToReturn);
  410. UA_StatusCode retval = setMonitoredItemSettings(server, mon, mon->monitoringMode,
  411. &request->requestedParameters,
  412. v.value.type);
  413. UA_DataValue_deleteMembers(&v);
  414. if(retval != UA_STATUSCODE_GOOD) {
  415. result->statusCode = retval;
  416. return;
  417. }
  418. result->revisedSamplingInterval = mon->samplingInterval;
  419. result->revisedQueueSize = mon->maxQueueSize;
  420. /* Remove some notifications if the queue is now too small */
  421. UA_MonitoredItem_ensureQueueSpace(server, mon);
  422. }
  423. void
  424. Service_ModifyMonitoredItems(UA_Server *server, UA_Session *session,
  425. const UA_ModifyMonitoredItemsRequest *request,
  426. UA_ModifyMonitoredItemsResponse *response) {
  427. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing ModifyMonitoredItemsRequest");
  428. if(server->config.maxMonitoredItemsPerCall != 0 &&
  429. request->itemsToModifySize > server->config.maxMonitoredItemsPerCall) {
  430. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYOPERATIONS;
  431. return;
  432. }
  433. /* Check if the timestampstoreturn is valid */
  434. if(request->timestampsToReturn > UA_TIMESTAMPSTORETURN_NEITHER) {
  435. response->responseHeader.serviceResult = UA_STATUSCODE_BADTIMESTAMPSTORETURNINVALID;
  436. return;
  437. }
  438. /* Get the subscription */
  439. UA_Subscription *sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  440. if(!sub) {
  441. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  442. return;
  443. }
  444. sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  445. response->responseHeader.serviceResult =
  446. UA_Server_processServiceOperations(server, session,
  447. (UA_ServiceOperation)Operation_ModifyMonitoredItem, sub,
  448. &request->itemsToModifySize, &UA_TYPES[UA_TYPES_MONITOREDITEMMODIFYREQUEST],
  449. &response->resultsSize, &UA_TYPES[UA_TYPES_MONITOREDITEMMODIFYRESULT]);
  450. }
  451. struct setMonitoringContext {
  452. UA_Subscription *sub;
  453. UA_MonitoringMode monitoringMode;
  454. };
  455. static void
  456. Operation_SetMonitoringMode(UA_Server *server, UA_Session *session,
  457. struct setMonitoringContext *smc,
  458. UA_UInt32 *monitoredItemId, UA_StatusCode *result) {
  459. UA_MonitoredItem *mon = UA_Subscription_getMonitoredItem(smc->sub, *monitoredItemId);
  460. if(!mon) {
  461. *result = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  462. return;
  463. }
  464. if(mon->monitoredItemType != UA_MONITOREDITEMTYPE_CHANGENOTIFY
  465. && mon->monitoredItemType != UA_MONITOREDITEMTYPE_EVENTNOTIFY) {
  466. *result = UA_STATUSCODE_BADNOTIMPLEMENTED;
  467. return;
  468. }
  469. /* Check if the MonitoringMode is valid or not */
  470. if(smc->monitoringMode > UA_MONITORINGMODE_REPORTING) {
  471. *result = UA_STATUSCODE_BADMONITORINGMODEINVALID;
  472. return;
  473. }
  474. if(mon->monitoringMode == smc->monitoringMode)
  475. return;
  476. mon->monitoringMode = smc->monitoringMode;
  477. if(mon->monitoringMode == UA_MONITORINGMODE_REPORTING) {
  478. UA_MonitoredItem_registerSampleCallback(server, mon);
  479. } else {
  480. UA_MonitoredItem_unregisterSampleCallback(server, mon);
  481. // TODO correctly implement SAMPLING
  482. /* Setting the mode to DISABLED or SAMPLING causes all queued Notifications to be deleted */
  483. UA_Notification *notification, *notification_tmp;
  484. TAILQ_FOREACH_SAFE(notification, &mon->queue, listEntry, notification_tmp) {
  485. UA_Notification_delete(smc->sub, mon, notification);
  486. }
  487. /* Initialize lastSampledValue */
  488. UA_ByteString_deleteMembers(&mon->lastSampledValue);
  489. UA_Variant_deleteMembers(&mon->lastValue);
  490. }
  491. }
  492. void
  493. Service_SetMonitoringMode(UA_Server *server, UA_Session *session,
  494. const UA_SetMonitoringModeRequest *request,
  495. UA_SetMonitoringModeResponse *response) {
  496. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing SetMonitoringMode");
  497. if(server->config.maxMonitoredItemsPerCall != 0 &&
  498. request->monitoredItemIdsSize > server->config.maxMonitoredItemsPerCall) {
  499. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYOPERATIONS;
  500. return;
  501. }
  502. /* Get the subscription */
  503. struct setMonitoringContext smc;
  504. smc.sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  505. if(!smc.sub) {
  506. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  507. return;
  508. }
  509. smc.sub->currentLifetimeCount = 0; /* Reset the subscription lifetime */
  510. smc.monitoringMode = request->monitoringMode;
  511. response->responseHeader.serviceResult =
  512. UA_Server_processServiceOperations(server, session,
  513. (UA_ServiceOperation)Operation_SetMonitoringMode, &smc,
  514. &request->monitoredItemIdsSize, &UA_TYPES[UA_TYPES_UINT32],
  515. &response->resultsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
  516. }
  517. /* TODO: Unify with senderror in ua_server_binary.c */
  518. static void
  519. subscriptionSendError(UA_SecureChannel *channel, UA_UInt32 requestHandle,
  520. UA_UInt32 requestId, UA_StatusCode error) {
  521. UA_PublishResponse err_response;
  522. UA_PublishResponse_init(&err_response);
  523. err_response.responseHeader.requestHandle = requestHandle;
  524. err_response.responseHeader.timestamp = UA_DateTime_now();
  525. err_response.responseHeader.serviceResult = error;
  526. UA_SecureChannel_sendSymmetricMessage(channel, requestId, UA_MESSAGETYPE_MSG,
  527. &err_response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
  528. }
  529. void
  530. Service_Publish(UA_Server *server, UA_Session *session,
  531. const UA_PublishRequest *request, UA_UInt32 requestId) {
  532. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing PublishRequest");
  533. /* Return an error if the session has no subscription */
  534. if(LIST_EMPTY(&session->serverSubscriptions)) {
  535. subscriptionSendError(session->header.channel, request->requestHeader.requestHandle,
  536. requestId, UA_STATUSCODE_BADNOSUBSCRIPTION);
  537. return;
  538. }
  539. /* Handle too many subscriptions to free resources before trying to allocate
  540. * resources for the new publish request. If the limit has been reached the
  541. * oldest publish request shall be responded */
  542. if((server->config.maxPublishReqPerSession != 0) &&
  543. (session->numPublishReq >= server->config.maxPublishReqPerSession)) {
  544. if(!UA_Subscription_reachedPublishReqLimit(server, session)) {
  545. subscriptionSendError(session->header.channel, requestId,
  546. request->requestHeader.requestHandle,
  547. UA_STATUSCODE_BADINTERNALERROR);
  548. return;
  549. }
  550. }
  551. /* Allocate the response to store it in the retransmission queue */
  552. UA_PublishResponseEntry *entry = (UA_PublishResponseEntry *)
  553. UA_malloc(sizeof(UA_PublishResponseEntry));
  554. if(!entry) {
  555. subscriptionSendError(session->header.channel, requestId,
  556. request->requestHeader.requestHandle,
  557. UA_STATUSCODE_BADOUTOFMEMORY);
  558. return;
  559. }
  560. /* Prepare the response */
  561. entry->requestId = requestId;
  562. UA_PublishResponse *response = &entry->response;
  563. UA_PublishResponse_init(response);
  564. response->responseHeader.requestHandle = request->requestHeader.requestHandle;
  565. /* Allocate the results array to acknowledge the acknowledge */
  566. if(request->subscriptionAcknowledgementsSize > 0) {
  567. response->results = (UA_StatusCode *)
  568. UA_Array_new(request->subscriptionAcknowledgementsSize,
  569. &UA_TYPES[UA_TYPES_STATUSCODE]);
  570. if(!response->results) {
  571. UA_free(entry);
  572. subscriptionSendError(session->header.channel, requestId,
  573. request->requestHeader.requestHandle,
  574. UA_STATUSCODE_BADOUTOFMEMORY);
  575. return;
  576. }
  577. response->resultsSize = request->subscriptionAcknowledgementsSize;
  578. }
  579. /* Delete Acknowledged Subscription Messages */
  580. for(size_t i = 0; i < request->subscriptionAcknowledgementsSize; ++i) {
  581. UA_SubscriptionAcknowledgement *ack = &request->subscriptionAcknowledgements[i];
  582. UA_Subscription *sub = UA_Session_getSubscriptionById(session, ack->subscriptionId);
  583. if(!sub) {
  584. response->results[i] = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  585. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  586. "Cannot process acknowledgements subscription %u",
  587. ack->subscriptionId);
  588. continue;
  589. }
  590. /* Remove the acked transmission from the retransmission queue */
  591. response->results[i] = UA_Subscription_removeRetransmissionMessage(sub, ack->sequenceNumber);
  592. }
  593. /* Queue the publish response. It will be dequeued in a repeated publish
  594. * callback. This can also be triggered right now for a late
  595. * subscription. */
  596. UA_Session_queuePublishReq(session, entry, false);
  597. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Queued a publication message");
  598. /* If there are late subscriptions, the new publish request is used to
  599. * answer them immediately. However, a single subscription that generates
  600. * many notifications must not "starve" other late subscriptions. Therefore
  601. * we keep track of the last subscription that got preferential treatment.
  602. * We start searching for late subscriptions **after** the last one. */
  603. UA_Subscription *immediate = NULL;
  604. if(session->lastSeenSubscriptionId > 0) {
  605. LIST_FOREACH(immediate, &session->serverSubscriptions, listEntry) {
  606. if(immediate->subscriptionId == session->lastSeenSubscriptionId) {
  607. immediate = LIST_NEXT(immediate, listEntry);
  608. break;
  609. }
  610. }
  611. }
  612. /* If no entry was found, start at the beginning and don't restart */
  613. UA_Boolean found = false;
  614. if(!immediate)
  615. immediate = LIST_FIRST(&session->serverSubscriptions);
  616. else
  617. found = true;
  618. repeat:
  619. while(immediate) {
  620. if(immediate->state == UA_SUBSCRIPTIONSTATE_LATE) {
  621. session->lastSeenSubscriptionId = immediate->subscriptionId;
  622. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  623. "Subscription %u | Response on a late subscription",
  624. immediate->subscriptionId);
  625. UA_Subscription_publish(server, immediate);
  626. return;
  627. }
  628. immediate = LIST_NEXT(immediate, listEntry);
  629. }
  630. /* Restart at the beginning of the list */
  631. if(found) {
  632. immediate = LIST_FIRST(&session->serverSubscriptions);
  633. found = false;
  634. goto repeat;
  635. }
  636. /* No late subscription this time */
  637. session->lastSeenSubscriptionId = 0;
  638. }
  639. static void
  640. Operation_DeleteSubscription(UA_Server *server, UA_Session *session, void *_,
  641. UA_UInt32 *subscriptionId, UA_StatusCode *result) {
  642. *result = UA_Session_deleteSubscription(server, session, *subscriptionId);
  643. if(*result == UA_STATUSCODE_GOOD) {
  644. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  645. "Subscription %u | Subscription deleted",
  646. *subscriptionId);
  647. } else {
  648. UA_LOG_DEBUG_SESSION(server->config.logger, session,
  649. "Deleting Subscription with Id %u failed with error code %s",
  650. *subscriptionId, UA_StatusCode_name(*result));
  651. }
  652. }
  653. void
  654. Service_DeleteSubscriptions(UA_Server *server, UA_Session *session,
  655. const UA_DeleteSubscriptionsRequest *request,
  656. UA_DeleteSubscriptionsResponse *response) {
  657. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing DeleteSubscriptionsRequest");
  658. response->responseHeader.serviceResult =
  659. UA_Server_processServiceOperations(server, session,
  660. (UA_ServiceOperation)Operation_DeleteSubscription, NULL,
  661. &request->subscriptionIdsSize, &UA_TYPES[UA_TYPES_UINT32],
  662. &response->resultsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
  663. /* The session has at least one subscription */
  664. if(LIST_FIRST(&session->serverSubscriptions))
  665. return;
  666. /* Send remaining publish responses if the last subscription was removed */
  667. UA_Subscription_answerPublishRequestsNoSubscription(server, session);
  668. }
  669. static void
  670. Operation_DeleteMonitoredItem(UA_Server *server, UA_Session *session, UA_Subscription *sub,
  671. UA_UInt32 *monitoredItemId, UA_StatusCode *result) {
  672. *result = UA_Subscription_deleteMonitoredItem(server, sub, *monitoredItemId);
  673. }
  674. void
  675. Service_DeleteMonitoredItems(UA_Server *server, UA_Session *session,
  676. const UA_DeleteMonitoredItemsRequest *request,
  677. UA_DeleteMonitoredItemsResponse *response) {
  678. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing DeleteMonitoredItemsRequest");
  679. if(server->config.maxMonitoredItemsPerCall != 0 &&
  680. request->monitoredItemIdsSize > server->config.maxMonitoredItemsPerCall) {
  681. response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYOPERATIONS;
  682. return;
  683. }
  684. /* Get the subscription */
  685. UA_Subscription *sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  686. if(!sub) {
  687. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  688. return;
  689. }
  690. /* Reset the subscription lifetime */
  691. sub->currentLifetimeCount = 0;
  692. response->responseHeader.serviceResult =
  693. UA_Server_processServiceOperations(server, session,
  694. (UA_ServiceOperation)Operation_DeleteMonitoredItem, sub,
  695. &request->monitoredItemIdsSize, &UA_TYPES[UA_TYPES_UINT32],
  696. &response->resultsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
  697. }
  698. UA_StatusCode
  699. UA_Server_deleteMonitoredItem(UA_Server *server, UA_UInt32 monitoredItemId) {
  700. UA_MonitoredItem *mon;
  701. LIST_FOREACH(mon, &server->localMonitoredItems, listEntry) {
  702. if(mon->monitoredItemId != monitoredItemId)
  703. continue;
  704. LIST_REMOVE(mon, listEntry);
  705. UA_MonitoredItem_delete(server, mon);
  706. return UA_STATUSCODE_GOOD;
  707. }
  708. return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
  709. }
  710. void
  711. Service_Republish(UA_Server *server, UA_Session *session, const UA_RepublishRequest *request,
  712. UA_RepublishResponse *response) {
  713. UA_LOG_DEBUG_SESSION(server->config.logger, session, "Processing RepublishRequest");
  714. /* Get the subscription */
  715. UA_Subscription *sub = UA_Session_getSubscriptionById(session, request->subscriptionId);
  716. if(!sub) {
  717. response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
  718. return;
  719. }
  720. /* Reset the subscription lifetime */
  721. sub->currentLifetimeCount = 0;
  722. /* Find the notification in the retransmission queue */
  723. UA_NotificationMessageEntry *entry;
  724. TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
  725. if(entry->message.sequenceNumber == request->retransmitSequenceNumber)
  726. break;
  727. }
  728. if(!entry) {
  729. response->responseHeader.serviceResult = UA_STATUSCODE_BADMESSAGENOTAVAILABLE;
  730. return;
  731. }
  732. response->responseHeader.serviceResult =
  733. UA_NotificationMessage_copy(&entry->message, &response->notificationMessage);
  734. }
  735. #endif /* UA_ENABLE_SUBSCRIPTIONS */