123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602 |
- /* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at http://mozilla.org/MPL/2.0/.
- *
- * Copyright 2015-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
- * Copyright 2015 (c) Chris Iatrou
- * Copyright 2015-2016 (c) Sten Grüner
- * Copyright 2017-2018 (c) Thomas Stalder, Blue Time Concept SA
- * Copyright 2015 (c) Joakim L. Gilje
- * Copyright 2016-2017 (c) Florian Palm
- * Copyright 2015-2016 (c) Oleksiy Vasylyev
- * Copyright 2017 (c) frax2222
- * Copyright 2017 (c) Stefan Profanter, fortiss GmbH
- * Copyright 2017 (c) Ari Breitkreuz, fortiss GmbH
- * Copyright 2017 (c) Mattias Bornhager
- * Copyright 2018 (c) Hilscher Gesellschaft für Systemautomation mbH (Author: Martin Lang)
- */
- #include "ua_server_internal.h"
- #include "ua_subscription.h"
- #ifdef UA_ENABLE_SUBSCRIPTIONS /* conditional compilation */
- UA_Subscription *
- UA_Subscription_new(UA_Session *session, UA_UInt32 subscriptionId) {
- /* Allocate the memory */
- UA_Subscription *newSub =
- (UA_Subscription*)UA_calloc(1, sizeof(UA_Subscription));
- if(!newSub)
- return NULL;
- /* Remaining members are covered by calloc zeroing out the memory */
- newSub->session = session;
- newSub->subscriptionId = subscriptionId;
- newSub->state = UA_SUBSCRIPTIONSTATE_NORMAL; /* The first publish response is sent immediately */
- /* Even if the first publish response is a keepalive the sequence number is 1.
- * This can happen by a subscription without a monitored item (see CTT test scripts). */
- newSub->nextSequenceNumber = 1;
- TAILQ_INIT(&newSub->retransmissionQueue);
- TAILQ_INIT(&newSub->notificationQueue);
- return newSub;
- }
- void
- UA_Subscription_deleteMembers(UA_Server *server, UA_Subscription *sub) {
- Subscription_unregisterPublishCallback(server, sub);
- /* Delete monitored Items */
- UA_MonitoredItem *mon, *tmp_mon;
- LIST_FOREACH_SAFE(mon, &sub->monitoredItems, listEntry, tmp_mon) {
- LIST_REMOVE(mon, listEntry);
- UA_LOG_INFO_SESSION(&server->config.logger, sub->session,
- "Subscription %u | MonitoredItem %i | "
- "Deleted the MonitoredItem", sub->subscriptionId,
- mon->monitoredItemId);
- UA_MonitoredItem_delete(server, mon);
- }
- sub->monitoredItemsSize = 0;
- /* Delete Retransmission Queue */
- UA_NotificationMessageEntry *nme, *nme_tmp;
- TAILQ_FOREACH_SAFE(nme, &sub->retransmissionQueue, listEntry, nme_tmp) {
- TAILQ_REMOVE(&sub->retransmissionQueue, nme, listEntry);
- UA_NotificationMessage_deleteMembers(&nme->message);
- UA_free(nme);
- --sub->session->totalRetransmissionQueueSize;
- --sub->retransmissionQueueSize;
- }
- UA_assert(sub->retransmissionQueueSize == 0);
- UA_LOG_INFO_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Deleted the Subscription",
- sub->subscriptionId);
- }
- UA_MonitoredItem *
- UA_Subscription_getMonitoredItem(UA_Subscription *sub, UA_UInt32 monitoredItemId) {
- UA_MonitoredItem *mon;
- LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
- if(mon->monitoredItemId == monitoredItemId)
- break;
- }
- return mon;
- }
- UA_StatusCode
- UA_Subscription_deleteMonitoredItem(UA_Server *server, UA_Subscription *sub,
- UA_UInt32 monitoredItemId) {
- /* Find the MonitoredItem */
- UA_MonitoredItem *mon;
- LIST_FOREACH(mon, &sub->monitoredItems, listEntry) {
- if(mon->monitoredItemId == monitoredItemId)
- break;
- }
- if(!mon)
- return UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
- UA_LOG_INFO_SESSION(&server->config.logger, sub->session,
- "Subscription %u | MonitoredItem %i | "
- "Delete the MonitoredItem", sub->subscriptionId,
- mon->monitoredItemId);
- /* Remove the MonitoredItem */
- LIST_REMOVE(mon, listEntry);
- sub->monitoredItemsSize--;
- /* Remove content and delayed free */
- UA_MonitoredItem_delete(server, mon);
- return UA_STATUSCODE_GOOD;
- }
- void
- UA_Subscription_addMonitoredItem(UA_Subscription *sub, UA_MonitoredItem *newMon) {
- sub->monitoredItemsSize++;
- LIST_INSERT_HEAD(&sub->monitoredItems, newMon, listEntry);
- }
- static void
- removeOldestRetransmissionMessage(UA_Session *session) {
- UA_NotificationMessageEntry *oldestEntry = NULL;
- UA_Subscription *oldestSub = NULL;
- UA_Subscription *sub;
- LIST_FOREACH(sub, &session->serverSubscriptions, listEntry) {
- UA_NotificationMessageEntry *first =
- TAILQ_LAST(&sub->retransmissionQueue, ListOfNotificationMessages);
- if(!first)
- continue;
- if(!oldestEntry || oldestEntry->message.publishTime > first->message.publishTime) {
- oldestEntry = first;
- oldestSub = sub;
- }
- }
- UA_assert(oldestEntry);
- UA_assert(oldestSub);
- TAILQ_REMOVE(&oldestSub->retransmissionQueue, oldestEntry, listEntry);
- UA_NotificationMessage_deleteMembers(&oldestEntry->message);
- UA_free(oldestEntry);
- --session->totalRetransmissionQueueSize;
- --oldestSub->retransmissionQueueSize;
- }
- static void
- UA_Subscription_addRetransmissionMessage(UA_Server *server, UA_Subscription *sub,
- UA_NotificationMessageEntry *entry) {
- /* Release the oldest entry if there is not enough space */
- if(server->config.maxRetransmissionQueueSize > 0 &&
- sub->session->totalRetransmissionQueueSize >= server->config.maxRetransmissionQueueSize) {
- UA_LOG_WARNING_SESSION(&server->config.logger, sub->session, "Subscription %u | "
- "Retransmission queue overflow", sub->subscriptionId);
- removeOldestRetransmissionMessage(sub->session);
- }
- /* Add entry */
- TAILQ_INSERT_TAIL(&sub->retransmissionQueue, entry, listEntry);
- ++sub->session->totalRetransmissionQueueSize;
- ++sub->retransmissionQueueSize;
- }
- UA_StatusCode
- UA_Subscription_removeRetransmissionMessage(UA_Subscription *sub, UA_UInt32 sequenceNumber) {
- /* Find the retransmission message */
- UA_NotificationMessageEntry *entry;
- TAILQ_FOREACH(entry, &sub->retransmissionQueue, listEntry) {
- if(entry->message.sequenceNumber == sequenceNumber)
- break;
- }
- if(!entry)
- return UA_STATUSCODE_BADSEQUENCENUMBERUNKNOWN;
- /* Remove the retransmission message */
- TAILQ_REMOVE(&sub->retransmissionQueue, entry, listEntry);
- --sub->session->totalRetransmissionQueueSize;
- --sub->retransmissionQueueSize;
- UA_NotificationMessage_deleteMembers(&entry->message);
- UA_free(entry);
- return UA_STATUSCODE_GOOD;
- }
- static UA_StatusCode
- prepareNotificationMessage(UA_Server *server, UA_Subscription *sub,
- UA_NotificationMessage *message, size_t notifications) {
- UA_assert(notifications > 0);
- /* Allocate an ExtensionObject for events and data */
- message->notificationData = (UA_ExtensionObject*)
- UA_Array_new(2, &UA_TYPES[UA_TYPES_EXTENSIONOBJECT]);
- if(!message->notificationData)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- message->notificationDataSize = 2;
- /* Pre-allocate DataChangeNotifications */
- size_t notificationDataIdx = 0;
- UA_DataChangeNotification *dcn = NULL;
- if(sub->dataChangeNotifications > 0) {
- dcn = UA_DataChangeNotification_new();
- if(!dcn) {
- UA_NotificationMessage_deleteMembers(message);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- message->notificationData->encoding = UA_EXTENSIONOBJECT_DECODED;
- message->notificationData->content.decoded.data = dcn;
- message->notificationData->content.decoded.type = &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION];
- size_t dcnSize = sub->dataChangeNotifications;
- if(dcnSize > notifications)
- dcnSize = notifications;
- dcn->monitoredItems = (UA_MonitoredItemNotification*)
- UA_Array_new(dcnSize, &UA_TYPES[UA_TYPES_MONITOREDITEMNOTIFICATION]);
- if(!dcn->monitoredItems) {
- UA_NotificationMessage_deleteMembers(message);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- dcn->monitoredItemsSize = dcnSize;
- notificationDataIdx++;
- }
- #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
- UA_EventNotificationList *enl = NULL;
- UA_StatusChangeNotification *scn = NULL;
- /* Pre-allocate either StatusChange or EventNotifications. Sending a
- * (single) StatusChangeNotification has priority. */
- if(sub->statusChangeNotifications > 0) {
- scn = UA_StatusChangeNotification_new();
- if(!scn) {
- UA_NotificationMessage_deleteMembers(message);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
- message->notificationData[notificationDataIdx].content.decoded.data = scn;
- message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION];
- notificationDataIdx++;
- } else if(sub->eventNotifications > 0) {
- enl = UA_EventNotificationList_new();
- if(!enl) {
- UA_NotificationMessage_deleteMembers(message);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- message->notificationData[notificationDataIdx].encoding = UA_EXTENSIONOBJECT_DECODED;
- message->notificationData[notificationDataIdx].content.decoded.data = enl;
- message->notificationData[notificationDataIdx].content.decoded.type = &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST];
- size_t enlSize = sub->eventNotifications;
- if(enlSize > notifications)
- enlSize = notifications;
- enl->events = (UA_EventFieldList*) UA_Array_new(enlSize, &UA_TYPES[UA_TYPES_EVENTFIELDLIST]);
- if(!enl->events) {
- UA_NotificationMessage_deleteMembers(message);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- enl->eventsSize = enlSize;
- notificationDataIdx++;
- }
- #endif
- UA_assert(notificationDataIdx > 0);
- message->notificationDataSize = notificationDataIdx;
- /* <-- The point of no return --> */
- size_t totalNotifications = 0; /* How many notifications were moved to the response overall? */
- size_t dcnPos = 0; /* How many DataChangeNotifications were put into the list? */
- #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
- size_t enlPos = 0; /* How many EventNotifications were moved into the list */
- #endif
- UA_Notification *notification, *notification_tmp;
- TAILQ_FOREACH_SAFE(notification, &sub->notificationQueue, globalEntry, notification_tmp) {
- if(totalNotifications >= notifications)
- break;
-
- UA_MonitoredItem *mon = notification->mon;
- /* Remove from the queues and decrease the counters */
- UA_Notification_dequeue(server, notification);
- /* Move the content to the response */
- if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_CHANGENOTIFY) {
- UA_assert(dcn != NULL); /* Have at least one change notification */
- /* Move the content to the response */
- UA_MonitoredItemNotification *min = &dcn->monitoredItems[dcnPos];
- min->clientHandle = mon->clientHandle;
- min->value = notification->data.value;
- UA_DataValue_init(¬ification->data.value); /* Reset after the value has been moved */
- dcnPos++;
- }
- #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
- else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_STATUSNOTIFY && scn) {
- // TODO: Handling of StatusChangeNotifications
- scn = NULL; /* At most one per PublishReponse */
- } else if(mon->monitoredItemType == UA_MONITOREDITEMTYPE_EVENTNOTIFY && enl) {
- UA_assert(enl != NULL); /* Have at least one event notification */
- /* Move the content to the response */
- UA_EventFieldList *efl = &enl->events[enlPos];
- *efl = notification->data.event.fields;
- UA_EventFieldList_init(¬ification->data.event.fields);
- efl->clientHandle = mon->clientHandle;
- enlPos++;
- }
- #endif
- else {
- UA_Notification_delete(notification);
- continue; /* Unknown type. Nothing to do */
- }
- UA_Notification_delete(notification);
- totalNotifications++;
- }
- /* Set sizes */
- if(dcn) {
- dcn->monitoredItemsSize = dcnPos;
- if(dcnPos == 0) {
- UA_free(dcn->monitoredItems);
- dcn->monitoredItems = NULL;
- }
- }
- #ifdef UA_ENABLE_SUBSCRIPTIONS_EVENTS
- if(enl) {
- enl->eventsSize = enlPos;
- if(enlPos == 0) {
- UA_free(enl->events);
- enl->events = NULL;
- }
- }
- #endif
- return UA_STATUSCODE_GOOD;
- }
- /* According to OPC Unified Architecture, Part 4 5.13.1.1 i) The value 0 is
- * never used for the sequence number */
- static UA_UInt32
- UA_Subscription_nextSequenceNumber(UA_UInt32 sequenceNumber) {
- UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
- if(nextSequenceNumber == 0)
- nextSequenceNumber = 1;
- return nextSequenceNumber;
- }
- static void
- publishCallback(UA_Server *server, UA_Subscription *sub) {
- sub->readyNotifications = sub->notificationQueueSize;
- UA_Subscription_publish(server, sub);
- }
- void
- UA_Subscription_publish(UA_Server *server, UA_Subscription *sub) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session, "Subscription %u | "
- "Publish Callback", sub->subscriptionId);
- /* Dequeue a response */
- UA_PublishResponseEntry *pre = UA_Session_dequeuePublishReq(sub->session);
- if(pre) {
- sub->currentLifetimeCount = 0; /* Reset the LifetimeCounter */
- } else {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | The publish queue is empty",
- sub->subscriptionId);
- ++sub->currentLifetimeCount;
- if(sub->currentLifetimeCount > sub->lifeTimeCount) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | End of lifetime "
- "for subscription", sub->subscriptionId);
- UA_Session_deleteSubscription(server, sub->session, sub->subscriptionId);
- /* TODO: send a StatusChangeNotification with Bad_Timeout */
- return;
- }
- }
- /* If there are several late publish responses... */
- if(sub->readyNotifications > sub->notificationQueueSize)
- sub->readyNotifications = sub->notificationQueueSize;
- /* Count the available notifications */
- UA_UInt32 notifications = sub->readyNotifications;
- if(!sub->publishingEnabled)
- notifications = 0;
- UA_Boolean moreNotifications = false;
- if(notifications > sub->notificationsPerPublish) {
- notifications = sub->notificationsPerPublish;
- moreNotifications = true;
- }
- /* Return if no notifications and no keepalive */
- if(notifications == 0) {
- ++sub->currentKeepAliveCount;
- if(sub->currentKeepAliveCount < sub->maxKeepAliveCount) {
- if(pre)
- UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
- return;
- }
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Sending a KeepAlive",
- sub->subscriptionId);
- }
- /* We want to send a response. Is the channel open? */
- UA_SecureChannel *channel = sub->session->header.channel;
- if(!channel || !pre) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Want to send a publish response but can't. "
- "The subscription is late.", sub->subscriptionId);
- sub->state = UA_SUBSCRIPTIONSTATE_LATE;
- if(pre)
- UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
- return;
- }
- /* Prepare the response */
- UA_PublishResponse *response = &pre->response;
- UA_NotificationMessage *message = &response->notificationMessage;
- UA_NotificationMessageEntry *retransmission = NULL;
- if(notifications > 0) {
- /* Allocate the retransmission entry */
- retransmission = (UA_NotificationMessageEntry*)UA_malloc(sizeof(UA_NotificationMessageEntry));
- if(!retransmission) {
- UA_LOG_WARNING_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Could not allocate memory for retransmission. "
- "The subscription is late.", sub->subscriptionId);
- sub->state = UA_SUBSCRIPTIONSTATE_LATE;
- UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
- return;
- }
- /* Prepare the response */
- UA_StatusCode retval = prepareNotificationMessage(server, sub, message, notifications);
- if(retval != UA_STATUSCODE_GOOD) {
- UA_LOG_WARNING_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Could not prepare the notification message. "
- "The subscription is late.", sub->subscriptionId);
- UA_free(retransmission);
- sub->state = UA_SUBSCRIPTIONSTATE_LATE;
- UA_Session_queuePublishReq(sub->session, pre, true); /* Re-enqueue */
- return;
- }
- }
- /* <-- The point of no return --> */
- /* Adjust the number of ready notifications */
- UA_assert(sub->readyNotifications >= notifications);
- sub->readyNotifications -= notifications;
- /* Set up the response */
- response->responseHeader.timestamp = UA_DateTime_now();
- response->subscriptionId = sub->subscriptionId;
- response->moreNotifications = moreNotifications;
- message->publishTime = response->responseHeader.timestamp;
- /* Set sequence number to message. Started at 1 which is given
- * during creating a new subscription. The 1 is required for
- * initial publish response with or without an monitored item. */
- message->sequenceNumber = sub->nextSequenceNumber;
- if(notifications > 0) {
- /* Put the notification message into the retransmission queue. This
- * needs to be done here, so that the message itself is included in the
- * available sequence numbers for acknowledgement. */
- retransmission->message = response->notificationMessage;
- UA_Subscription_addRetransmissionMessage(server, sub, retransmission);
- /* Only if a notification was created, the sequence number must be increased.
- * For a keepalive the sequence number can be reused. */
- sub->nextSequenceNumber = UA_Subscription_nextSequenceNumber(sub->nextSequenceNumber);
- }
- /* Get the available sequence numbers from the retransmission queue */
- size_t available = sub->retransmissionQueueSize;
- UA_STACKARRAY(UA_UInt32, seqNumbers, available);
- if(available > 0) {
- response->availableSequenceNumbers = seqNumbers;
- response->availableSequenceNumbersSize = available;
- size_t i = 0;
- UA_NotificationMessageEntry *nme;
- TAILQ_FOREACH(nme, &sub->retransmissionQueue, listEntry) {
- response->availableSequenceNumbers[i] = nme->message.sequenceNumber;
- ++i;
- }
- }
- /* Send the response */
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Sending out a publish response "
- "with %u notifications", sub->subscriptionId,
- (UA_UInt32)notifications);
- UA_SecureChannel_sendSymmetricMessage(sub->session->header.channel, pre->requestId,
- UA_MESSAGETYPE_MSG, response,
- &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
- /* Reset subscription state to normal */
- sub->state = UA_SUBSCRIPTIONSTATE_NORMAL;
- sub->currentKeepAliveCount = 0;
- /* Free the response */
- UA_Array_delete(response->results, response->resultsSize, &UA_TYPES[UA_TYPES_UINT32]);
- UA_free(pre); /* No need for UA_PublishResponse_deleteMembers */
- /* Repeat sending responses if there are more notifications to send */
- if(moreNotifications)
- UA_Subscription_publish(server, sub);
- }
- UA_Boolean
- UA_Subscription_reachedPublishReqLimit(UA_Server *server, UA_Session *session) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, session,
- "Reached number of publish request limit");
- /* Dequeue a response */
- UA_PublishResponseEntry *pre = UA_Session_dequeuePublishReq(session);
- /* Cannot publish without a response */
- if(!pre) {
- UA_LOG_FATAL_SESSION(&server->config.logger, session, "No publish requests available");
- return false;
- }
- /* <-- The point of no return --> */
- UA_PublishResponse *response = &pre->response;
- UA_NotificationMessage *message = &response->notificationMessage;
- /* Set up the response. Note that this response has no related subscription id */
- response->responseHeader.timestamp = UA_DateTime_now();
- response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS;
- response->subscriptionId = 0;
- response->moreNotifications = false;
- message->publishTime = response->responseHeader.timestamp;
- message->sequenceNumber = 0;
- response->availableSequenceNumbersSize = 0;
- /* Send the response */
- UA_LOG_DEBUG_SESSION(&server->config.logger, session,
- "Sending out a publish response triggered by too many publish requests");
- UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId,
- UA_MESSAGETYPE_MSG, response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
- /* Free the response */
- UA_Array_delete(response->results, response->resultsSize, &UA_TYPES[UA_TYPES_UINT32]);
- UA_free(pre); /* no need for UA_PublishResponse_deleteMembers */
- return true;
- }
- UA_StatusCode
- Subscription_registerPublishCallback(UA_Server *server, UA_Subscription *sub) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session,
- "Subscription %u | Register subscription "
- "publishing callback", sub->subscriptionId);
- if(sub->publishCallbackIsRegistered)
- return UA_STATUSCODE_GOOD;
- UA_StatusCode retval =
- UA_Server_addRepeatedCallback(server, (UA_ServerCallback)publishCallback,
- sub, (UA_UInt32)sub->publishingInterval, &sub->publishCallbackId);
- if(retval != UA_STATUSCODE_GOOD)
- return retval;
- sub->publishCallbackIsRegistered = true;
- return UA_STATUSCODE_GOOD;
- }
- void
- Subscription_unregisterPublishCallback(UA_Server *server, UA_Subscription *sub) {
- UA_LOG_DEBUG_SESSION(&server->config.logger, sub->session, "Subscription %u | "
- "Unregister subscription publishing callback", sub->subscriptionId);
- if(!sub->publishCallbackIsRegistered)
- return;
- UA_Server_removeRepeatedCallback(server, sub->publishCallbackId);
- sub->publishCallbackIsRegistered = false;
- }
- /* When the session has publish requests stored but the last subscription is
- * deleted... Send out empty responses */
- void
- UA_Subscription_answerPublishRequestsNoSubscription(UA_Server *server, UA_Session *session) {
- /* No session or there are remaining subscriptions */
- if(!session || LIST_FIRST(&session->serverSubscriptions))
- return;
- /* Send a response for every queued request */
- UA_PublishResponseEntry *pre;
- while((pre = UA_Session_dequeuePublishReq(session))) {
- UA_PublishResponse *response = &pre->response;
- response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
- response->responseHeader.timestamp = UA_DateTime_now();
- UA_SecureChannel_sendSymmetricMessage(session->header.channel, pre->requestId, UA_MESSAGETYPE_MSG,
- response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
- UA_PublishResponse_deleteMembers(response);
- UA_free(pre);
- }
- }
- #endif /* UA_ENABLE_SUBSCRIPTIONS */
|