Merge pull request #6 from tilfin/feature/update-modules

tilfin · web-flow · commit b1552e6a4169 · 2018-11-29T00:17:18.000+09:00
Update modules
diff --git a/.npmignore b/.npmignore
@@ -1,6 +1,7 @@
 coverage
 example
 test
+tmp
 .gitignore
 .travis.yml
 Makefile
diff --git a/README.md b/README.md
@@ -10,7 +10,7 @@ kinesis-stream-lambda
 
 * 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.expandArray 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
@@ -27,33 +27,59 @@ furthermore,
 $ npm install -save aws-kinesis-agg
 ```
 
-## Lambda handler example
+## Lambda handler examples
+
+### async/await style
 
 ```javascript
-const es = require('event-stream');
+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
+
+```javascript
+const StreamUtils = require('@tilfin/stream-utils');
+const KSL = require('kinesis-stream-lambda');
 
-exports.handler = function(event, context) {
+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', function() {
+  stream.on('end', () => {
     console.dir(result);
-    context.done();
+    callback();
   });
 
-  stream.on('error', function(err) {
-    context.fail(err);
+  stream.on('error', err => {
+    callback(err);
   });
 
   stream
-  .pipe(KSL.parseJSON({ expandArray: false }))
-  .pipe(es.map(function(data, callback) {
+  .pipe(KSL.parseJSON({ flatArray: false }))
+  .pipe(StreamUtils.map(function(data, cb) {
     result.push(data);
-    callback(null, data)
+    cb(null, data)
   }));
 }
 ```
