Subscriptions and Monitored Items

An OPC UA Subscription lets a Client receive changes without repeatedly calling client:read. A Subscription defines when the Server may publish notifications. One or more MonitoredItems select the nodes and attributes to sample and report.

Use a Subscription for values that must be observed continuously, such as temperatures, counters, alarms, or server status. Use Read when only a current snapshot is needed. Classic Subscriptions are part of the Client/Server model; they are different from OPC UA PubSub over MQTT.

The Lua client exposes the Subscription and MonitoredItem services directly. The application therefore manages this lifecycle:

Create a Subscription in an active Session.
Create one or more MonitoredItems in that Subscription.
Keep sending Publish requests and process the returned notifications.
Acknowledge notification sequence numbers so the Server can release them.
Delete MonitoredItems and the Subscription when they are no longer needed.

The current compact implementation supports data-change notifications. Event MonitoredItems are not supported.

Subscribe to Server Time

Every OPC UA Server exposes ServerStatus.CurrentTime as NodeId i=2258. The following example creates a Subscription and monitors the Value attribute of that node.

Create a Subscription

Create the Subscription after connecting, opening a SecureChannel, and activating a Session. The Server may revise the requested publishing interval, lifetime count, and keep-alive count. Their revised values are returned in the CreateSubscription response.

-- Create a Subscription that publishes at most once every 250 milliseconds.
response, err = client:createSubscription({
  RequestedPublishingInterval = 250,
  RequestedLifetimeCount = 60,
  RequestedMaxKeepAliveCount = 10,
  MaxNotificationsPerPublish = 10,
  PublishingEnabled = true,
  Priority = 0,
})
checkError("CreateSubscription", err)

local subscriptionId = response.SubscriptionId

Full source

The most important settings are:

RequestedPublishingInterval: Minimum interval, in milliseconds, between publishing cycles.
RequestedMaxKeepAliveCount: Number of publishing cycles without notifications before the Server sends a keep-alive response.
RequestedLifetimeCount: Number of publishing cycles without an available Publish request before the Subscription expires. The Server revises this to at least three times the revised keep-alive count.
MaxNotificationsPerPublish: Maximum number of notifications in one Publish response. Zero means no client-requested limit.
PublishingEnabled: Whether the Server may report queued notifications.

Create a MonitoredItem

A MonitoredItem identifies an attribute to sample. ClientHandle is chosen by the application and is returned with each value, allowing notifications to be associated with application state without comparing NodeIds.

local currentTimeNodeId = "i=2258"

-- Monitor the Value attribute of the standard ServerStatus.CurrentTime node.
response, err = client:createMonitoredItems({
  SubscriptionId = subscriptionId,
  TimestampsToReturn = ua.TimestampsToReturn.Both,
  ItemsToCreate = {
    {
      ItemToMonitor = {
        NodeId = currentTimeNodeId,
        AttributeId = ua.AttributeId.Value,
      },
      MonitoringMode = ua.MonitoringMode.Reporting,
      RequestedParameters = {
        ClientHandle = 1,
        SamplingInterval = 250,
        QueueSize = 2,
        DiscardOldest = true,
      },
    },
  },
})
checkError("CreateMonitoredItems", err)
checkStatus("CreateMonitoredItems", response.Results[1].StatusCode)

local monitoredItemId = response.Results[1].MonitoredItemId

Full source

SamplingInterval is the requested interval, in milliseconds, at which the Server checks the value. Zero requests the fastest practical sampling, and a negative value requests the Subscription publishing interval. QueueSize controls how many changed values the Server retains, and DiscardOldest selects which value is removed if that queue is full.

TimestampsToReturn selects the timestamps included in each DataValue. Use ua.TimestampsToReturn.Source, Server, Both, or Neither.

When no filter is supplied, a value is queued when its StatusCode or value changes. A DataChangeFilter can instead select status-only or status/value/timestamp triggering and can apply an absolute deadband. Percent deadband and event filters are not supported by the compact server.

MonitoringMode controls sampling and reporting:

ua.MonitoringMode.Disabled: Do not sample or report the item.
ua.MonitoringMode.Sampling: Sample and queue changes, but do not add them to Publish responses.
ua.MonitoringMode.Reporting: Sample, queue, and report changes through Publish responses.

Always inspect each entry in response.Results. A service call can succeed while an individual MonitoredItem fails, for example because its NodeId or AttributeId is invalid.

Receive Changed Values

Publish is initiated by the Client. The Server holds a Publish request until a notification or keep-alive is ready. A synchronous client normally sends the next request after processing the previous response. Applications using callback mode can keep multiple Publish requests outstanding.

