module import { EventEmitter } from "events" ; import * as chalk from "chalk" ; import { AddressSpace , BaseNode , Duration , UAObjectType } from "node-opcua-address-space" ; import { checkSelectClauses } from "node-opcua-address-space" ; import { SessionContext } from "node-opcua-address-space" ; import { assert } from "node-opcua-assert" ; import { Byte } from "node-opcua-basic-types" ; import { SubscriptionDiagnosticsDataType } from "node-opcua-common" ; import { NodeClass , AttributeIds , isValidDataEncoding } from "node-opcua-data-model" ; import { TimestampsToReturn } from "node-opcua-data-value" ; import { checkDebugFlag , make_debugLog , make_warningLog } from "node-opcua-debug" ; import { NodeId } from "node-opcua-nodeid" ; import { ObjectRegistry } from "node-opcua-object-registry" ; import { SequenceNumberGenerator } from "node-opcua-secure-channel" ; import { EventFilter } from "node-opcua-service-filter" ; import { AggregateFilter } from "node-opcua-service-subscription" ; import { DataChangeNotification , EventNotificationList , MonitoringMode , MonitoredItemCreateResult , MonitoredItemNotification , PublishResponse , NotificationMessage , StatusChangeNotification , DataChangeFilter , MonitoredItemCreateRequest } from "node-opcua-service-subscription" ; import { StatusCode , StatusCodes } from "node-opcua-status-code" ; import { AggregateFilterResult , ContentFilterResult , EventFieldList , EventFilterResult , NotificationData } from "node-opcua-types" ; import { Queue } from "./queue" ; import { MonitoredItem , MonitoredItemOptions , QueueItem } from "./monitored_item" ; import { ServerSession } from "./server_session" ; import { validateFilter } from "./validate_filter" ; import { IServerSidePublishEngine , TransferredSubscription } from "./i_server_side_publish_engine" ; const debugLog = make_debugLog ( __filename ) ; const doDebug = checkDebugFlag ( __filename ) ; const warningLog = make_warningLog ( __filename ) ; const maxNotificationMessagesInQueue = 100 ; export interface SubscriptionDiagnosticsDataTypePriv extends SubscriptionDiagnosticsDataType { $subscription : Subscription ; } export enum SubscriptionState { CLOSED = 1 , CREATING = 2 , NORMAL = 3 , LATE = 4 , KEEPALIVE = 5 , TERMINATED = 6 } function _adjust_publishing_interval ( publishingInterval ? ) { publishingInterval = publishingInterval === undefined || Number . isNaN ( publishingInterval ) ? Subscription . defaultPublishingInterval : publishingInterval ; publishingInterval = Math . max ( publishingInterval , Subscription . minimumPublishingInterval ) ; publishingInterval = Math . min ( publishingInterval , Subscription . maximumPublishingInterval ) ; return publishingInterval ; } const minimumMaxKeepAliveCount = 2 ; const maximumMaxKeepAliveCount = 12000 ; function _adjust_maxKeepAliveCount ( maxKeepAliveCount ? ) { maxKeepAliveCount = maxKeepAliveCount || minimumMaxKeepAliveCount ; maxKeepAliveCount = Math . max ( maxKeepAliveCount , minimumMaxKeepAliveCount ) ; maxKeepAliveCount = Math . min ( maxKeepAliveCount , maximumMaxKeepAliveCount ) ; return maxKeepAliveCount ; } function _adjust_lifeTimeCount ( lifeTimeCount , maxKeepAliveCount , publishingInterval ) { lifeTimeCount = lifeTimeCount || 1 ; lifeTimeCount = Math . max ( lifeTimeCount , maxKeepAliveCount * 3 ) ; const minTicks = Math . ceil ( ( 5 * 1000 ) / publishingInterval ) ; lifeTimeCount = Math . max ( minTicks , lifeTimeCount ) ; return lifeTimeCount ; } function _adjust_publishingEnable ( publishingEnabled ? ) { return publishingEnabled === null || publishingEnabled === undefined ? true : ! ! publishingEnabled ; } function _adjust_maxNotificationsPerPublish ( maxNotificationsPerPublish ? ) { assert ( Subscription . maxNotificationPerPublishHighLimit > 0 , "Subscription.maxNotificationPerPublishHighLimit must be positive" ) ; maxNotificationsPerPublish = maxNotificationsPerPublish || 0 ; assert ( typeof maxNotificationsPerPublish === "number" ) ; maxNotificationsPerPublish = maxNotificationsPerPublish >= 0 ? maxNotificationsPerPublish : 0 ; if ( maxNotificationsPerPublish === 0 ) { maxNotificationsPerPublish = Subscription . maxNotificationPerPublishHighLimit ; } else { maxNotificationsPerPublish = Math . min ( Subscription . maxNotificationPerPublishHighLimit , maxNotificationsPerPublish ) ; } assert ( maxNotificationsPerPublish !== 0 && maxNotificationsPerPublish <= Subscription . maxNotificationPerPublishHighLimit ) ; return maxNotificationsPerPublish ; } function w ( s , length ) { return ( "000" + s ) . substr ( - length ) ; } function t ( d ) { return w ( d . getHours ( ) , 2 ) + ":" + w ( d . getMinutes ( ) , 2 ) + ":" + w ( d . getSeconds ( ) , 2 ) + ":" + w ( d . getMilliseconds ( ) , 3 ) ; } function _getSequenceNumbers ( arr ) { return arr . map ( ( notificationMessage ) => notificationMessage . sequenceNumber ) ; } function analyseEventFilterResult ( node , eventFilter ) { if ( ! ( eventFilter instanceof EventFilter ) ) { throw new Error ( "Internal Error" ) ; } const selectClauseResults = checkSelectClauses ( node as UAObjectType , eventFilter . selectClauses || [ ] ) ; const whereClauseResult = new ContentFilterResult ( ) ; return new EventFilterResult ( { selectClauseDiagnosticInfos : [ ] , selectClauseResults , whereClauseResult } ) ; } function analyseDataChangeFilterResult ( node , dataChangeFilter ) { assert ( dataChangeFilter instanceof DataChangeFilter ) ; return null ; } function analyseAggregateFilterResult ( node , aggregateFilter ) { assert ( aggregateFilter instanceof AggregateFilter ) ; return new AggregateFilterResult ( { } ) ; } function _process_filter ( node , filter ) { if ( ! filter ) { return null ; } if ( filter instanceof EventFilter ) { return analyseEventFilterResult ( node , filter ) ; } else if ( filter instanceof DataChangeFilter ) { return analyseDataChangeFilterResult ( node , filter ) ; } else if ( filter instanceof AggregateFilter ) { return analyseAggregateFilterResult ( node , filter ) ; } throw new Error ( "invalid filter" ) ; } private function createSubscriptionDiagnostics ( subscription ) { assert ( subscription instanceof Subscription ) ; const subscriptionDiagnostics = new SubscriptionDiagnosticsDataType ( { } ) ; const subscription_subscriptionDiagnostics = subscriptionDiagnostics as any ; subscription_subscriptionDiagnostics . $subscription = subscription ; subscription_subscriptionDiagnostics . __defineGetter__ ( "sessionId" , function ( this ) : NodeId { if ( ! this . $subscription ) { return NodeId . nullNodeId ; } return this . $subscription . getSessionId ( ) ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "subscriptionId" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . id ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "priority" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . priority ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "publishingInterval" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . publishingInterval ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "maxLifetimeCount" , function ( this ) { return this . $subscription . lifeTimeCount ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "maxKeepAliveCount" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . maxKeepAliveCount ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "maxNotificationsPerPublish" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . maxNotificationsPerPublish ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "publishingEnabled" , function ( this ) : boolean { if ( ! this . $subscription ) { return false ; } return this . $subscription . publishingEnabled ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "monitoredItemCount" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . monitoredItemCount ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "nextSequenceNumber" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . _get_future_sequence_number ( ) ; } ) ; subscription_subscriptionDiagnostics . __defineGetter__ ( "disabledMonitoredItemCount" , function ( this ) : number { if ( ! this . $subscription ) { return 0 ; } return this . $subscription . disabledMonitoredItemCount ; } ) ; return subscriptionDiagnostics as SubscriptionDiagnosticsDataTypePriv ; } export interface SubscriptionOptions { sessionId ? : NodeId ; /**
     * (default:1000) the publishing interval.
     */ publishingInterval ? : number ; /**
     * (default:10) the max Life Time Count
     */ maxKeepAliveCount ? : number ; lifeTimeCount ? : number ; /**
     * (default:true)
     */ publishingEnabled ? : boolean ; /**
     * (default:0)
     */ maxNotificationsPerPublish ? : number ; /**
     * subscription priority Byte:(0-255)
     */ priority ? : number ; publishEngine ? : IServerSidePublishEngine ; /**
     *  a unique identifier
     */ id ? : number ; } let g_monitoredItemId = Math . ceil ( Math . random ( ) * 100000 ) ; function getNextMonitoredItemId ( ) { return g_monitoredItemId ++ ; } const INVALID_ID = - 1 ; export type Notification = DataChangeNotification | EventNotificationList | StatusChangeNotification ; export type Counter = number ; export interface ModifySubscriptionParameters { /**
     *     requestedPublishingInterval =0 means fastest possible
     */ requestedPublishingInterval ? : Duration ; requestedLifetimeCount ? : Counter ; /**
     * requestedMaxKeepAliveCount  ===0 means no change
     */ requestedMaxKeepAliveCount ? : Counter ; maxNotificationsPerPublish ? : Counter ; priority ? : Byte ; } export interface GetMonitoredItemsResult { /**
     * array of serverHandles for all MonitoredItems of the subscription
     * identified by subscriptionId.
     */ serverHandles : Uint32Array ; /**
     *  array of clientHandles for all MonitoredItems of the subscription
     *  identified by subscriptionId.
     */ clientHandles : Uint32Array ; statusCode : StatusCode ; } export interface InternalNotification { monitoredItemId ? : number ; notification : QueueItem | StatusChangeNotification ; publishTime : Date ; start_tick : number ; } export interface InternalCreateMonitoredItemResult { monitoredItem ? : MonitoredItem ; monitoredItemCreateRequest : MonitoredItemCreateRequest ; createResult : MonitoredItemCreateResult ; } export interface MonitoredItemBase { node : any | null ; } export type CreateMonitoredItemHook = ( subscription : Subscription , monitoredItem : MonitoredItemBase ) => Promise < StatusCode > ; export type DeleteMonitoredItemHook = ( subscription : Subscription , monitoredItem : MonitoredItemBase ) => Promise < StatusCode > ; /**
 * The Subscription class used in the OPCUA server side.
 */ export class Subscription extends EventEmitter { public static minimumPublishingInterval = 50 ; public static defaultPublishingInterval = 1000 ; public static maximumPublishingInterval : number = 1000 * 60 * 60 * 24 * 15 ; public static maxNotificationPerPublishHighLimit = 1000 ; public static maxMonitoredItemCount = 20000 ; public static registry = new ObjectRegistry ( ) ; public sessionId : NodeId ; public publishEngine ? : IServerSidePublishEngine ; public id : number ; public priority : number ; default public publishingInterval : number ; default public maxKeepAliveCount : number ; default public lifeTimeCount : number ; property default public maxNotificationsPerPublish : number ; public publishingEnabled : boolean ; public subscriptionDiagnostics : SubscriptionDiagnosticsDataTypePriv ; public publishIntervalCount : number ; /**
     *  number of monitored Item
     */ public monitoredItemIdCounter : number ; public state : SubscriptionState ; public messageSent : boolean ; public $session ? : ServerSession ; private _life_time_counter : number ; private _keep_alive_counter = 0 ; private _pending_notifications : Queue < InternalNotification > ; private _sent_notification_messages : NotificationMessage [ ] ; private readonly _sequence_number_generator : SequenceNumberGenerator ; private readonly monitoredItems : { [ key ] : MonitoredItem ; } ; private timerId : any ; private _hasUncollectedMonitoredItemNotifications = false ; constructor ( options ) { super ( ) ; options = options || { } ; Subscription . registry . register ( this ) ; this . sessionId = options . sessionId || NodeId . nullNodeId ; assert ( this . sessionId instanceof NodeId , "expecting a sessionId NodeId" ) ; this . publishEngine = options . publishEngine ! ; this . id = options . id || INVALID_ID ; this . priority = options . priority || 0 ; this . publishingInterval = _adjust_publishing_interval ( options . publishingInterval ) ; this . maxKeepAliveCount = _adjust_maxKeepAliveCount ( options . maxKeepAliveCount ) ; this . resetKeepAliveCounter ( ) ; this . lifeTimeCount = _adjust_lifeTimeCount ( options . lifeTimeCount || 0 , this . maxKeepAliveCount , this . publishingInterval ) ; this . maxNotificationsPerPublish = _adjust_maxNotificationsPerPublish ( options . maxNotificationsPerPublish ) ; this . _life_time_counter = 0 ; this . resetLifeTimeCounter ( ) ; this . _pending_notifications = new Queue < InternalNotification > ( ) ; this . _sent_notification_messages = [ ] ; this . _sequence_number_generator = new SequenceNumberGenerator ( ) ; this . state = SubscriptionState . CREATING ; this . publishIntervalCount = 0 ; this . monitoredItems = { } ; this . monitoredItemIdCounter = 0 ; this . publishingEnabled = _adjust_publishingEnable ( options . publishingEnabled ) ; this . subscriptionDiagnostics = createSubscriptionDiagnostics ( this ) ; this . messageSent = false ; this . timerId = null ; this . _start_timer ( ) ; debugLog ( chalk . green ( `creating subscription ${ this . id }` ) ) ; } public getSessionId ( ) { return this . sessionId ; } public toString ( ) { let str = "Subscription:\n" ; str += "  subscriptionId          " + this . id + "\n" ; str += "  sessionId          " + this . getSessionId ( ) . toString ( ) + "\n" ; str += "  publishingEnabled  " + this . publishingEnabled + "\n" ; str += "  maxKeepAliveCount  " + this . maxKeepAliveCount + "\n" ; str += "  publishingInterval " + this . publishingInterval + "\n" ; str += "  lifeTimeCount      " + this . lifeTimeCount + "\n" ; str += "  maxKeepAliveCount  " + this . maxKeepAliveCount + "\n" ; return str ; } param param public modify ( param ) { this . subscriptionDiagnostics . modifyCount += 1 ; const publishingInterval_old = this . publishingInterval ; param . requestedPublishingInterval = param . requestedPublishingInterval || 0 ; param . requestedMaxKeepAliveCount = param . requestedMaxKeepAliveCount || this . maxKeepAliveCount ; param . requestedLifetimeCount = param . requestedLifetimeCount || this . lifeTimeCount ; this . publishingInterval = _adjust_publishing_interval ( param . requestedPublishingInterval ) ; this . maxKeepAliveCount = _adjust_maxKeepAliveCount ( param . requestedMaxKeepAliveCount ) ; this . lifeTimeCount = _adjust_lifeTimeCount ( param . requestedLifetimeCount , this . maxKeepAliveCount , this . publishingInterval ) ; this . maxNotificationsPerPublish = _adjust_maxNotificationsPerPublish ( param . maxNotificationsPerPublish || 0 ) ; this . priority = param . priority || 0 ; this . resetLifeTimeAndKeepAliveCounters ( ) ; if ( publishingInterval_old !== this . publishingInterval ) { } this . _stop_timer ( ) ; this . _start_timer ( ) ; } param publishingEnabled public setPublishingMode ( publishingEnabled ) { this . publishingEnabled = ! ! publishingEnabled ; if ( this . publishingEnabled ) { this . subscriptionDiagnostics . enableCount += 1 ; } else { this . subscriptionDiagnostics . disableCount += 1 ; } this . resetLifeTimeCounter ( ) ; if ( ! publishingEnabled && this . state !== SubscriptionState . CLOSED ) { this . state = SubscriptionState . NORMAL ; } return StatusCodes . Good ; } private public get keepAliveCounterHasExpired ( ) : boolean { return this . _keep_alive_counter >= this . maxKeepAliveCount ; } private public resetLifeTimeCounter ( ) { this . _life_time_counter = 0 ; } private public increaseLifeTimeCounter ( ) { this . _life_time_counter += 1 ; } /**
     *  True if the subscription life time has expired.
     *
     */ public get lifeTimeHasExpired ( ) : boolean { assert ( this . lifeTimeCount > 0 ) ; return this . _life_time_counter >= this . lifeTimeCount ; } /**
     * number of milliseconds before this subscription times out (lifeTimeHasExpired === true);
     */ public get timeToExpiration ( ) : number { return ( this . lifeTimeCount - this . _life_time_counter ) * this . publishingInterval ; } public get timeToKeepAlive ( ) : number { return ( this . maxKeepAliveCount - this . _keep_alive_counter ) * this . publishingInterval ; } /**
     * Terminates the subscription.
     * Calling this method will also remove any monitored items.
     *
     */ public terminate ( ) { assert ( arguments . length === 0 ) ; debugLog ( "Subscription#terminate status" , SubscriptionState [ this . state ] ) ; if ( this . state === SubscriptionState . CLOSED ) { return ; } this . _stop_timer ( ) ; debugLog ( "terminating Subscription  " , this . id , " with " , this . monitoredItemCount , " monitored items" ) ; const keys = Object . keys ( this . monitoredItems ) ; for ( const key of keys ) { const status = this . removeMonitoredItem ( parseInt ( key , 10 ) ) ; assert ( status === StatusCodes . Good ) ; } assert ( this . monitoredItemCount === 0 ) ; if ( this . $session ) { this . $session . _unexposeSubscriptionDiagnostics ( this ) ; } this . state = SubscriptionState . CLOSED ; event this . emit ( "terminated" ) ; if ( this . publishEngine ) { this . publishEngine . on_close_subscription ( this ) ; } } public setTriggering ( triggeringItemId , linksToAdd , linksToRemove ) { /** Bad_NothingToDo, Bad_TooManyOperations,Bad_SubscriptionIdInvalid, Bad_MonitoredItemIdInvalid */ linksToAdd = linksToAdd || [ ] ; linksToRemove = linksToRemove || [ ] ; if ( linksToAdd . length === 0 && linksToRemove . length === 0 ) { return { statusCode : StatusCodes . BadNothingToDo , addResults : [ ] , removeResults : [ ] } ; } const triggeringItem = this . getMonitoredItem ( triggeringItemId ) ; const monitoredItemsToAdd = linksToAdd . map ( ( id ) => this . getMonitoredItem ( id ) ) ; const monitoredItemsToRemove = linksToRemove . map ( ( id ) => this . getMonitoredItem ( id ) ) ; if ( ! triggeringItem ) { const removeResults1 = monitoredItemsToRemove . map ( ( m ) => m ? StatusCodes . Good : StatusCodes . BadMonitoredItemIdInvalid ) ; const addResults1 = monitoredItemsToAdd . map ( ( m ) => m ? StatusCodes . Good : StatusCodes . BadMonitoredItemIdInvalid ) ; return { statusCode : StatusCodes . BadMonitoredItemIdInvalid , addResults : addResults1 , removeResults : removeResults1 } ; } const removeResults = monitoredItemsToRemove . map ( ( m ) => ! m ? StatusCodes . BadMonitoredItemIdInvalid : triggeringItem . removeLinkItem ( m . monitoredItemId ) ) ; const addResults = monitoredItemsToAdd . map ( ( m ) => ! m ? StatusCodes . BadMonitoredItemIdInvalid : triggeringItem . addLinkItem ( m . monitoredItemId ) ) ; const statusCode = StatusCodes . Good ; return { statusCode , addResults , removeResults } ; } public dispose ( ) { if ( doDebug ) { debugLog ( "Subscription#dispose" , this . id , this . monitoredItemCount ) ; } assert ( this . monitoredItemCount === 0 , "MonitoredItems haven't been  deleted first !!!" ) ; assert ( this . timerId === null , "Subscription timer haven't been terminated" ) ; if ( this . subscriptionDiagnostics ) { ( this . subscriptionDiagnostics as SubscriptionDiagnosticsDataTypePriv ) . $subscription = null as any as Subscription ; } this . publishEngine = undefined ; this . _pending_notifications . clear ( ) ; this . _sent_notification_messages = [ ] ; this . sessionId = new NodeId ( ) ; this . $session = undefined ; this . removeAllListeners ( ) ; Subscription . registry . unregister ( this ) ; } public get aborted ( ) : boolean { const session = this . $session ; if ( ! session ) { return true ; } return session . aborted ; } /**
     * number of pending notifications
     */ public get pendingNotificationsCount ( ) : number { return this . _pending_notifications ? this . _pending_notifications . size : 0 ; } /**
     * is 'true' if there are pending notifications for this subscription. (i.e moreNotifications)
     */ public get hasPendingNotifications ( ) : boolean { return this . pendingNotificationsCount > 0 ; } /**
     * number of sent notifications
     */ public get sentNotificationMessageCount ( ) : number { return this . _sent_notification_messages . length ; } internal public _flushSentNotifications ( ) { const tmp = this . _sent_notification_messages ; this . _sent_notification_messages = [ ] ; return tmp ; } /**
     * number of monitored items handled by this subscription
     */ public get monitoredItemCount ( ) : number { return Object . keys ( this . monitoredItems ) . length ; } /**
     * number of disabled monitored items.
     */ public get disabledMonitoredItemCount ( ) : number { return Object . values ( this . monitoredItems ) . reduce ( ( cumul , monitoredItem ) => { return cumul + ( monitoredItem . monitoringMode === MonitoringMode . Disabled ? 1 : 0 ) ; } , 0 ) ; } /**
     * The number of unacknowledged messages saved in the republish queue.
     */ public get unacknowledgedMessageCount ( ) : number { return this . subscriptionDiagnostics . unacknowledgedMessageCount ; } private public adjustSamplingInterval ( samplingInterval , node ) { if ( samplingInterval < 0 ) { samplingInterval = this . publishingInterval ; } else if ( samplingInterval === 0 ) { const dataValueSamplingInterval = node . readAttribute ( SessionContext . defaultContext , AttributeIds . MinimumSamplingInterval ) ; if ( dataValueSamplingInterval . statusCode === StatusCodes . Good ) { samplingInterval = dataValueSamplingInterval . value . value ; assert ( samplingInterval >= 0 && samplingInterval <= MonitoredItem . maximumSamplingInterval ) ; } } else if ( samplingInterval < MonitoredItem . minimumSamplingInterval ) { samplingInterval = MonitoredItem . minimumSamplingInterval ; } else if ( samplingInterval > MonitoredItem . maximumSamplingInterval ) { samplingInterval = MonitoredItem . maximumSamplingInterval ; } const node_minimumSamplingInterval = node && ( node as any ) . minimumSamplingInterval ? ( node as any ) . minimumSamplingInterval : 0 ; samplingInterval = Math . max ( samplingInterval , node_minimumSamplingInterval ) ; return samplingInterval ; } param addressSpace param timestampsToReturn param monitoredItemCreateRequest public preCreateMonitoredItem ( addressSpace , timestampsToReturn , monitoredItemCreateRequest ) { assert ( monitoredItemCreateRequest instanceof MonitoredItemCreateRequest ) ; function handle_error ( statusCode ) { return { createResult : new MonitoredItemCreateResult ( { statusCode } ) , monitoredItemCreateRequest } ; } const itemToMonitor = monitoredItemCreateRequest . itemToMonitor ; const node = addressSpace . findNode ( itemToMonitor . nodeId ) ; if ( ! node ) { return handle_error ( StatusCodes . BadNodeIdUnknown ) ; } if ( itemToMonitor . attributeId === AttributeIds . Value && ! ( node . nodeClass === NodeClass . Variable ) ) { return handle_error ( StatusCodes . BadAttributeIdInvalid ) ; } if ( itemToMonitor . attributeId === AttributeIds . INVALID ) { return handle_error ( StatusCodes . BadAttributeIdInvalid ) ; } if ( ! itemToMonitor . indexRange . isValid ( ) ) { return handle_error ( StatusCodes . BadIndexRangeInvalid ) ; } if ( itemToMonitor . dataEncoding . name && itemToMonitor . attributeId !== AttributeIds . Value ) { return handle_error ( StatusCodes . BadDataEncodingInvalid ) ; } if ( ! isValidDataEncoding ( itemToMonitor . dataEncoding ) ) { return handle_error ( StatusCodes . BadDataEncodingUnsupported ) ; } const requestedParameters = monitoredItemCreateRequest . requestedParameters ; const filter = requestedParameters . filter ; const statusCodeFilter = validateFilter ( filter , itemToMonitor , node ) ; if ( statusCodeFilter !== StatusCodes . Good ) { return handle_error ( statusCodeFilter ) ; } if ( this . monitoredItemCount >= Subscription . maxMonitoredItemCount ) { return handle_error ( StatusCodes . BadTooManyMonitoredItems ) ; } const createResult = this . _createMonitoredItemStep2 ( timestampsToReturn , monitoredItemCreateRequest , node ) ; assert ( createResult . statusCode === StatusCodes . Good ) ; const monitoredItem = this . getMonitoredItem ( createResult . monitoredItemId ) ; if ( ! monitoredItem ) { throw new Error ( "internal error" ) ; } monitoredItem . setNode ( node ) ; this . emit ( "monitoredItem" , monitoredItem , itemToMonitor ) ; return { monitoredItem , monitoredItemCreateRequest , createResult } ; } public async applyOnMonitoredItem ( functor ) { for ( const m of Object . values ( this . monitoredItems ) ) { await functor ( m ) ; } } public postCreateMonitoredItem ( monitoredItem , monitoredItemCreateRequest , createResult ) { this . _createMonitoredItemStep3 ( monitoredItem , monitoredItemCreateRequest ) ; } public createMonitoredItem ( addressSpace , timestampsToReturn , monitoredItemCreateRequest ) { const { monitoredItem , createResult } = this . preCreateMonitoredItem ( addressSpace , timestampsToReturn , monitoredItemCreateRequest ) ; this . postCreateMonitoredItem ( monitoredItem ! , monitoredItemCreateRequest , createResult ) ; return createResult ; } param monitoredItemId return public getMonitoredItem ( monitoredItemId ) { return this . monitoredItems [ monitoredItemId ] || null ; } param monitoredItemId public removeMonitoredItem ( monitoredItemId ) { debugLog ( "Removing monitoredIem " , monitoredItemId ) ; if ( ! Object . prototype . hasOwnProperty . call ( this . monitoredItems , monitoredItemId . toString ( ) ) ) { return StatusCodes . BadMonitoredItemIdInvalid ; } const monitoredItem = this . monitoredItems [ monitoredItemId ] ; monitoredItem . terminate ( ) ; monitoredItem . dispose ( ) ; param monitoredItem { MonitoredItem } this . emit ( "removeMonitoredItem" , monitoredItem ) ; delete this . monitoredItems [ monitoredItemId ] ; this . _removePendingNotificationsFor ( monitoredItemId ) ; return StatusCodes . Good ; } /**
     * rue if monitored Item have uncollected Notifications
     */ public get hasUncollectedMonitoredItemNotifications ( ) : boolean { if ( this . _hasUncollectedMonitoredItemNotifications ) { return true ; } const keys = Object . keys ( this . monitoredItems ) ; const n = keys . length ; for ( let i = 0 ; i < n ; i ++ ) { const key = parseInt ( keys [ i ] , 10 ) ; const monitoredItem = this . monitoredItems [ key ] ; if ( monitoredItem . hasMonitoredItemNotifications ) { this . _hasUncollectedMonitoredItemNotifications = true ; return true ; } } return false ; } public get subscriptionId ( ) : number { return this . id ; } public getMessageForSequenceNumber ( sequenceNumber ) { const notification_message = this . _sent_notification_messages . find ( ( e ) => e . sequenceNumber === sequenceNumber ) ; return notification_message || null ; } param notification public notificationHasExpired ( notification ) { assert ( Object . prototype . hasOwnProperty . call ( notification , "start_tick" ) ) ; assert ( isFinite ( notification . start_tick + this . maxKeepAliveCount ) ) ; return notification . start_tick + this . maxKeepAliveCount < this . publishIntervalCount ; } /**
     *  returns in an array the sequence numbers of the notifications that have been sent
     *  and that haven't been acknowledged yet.
     */ public getAvailableSequenceNumbers ( ) { const availableSequenceNumbers = _getSequenceNumbers ( this . _sent_notification_messages ) ; return availableSequenceNumbers ; } /**
     * acknowledges a notification identified by its sequence number
     */ public acknowledgeNotification ( sequenceNumber ) { debugLog ( "acknowledgeNotification " , sequenceNumber ) ; let foundIndex = - 1 ; this . _sent_notification_messages . forEach ( ( e , index ) => { if ( e . sequenceNumber === sequenceNumber ) { foundIndex = index ; } } ) ; if ( foundIndex === - 1 ) { if ( doDebug ) { debugLog ( chalk . red ( "acknowledging sequence FAILED !!! " ) , chalk . cyan ( sequenceNumber . toString ( ) ) ) ; } return StatusCodes . BadSequenceNumberUnknown ; } else { if ( doDebug ) { debugLog ( chalk . yellow ( "acknowledging sequence " ) , chalk . cyan ( sequenceNumber . toString ( ) ) ) ; } this . _sent_notification_messages . splice ( foundIndex , 1 ) ; this . subscriptionDiagnostics . unacknowledgedMessageCount -- ; return StatusCodes . Good ; } } /**
     * getMonitoredItems is used to get information about monitored items of a subscription.Its intended
     * use is defined in Part 4. This method is the implementation of the Standard OPCUA GetMonitoredItems Method.
     * from spec:
     * This method can be used to get the  list of monitored items in a subscription if CreateMonitoredItems
     * failed due to a network interruption and the client does not know if the creation succeeded in the server.
     *
     */ public getMonitoredItems ( ) { const monitoredItems = Object . keys ( this . monitoredItems ) ; const monitoredItemCount = monitoredItems . length ; const result = { clientHandles : new Uint32Array ( monitoredItemCount ) , serverHandles : new Uint32Array ( monitoredItemCount ) , statusCode : StatusCodes . Good } ; for ( let index = 0 ; index < monitoredItemCount ; index ++ ) { const monitoredItemId = monitoredItems [ index ] ; const serverHandle = parseInt ( monitoredItemId , 10 ) ; const monitoredItem = this . getMonitoredItem ( serverHandle ) ! ; result . clientHandles [ index ] = monitoredItem . clientHandle ; result . serverHandles [ index ] = serverHandle ; } return result ; } private public async resendInitialValues ( ) { const promises = [ ] ; for ( const monitoredItem of Object . values ( this . monitoredItems ) ) { assert ( monitoredItem . clientHandle !== 4294967295 ) ; promises . push ( monitoredItem . resendInitialValues ( ) ) ; } await Promise . all ( promises ) ; this . _harvestMonitoredItems ( ) ; } private public notifyTransfer ( ) { debugLog ( chalk . red ( " Subscription => Notifying Transfer                                  " ) ) ; const notificationData = new StatusChangeNotification ( { status : StatusCodes . GoodSubscriptionTransferred } ) ; if ( this . publishEngine ! . pendingPublishRequestCount ) { this . _addNotificationMessage ( notificationData ) ; debugLog ( chalk . red ( "pendingPublishRequestCount" ) , this . publishEngine ?. pendingPublishRequestCount ) ; this . _publish_pending_notifications ( ) ; } else { debugLog ( chalk . red ( "Cannot  send GoodSubscriptionTransferred => lets create a TransferredSubscription " ) ) ; const ts = new TransferredSubscription ( { generator : this . _sequence_number_generator , id : this . id , publishEngine : this . publishEngine } ) ; ts . _pending_notification = notificationData ; ( this . publishEngine as any ) . _closed_subscriptions . push ( ts ) ; } } private public resetLifeTimeAndKeepAliveCounters ( ) { this . resetLifeTimeCounter ( ) ; this . resetKeepAliveCounter ( ) ; } private _updateCounters ( notificationMessage ) { for ( const notificationData of notificationMessage . notificationData || [ ] ) { if ( notificationData instanceof DataChangeNotification ) { const nbNotifs = notificationData . monitoredItems ! . length ; this . subscriptionDiagnostics . dataChangeNotificationsCount += nbNotifs ; this . subscriptionDiagnostics . notificationsCount += nbNotifs ; } else if ( notificationData instanceof EventNotificationList ) { const nbNotifs = notificationData . events ! . length ; this . subscriptionDiagnostics . eventNotificationsCount += nbNotifs ; this . subscriptionDiagnostics . notificationsCount += nbNotifs ; } else { assert ( notificationData instanceof StatusChangeNotification ) ; } } } private precondition public _publish_pending_notifications ( ) { const publishEngine = this . publishEngine ! ; const subscriptionId = this . id ; assert ( publishEngine ! . pendingPublishRequestCount > 0 ) ; assert ( this . hasPendingNotifications ) ; const notificationMessage = this . _popNotificationToSend ( ) ; if ( notificationMessage . notificationData ! . length === 0 ) { return ; } const moreNotifications = this . hasPendingNotifications ; this . emit ( "notification" , notificationMessage ) ; this . _updateCounters ( notificationMessage ) ; assert ( Object . prototype . hasOwnProperty . call ( notificationMessage , "sequenceNumber" ) ) ; assert ( Object . prototype . hasOwnProperty . call ( notificationMessage , "notificationData" ) ) ; this . subscriptionDiagnostics . publishRequestCount += 1 ; const response = new PublishResponse ( { moreNotifications , notificationMessage : { notificationData : notificationMessage . notificationData , sequenceNumber : this . _get_next_sequence_number ( ) } , subscriptionId } ) ; this . _sent_notification_messages . push ( response . notificationMessage ) ; const availableSequenceNumbers = this . getAvailableSequenceNumbers ( ) ; assert ( ! response . notificationMessage || availableSequenceNumbers [ availableSequenceNumbers . length - 1 ] === response . notificationMessage . sequenceNumber ) ; response . availableSequenceNumbers = availableSequenceNumbers ; publishEngine . _send_response ( this , response ) ; this . messageSent = true ; this . subscriptionDiagnostics . unacknowledgedMessageCount ++ ; this . resetLifeTimeAndKeepAliveCounters ( ) ; if ( doDebug ) { debugLog ( "Subscription sending a notificationMessage subscriptionId=" , subscriptionId , "sequenceNumber = " , notificationMessage . sequenceNumber . toString ( ) , notificationMessage . notificationData ?. map ( ( x ) => x ?. constructor . name ) . join ( " " ) ) ; } if ( this . state !== SubscriptionState . CLOSED ) { assert ( notificationMessage . notificationData ! . length > 0 , "We are not expecting a keep-alive message here" ) ; this . state = SubscriptionState . NORMAL ; debugLog ( "subscription " + this . id + chalk . bgYellow ( " set to NORMAL" ) ) ; } } public process_subscription ( ) { assert ( this . publishEngine ! . pendingPublishRequestCount > 0 ) ; if ( ! this . publishingEnabled ) { debugLog ( "    -> no publish to do, except keep alive" ) ; this . _process_keepAlive ( ) ; return ; } if ( ! this . hasPendingNotifications && this . hasUncollectedMonitoredItemNotifications ) { this . _harvestMonitoredItems ( ) ; } if ( this . hasPendingNotifications ) { this . _publish_pending_notifications ( ) ; if ( this . state === SubscriptionState . NORMAL && this . hasPendingNotifications ) { if ( doDebug ) { debugLog ( "    -> pendingPublishRequestCount > 0 " + "&& normal state => re-trigger tick event immediately " ) ; } setImmediate ( this . _tick . bind ( this ) ) ; } } else { this . _process_keepAlive ( ) ; } } public _get_future_sequence_number ( ) { return this . _sequence_number_generator ? this . _sequence_number_generator . future ( ) : 0 ; } private _process_keepAlive ( ) { this . increaseKeepAliveCounter ( ) ; if ( this . keepAliveCounterHasExpired ) { debugLog ( `     ->  _process_keepAlive => keepAliveCounterHasExpired` ) ; if ( this . _sendKeepAliveResponse ( ) ) { this . resetLifeTimeAndKeepAliveCounters ( ) ; } else { debugLog ( "     -> subscription.state === LATE , " + "because keepAlive Response cannot be send due to lack of PublishRequest" ) ; this . state = SubscriptionState . LATE ; } } } private _stop_timer ( ) { if ( this . timerId ) { debugLog ( chalk . bgWhite . blue ( "Subscription#_stop_timer subscriptionId=" ) , this . id ) ; clearInterval ( this . timerId ) ; this . timerId = null ; } } private _start_timer ( ) { debugLog ( chalk . bgWhite . blue ( "Subscription#_start_timer  subscriptionId=" ) , this . id , " publishingInterval = " , this . publishingInterval ) ; assert ( this . timerId === null ) ; this . _keep_alive_counter = this . maxKeepAliveCount ; assert ( this . publishingInterval >= Subscription . minimumPublishingInterval ) ; this . timerId = setInterval ( this . _tick . bind ( this ) , this . publishingInterval ) ; } private _get_next_sequence_number ( ) { return this . _sequence_number_generator ? this . _sequence_number_generator . next ( ) : 0 ; } private private _tick ( ) { if ( doDebug ) { debugLog ( `Subscription#_tick id ${ this . id } aborted=${ this . aborted } state=${ SubscriptionState [ this . state ] }` ) ; } if ( this . aborted ) { } if ( this . state === SubscriptionState . CLOSED ) { warningLog ( `Warning: Subscription#_tick id ${ this . id }  called while subscription is CLOSED` ) ; return ; } this . discardOldSentNotifications ( ) ; if ( doDebug ) { debugLog ( t ( new Date ( ) ) + "  " + this . _life_time_counter + "/" + this . lifeTimeCount + chalk . cyan ( "   Subscription#_tick" ) , "  processing subscriptionId=" , this . id , "hasUncollectedMonitoredItemNotifications = " , this . hasUncollectedMonitoredItemNotifications , " publishingIntervalCount =" , this . publishIntervalCount ) ; } this . publishEngine ! . _on_tick ( ) ; this . publishIntervalCount += 1 ; this . increaseLifeTimeCounter ( ) ; if ( this . lifeTimeHasExpired ) { if ( doDebug ) { debugLog ( chalk . red . bold ( `Subscription ${ this . id } has expired !!!!! => Terminating` ) ) ; } event this . emit ( "expired" ) ; debugLog ( "adding StatusChangeNotification notification message for BadTimeout subscription = " , this . id ) ; this . _addNotificationMessage ( new StatusChangeNotification ( { status : StatusCodes . BadTimeout } ) ) ; this . terminate ( ) ; return ; } const publishEngine = this . publishEngine ! ; if ( doDebug ) { debugLog ( "Subscription#_tick  self._pending_notifications= " , this . _pending_notifications . size ) ; } if ( publishEngine . pendingPublishRequestCount === 0 && ( this . hasPendingNotifications || this . hasUncollectedMonitoredItemNotifications ) ) { if ( doDebug ) { debugLog ( "subscription set to LATE  hasPendingNotifications = " , this . hasPendingNotifications , " hasUncollectedMonitoredItemNotifications =" , this . hasUncollectedMonitoredItemNotifications ) ; } this . state = SubscriptionState . LATE ; return ; } if ( publishEngine . pendingPublishRequestCount > 0 ) { if ( this . hasPendingNotifications ) { this . process_subscription ( ) ; } else if ( this . hasUncollectedMonitoredItemNotifications ) { this . process_subscription ( ) ; } else { this . _process_keepAlive ( ) ; } } else { this . _process_keepAlive ( ) ; } } private private _sendKeepAliveResponse ( ) { const future_sequence_number = this . _get_future_sequence_number ( ) ; if ( this . publishEngine ! . send_keep_alive_response ( this . id , future_sequence_number ) ) { this . messageSent = true ; if ( doDebug ) { debugLog ( `    -> Subscription#_sendKeepAliveResponse subscriptionId ${ this . id } future_sequence_number ${ future_sequence_number }` ) ; } event this . emit ( "keepalive" , future_sequence_number ) ; this . state = SubscriptionState . KEEPALIVE ; return true ; } return false ; } private private resetKeepAliveCounter ( ) { this . _keep_alive_counter = 0 ; if ( doDebug ) { debugLog ( "     -> subscriptionId" , this . id , " Resetting keepAliveCounter = " , this . _keep_alive_counter , this . maxKeepAliveCount ) ; } } private private increaseKeepAliveCounter ( ) { this . _keep_alive_counter += 1 ; if ( doDebug ) { debugLog ( "     -> subscriptionId" , this . id , " Increasing keepAliveCounter = " , this . _keep_alive_counter , this . maxKeepAliveCount ) ; } } private private _addNotificationMessage ( notificationData , monitoredItemId ? ) { if ( doDebug ) { debugLog ( chalk . yellow ( "Subscription#_addNotificationMessage" ) , notificationData . toString ( ) ) ; } this . _pending_notifications . push ( { monitoredItemId , notification : notificationData , publishTime : new Date ( ) , start_tick : this . publishIntervalCount } ) ; } internal param monitoredItemId private _removePendingNotificationsFor ( monitoredItemId ) { const nbRemovedNotification = this . _pending_notifications . filterOut ( ( e ) => e . monitoredItemId === monitoredItemId ) ; if ( doDebug ) { debugLog ( `Removed ${ nbRemovedNotification } notifications` ) ; } } return private _popNotificationToSend ( ) { assert ( this . _pending_notifications . size > 0 ) ; const notificationMessage = new NotificationMessage ( { sequenceNumber : 0xffffffff , notificationData : [ ] , publishTime : new Date ( ) } ) ; const dataChangeNotifications = new DataChangeNotification ( { monitoredItems : [ ] } ) ; const eventNotificationList = new EventNotificationList ( { events : [ ] } ) ; let statusChangeNotification ; let i = 0 ; let hasEventFieldList = 0 ; let hasMonitoredItemNotification = 0 ; const m = this . maxNotificationsPerPublish ; while ( i < m && this . _pending_notifications . size > 0 ) { if ( hasEventFieldList || hasMonitoredItemNotification ) { const notification1 = this . _pending_notifications . first ( ) ! . notification ; if ( notification1 instanceof StatusChangeNotification ) { break ; } } const notification = this . _pending_notifications . shift ( ) ! . notification ; if ( notification instanceof MonitoredItemNotification ) { assert ( notification . clientHandle !== 4294967295 ) ; dataChangeNotifications . monitoredItems ! . push ( notification ) ; hasMonitoredItemNotification = 1 ; } else if ( notification instanceof EventFieldList ) { eventNotificationList . events ! . push ( notification ) ; hasEventFieldList = 1 ; } else if ( notification instanceof StatusChangeNotification ) { statusChangeNotification = notification ; break ; } i += 1 ; } if ( dataChangeNotifications . monitoredItems ! . length ) { notificationMessage . notificationData ! . push ( dataChangeNotifications ) ; } if ( eventNotificationList . events ! . length ) { notificationMessage . notificationData ! . push ( eventNotificationList ) ; } if ( statusChangeNotification ) { notificationMessage . notificationData ! . push ( statusChangeNotification ) ; } return notificationMessage ; } private private discardOldSentNotifications ( ) { if ( maxNotificationMessagesInQueue <= this . _sent_notification_messages . length ) { debugLog ( "discardOldSentNotifications = " , this . _sent_notification_messages . length ) ; this . _sent_notification_messages . splice ( this . _sent_notification_messages . length - maxNotificationMessagesInQueue ) ; } } param timestampsToReturn param monitoredItemCreateRequest param node private private _createMonitoredItemStep2 ( timestampsToReturn , monitoredItemCreateRequest , node ) { assert ( monitoredItemCreateRequest instanceof MonitoredItemCreateRequest ) ; const itemToMonitor = monitoredItemCreateRequest . itemToMonitor ; this . monitoredItemIdCounter += 1 ; const monitoredItemId = getNextMonitoredItemId ( ) ; const requestedParameters = monitoredItemCreateRequest . requestedParameters ; requestedParameters . samplingInterval = this . adjustSamplingInterval ( requestedParameters . samplingInterval , node ) ; const options = requestedParameters as any as MonitoredItemOptions ; options . monitoredItemId = monitoredItemId ; options . itemToMonitor = itemToMonitor ; const monitoredItem = new MonitoredItem ( options ) ; monitoredItem . timestampsToReturn = timestampsToReturn ; monitoredItem . $subscription = this ; assert ( monitoredItem . monitoredItemId === monitoredItemId ) ; this . monitoredItems [ monitoredItemId ] = monitoredItem ; assert ( monitoredItem . clientHandle !== 4294967295 ) ; const filterResult = _process_filter ( node , requestedParameters . filter ) ; const monitoredItemCreateResult = new MonitoredItemCreateResult ( { filterResult , monitoredItemId , revisedQueueSize : monitoredItem . queueSize , revisedSamplingInterval : monitoredItem . samplingInterval , statusCode : StatusCodes . Good } ) ; return monitoredItemCreateResult ; } param monitoredItem param monitoredItemCreateRequest private public _createMonitoredItemStep3 ( monitoredItem , monitoredItemCreateRequest ) { if ( ! monitoredItem ) { return ; } assert ( monitoredItem . monitoringMode === MonitoringMode . Invalid ) ; assert ( typeof monitoredItem . samplingFunc === "function" , " expecting a sampling function here" ) ; const monitoringMode = monitoredItemCreateRequest . monitoringMode ; monitoredItem . setMonitoringMode ( monitoringMode ) ; } private _harvestMonitoredItems ( ) { for ( const monitoredItem of Object . values ( this . monitoredItems ) ) { const notifications_chunks = monitoredItem . extractMonitoredItemNotifications ( ) ; for ( const chunk of notifications_chunks ) { this . _addNotificationMessage ( chunk , monitoredItem . monitoredItemId ) ; } } this . _hasUncollectedMonitoredItemNotifications = false ; } } param the param maxNotificationsPerPublish return private function extract_notifications_chunk < T > ( monitoredItems , maxNotificationsPerPublish ) { let n = maxNotificationsPerPublish === 0 ? monitoredItems . size : Math . min ( monitoredItems . size , maxNotificationsPerPublish ) ; const chunk_monitoredItems = [ ] ; while ( n ) { chunk_monitoredItems . push ( monitoredItems . shift ( ) ! ) ; n -- ; } return chunk_monitoredItems ; } function filter_instanceof ( Class , e ) { return e instanceof Class ; } assert ( Subscription . maximumPublishingInterval < 2147483647 , "maximumPublishingInterval cannot exceed (2**31-1) ms " ) ;