Rorschach

Rorschach is a Node.js module for high-level interaction with ZooKeeper. Currently it provides all the basic ZooKeeper operations. Refer to the Rorschach API doc below. And in a mean time more recipes will be added.

Rorschach is meant to provide functionality similar to Apache Curator. It was started as implementation of distributed locking recipe and most of algorithms are taken from Curator sources.

While Curator is a ZooKeeper Keeper, Rorschach is a ZooKeeper Watchman.

Installation

npm install rorschach --save

Example

'use strict';

var Rorschach = require('rorschach');
var zookeeper = require('node-zookeeper-client');

var ACL = Rorschach.ACL;
var CreateMode = Rorschach.CreateMode;
var Event = Rorschach.Event;

var Lock = Rorschach.Lock;
var ReadWriteLock = Rorschach.ReadWriteLock;

var client = new Rorschach('127.0.0.1:2181');

client.on('connected', doTheWork);
client.on('error', onerror);

function doTheWork() {
  client
    .create()
    .withMode(CreateMode.EPHEMERAL)
    .withACL(ACL.READ_ACL_UNSAFE)
    .forPath('/a', /*afterCreate(err, path)*/);

  client
   .create()
   .withProtection()
   .creatingParentsIfNeeded()
   .forPath('/my/cool/znode/foo/bar', /*afterCreate(err, path)*/);

  client
   .delete()
   .deleteChildrenIfNeeded()
   .guaranteed()
   .forPath('/my/cool/znode', /*afterDelete(err)*/);

  client
   .getData()
   .usingWatcher(function watcher(event) {
     if (event.getType() === Event.NODE_DELETED) {
       // handle deleted
      }
      else if (event.getType() === Event.NODE_DATA_CHANGED) {
        // handle data change
      }
      else {
        console.warn('Wow, really?');
      }
    })
    .forPath('/some/path', /*afterGetData(err, data, stat)*/);

  var lock = new Lock(client, '/my/znodes/locks/myResource');
  lock.acquire(500, function afterAcquire(err) {
    // handle error

    // do the work with `myResource`

    lock.release(/*callback(err)*/);
  });

  var rwLock = new ReadWriteLock(client, '/my/znodes/locks/anotherResource');
  rwLock.writeLock().acquire(1000, function afterWriteAcquire(err) {
    // handle error

    // update `anotherResource`

    rwLock.writeLock().release(/*callback(err)*/);
  });
}

function onerror(err) {
  console.warn('[Error: %d]: %s', err.getCode(), err.stack);
}

API

• Rorschach
  • Rorschach(connectionString, [options])
  • Event: 'connected'
  • Event: 'connectionStateChanged'
  • Event: 'error'
  • #close([callback])
  • #create()
  • #delete()
  • #exists()
  • #getACL()
  • #getChildren()
  • #getData()
  • #setACL()
  • #setData()
• ConnectionState
  • .CONNECTED
  • .READ_ONLY
  • .RECONNECTED
  • .SUSPENDED
  • .LOST
  • #isConnected()
• Utils
  • .deleteChildren(zk, path, [deleteSelf], callback)
  • .join(args...)
  • .mkdirs(zk, path, makeLastNode, acl, callback)
• RetryPolicy
  • RetryPolicy([options])
  • .DEFAULT_MAX_ATTEMPTS
  • .DEFAULT_RETRYABLE_ERRORS
  • #isRetryable(err)
• CreateBuilder
  • #creatingParentsIfNeeded()
  • #forPath(path, [data], callback)
  • #withACL(acls)
  • #withMode(mode)
  • #withProtection()
• DeleteBuilder
  • #deleteChildrenIfNeeded()
  • #forPath(path, callback)
  • #guaranteed()
  • #withVersion(version)
• ExistsBuilder
  • #forPath(path, callback)
  • #usingWatcher(watcher)
• GetACLBuilder
  • #forPath(path, callback)
• GetChildrenBuilder
  • #forPath(path, callback)
  • #usingWatcher(watcher)
• GetDataBuilder
  • #forPath(path, callback)
  • #usingWatcher(watcher)
