import {SubscribeStream} from 'atscm/src/lib/server/Watcher.js'

public class | source

SubscribeStream

Extends:

Stream → QueueStream → SubscribeStream

A stream that monitors changes in the read nodes.

Test:

SubscribeStream

Constructor Summary

Public Constructor
public	constructor(options: Object) Creates a new SubscribeStream based on some options.

Member Summary

Public Members
public	subscription: node-opcua~ClientSubscription The node-opcua~ClientSubscription to use to monitor nodes.

Private Members
private	_trackChanges: Boolean Set to true once all nodes are monitored.

Method Summary

Public Methods
public	createSubscription() Creates the server subscription to monitor nodes with.
public	processChunk(referenceDescription: node-opcua~ReferenceDescription, handleErrors: function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)) Monitors the nodes specified by a node-opcua~ReferenceDescription.
public	processErrorMessage(referenceDescription: node-opcua~ReferenceDescription): string Returns an error message specifically for the given reference description.

Private Methods
private	_flush(callback: function(err: ?Error)) Starts tracking node changes.
private	_transform(desc: node-opcua~ReferenceDescription, enc: string, callback: function(err: ?Error)) Buffers all read node descriptions until the subscription started, then calls QueueStream#_enqueueChunk with them.

Inherited Summary

From class Stream
private	_keepSessionAlive: Boolean `true` if the stream's atvise server session should be kept alive once the stream ends.
private	_flush(callback: function(err: ?Error, data: Object)) Called just before the stream is closed: Closes the open session.

From class QueueStream
public get	hasPending: boolean `true` if there are queued operations or an operation is running right now.
public get	opsPerSecond: number The number of processed chunks per second.
public get	processed: number The number of chunks already processed.
public get	queueEmpty: boolean `true` if there are no queued operations.
private	_maxParallel: number The maximum of parallel tasks to execute
private	_processed: Number The number of chunks processed so far.
private	_processing: Number The number of running operations.
private	_queued: *[] The queued chunks.
private	_start: Number The timestamp of the date when the stream was created.
public abstract	processChunk(chunk: *, handleErrors: function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)) The function to call when a chunk is ready to be processed.
public abstract	processErrorMessage(chunk: *): string The error message to use when processing a chunk fails.
private	_enqueueChunk(chunk: *) Enqueues the given chunk for processing.
private	_flush(callback: Function) Waits for pending operations to complete.
private	_processChunk(chunk: *) Calls QueueStream#processChunk and handles errors and invalid status codes.
private	_transform(chunk: *, enc: string, callback: Function) Calls QueueStream#_enqueueChunk as soon as the stream's session is opened.

Public Constructors

public constructor(options: Object) source

Creates a new SubscribeStream based on some options.

Override:

QueueStream#constructor

Params:

Name	Type	Attribute	Description
options	Object		The stream options to apply.

Test:

SubscribeStream #constructor

Public Members

public subscription: node-opcua~ClientSubscription source

The node-opcua~ClientSubscription to use to monitor nodes.

Private Members

private _trackChanges: Boolean source

Set to true once all nodes are monitored.

Public Methods

public createSubscription() source

Creates the server subscription to monitor nodes with.

Test:

SubscribeStream #createSubscription

public processChunk(referenceDescription: node-opcua~ReferenceDescription, handleErrors: function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)) source

Monitors the nodes specified by a node-opcua~ReferenceDescription.

Override:

QueueStream#processChunk

Params:

Name	Type	Attribute	Description
referenceDescription	node-opcua~ReferenceDescription		The refernce description to use.
handleErrors	function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)		The error handler to call. See QueueStream#processChunk for details.

Test:

SubscribeStream #processChunk

public processErrorMessage(referenceDescription: node-opcua~ReferenceDescription): string source

Returns an error message specifically for the given reference description.

Override:

QueueStream#processErrorMessage

Params:

Name	Type	Attribute	Description
referenceDescription	node-opcua~ReferenceDescription		The reference description to get the error message for.

Return:

string

The specific error message.

Test:

SubscribeStream #processErrorMessage

Private Methods

private _flush(callback: function(err: ?Error)) source

Starts tracking node changes.

Override:

QueueStream#_flush

Params:

Name	Type	Attribute	Description
callback	function(err: ?Error)		Called once flushing is complete.

private _transform(desc: node-opcua~ReferenceDescription, enc: string, callback: function(err: ?Error)) source

Buffers all read node descriptions until the subscription started, then calls QueueStream#_enqueueChunk with them.

Override:

QueueStream#_transform

Params:

Name	Type	Attribute	Description
desc	node-opcua~ReferenceDescription		Reference description of the node to transform.
enc	string		The encoding used.
callback	function(err: ?Error)		Called with the error that occured while trying to monitor the given node.

Test:

SubscribeStream #_transform