module import { EventEmitter } from "events" ; import * as chalk from "chalk" ; import { partition , sortBy } from "lodash" ; import { assert } from "node-opcua-assert" ; import { checkDebugFlag , make_debugLog } from "node-opcua-debug" ; import { ObjectRegistry } from "node-opcua-object-registry" ; import { StatusCode , StatusCodes } from "node-opcua-status-code" ; import { PublishRequest , PublishResponse , SubscriptionAcknowledgement } from "node-opcua-types" ; import { Subscription } from "./server_subscription" ; import { SubscriptionState } from "./server_subscription" ; import { IServerSidePublishEngine , INotifMsg , IClosedOrTransferredSubscription } from "./i_server_side_publish_engine" ; const debugLog = make_debugLog ( __filename ) ; const doDebug = checkDebugFlag ( __filename ) ; function traceLog ( ... args ) { if ( ! doDebug ) { return ; } const a = args . map ( ( x ? ) => x ! ) ; a . unshift ( chalk . yellow ( " TRACE " ) ) ; debugLog ( ... a ) ; } export interface ServerSidePublishEngineOptions { maxPublishRequestInQueue ? : number ; } interface PublishData { request : PublishRequest ; serverTimeWhenReceived : number ; results : StatusCode [ ] ; callback : ( request : PublishRequest , response : PublishResponse ) => void ; } function _assertValidPublishData ( publishData ) { assert ( publishData . request instanceof PublishRequest ) ; assert ( typeof publishData . serverTimeWhenReceived === "number" ) ; assert ( Array . isArray ( publishData . results ) ) ; assert ( typeof publishData . callback === "function" ) ; } function dummy_function ( ) { } function addDate ( date , delta ) { return new Date ( date . getTime ( ) + delta ) ; } function timeout_filter ( publishData ) { const request = publishData . request ; const results = publishData . results ; if ( ! request . requestHeader . timeoutHint ) { return false ; } const serverTimeWhenReceived = publishData . serverTimeWhenReceived ; const expected_timeout_time = addDate ( new Date ( serverTimeWhenReceived ) , request . requestHeader . timeoutHint ) ; return expected_timeout_time . getTime ( ) < Date . now ( ) ; } /***
 *  a Publish Engine for a given session
 */ export class ServerSidePublishEngine extends EventEmitter implements IServerSidePublishEngine { public static registry = new ObjectRegistry ( ) ; private public static transferSubscriptionsToOrphan ( srcPublishEngine , destPublishEngine ) { debugLog ( chalk . yellow ( "ServerSidePublishEngine#transferSubscriptionsToOrphan! " + "start transferring long live subscriptions to orphan" ) ) ; for ( const subscription of Object . values ( srcPublishEngine . _subscriptions ) ) { assert ( ( subscription . publishEngine as any ) === srcPublishEngine ) ; if ( subscription . $session ) { subscription . $session . _unexposeSubscriptionDiagnostics ( subscription ) ; } else { console . warn ( "Warning:  subscription" , subscription . id , " has no session attached!!!" ) ; } ServerSidePublishEngine . transferSubscription ( subscription , destPublishEngine , false ) ; } assert ( srcPublishEngine . subscriptionCount === 0 ) ; debugLog ( chalk . yellow ( "ServerSidePublishEngine#transferSubscriptionsToOrphan! " + "end transferring long lived subscriptions to orphan" ) ) ; } param subscription param destPublishEngine param sendInitialValues private public static async transferSubscription ( subscription , destPublishEngine , sendInitialValues ) { const srcPublishEngine = subscription . publishEngine as any as ServerSidePublishEngine ; assert ( ! destPublishEngine . getSubscriptionById ( subscription . id ) ) ; assert ( srcPublishEngine . getSubscriptionById ( subscription . id ) ) ; destPublishEngine . _purge_dangling_subscription ( subscription . id ) ; debugLog ( chalk . cyan ( "ServerSidePublishEngine.transferSubscription live subscriptionId =" ) , subscription . subscriptionId ) ; debugLog ( chalk . cyan ( "ServerSidePublishEngine.transferSubscription with  = " ) , subscription . getAvailableSequenceNumbers ( ) ) ; subscription . notifyTransfer ( ) ; destPublishEngine . add_subscription ( srcPublishEngine . detach_subscription ( subscription ) ) ; subscription . resetLifeTimeCounter ( ) ; if ( sendInitialValues ) { debugLog ( "Resending initial values" ) ; await subscription . resendInitialValues ( ) ; } assert ( destPublishEngine . getSubscriptionById ( subscription . id ) ) ; assert ( ! srcPublishEngine . getSubscriptionById ( subscription . id ) ) ; } public maxPublishRequestInQueue = 0 ; public isSessionClosed = false ; private _publish_request_queue : PublishData [ ] = [ ] ; private _subscriptions : { [ key ] : Subscription ; } ; private _closed_subscriptions : IClosedOrTransferredSubscription [ ] = [ ] ; constructor ( options ? ) { super ( ) ; options = options || { } ; ServerSidePublishEngine . registry . register ( this ) ; this . _publish_request_queue = [ ] ; this . _subscriptions = { } ; this . _closed_subscriptions = [ ] ; this . maxPublishRequestInQueue = options . maxPublishRequestInQueue || 100 ; this . isSessionClosed = false ; } public toString ( ) { let str = "" ; str += `maxPublishRequestInQueue ${ this . maxPublishRequestInQueue }\n` ; str += `subscriptions ${ Object . keys ( this . _subscriptions ) . join ( ) }\n` ; str += `closed subscriptions ${ this . _closed_subscriptions . map ( ( s ) => s . id ) . join ( ) }\n` ; return str ; } public dispose ( ) { debugLog ( "ServerSidePublishEngine#dispose" ) ; assert ( Object . keys ( this . _subscriptions ) . length === 0 , "self._subscriptions count!=0" ) ; this . _subscriptions = { } ; assert ( this . _closed_subscriptions . length === 0 , "self._closed_subscriptions count!=0" ) ; this . _closed_subscriptions = [ ] ; ServerSidePublishEngine . registry . unregister ( this ) ; } public process_subscriptionAcknowledgements ( subscriptionAcknowledgements ) { subscriptionAcknowledgements = subscriptionAcknowledgements || [ ] ; debugLog ( "process_subscriptionAcknowledgements = " , subscriptionAcknowledgements ) ; const results = subscriptionAcknowledgements . map ( ( subscriptionAcknowledgement ) => { const subscription = this . getSubscriptionById ( subscriptionAcknowledgement . subscriptionId ) ; if ( ! subscription ) { return StatusCodes . BadSubscriptionIdInvalid ; } return subscription . acknowledgeNotification ( subscriptionAcknowledgement . sequenceNumber ) ; } ) ; return results ; } /**
     * get a array of subscription handled by the publish engine.
     */ public get subscriptions ( ) : Subscription [ ] { return Object . values ( this . _subscriptions ) ; } /**
     */ public add_subscription ( subscription ) { assert ( subscription instanceof Subscription ) ; assert ( isFinite ( subscription . id ) ) ; subscription . publishEngine = ( subscription . publishEngine || this ) as any ; assert ( ( subscription . publishEngine as any ) === this ) ; assert ( ! this . _subscriptions [ subscription . id ] ) ; debugLog ( "ServerSidePublishEngine#add_subscription -  adding subscription with Id:" , subscription . id ) ; this . _subscriptions [ subscription . id ] = subscription ; return subscription ; } public detach_subscription ( subscription ) { assert ( subscription instanceof Subscription ) ; assert ( isFinite ( subscription . id ) ) ; assert ( ( subscription . publishEngine as any ) === this ) ; assert ( this . _subscriptions [ subscription . id ] === subscription ) ; delete this . _subscriptions [ subscription . id ] ; subscription . publishEngine = null as any ; debugLog ( "ServerSidePublishEngine#detach_subscription detaching subscription with Id:" , subscription . id ) ; return subscription ; } /**
     */ public shutdown ( ) { if ( this . subscriptionCount !== 0 ) { debugLog ( chalk . red ( "Shutting down pending subscription" ) ) ; this . subscriptions . map ( ( subscription ) => subscription . terminate ( ) ) ; } assert ( this . subscriptionCount === 0 , "subscription shall be removed first before you can shutdown a publish engine" ) ; debugLog ( "ServerSidePublishEngine#shutdown" ) ; this . _publish_request_queue = [ ] ; this . _closed_subscriptions . map ( ( subscription ) => subscription . dispose ( ) ) ; this . _closed_subscriptions = [ ] ; } /**
     * number of pending PublishRequest available in queue
     */ public get pendingPublishRequestCount ( ) : number { return this . _publish_request_queue . length ; } /**
     * number of subscriptions
     */ public get subscriptionCount ( ) : number { return Object . keys ( this . _subscriptions ) . length ; } public get pendingClosedSubscriptionCount ( ) : number { return this . _closed_subscriptions . length ; } public get currentMonitoredItemCount ( ) : number { const subscriptions = Object . values ( this . _subscriptions ) ; const result = subscriptions . reduce ( ( cumul , subscription ) => { return cumul + subscription . monitoredItemCount ; } , 0 ) ; assert ( isFinite ( result ) ) ; return result ; } public _purge_dangling_subscription ( subscriptionId ) { this . _closed_subscriptions = this . _closed_subscriptions . filter ( ( s ) => s . id !== subscriptionId ) ; } public on_close_subscription ( subscription ) { debugLog ( "ServerSidePublishEngine#on_close_subscription" , subscription . id ) ; if ( subscription . hasPendingNotifications ) { debugLog ( "ServerSidePublishEngine#on_close_subscription storing subscription" , subscription . id , " to _closed_subscriptions because it has pending notification" ) ; this . _closed_subscriptions . push ( subscription ) ; } else { debugLog ( "ServerSidePublishEngine#on_close_subscription disposing subscription" , subscription . id ) ; subscription . dispose ( ) ; } delete this . _subscriptions [ subscription . id ] ; while ( this . _feed_closed_subscription ( ) ) { } if ( this . subscriptionCount === 0 && this . _closed_subscriptions . length === 0 ) { this . cancelPendingPublishRequest ( ) ; } } param subscriptionId return public getSubscriptionById ( subscriptionId ) { return this . _subscriptions [ subscriptionId . toString ( ) ] ; } public findLateSubscriptions ( ) { const subscriptions = Object . values ( this . _subscriptions ) ; return subscriptions . filter ( ( subscription ) => { return subscription . state === SubscriptionState . LATE && subscription . publishingEnabled ; } ) ; } public get hasLateSubscriptions ( ) : boolean { return this . findLateSubscriptions ( ) . length > 0 ; } public findLateSubscriptionsSortedByAge ( ) { let late_subscriptions = this . findLateSubscriptions ( ) ; late_subscriptions = sortBy ( late_subscriptions , "timeToExpiration" ) ; return late_subscriptions ; } public cancelPendingPublishRequestBeforeChannelChange ( ) { this . _cancelPendingPublishRequest ( StatusCodes . BadSecureChannelClosed ) ; } public onSessionClose ( ) { this . isSessionClosed = true ; this . _cancelPendingPublishRequest ( StatusCodes . BadSessionClosed ) ; } private public cancelPendingPublishRequest ( ) { assert ( this . subscriptionCount === 0 ) ; this . _cancelPendingPublishRequest ( StatusCodes . BadNoSubscription ) ; } param request param callback private internal public _on_PublishRequest ( request , callback ? ) { callback = callback || dummy_function ; assert ( typeof callback === "function" ) ; if ( ! ( request instanceof PublishRequest ) ) { throw new Error ( "Internal error : expecting a Publish Request here" ) ; } const subscriptionAckResults = this . process_subscriptionAcknowledgements ( request . subscriptionAcknowledgements || [ ] ) ; const currentTime = Date . now ( ) ; const publishData = { callback , request , results : subscriptionAckResults , serverTimeWhenReceived : currentTime } ; if ( this . isSessionClosed ) { traceLog ( "server has received a PublishRequest but session is Closed" ) ; this . _send_error_for_request ( publishData , StatusCodes . BadSessionClosed ) ; } else if ( this . subscriptionCount === 0 ) { if ( this . _closed_subscriptions . length > 0 && this . _closed_subscriptions [ 0 ] . hasPendingNotifications ) { const verif = this . _publish_request_queue . length ; this . _publish_request_queue . push ( publishData ) ; const processed = this . _feed_closed_subscription ( ) ; return ; } traceLog ( "server has received a PublishRequest but has no subscription opened" ) ; this . _send_error_for_request ( publishData , StatusCodes . BadNoSubscription ) ; } else { this . _publish_request_queue . push ( publishData ) ; assert ( this . pendingPublishRequestCount > 0 ) ; debugLog ( chalk . bgWhite . red ( "Adding a PublishRequest to the queue " ) , this . _publish_request_queue . length ) ; this . _feed_closed_subscription ( ) ; this . _feed_late_subscription ( ) ; this . _handle_too_many_requests ( ) ; } } private _find_starving_subscription ( ) { const late_subscriptions = this . findLateSubscriptions ( ) ; function compare_subscriptions ( s1 , s2 ) { if ( s1 . priority === s2 . priority ) { return s1 . timeToExpiration < s2 . timeToExpiration ? 1 : 0 ; } return s1 . priority > s2 . priority ? 1 : 0 ; } function findLateSubscriptionSortedByPriority ( ) { if ( late_subscriptions . length === 0 ) { return null ; } late_subscriptions . sort ( compare_subscriptions ) ; if ( doDebug ) { debugLog ( late_subscriptions . map ( ( s ) => "[ id = " + s . id + " prio=" + s . priority + " t=" + s . timeToExpiration + " ka=" + s . timeToKeepAlive + " m?=" + s . hasUncollectedMonitoredItemNotifications + "]" ) . join ( " \n" ) ) ; } return late_subscriptions [ late_subscriptions . length - 1 ] ; } if ( this . _closed_subscriptions ) { } const starving_subscription = findLateSubscriptionSortedByPriority ( ) ; return starving_subscription ; } private _feed_late_subscription ( ) { setImmediate ( ( ) => { if ( ! this . pendingPublishRequestCount ) { return ; } const starving_subscription = this . _find_starving_subscription ( ) ; if ( starving_subscription ) { debugLog ( chalk . bgWhite . red ( "feeding most late subscription subscriptionId  = " ) , starving_subscription . id ) ; starving_subscription . process_subscription ( ) ; } } ) ; } private _feed_closed_subscription ( ) { if ( ! this . pendingPublishRequestCount ) { return false ; } if ( this . _closed_subscriptions . length === 0 ) { debugLog ( "ServerSidePublishEngine#_feed_closed_subscription  -> nothing to do" ) ; return false ; } const closed_subscription = this . _closed_subscriptions [ 0 ] ! ; assert ( closed_subscription . hasPendingNotifications ) ; debugLog ( "ServerSidePublishEngine#_feed_closed_subscription for closed_subscription " , closed_subscription . id ) ; closed_subscription ?. _publish_pending_notifications ( ) ; if ( ! closed_subscription ?. hasPendingNotifications ) { closed_subscription . dispose ( ) ; this . _closed_subscriptions . shift ( ) ; } return true ; } private _send_error_for_request ( publishData , statusCode ) { _assertValidPublishData ( publishData ) ; const publishResponse = new PublishResponse ( { responseHeader : { serviceResult : statusCode } } ) ; this . _send_response_for_request ( publishData , publishResponse ) ; } private _cancelPendingPublishRequest ( statusCode ) { if ( this . _publish_request_queue ) { debugLog ( chalk . red ( "Cancelling pending PublishRequest with statusCode  " ) , statusCode . toString ( ) , " length =" , this . _publish_request_queue . length ) ; } else { debugLog ( chalk . red ( "No pending PublishRequest to cancel" ) ) ; } for ( const publishData of this . _publish_request_queue ) { this . _send_error_for_request ( publishData , statusCode ) ; } this . _publish_request_queue = [ ] ; } private _handle_too_many_requests ( ) { if ( this . pendingPublishRequestCount > this . maxPublishRequestInQueue ) { traceLog ( "server has received too many PublishRequest" , this . pendingPublishRequestCount , "/" , this . maxPublishRequestInQueue ) ; assert ( this . pendingPublishRequestCount === this . maxPublishRequestInQueue + 1 ) ; const publishData = this . _publish_request_queue . shift ( ) ! ; this . _send_error_for_request ( publishData , StatusCodes . BadTooManyPublishRequests ) ; } } method param subscriptionId param future_sequence_number return public send_keep_alive_response ( subscriptionId , future_sequence_number ) { const subscription = this . getSubscriptionById ( subscriptionId ) ; if ( ! subscription ) { traceLog ( "send_keep_alive_response  => invalid subscriptionId = " , subscriptionId ) ; return false ; } if ( this . pendingPublishRequestCount === 0 || subscription . hasPendingNotifications ) { traceLog ( "send_keep_alive_response  => cannot send keep-alive  (no PublishRequest left) subscriptionId = " , subscriptionId ) ; return false ; } debugLog ( `Sending keep alive response for subscription id ${ subscription . id } ${ subscription . publishingInterval } ${ subscription . maxKeepAliveCount }` ) ; this . _send_response ( subscription , new PublishResponse ( { availableSequenceNumbers : subscription . getAvailableSequenceNumbers ( ) , moreNotifications : false , notificationMessage : { sequenceNumber : future_sequence_number } , subscriptionId } ) ) ; return true ; } public _send_response ( subscription , response ) { assert ( this . pendingPublishRequestCount > 0 ) ; assert ( response . subscriptionId !== 0xffffff ) ; const publishData = this . _publish_request_queue . shift ( ) ! ; this . _send_response_for_request ( publishData , response ) ; } public _on_tick ( ) { this . _cancelTimeoutRequests ( ) ; } private _cancelTimeoutRequests ( ) { if ( this . _publish_request_queue . length === 0 ) { return ; } const parts = partition ( this . _publish_request_queue , timeout_filter ) ; this . _publish_request_queue = parts [ 1 ] ; const invalid_published_request = parts [ 0 ] ; for ( const publishData of invalid_published_request ) { if ( doDebug ) { debugLog ( chalk . cyan ( " CANCELING TIMEOUT PUBLISH REQUEST " ) ) ; } this . _send_error_for_request ( publishData , StatusCodes . BadTimeout ) ; } } public _send_response_for_request ( publishData , response ) { if ( doDebug ) { debugLog ( "_send_response_for_request " , response . toString ( ) ) ; } _assertValidPublishData ( publishData ) ; response . results = publishData . results ; response . responseHeader . requestHandle = publishData . request . requestHeader . requestHandle ; publishData . callback ( publishData . request , response ) ; } }