Kinesis-stream-lambda NPM

kinesis-stream-lambda

Features

Easily reads a Lambda event of Kinesis Stream as a stream handling the chunk as Buffer
Supports KPL aggregation (set opts.isAgg true)
Provides KSL.parseJSON transform to handle items expanded array data in one record (set opts.flatArray true)
Node.js 6.10 or Later

How to install

$ npm install -save kinesis-stream-lambda

KPL aggregation only

furthermore,

$ npm install -save aws-kinesis-agg

Lambda handler examples

async/await style

const StreamUtils = require('@tilfin/stream-utils');
const KSL = require('kinesis-stream-lambda');
const PromisedLife = require('promised-lifestream');

exports.handler = async function (event) {
  console.log('event: ', JSON.stringify(event, null, 2));

  const result = [];

  await PromisedLife([
    KSL.reader(event, { isAgg: false }),
    KSL.parseJSON({ flatArray: false }),
    StreamUtils.map(function(data, cb) {
      result.push(data);
      cb(null, data)
    })
  ])

  console.dir(result);
}

normal style

const StreamUtils = require('@tilfin/stream-utils');
const KSL = require('kinesis-stream-lambda');

exports.handler = function (event, context, callback) {
  console.log('event: ', JSON.stringify(event, null, 2));

  const result = [];
  const stream = KSL.reader(event, { isAgg: false });

  stream.on('end', () => {
    console.dir(result);
    callback();
  });

  stream.on('error', err => {
    callback(err);
  });

  stream
  .pipe(KSL.parseJSON({ flatArray: false }))
  .pipe(StreamUtils.map(function(data, cb) {
    result.push(data);
    cb(null, data)
  }));
}