-- Send Publish requests and acknowledge the previous notification message.
local acknowledgements = {}
local receivedValues = 0
while receivedValues < 3 do
  -- Wait for the next data-change notification or keep-alive.
  response, err = client:publish({
    TimeoutHint = 5000,
    SubscriptionAcknowledgements = acknowledgements,
  })
  checkError("Publish", err)

  local message = response.NotificationMessage
  local notification = message.NotificationData[1]
  acknowledgements = {}

  if notification then
    local monitoredItem = notification.Body.MonitoredItems[1]
    trace("Server time: " .. tostring(monitoredItem.Value.Value))
    receivedValues = receivedValues + 1
    acknowledgements[1] = {
      SubscriptionId = subscriptionId,
      SequenceNumber = message.SequenceNumber,
    }
  end
end

Full source

NotificationData is empty for a keep-alive. A data-change notification contains Body.MonitoredItems. Each entry contains the configured ClientHandle and a Value DataValue.

A non-empty NotificationMessage has a sequence number. Include that number in SubscriptionAcknowledgements on a later Publish request. Until it is acknowledged, the message may be recovered with client:republish if it was lost in transit.

Delete the Resources

Delete MonitoredItems when the application no longer needs their values, then delete the Subscription. Closing a Session with subscriptions deleted also releases them, but explicit deletion makes the resource lifetime clear.

-- Delete the MonitoredItem when its values are no longer needed.
response, err = client:deleteMonitoredItems({
  SubscriptionId = subscriptionId,
  MonitoredItemIds = {monitoredItemId},
})
checkError("DeleteMonitoredItems", err)
checkStatus("DeleteMonitoredItems", response.Results[1])

-- Delete the now-empty Subscription.
response, err = client:deleteSubscriptions(subscriptionId)
checkError("DeleteSubscriptions", err)
checkStatus("DeleteSubscriptions", response.Results[1])

-- Close the active Session and release its Server resources.
client:closeSession()

-- Disconnect the Client transport.
client:disconnect()

Full source

Manage Subscriptions

Change the negotiated parameters with modifySubscription. The response contains the revised values:

-- Change the publishing parameters of the Subscription.
response, err = client:modifySubscription({
  SubscriptionId = subscriptionId,
  RequestedPublishingInterval = 500,
  RequestedLifetimeCount = 60,
  RequestedMaxKeepAliveCount = 10,
  MaxNotificationsPerPublish = 20,
  Priority = 0,
})
checkError("ModifySubscription", err)

Full source

Temporarily stop or resume notification publishing with setPublishingMode without deleting queued MonitoredItem values:

-- Pause notification publishing without stopping MonitoredItem sampling.
response, err = client:setPublishingMode({
  PublishingEnabled = false,
  SubscriptionIds = {subscriptionId},
})
checkError("SetPublishingMode", err)
checkStatus("SetPublishingMode", response.Results[1])

-- Resume notification publishing for the Subscription.
response, err = client:setPublishingMode({
  PublishingEnabled = true,
  SubscriptionIds = {subscriptionId},
})
checkError("SetPublishingMode", err)
checkStatus("SetPublishingMode", response.Results[1])

Full source

Delete one or more Subscriptions with deleteSubscriptions. Its Results array contains one StatusCode for every requested SubscriptionId.

Manage Monitored Items

Change sampling and queue parameters with modifyMonitoredItems without recreating a MonitoredItem:

-- Change the sampling and queue parameters of the MonitoredItem.
response, err = client:modifyMonitoredItems({
  SubscriptionId = subscriptionId,
  TimestampsToReturn = ua.TimestampsToReturn.Both,
  ItemsToModify = {
    {
      MonitoredItemId = monitoredItemId,
      RequestedParameters = {
        ClientHandle = 1,
        SamplingInterval = 500,
        QueueSize = 4,
        DiscardOldest = true,
      },
    },
  },
})
checkError("ModifyMonitoredItems", err)
checkStatus("ModifyMonitoredItems", response.Results[1].StatusCode)

Full source

Pause sampling or reporting with setMonitoringMode:

-- Continue sampling but pause reporting the MonitoredItem's queued values.
response, err = client:setMonitoringMode({
  SubscriptionId = subscriptionId,
  MonitoringMode = ua.MonitoringMode.Sampling,
  MonitoredItemIds = {monitoredItemId},
})
checkError("SetMonitoringMode", err)
checkStatus("SetMonitoringMode", response.Results[1])

-- Resume reporting queued values through Publish responses.
response, err = client:setMonitoringMode({
  SubscriptionId = subscriptionId,
  MonitoringMode = ua.MonitoringMode.Reporting,
  MonitoredItemIds = {monitoredItemId},
})
checkError("SetMonitoringMode", err)
checkStatus("SetMonitoringMode", response.Results[1])

Full source

As with creation and deletion, inspect every StatusCode in response.Results after modifying modes or parameters.