riakdb

Riak client with stream interfaces using protocol buffers

Installation

npm install riakdb

Example

Using riakdb is simple. Create the client, call .connect and make the requests. It is important to note that riakdb comes with two interfaces.

• A low level interface, there directly maps to the Riak protocol
• A high level interface, made for node.js

var riakdb = require('riakdb');
var client = riakdb({
  nodes: [{
    host: '127.0.0.1',
    port: 8087
  }]
});
 
client.connect();
client.getKeys({ bucket: 'examples' })
  .on('data', function (key) {
    console.log(key);
  })
  .once('end', function () {
    client.close();
  });

Documentation

The documentation is separated intro three parts:

Connection

Discusses how to setup and configure the Riak connection pool and the associated events.

High level interface

riakdb comes with two interfaces for communicating with Riak. Generally there is no difference, however in some cases a convenient abstraction is put around the underlying low level interace, to make it easier to use.

An example of this is the .getKeys(request) (stream of all keys in bucket). In this case Riak returns a paginated stream, meaning that each item can contain multiply keys. The high level interface performs a depagination such that each item contain just a single key.

Because of the wast amount of features in Riak, some high level methods may not support all the features of the corresponding low level method.

Low level interface

To give better backward compatibility and ensure that all features in Riak are supported a low level interface is also provided. This is a direct mapping to the Riak protocol buffer definition, with nothing on top. This uses the protocol-buffers module, so the encoding and decoding behaviour is defined by that.

Connection

client = RiakClient(settings)

The main export from riakdb is the RiakClient constructor function. The constructor takes a settings object with some required and some optional properties.

var riakdb = require('riakdb');
 
var client = riakdb({
  nodes: [{
    host: '127.0.0.1',
    port: 8087
  }]
});

The settings object takes the following properties:

• nodes (required): An array of address objects, these objects are used in the net.connect function and thus takes the same arguments. But usually you only need host and port.
• minConnections (default 0): The minimum amount of connection there should always exists. It is possible to have more connections, but these will only be created when the amount of parallel requests exceeds minConnections.
• maxConnections (default 20): The maximum amount of requests. If all connections are in use and maxConnections prevents more connections from being created. Then the requests will be buffered.
• connectionTimeout (default 60000, 1 min): If a connection haven't made any requests within connectionTimeout ms and there is than minConnections active connections, then this connection will be closed.

client.connect()

client.connect();

The client won't start connecting before client.connect() is called. Until then requests will be buffered.

client.close()

client.close();

When client.close() is called, all connections are closed. This means that active requests won't get a response, but they will get an error. Buffered requests will also not be send.

client.enums

client.enums contains the enum objects associated with the riak protocol buffer message definitions. The enum objects are:

• RpbReplMode
• MapFieldType
• DataType
• FlagOp
• IndexQueryType

client.on('connect')

When the minConnections amount of connections are created then this event is fired. If minConnections is zero, then the connect event is fired on the next tick.

client.on('close')

When all connections are closed after a client.close() call, then this event will fire.

client.on('error')

Errors there happens on the connection sockets will be emitted here. If there is an active request on that connection then the stream or callback will also be notified.

High level interface

Note that if there is no defined high level function, then the same method definitions there is in client.low[method] also exists in client[method].

client.getKeys(request)

A depaginized version of client.low.getKeys. This means that each stream item contains only one key. This is different from the direct protocol mapping, where each item can contain multiply keys.

client.getKeys({
  bucket: 'bucket-name', // required, the bucket name within the bucket type
  type: 'type-name', // optional, default is 'default'
  timeout: Number, // undocumented riak option
}).on('data', function (key) {
  // `key` is buffer
});

client.getIndex(request)

A depaginized version of client.low.getIndex. By default each item will contain a key. This abstraction does not support continuation. If you need this please make a pull request or use client.low.getIndex.

client.getIndex({
  bucket: 'riakdb-client-getindex', // required
  index: '2i_bin', // required, the secondary index to lookup in
  qtype: 'range', // required, automatically converted to a number
  min_range: 'A',
  max_range: 'Z'
}).on('data', function (key) {
  // `key` is a buffer with the item key
});