diff --git a/example/async.js b/example/async.js
@@ -0,0 +1,28 @@
+const StreamUtils = require('@tilfin/stream-utils');
+const KSL = require('../lib');
+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);
+}
+
+exports.handler(require('./data'))
+.then(() => {
+  console.info('<<< END >>>')
+})
+.catch(err => {
+  console.error('Failed:', err)
+})
diff --git a/example/data.json b/example/data.json
@@ -0,0 +1,52 @@
+{
+    "Records": [
+        {
+            "kinesis": {
+                "kinesisSchemaVersion": "1.0",
+                "partitionKey": "485bd0f3c7f8b6ec3ddc6ada18abcf8d",
+                "sequenceNumber": "49565441425403787852263465772329791873704550161163747330",
+                "data": "W3siaWQiOjEsInZhbHVlIjoxMH0seyJpZCI6MiwidmFsdWUiOjIwfSx7ImlkIjozLCJ2YWx1ZSI6MzB9LHsiaWQiOjQsInZhbHVlIjoxNX0seyJpZCI6NSwidmFsdWUiOjI1fV0=",
+                "approximateArrivalTimestamp": 1473002069.264
+            },
+            "eventSource": "aws:kinesis",
+            "eventVersion": "1.0",
+            "eventID": "shardId-000000000000:49565441425403787852263465772329791873704550161163747330",
+            "eventName": "aws:kinesis:record",
+            "invokeIdentityArn": "arn:aws:iam::986450800000:role/lambda_basic_execution",
+            "awsRegion": "ap-northeast-1",
+            "eventSourceARN": "arn:aws:kinesis:ap-northeast-1:986450800000:stream/kpl-stream"
+        },
+        {
+            "kinesis": {
+                "kinesisSchemaVersion": "1.0",
+                "partitionKey": "485bd0f3c7f8b6ec3ddc6ada18abcf8d",
+                "sequenceNumber": "49565441425403787852263465772329791873704550161163747330",
+                "data": "W3siaWQiOjEsInZhbHVlIjoxMH0seyJpZCI6MiwidmFsdWUiOjIwfSx7ImlkIjozLCJ2YWx1ZSI6MzB9LHsiaWQiOjQsInZhbHVlIjoxNX0seyJpZCI6NSwidmFsdWUiOjI1fV0=",
+                "approximateArrivalTimestamp": 1473002069.264
+            },
+            "eventSource": "aws:kinesis",
+            "eventVersion": "1.0",
+            "eventID": "shardId-000000000000:49565441425403787852263465772329791873704550161163747330",
+            "eventName": "aws:kinesis:record",
+            "invokeIdentityArn": "arn:aws:iam::986450800000:role/lambda_basic_execution",
+            "awsRegion": "ap-northeast-1",
+            "eventSourceARN": "arn:aws:kinesis:ap-northeast-1:986450800000:stream/kpl-stream"
+        },
+        {
+            "kinesis": {
+                "kinesisSchemaVersion": "1.0",
+                "partitionKey": "485bd0f3c7f8b6ec3ddc6ada18abcf8d",
+                "sequenceNumber": "49565441425403787852263465772329791873704550161163747330",
+                "data": "W3siaWQiOjEsInZhbHVlIjoxMH0seyJpZCI6MiwidmFsdWUiOjIwfSx7ImlkIjozLCJ2YWx1ZSI6MzB9LHsiaWQiOjQsInZhbHVlIjoxNX0seyJpZCI6NSwidmFsdWUiOjI1fV0=",
+                "approximateArrivalTimestamp": 1473002069.264
+            },
+            "eventSource": "aws:kinesis",
+            "eventVersion": "1.0",
+            "eventID": "shardId-000000000000:49565441425403787852263465772329791873704550161163747330",
+            "eventName": "aws:kinesis:record",
+            "invokeIdentityArn": "arn:aws:iam::986450800000:role/lambda_basic_execution",
+            "awsRegion": "ap-northeast-1",
+            "eventSourceARN": "arn:aws:kinesis:ap-northeast-1:986450800000:stream/kpl-stream"
+        }
+    ]
+}
diff --git a/example/normal.js b/example/normal.js
@@ -0,0 +1,30 @@
+const StreamUtils = require('@tilfin/stream-utils');
+const KSL = require('../lib');
+
+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)
+  }));
+}
+
+exports.handler(require('./data'), null, function (err) {
+  if (err) console.error('Failed:', err)
+  else console.info('<<< END >>>')
+})
diff --git a/lib/json_transform.js b/lib/json_transform.js
@@ -14,13 +14,15 @@ class JSONTransform extends Transform {
    *
    * @param {Object} options - Transform options
    * @param {Number} options.countBy - number of block items
-   * @param {Boolean} options.expandArray - Flatten an array into items
+   * @param {Boolean} options.flatArray - Flatten an array into items
+   * @param {Boolean} options.expandArray - Deprecated. use flatArray
    */
   constructor(options) {
     const opts = options || {};
     const countBy = opts.countBy || 1;
-    const expanding = opts.expandArray || false;
+    const expanding = opts.flatArray || opts.expandArray || false;
     delete opts.countBy;
+    delete opts.flatArray;
     delete opts.expandArray;
     opts.objectMode = true;
     super(opts);
diff --git a/package.json b/package.json
@@ -1,11 +1,11 @@
 {
   "name": "kinesis-stream-lambda",
-  "version": "0.8.1",
+  "version": "0.9.0",
   "description": "Readable stream in Lambda for Kinesis Stream",
   "main": "lib/index.js",
   "scripts": {
     "test": "mocha test/*.test.js",
-    "coverage": "./node_modules/.bin/istanbul cover _mocha -r test/*.test.js"
+    "coverage": "istanbul cover _mocha -r test/*.test.js"
   },
   "keywords": [
     "kinesis",
@@ -19,16 +19,19 @@
     "url": "https://github.com/tilfin/kinesis-stream-lambda"
   },
   "devDependencies": {
-    "aws-kinesis-agg": "^2.2.2",
-    "chai": "^3.5.0",
-    "coveralls": "^2.11.12",
-    "event-stream": "=3.3.4",
+    "@tilfin/stream-utils": "^0.6.0",
+    "aws-kinesis-agg": "^4.0.2",
+    "chai": "^4.2.0",
+    "coveralls": "^3.0.2",
     "istanbul": "^0.4.5",
-    "mocha": "^3.0.2",
-    "mocha-lcov-reporter": "^1.2.0",
-    "through2": "^2.0.1"
+    "mocha": "^5.2.0",
+    "mocha-lcov-reporter": "^1.3.0",
+    "through2": "^3.0.0"
   },
   "engines": {
-    "node": ">=6"
+    "node": ">=6.10"
+  },
+  "dependencies": {
+    "promised-lifestream": "^0.7.0"
   }
 }
diff --git a/test/json_stream.test.js b/test/json_stream.test.js
@@ -1,10 +1,10 @@
-'use strict';
+'use strict'
+
+const StreamUtils = require('@tilfin/stream-utils')
+const { assert } = require('chai')
 
 const fs = require('fs');
-const es = require('event-stream');
 const through2 = require('through2');
-const chai = require('chai');
-const assert = chai.assert;
 
 const JSONTransform = require('../lib/json_transform');
 
@@ -30,13 +30,13 @@ describe('JSONTransform', () => {
   context('passed 9 items JSON with count by 2 and expanding array', () => {
     it('flush 5 items', (done) => {
       const readStream = fs.createReadStream(__dirname + '/fixtures/data/data_json.txt');
-      const jsonStream = new JSONTransform({ expandArray: true, countBy: 2 })
+      const jsonStream = new JSONTransform({ flatArray: true, countBy: 2 })
         .on('error', function(err) {
           assert.ifError(err);
           done();
         });
 
-      const writeStream = es.writeArray(function (err, array) {
+      const writeStream = StreamUtils.writeArray(function (err, array) {
         assert.deepEqual(array[0], [{ id: 1 }, { id: 2 }]);
         assert.deepEqual(array[1], [{ id: 3 }, { id: 4 }]);
         assert.deepEqual(array[2], [{ id: 5 }, { id: 6 }]);
@@ -52,7 +52,7 @@ describe('JSONTransform', () => {
   context('highWaterMark is less than item count of 1 JSON line', () => {
     it('reads rightly', (done) => {
       const readStream = fs.createReadStream(__dirname + '/fixtures/data/data_json.txt');
-      const jsonStream = new JSONTransform({ highWaterMark: 7, expandArray: true, countBy: 1 })
+      const jsonStream = new JSONTransform({ highWaterMark: 7, flatArray: true, countBy: 1 })
         .on('error', function(err) {
           assert.ifError(err);
           done();
@@ -70,7 +70,7 @@ describe('JSONTransform', () => {
           cb();
         });
 
-      const writeStream = es.writeArray(function (err, array) {
+      const writeStream = StreamUtils.writeArray(function (err, array) {
         assert.equal(array.length, 9);
         done();
       });
@@ -84,7 +84,7 @@ describe('JSONTransform', () => {
       it(`flush valid items through '${errMsg}'`, (done) => {
         const readStream = fs.createReadStream(__dirname + '/fixtures/data/multiline_json.txt', { highWaterMark });
         const jsonStream = new JSONTransform()
-        const writeStream = es.writeArray(function (err, array) {
+        const writeStream = StreamUtils.writeArray(function (err, array) {
           assert.deepEqual(array[0], { color: "red", value: "#f00" });
           assert.deepEqual(array[1], { color: "green", value: "#0f0" });
           assert.deepEqual(array[2], { color: "blue", value: "#00f" });
diff --git a/test/main.test.js b/test/main.test.js
diff --git a/test/read_stream.test.js b/test/read_stream.test.js