• SetACLBuilder
  • #forPath(path, acls, callback)
  • #withVersion(version)
• SetDataBuilder
  • #forPath(path, data, callback)
  • #withVersion(version)
• LeaderElection
  • LeaderElection(client, path, id)
  • Event: 'error'
  • Event: 'isLeader'
  • Event: 'notLeader'
  • #client
  • #path
  • #id
  • #hasLeadership()
  • #start(callback)
  • #stop(callback)
• Lock
  • Lock(client, basePath, [lockName], [lockDriver])
  • .LOCK_NAME
  • #client
  • #basePath
  • #lockName
  • #driver
  • #lockPath
  • #maxLeases
  • #acquires
  • #acquire([timeout], callback)
  • #destroy([callback])
  • #isOwner()
  • #release([callback])
  • #setMaxLeases(maxLeases)
• ReadWriteLock
  • ReadWriteLock(client, basePath)
  • .READ_LOCK_NAME
  • .WRITE_LOCK_NAME
  • #writeMutex
  • #readMutex
  • #readLock()
  • #writeLock()
• ExecutionError
  • #original
  • #operation
  • #operationArgs
  • #getCode()

Rorschach

Main class and, better to add, namespace.

For convenience, following node-zookeeper-client classes & constants were referenced by Rorschach:

• ACL
• CreateMode
• Event
• Exception
• Id
• Permission
• State

So you can use them like following:

function watcher(event) {
  if (event.getType() === Rorschach.Event.NODE_DELETED) {
     console.log('At last...');
  }
}

Rorschach(connectionString, [options])

Create instance.

Arguments

• connectionString String ZooKeeper connection string
• options Object Options:
  • retryPolicy Object|RetryPolicy RetryPolicy instance or options
  • zookeeper Object ZooKeeper client options

Event: connected

function onConnected() { }

Emitted when connection to ZooKeeper server is established.

Event: connectionStateChanged

function onStateChanged(state) {
  if (state === Rorschach.ConnectionState.CONNECTED || state === Rorschach.ConnectionState.RECONNECTED) {
    console.info('Let\'s rock!');
  }
  /* else if ... */
}

Emitted whenever ZooKeeper client connection state changes. The only argument is state which is one of ConnectionStates.

Event: error

function onerror(err) {
  /* whatTheFussIsGoingOn(err); */
}

Currently, this event is emitted only when some operation fails in retry loop. It is emitted only if error event listener is added to Rorschach instance - to save user from Unhandled 'error' event.

void close([callback])

Close connection to ZooKeeper.

Arguments

• callback function Callback function

CreateBuilder create()

Instantiate create operation builder.

Returns

• CreateBuilder Builder instance

DeleteBuilder delete()

Instantiate delete operation builder.

Returns

• DeleteBuilder Builder instance

ExistsBuilder exists()

Instantiate exists operation builder.

Returns

• ExistsBuilder Builder instance

GetACLBuilder getACL()

Instantiate get ACL builder.

Returns

• GetACLBuilder Builder instance

GetChildrenBuilder getChildren()

Instantiate get children builder.

Returns

• GetChildrenBuilder Builder instance

GetDataBuilder getData()

Instantiate get data builder.

Returns

• GetDataBuilder Builder instance

SetACLBuilder setACL()

Instantiate set ACL builder.

Returns

• SetACLBuilder Builder instance

SetDataBuilder setData()

Instantiate set data builder.

Returns

• SetDataBuilder Builder instance

ConnectionState (Rorschach.ConnectionState)

Represents high-level connection state.

static ConnectionState CONNECTED

Connected state. Emitted only once for each Rorschach instance.

static ConnectionState READ_ONLY

Connected in read-only mode.

static ConnectionState RECONNECTED

Connection was re-established.

static ConnectionState SUSPENDED

Connection was lost, but we're waiting to re-connect.

static ConnectionState LOST

Connection was lost. Bye-bye, white pony.

void isConnected()

Return "connected" state.

Utils (Rorschach.Utils)

