node-dataengine-client

Install

npm install node-dataengine-client

DataEngine

const DataEngine = require('node-dataengine-client').DataEngine;

Communicates with Data Engine, maintains a persistent connection, handles data store operations and event signaling. If the persistent connection (websocket) is not connected, REST will be used. Events can be registered without having the connection active, the client will automatically register them with the server when connection is established. Works with node.js and in browser.

Constructor

new DataEngine(options);

DataEngine Options

Property	Type	Default	Description
host	string	127.0.0.1	Data Engine Server host or ip
port	number	4300	Data Engine Server port
path	string		Path to api if behind proxy
apiKey	string		Secret key to access API (if set in server)
secure	boolean	false	Use TLS
enablePing	boolean	false
pingInterval	number	5000	Time in milliseconds
clientId	string		Name of client, visible in admin UI status section
bucket	string		Default bucket for operations

Methods

read(key, [opts])

Returns promise with read object. If empty string is passed as key, an array of all objects in bucket will be returned.

options (optional)

Property	Type	Default	Description
bucket	string	null	Optionally override default bucket
forceHttp	boolean	false	Force request to be sent over REST api
async	boolean	true

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.read('myKey')
  .then(value => console.log(value));

list([opts])

List objects in bucket. Returns promise with array.

options (optional)

Property	Type	Default	Description
bucket	string	null	Optionally override default bucket
keysOnly	boolean	true	Set to false to retrieve object content
forceHttp	boolean	false	Force request to be sent over REST api
async	boolean	true

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.list()
  .then(objects => console.log(objects));

write(key, data, [opts])

Returns promise

options (optional)

