import QueueStream from 'atscm/src/lib/server/QueueStream.js'

public class | source

QueueStream

Extends:

Stream → QueueStream

Direct Subclass:

CallMethodStream

Indirect Subclass:

CallScriptStream, CreateNodeStream, ImportStream

A stream that process atvise server requests in parallel.

Test:

QueueStream

Constructor Summary

Public Constructor
public	constructor(options: Object) Creates a new QueueStream with the given options.

Member Summary

Public Members
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 Members
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.

Method Summary

Public Methods
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 Methods
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.

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.

Public Constructors

public constructor(options: Object) source

Creates a new QueueStream with the given options.

Override:

Stream#constructor

Params:

Name	Type	Attribute	Description
options	Object	optional	The options to use.
options.maxParallel	number	optional	The maximum of parallel tasks to execute.

Test:

QueueStream #constructor

Public Members

public get hasPending: boolean source

true if there are queued operations or an operation is running right now.

Test:

QueueStream #hasPending

public get opsPerSecond: number source

The number of processed chunks per second.

Test:

QueueStream #opsPerSecond

public get processed: number source

The number of chunks already processed.

Test:

QueueStream #processed

public get queueEmpty: boolean source

true if there are no queued operations.

Test:

QueueStream #queueEmpty

Private Members

private _maxParallel: number source

The maximum of parallel tasks to execute

private _processed: Number source

The number of chunks processed so far.

private _processing: Number source

The number of running operations.

private _queued: *[] source

The queued chunks.

private _start: Number source

The timestamp of the date when the stream was created.

Public Methods

public abstract processChunk(chunk: *, handleErrors: function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)) source

The function to call when a chunk is ready to be processed. Must be overridden by all subclasses..

Params:

Name	Type	Attribute	Description
chunk	*		The chunk to process.
handleErrors	function(err: Error, statusCode: node-opcua~StatusCodes, onSuccess: function)		Call this function to handle errors and bad status codes. When no error occured and the status code received is fine, `onSuccess` is called. Further processing of valid chunks, for example Recursions should happen in `onSuccess`. Note that `onSuccess` is an asynchronous function with a callback as an argument..

Example:

Basic implementation

class MyQueueStream extends QueueStream {
  ...
  processChunk(chunk, handle) {
    client.session.doSomething((err, result, statusCode) => handle(err, statusCode, done => {
      // This is called if err is falsy and status code is node-opcua~StatusCodes.Good
      doSomethingWith(result);
      done();
    }));
  }
  ...
}

Implement a recursion

class RecursiveQueueStream extends QueueStream {
  ...
  processChunk(chunk, handle) {
    client.session.doSomething((err, result, statusCode) => handle(err, statusCode, done => {
      // Write the result back to the stream.
      // This means, that `result` will be queued and, as soon as possible, #processChunk will
      // be called with `result` as the `chunk` argument.
      this.write(result, null, done);
    }));
  }
  ...
}

Allowing some invalid status codes

import { StatusCodes } from 'node-opcua';

class FriendlyQueueStream extends QueueStream {
  ...
  processChunk(chunk, handle) {
    client.session.doSomething((err, result, statusCode) => {
      if (statusCode === StatusCodes.BadUserAccessDenied) {
        Logger.warn(`Ignored invalid status: ${statusCode.description}`);
        handle(err, StatusCodes.Good, done => done());
      } else {
        handle(err, statusCode, done => done());
      }
    });
  }
  ...
}

Test:

public abstract processErrorMessage(chunk: *): string source

The error message to use when processing a chunk fails. Must be overridden by all subclasses!.

Params:

Name	Type	Attribute	Description
chunk	*		The chunk being processed.

Return:

string

The error message to use.

Test:

QueueStream #processErrorMessage

Private Methods

private _enqueueChunk(chunk: *) source

Enqueues the given chunk for processing.

Params:

Name	Type	Attribute	Description
chunk	*		The chunk to enqueue.

Test:

QueueStream #_enqueueChunk

private _flush(callback: Function) source

Waits for pending operations to complete.

Override:

Stream#_flush

Params:

Name	Type	Attribute	Description
callback	Function		Called once all queued chunks have been processed.

Test:

QueueStream #_flush

private _processChunk(chunk: *) source

Calls QueueStream#processChunk and handles errors and invalid status codes.

Params:

Name	Type	Attribute	Description
chunk	*		The chunk to process.

Emit:

*	Emits a `processed-chunk` event once a chunk was processed.

Test:

QueueStream #_processChunk

private _transform(chunk: *, enc: string, callback: Function) source

Calls QueueStream#_enqueueChunk as soon as the stream's session is opened.

Params:

Name	Type	Attribute	Description
chunk	*		The chunk to transform.
enc	string		The encoding used.
callback	Function		Called once the chunk has been enqueued.

Test:

QueueStream #_transform