Utility functions.

static void deleteChildren(zk, path, [deleteSelf], callback)

Delete children for given path and maybe given znode.

Arguments

• zk Client ZooKeeper client
• path String Node path
• deleteSelf Boolean Delete node itself Default: false
• callback function Callback function: (err)

static String join(args...)

Join paths.

Arguments

• args String Paths

Returns

• String

static void mkdirs(zk, path, makeLastNode, acl, callback)

Create path and parent nodes if needed.

Arguments

• zk Client ZooKeeper client
• path String
• makeLastNode Boolean
• acl Array.<ACL>
• callback function Callback function: (err)

RetryPolicy (Rorschach.RetryPolicy)

Retry policy controls behavior of Rorschach in case of operational errors.

RetryPolicy([options])

Instantiate policy.

Arguments

• options Object Options:
  • maxAttempts Number Max number of attempts
  • codes Array.<String>|function Error codes or error validation function

static const Number DEFAULT_MAX_ATTEMPTS

Default number of operation attempts.

static const Array.<Number> DEFAULT_RETRYABLE_ERRORS

Default codes of errors allowing re-try in case of no. of attempts < maxRetries.

Boolean isRetryable(err)

Check if error is retryable.

Arguments

• err Error|Exception ZooKeeper client error

Returns

• Boolean

CreateBuilder

Create request builder.

CreateBuilder creatingParentsIfNeeded()

If path create operation will receive NO_NODE error then builder will make an attempt to create parent nodes.

Returns

• CreateBuilder

void forPath(path, [data], callback)

Execute create op.

Arguments

• path String Path to znode
• data Buffer ZNode data to set Default: null
• callback function Callback function: (executionError, path)

CreateBuilder withACL(acls)

Set ACLs.

Arguments

• acls Array.<ACL>

Returns

• CreateBuilder

CreateBuilder withMode(mode)

Set create mode.

Arguments

• mode Number CreateMode

Returns

• CreateBuilder

CreateBuilder withProtection()

See this page for explanation.

Returns

• CreateBuilder

DeleteBuilder

Delete request builder.

DeleteBuilder deleteChildrenIfNeeded()

If delete operation receives NOT_EMPTY error then make an attempt to delete child nodes.

Returns

• DeleteBuilder

void forPath(path, callback)

Execute delete.

Arguments

• path String Node path
• callback function Callback function: (err)

DeleteBuilder guaranteed()

Mark delete op. as guaranteed.

Returns

• DeleteBuilder

DeleteBuilder withVersion(version)

Set node version to delete.

Arguments

• version Number

Returns

• DeleteBuilder

ExistsBuilder

Exists request builder.

void forPath(path, callback)

Execute exists().

Arguments

• path String Node path
• callback function Callback function: (executionError, exists, stat)

ExistsBuilder usingWatcher(watcher)

Add watcher to operation request.

Arguments

• watcher function Watch function: (event)

Returns

• ExistsBuilder

GetACLBuilder

Get ACL request builder.

void forPath(path, callback)

Execute getACL().

Arguments

• path String Node path
• callback function Callback function: (executionError, acls, stat)

GetChildrenBuilder

Get children request builder.

void forPath(path, callback)

Execute getChildren().

Arguments

• path String Node path
• callback function Callback function: (executionError, data, stat)

GetChildrenBuilder usingWatcher(watcher)

Add watcher to operation request.

Arguments

• watcher function Watch function: (event)

Returns

• GetChildrenBuilder

GetDataBuilder

Get data request builder.

void forPath(path, callback)

Execute getData().

Arguments

• path String Node path
• callback function Callback function: (executionError, data, stat)

GetDataBuilder usingWatcher(watcher)

Add watcher to operation request.

Arguments

• watcher function Watch function: (event)

Returns

• GetDataBuilder

SetACLBuilder

Set ACL request builder.

void forPath(path, acls, callback)

Execute setACL().

Arguments

• path String Node path
• acls Array.<ACL> ACLs
• callback function Callback function: (executionError, stat)

SetACLBuilder withVersion(version)

