kinesis-decoded.js

const kinesis = require('kinesis');
const {Transform} = require('stream');

class KinesisDecode extends Transform {
	constructor(options) {
		super(Object.assign({objectMode: true}, options));
	}

	_transform(record, encoding, cb) {
		try {
			cb(null, JSON.parse(record.Data.toString('utf8')));
		} catch(err) {
			cb(err);
		}
	}
}

module.exports = options => kinesis.stream(options).pipe(new KinesisDecode());