Node class

Node class

constructor(options, eventemitter)

Node constructor

-options: [any] - optional, any information you want to pass to node instance

-eventemitter: [Eventemitter] - optional, if you want the node to share a single event emitter use this parameter

Downstream API

push(signal)

Push a signal to the node.

-signal: [any] - the signal you want to push to node

-return: current node

onReceive(signal)

Handle received signal. Usually you don't need to override this method, the default implementation delegates the signal handling to three methods:

• onSignal(signal) : handles the signal
• onError(error) : handles the error
• onEnd(signal) : handles the end

-signal: [any] - the signal to handle

-return: current node

send(signal)

Send a signal to the downstream nodes

-signal : the signal to send

-return: current node

onSignal(signal)

Handle received signal. Override this method to implement your operator

-signal: [any] - the signal to handle

-return: current node

onError(error)

Handle errors.

Note: the argument may not be an Error object. You can override throwError(error) method to throw any object to represent an error. This is very useful if your system have a customised error structure.

-error: [any] - the object that represented an error. See throwError(error)

-return: current node

onEnd(signal)

Handle END signal.

-signal: [any] - the END signal to handle

-return: current node

isErrorSignal(signal)

Check if an signal represents an error

-signal: [any] - the signal to check

-return: true if it represents an error otherwise false

isEndSignal(signal)

Check if an signal represents an END of stream

-signal: [any] - the signal to check

-return: true if it represents an END otherwise false

throwError(error, signal)

Throw an object to represent the error.

When node throws an error in onReceive(), it will pass the error and the signal to **throwError(error, signal). You can call this.send() to put an error to stream, or directly throw it to stop the stream.

-error: [Error] - the error

-signal: [any] - the signal who cause this error

-return: current node

observeSignal(signal, node)

observe an signal

-signal: [any] - the signal

-node: [Node] - the current node who send the signal

-return: current node

Upstream API

pull(cmd)

Request a signal to the upstream node by sending a command.

-cmd: [any] - the cmd in your request

-return: current node

onRequest(cmd)

Handle received request from downstream nodes. For most operator, you don't need to override this method, by default it calls request() method to pass the request to upstream.

-cmd: [any] - the command to handle

-return: current node

request(cmd)

Make a request to upstream

-cmd : the cmd to request

-return: current node

observeCommand(cmd, node)

observe an command

-cmd: [any] - the command

-node: [Node] - the current node who request the cmd

-return: current node

Utils

invokeSignalObservers(signal, when, ...data)

invoke all signal observers

-signal: [any] - the current signal

-when: [string] - a string to indicate when the observer is invoked. For example, "send" indicates the observer is called when the signal is sent by node

-data: [...] - optional, any data you want to pass to the observer

invokeCommandObservers(cmd, when, ...data)

invoke all command observers

-command: [any] - the current command

-when: [string] - a string to indicate when the observer is invoked. For example, "request" indicates the observer is called when the cmd is requested by node

-data: [...] - optional, any data you want to pass to the observer