Set node ACL version (number of changes of ACL).

It's not the same as node version (number of data changes).

Arguments

• version Number

Returns

• SetACLBuilder

SetDataBuilder

Set data request builder.

void forPath(path, data, callback)

Execute setData().

Arguments

• path String Node path
• data Buffer Data to set
• callback function Callback function: (executionError, stat)

SetDataBuilder withVersion(version)

Set node version.

Arguments

• version Number

Returns

• SetDataBuilder

LeaderElection (Rorschach.LeaderElection)

Leader election recipe implementation.

LeaderElection(client, path, id)

Leader election participant.

Arguments

• client Rorschach Rorschach instance
• path String Election path
• id String Participant id

Event: error

function onerror(err) {
  // do smth. with error
}

Emitted when some background operation fails. You must always set listener for this event.

Event: isLeader

function leaderListener() {
  // start tasks for which leader is responsible
}

Emitted when leadership is obtained.

Event: notLeader

function notLeaderListener() {
  // stop tasks for which leader is responsible
}

Emitted when leadership is lost.

Rorschach client

Ref. to client.

String path

ZooKeeper path where participants' nodes exist.

String id

Id of participant. It's kept in node.

Boolean hasLeadership()

Check if our node is leader.

Returns

• Boolean

void start(callback)

Start taking part in election process.

Arguments

• callback function Callback function: (err)

void stop(callback)

Leave the game for youngz.

Arguments

• callback function Callback function: (err)

Lock (Rorschach.Lock)

Distributed re-entrant lock.

Lock(client, basePath, [lockName], [lockDriver])

Distributed lock implementation

Arguments

• client Rorschach Rorschach instance
• basePath String Base lock path
• lockName String Ephemeral node name
• lockDriver LockDriver Lock utilities Default: new LockDriver()

static const String LOCK_NAME

Default lock node name.

Rorschach client

Keep ref to client as all the low-level operations are done through it.

String basePath

Base path should be valid ZooKeeper path.

String lockName

Node name.

LockDriver driver

Lock driver.

String lockPath

Sequence node name (set when acquired).

Number maxLeases

Number of max leases.

Number acquires

Number of acquires.

void acquire([timeout], callback)

Acquire a lock.

Arguments

• timeout Number Time to wait for (milliseconds).
• callback function Callback function: (err)

void destroy([callback])

Destroy lock i.e. remove node and set acquires counter to zero.

Arguments

• callback function Optional callback function

Boolean isOwner()

Check lock is owned by process.

Returns

• Boolean

void release([callback])

Release the lock.

Arguments

• callback function Callback function: (err)

Lock setMaxLeases(maxLeases)

Set max leases.

Arguments

• maxLeases Number Max # of participants holding the lock at one time

Returns

• Lock Instance is returned for chaining

ReadWriteLock (Rorschach.ReadWriteLock)

Readers-writer lock implementation.

ReadWriteLock(client, basePath)

Initialize read and write mutexes.

Arguments

• client Rorschach Rorschach instance
• basePath String Base lock path

static const String READ_LOCK_NAME

Read lock node name.

static const string WRITE_LOCK_NAME

Write lock node name.

Lock writeMutex

Write mutex.

Lock readMutex

Read mutex.

Lock readLock()

Return read mutex.

Returns

• Lock Read lock

Lock writeLock()

Return write mutex.

Returns

• Lock Write lock

ExecutionError (Rorschach.Errors.ExecutionError)

Error which serves a wrapper to actual ZooKeeper client error. It is returned by operation builders in case of failure.

Error original

Original error instance.

String operation

Operation type: create, remove, etc.

Array operationArgs

Arguments passed to original ZK client method.

Number getCode()

Defined only if original error is instance of Exception. Returns error code of original error.

Changelog

See CHANGELOG.md.

Roadmap

• Implement service discovery
• Finalize implementation of distributed locks:
  • clone of InterProcessSemaphoreV2;
  • clone of InterProcessSemaphoreMutex;
  • clone of InterProcessMultiLock;

License

See LICENSE.md.