Property	Type	Default	Description
bucket	string	null	Optionally override default bucket
forceHttp	boolean	false	Force request to be sent over REST api
async	boolean	true
failIfExists	boolean	false	Operation will fail if key already exists in data store
expire	string		Set expiration for object, possible values: 'session' - Delete object when client disconnects

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.write('myKey', { text: 'hello!' })
  .then(() => console.log('Written!');

patch(key, patch, [opts])

Returns promise. JSON patch existing object.

options (optional)

Property	Type	Default	Description
bucket	string	null	Optionally override default bucket
removeDataFromReply	boolean	false	Optionally remove patch result from reply
forceHttp	boolean	false	Force request to be sent over REST api
async	boolean	true

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.patch('myKey', [{ op: 'replace', path: '/text', value: 'hey!' }])
  .then(() => console.log('Patched!');

remove(key, [opts])

Returns promise. Removes an object from data store.

options (optional)

Property	Type	Default	Description
bucket	string	null	Optionally override default bucket
forceHttp	boolean	false	Force request to be sent over REST api
async	boolean	true

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.remove('myKey')
  .then(() => console.log('Removed!');

Properties

Connection: PersistentConnection

PersistentConnection

PersistentConnection is automatically created in the DataEngine instance as property connection

Methods

addListener(bucket, key, sendValues, callbackFn(eventObject))

Add listener for specific bucket and key. Wildcard * can be used to listen to all events in a bucket. If * is used for bucket, events will be triggered for all buckets.

eventObject

Property	Type	Description
bucket	string	bucket of event origin
key	string	key of event origin
value	object	written object if sendEvent was true in addListener call

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.connection.addListener('myBucket', 'myKey', true, (event) => {
  console.log(`Event received for 
    key ${event.key} in
    bucket ${event.bucket},
    value=${event.value}`); 
});

signalEvent(bucket, key, value)

Send data to Data Engine only to be propagated as an event to other clients.

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.connection.signalEvent(
  'myEventBucket',
  'myEventKey',
  { message: 'hello yall!' }
});

Events (callbacks)

onConnect(persistentConnection)

Callback triggered when connection is established.

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.connection.onConnect = (serverId) => {
  console.log(`Client connected to server with id ${serverId}`);
};

onDisconnect()

Callback triggered when connection is closed.

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.connection.onDisconnect = () => {
  console.log('Client disconnected');
};

onPong()

Callback triggered when pong packet is received from server.

Example

const de = new DataEngine({ bucket: 'myBucket' });
de.connection.onPong = () => {
  console.log('Server is responding!');
};

Messaging.RequestReply

const RequestReply = require('node-dataengine-client').Messaging.RequestReply;

Thin abstraction for sending and receiving messages between two clients connected to same Data Engine. Data Engine messaging is using the signalEvent method of PersistentConnection.

Requestor

Requestor is used to send messages to a receiver. Request will fail if no response is received within timeoutInMs or if no progress update is received within progressTimeoutInMs.

Constructor

new RequestReply.Requestor(DataEngineInstance, channel, timeoutInMs, progressTimeoutInMs);

Methods

request(msg)

Returns promise with reply.

Example

const de = new DataEngine();
 
// We cannot send request before we are connected
de.connection.onConnect = (serverId) => {
  // fail if response > 30 sec or if progress is not received < every 3 sec
  const requestor = new Requestor(de, 'my-channel', 30000, 3000);
  requestor.request({ msg: 'hello?' })
  .then(reply => console.log(`Got reply! ${reply}`);
  .catch(err => console.error(err));
};

Replier

Replier is used to receive and reply to messages.

Constructor

new RequestReply.Replier(DataEngineInstance, channel);

Events

onMessage(msg, replyFn(err, obj), progressFn(obj))

Called when a message is received. Reply is sent by calling replyFn with either an error or some data. In case of lengthy processing before reply can be sent, replier can call progressFn to avoid timeout in requestor.

Example

const de = new DataEngine();
 
const replier = new Replier(de, 'my-channel');
replier.onMessage = (msg, reply, progress) => {
  console.log(`Got request: ${msg}`);
  // send progress after 2 sec, and reply after 5 sec
  setTimeout(() => progress({ someprop: 'some value' }), 2000);
  setTimeout(() => reply(null, { someProperty: 'some value' }), 5000);
};

Messaging.CompetingConsumers

const CompetingConsumers = require('node-dataengine-client').Messaging.CompetingConsumers;

This pattern is used when one producer generates jobs that many listening consumers can claim. Only one of the consumers can claim each job.

Producer

Producer creates jobs.

Constructor

new CompetingConsumers.Producer(DataEngineInstance, channel, timeoutInMs, progressTimeoutInMs);

Methods

addJob(job, progressFn?, timoutInMs?, progressTimeoutInMs?)

Returns promise, resolved if consumer processed it successfully. Consumer can optionally call a progress callback (passed in progressFn) during its job processing.

Example

const de = new DataEngine();
 
const producer = new Producer(de, 'render-jobs', 10000);
de.connection.onConnect = (serverId) => {
  producer.addJob({ someProperty: 'value' }, (progress) => console.log(progress))
  .then((data) => {
    console.log(`Job done: ${data.result}`);
  })
  .catch(err => console.error(err));
}

CompetingConsumer

Consumer of jobs.

Constructor

new CompetingConsumers.CompetingConsumer(DataEngineInstance, channel);

Events

onJobQuery(job, callback(errno, worker))

Called when a job is requested by producer. errno should have one of values:

const JOB_NOT_SUPPORTED = 100;
const NO_WORKER_AVAILABLE = 200;
const JOB_OK = null;

Worker object must implement following:

class Worker {
  constructor(id) {
    this.id = id;
    this.available = true;
  }
 
  work(job, done, progress) {
    // do things
    // signal progress with progress();
    done(err, result);
  }
  
  release() {
    this.available = true;
  }
}

Example

class MyWorker {
  constructor(id) {
  this.id = id;
  this.available = true;
  }
 
  work(job, done, progress) {
    console.log('working..');
    // send progress every sec
    setTimeout(() => progress({ someProperty: 'value' }), 1000);
    // signal done after 10 sec
    setTimeout(() => done(null, { someProperty: 'value' }), 10000);
  }
  
  release() {
    this.available = true;
  }
}
 
const de = new DataEngine();
 
const consumer = new CompetingConsumer(de, 'my-channel');
const worker = new MyWorker('myworker1');
 
consumer.onJobQuery = (job, callback) => {
  if (worker.available) {
    callback(null, worker);
  } else {
    callback(200); // no worker available
  }
};