If return_terms is used each item is an object with the getIndex pair.

client.getIndex({
  bucket: 'riakdb-client-getindex', // required
  index: '2i_bin', // required, the secondary index to lookup in
  qtype: 'range', // required, automatically converted to a number
  min_range: 'A',
  max_range: 'Z',
  return_terms: true // optional, result pairs are returned instead of keys
}).on('data', function (pair) {
  // `pair.key` is a buffer with the secondary index value
  // `pair.value` is a buffer with the item key
});

client.mapred(request)

client.mapred takes a request object, this will be used as the request filed in RpbMapRedReq where content_type is also set to application/json. The other content_type, application/x-erlang-binary is not supported, if you need this make a pull request or use client.low.mapred.

client.mapred expects the response to be JSON encoded. The response is parsed and depaginized, such that each stream item contains the phase number and a single response object.

client.mapred({
  inputs: 'riakdb-client-mapred',
  query: [{
    map: {
      language: "javascript",
      name: "Riak.mapValuesJson"
    }
  }]
}).on('data', function (item) {
  // item.phase
  // item.response
});

Low level interface

The low level interface have two response types callback and stream. In both cases the function takes a request object as the first argument. There are a few exceptions/details to this pattern:

• In some cases (e.q. ping) there is no request parameters and thus there is no request argument.

• Some stream requests, requires you to set a request parameter there makes Riak return a stream and not a single message. These are marked with stream (set).

Example on a callback response with a request argument

client.low.get({
  key: new Buffer('some key'),
  bucket: new Buffer('some bucket')
}, function (err, response) {
  // response is contains the full content
});

Example on a callback response with no request argument

client.low.ping(function (err, response) {
  // response is null as there is also no response for `.ping`
});

Example on a stream response with a request argument

client.low.getKeys({
  bucket: new Buffer('some bucket')
}).pipe(output);

Example on a stream response with a required stream parameter

client.low.getBuckets({
  stream: true
}).pipe(output);

Full list of methods

This is a complete list of all the documented Riak requests, with a mapping to the method name, response type and link to the request and response structure (protocol).

Name	Method	Response Type	Protocol
Bucket Operations
List Buckets	getBuckets	stream (set)	RpbListBucketsReq
List Keys	getKeys	stream	RpbListKeysReq
Get Bucket Properties	getBucket	callback	RpbGetBucketReq
Set Bucket Properties	setBucket	callback	RpbSetBucketReq
Reset Bucket Properties	resetBucket	callback	RpbResetBucketReq
Object/Key Operations
Fetch Object	get	callback	RpbGetReq
Store Object	put	callback	RpbPutReq
Delete Object	del	callback	RpbDelReq
Query Operations
MapReduce	mapred	stream	RpbMapRedReq
Secondary Indexes	getIndex	stream (set)	RpbIndexReq
Search	search	callback	RpbSearchQueryReq
Server Operations
Ping	ping	callback	RpbPingReq
Server Info	getServerInfo	callback	RpbGetServerInfoReq
Bucket Type Operations
Get Bucket Type	getBucketType	callback	RpbGetBucketTypeReq
Set Bucket Type	setBucketType	callback	RpbSetBucketTypeReq
Data Type Operations
Data Type Fetch	getCrdt	callback	DtFetchReq
Data Type Store	putCrdt	callback	DtUpdateReq
Yokozuna Operations
Yokozuna Index Get	getSearchIndex	callback	RpbYokozunaIndexGetReq
Yokozuna Index Put	putSearchIndex	callback	RpbYokozunaIndexPutReq
Yokozuna Index Delete	delSearchIndex	callback	RpbYokozunaIndexDeleteReq
Yokozuna Schema Get	getSearchSchema	callback	RpbYokozunaSchemaGetReq
Yokozuna Schema Put	putSearchSchema	callback	RpbYokozunaSchemaPutReq

License

This software is licensed under "MIT"

Copyright (c) 2015 Andreas Madsen

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.