Rorschach

Rorschach is a Node.js module for high-level interaction with ZooKeeper.

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.

Hope, in some time more APIs will be added.

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

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...');
  }
}

Event: connected

function onConnected() { }

Emitted when connection to ZooKeeper server is established.

Event: connectionStateChanged

function onStateChanged(state) {
  if (state === Rorschach.State.SYNC_CONNECTED) {
    console.info('Let\'s rock!');
  }
  /* else if ... */
}

Emitted whenever ZooKeeper client connection state changes. The only argument is state which is one of the Rorschach.State.* constants.

P.S.: at the moment of writing, there are typos in mentioned documentation and actual state names are available here.

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.

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

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

GetChildrenBuilder getChildren()

Instantiate get children builder.

Returns

• GetChildrenBuilder Builder instance

GetDataBuilder getData()

Instantiate get data builder.

Returns

• GetDataBuilder Builder instance

SetDataBuilder setData()

Instantiate set data builder.

Returns

• SetDataBuilder Builder instance

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: (err, 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: (err, exists, stat)

ExistsBuilder usingWatcher(watcher)

Add watcher to operation request.

Arguments

• watcher function Watch function: (event)

Returns

• ExistsBuilder

GetChildrenBuilder

Get children request builder.

void forPath(path, callback)

Execute getChildren().

Arguments

• path String Node path
• callback function Callback function: (err, 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: (err, data, stat)

GetDataBuilder usingWatcher(watcher)

Add watcher to operation request.

Arguments

• watcher function Watch function: (event)

Returns

• GetDataBuilder

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: (err, stat)

SetDataBuilder withVersion(version)

Set node version.

Arguments

• version Number

Returns

• SetDataBuilder

Lock (Rorschach.Lock)

Distributed re-entrant lock.

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.

static const String LOCK_NAME

Default lock node name.

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()

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

Lock writeMutex

Write mutex.

Lock readMutex

Read mutex.

static const String READ_LOCK_NAME

Read lock node name.

static const string WRITE_LOCK_NAME

Write lock node name.

Lock readLock()

Return read mutex.

Returns

• Lock Read lock

Lock writeLock()

Return write mutex.

Returns

• Lock Write lock

Changelog

See CHANGELOG.md.

Roadmap

• Implement Rorschach#getACL() and Rorschach#setACL();
• Finalize implementation of distributed locks:
  • clone of InterProcessSemaphoreV2;
  • clone of InterProcessSemaphoreMutex;
  • clone of InterProcessMultiLock;
• Implement Leader Election recipe.

License

See LICENSE.md.