lifion-kinesis

Lifion's Node.js client for Amazon Kinesis Data Streams.

Getting Started

To install the module:

npm install lifion-kinesis --save

The main module export is a Kinesis class that instantiates as a readable stream.

const Kinesis = require('lifion-kinesis');

const kinesis = new Kinesis({
  streamName: 'sample-stream'
  /* other options from AWS.Kinesis */
});
kinesis.on('data', data => {
  console.log('Incoming data:', data);
});
kinesis.startConsumer();

To take advantage of back-pressure, the client can be piped to a writable stream:

const { promisify } = require('util');
const Kinesis = require('lifion-kinesis');
const stream = require('stream');

const asyncPipeline = promisify(stream.pipeline);
const kinesis = new Kinesis({
  streamName: 'sample-stream'
  /* other options from AWS.Kinesis */
});

asyncPipeline(
  kinesis,
  new stream.Writable({
    objectMode: true,
    write(data, encoding, callback) {
      console.log(data);
      callback();
    }
  })
).catch(console.error);
kinesis.startConsumer();

Features

• Standard Node.js stream abstraction of Kinesis streams.
• Node.js implementation of the new enhanced fan-out feature.
• Optional auto-creation, encryption, and tagging of Kinesis streams.
• Support for a polling mode, using the GetRecords API, with automatic checkpointing.
• Support for multiple concurrent consumers through automatic assignment of shards.
• Support for sending messages to streams, with auto-retries.

API Reference

• lifion-kinesis
  • Kinesis ⇐ PassThrough ⏏
    • new Kinesis(options)
    • instance
      • .startConsumer() ⇒ Promise
      • .stopConsumer()
      • .putRecord(params) ⇒ Promise
      • .listShards(params) ⇒ Promise
      • .putRecords(params) ⇒ Promise
      • .getStats() ⇒ Object
    • static
      • .getStats() ⇒ Object

Kinesis ⇐ PassThrough ⏏

A pass-through stream class specialization implementing a consumer of Kinesis Data Streams using the AWS SDK for JavaScript. Incoming data can be retrieved through either the data event or by piping the instance to other streams.

Kind: Exported class
Extends: PassThrough

new Kinesis(options)

Initializes a new instance of the Kinesis client.

kinesis.startConsumer() ⇒ Promise

Starts the stream consumer, by ensuring that the stream exists, that it's ready, and configured as requested. The internal managers that deal with heartbeats, state, and consumers will also be started.

Kind: instance method of Kinesis
Fulfil: undefined - Once the consumer has successfully started.
Reject: Error - On any unexpected error while trying to start.

kinesis.stopConsumer()

Stops the stream consumer. The internal managers will also be stopped.

Kind: instance method of Kinesis

kinesis.putRecord(params) ⇒ Promise

Writes a single data record into a stream.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

kinesis.listShards(params) ⇒ Promise

List the shards of a stream.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

kinesis.putRecords(params) ⇒ Promise

Writes multiple data records into a stream in a single call.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

kinesis.getStats() ⇒ Object

Returns statistics for the instance of the client.

Kind: instance method of Kinesis
Returns: Object - An object with the statistics.

Kinesis.getStats() ⇒ Object

Returns the aggregated statistics of all the instances of the client.

Kind: static method of Kinesis
Returns: Object - An object with the statistics.

License

MIT