Node.js stream interface for reading records from AWS Kinesis.
var AWS = new AWS.Kinesis({
region: 'us-east-1',
params: { StreamName: 'my-stream' }
});
// see below for options
var readable = require('kinesis-readable')(client, options);
readable
// 'data' events will trigger for a set of records in the stream
.on('data', function(records) {
console.log(records);
})
// each time a records are passed downstream, the 'checkpoint' event will provide
// the last sequence number that has been read
.on('checkpoint', function(sequenceNumber) {
console.log(sequenceNumber);
})
.on('error', function(err) {
console.error(err);
})
.on('end', function() {
console.log('all done!');
});
// Calling .close() will finish all pending GetRecord requests before emitting
// the 'end' event.
// Because the kinesis stream persists, the readable stream will not
// 'end' until you explicitly close it
setTimeout(function() {
readable.close();
}, 60 * 60 * 1000);
You can pass options to create the readable stream, all parameters are optional:
var options = {
shardId: 'shard-identifier', // defaults to first shard in the stream
iterator: 'LATEST', // default to TRIM_HORIZON
startAfter: '12345678901234567890', // start reading after this sequence number
startAt: '12345678901234567890', // start reading from this sequence number
timestamp: '2016-04-04T19:58:46.480-00:00', // start reading from this timestamp
limit: 100 // number of records per `data` event
};