Merge pull request #3 from tilfin/refactor/use-class

tilfin · web-flow · commit c3247fdd8601 · 2016-09-19T11:34:17.000+09:00
Use class define
diff --git a/lib/json_transform.js b/lib/json_transform.js
@@ -1,69 +1,70 @@
 'use strict';
 
 const Transform = require('stream').Transform;
-const util = require('util');
-
 
 /**
  * JSONTransform
- *
- * @param {Object} opts - Transform options
- * @param {Number} opts.countBy - number of block items
- * @param {Boolean} opts.expandArray - Flatten an array into items
  */
-function JSONTransform(opts) {
-  if (!(this instanceof JSONTransform))
-    return new JSONTransform(opts);
+class JSONTransform extends Transform {
 
-  const opts_ = opts || {};
-  this._countBy = opts_.countBy || 1;
-  this._expanding = opts_.expandArray || false;
-  delete opts_.countBy;
-  delete opts_.expandArray;
+  /**
+   * Contructor
+   *
+   * @param {Object} options - Transform options
+   * @param {Number} options.countBy - number of block items
+   * @param {Boolean} options.expandArray - Flatten an array into items
+   */
+  constructor(options) {
+    const opts = options || {};
+    const countBy = opts.countBy || 1;
+    const expanding = opts.expandArray || false;
+    delete opts.countBy;
+    delete opts.expandArray;
+    opts.objectMode = true;
+    super(opts);
 
-  this._items = [];
+    this._countBy = countBy;
+    this._expanding = expanding;
+    this._items = [];
+  }
 
-  opts_.objectMode = true;
-  Transform.call(this, opts_);
-}
-util.inherits(JSONTransform, Transform);
+  _transform(chunk, encoding, cb) {
+    try {
+      const data = JSON.parse(chunk);
+      if (this._expanding && (data instanceof Array)) {
+        this._items = this._items.concat(data);
+      } else {
+        this._items.push(data);
+      }
+    } catch (err) {
+      cb(err, null);
+      return;
+    }
 
-JSONTransform.prototype._transform = function(chunk, encoding, cb) {
-  try {
-    const data = JSON.parse(chunk);
-    if (this._expanding && (data instanceof Array)) {
-      this._items = this._items.concat(data);
-    } else {
-      this._items.push(data);
+    while (this._items.length >= this._countBy) {
+      if (!this._pushItems(this._shiftItems())) break;
     }
-  } catch (err) {
-    cb(err, null);
-    return;
+    cb();
   }
 
-  while (this._items.length >= this._countBy) {
-    if (!this._enqueue(this._shiftItems())) break;
-  }
-  cb();
-}
+  _flush(cb) {
+    const items = this._items;
 
-JSONTransform.prototype._flush = function(cb) {
-  const items = this._items;
+    while (items.length >= this._countBy) {
+      this._pushItems(this._shiftItems());
+    }
 
-  while (items.length >= this._countBy) {
-    this._enqueue(this._shiftItems());
+    if (items.length) this._pushItems(items);
+    cb();
   }
 
-  if (items.length) this._enqueue(items);
-  cb();
-}
-
-JSONTransform.prototype._enqueue = function(items) {
-  return this.push(this._countBy > 1 ? items : items[0]);
-}
+  _pushItems(items) {
+    return this.push(this._countBy > 1 ? items : items[0]);
+  }
 
-JSONTransform.prototype._shiftItems = function() {
-  return this._items.splice(0, this._countBy);
+  _shiftItems() {
+    return this._items.splice(0, this._countBy);
+  }
 }
 
 module.exports = JSONTransform;
diff --git a/lib/read_stream.js b/lib/read_stream.js
@@ -1,61 +1,66 @@
 'use strict';
 
-const util = require('util');
 const Readable = require('stream').Readable;
 
+/**
+ * KLReadStream
+ */
+class KLReadStream extends Readable {
 
-function KLReadStream(event, options) {
-  if (!(this instanceof KLReadStream))
-    return new KLReadStream(event, options);
+  /**
+   * Contructor
+   *
+   * @param {Object} event - Lambda event
+   * @param {Object} options - Readable stream options
+   * @param {Boolean} options.isAgg - Use KPL aggregation or not
+   */
+  constructor(event, options) {
+    const opts = options || {};
+    const isAgg = opts.isAgg;
+    delete opts.isAgg;
+    opts.objectMode = true;
+    super(opts);
 
-  this._records = event.Records;
-  this._recCnt = this._records.length;
-  this._recIdx = 0;
-
-  const opts = options || {};
-  if (opts.isAgg) {
-    this._agg = require('aws-kinesis-agg');
-  } else {
-    this._agg = null;
+    this._records = event.Records;
+    this._recCnt = this._records.length;
+    this._recIdx = 0;
+    this._agg = isAgg ? require('aws-kinesis-agg') : null;
   }
-  delete opts.isAgg;
 
-  opts.objectMode = true;
-  Readable.call(this, opts);
-}
-util.inherits(KLReadStream, Readable);
-KLReadStream.prototype._read = function(m) {
-  this._nextRecord();
-}
-KLReadStream.prototype._nextRecord = function() {
-  if (this._recIdx === this._recCnt) {
-    this.push(null);
-    return;
+  _read(m) {
+    this._nextRecord();
   }
 
-  const record = this._records[this._recIdx++];
-  if (this._agg) {
-    this._agg.deaggregateSync(record.kinesis, true, (err, userRecords) => {
-        if (err) {
-          this.emit('error', err);
-        } else {
-          userRecords.forEach((usrRec) => {
-            this._processRecord(usrRec.data);
-          });
-        }
-      });
-  } else {
-    this._processRecord(record.kinesis.data);
+  _nextRecord() {
+    if (this._recIdx === this._recCnt) {
+      this.push(null);
+      return;
+    }
+
+    const record = this._records[this._recIdx++];
+    if (this._agg) {
+      this._agg.deaggregateSync(record.kinesis, true, (err, userRecords) => {
+          if (err) {
+            this.emit('error', err);
+          } else {
+            userRecords.forEach((usrRec) => {
+              this._processRecord(usrRec.data);
+            });
+          }
+        });
+    } else {
+      this._processRecord(record.kinesis.data);
+    }
   }
-}
-KLReadStream.prototype._processRecord = function(b64Data) {
-  try {
-    const buf = new Buffer(b64Data, 'base64');
-    this.push(buf);
-  } catch (err) {
-    this.emit('error', Error('invalid Base64 data'));
+
+  _processRecord(b64Data) {
+    try {
+      const buf = new Buffer(b64Data, 'base64');
+      this.push(buf);
+    } catch (err) {
+      this.emit('error', Error('invalid Base64 data'));
+    }
   }
 }
 
-
 module.exports = KLReadStream;
diff --git a/test/json_stream.test.js b/test/json_stream.test.js
@@ -13,7 +13,7 @@ describe('JSONTransform', () => {
   context('passed valid JSON', () => {
     it('raises finish event', (done) => {
       const readStream = fs.createReadStream(__dirname + '/fixtures/data/valid_json.txt');
-      const jsonStream = JSONTransform()
+      const jsonStream = new JSONTransform()
         .on('finish', function() {
           assert.isOk(true);
           done();
@@ -30,7 +30,7 @@ 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 = JSONTransform({ expandArray: true, countBy: 2 })
+      const jsonStream = new JSONTransform({ expandArray: true, countBy: 2 })
         .on('error', function(err) {
           assert.ifError(err);
           done();
@@ -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 = JSONTransform({ highWaterMark: 7, expandArray: true, countBy: 1 })
+      const jsonStream = new JSONTransform({ highWaterMark: 7, expandArray: true, countBy: 1 })
         .on('error', function(err) {
           assert.ifError(err);
           done();
@@ -82,7 +82,7 @@ describe('JSONTransform', () => {
   context('passed invalid JSON', () => {
     it('raises error event', (done) => {
       const readStream = fs.createReadStream(__dirname + '/fixtures/data/invalid_json.txt');
-      const jsonStream = JSONTransform()
+      const jsonStream = new JSONTransform()
         .on('finish', function() {
           assert.isOk(false);
           done();
diff --git a/test/read_stream.test.js b/test/read_stream.test.js
@@ -4,31 +4,15 @@ const es = require('event-stream');
 const chai = require('chai');
 const assert = chai.assert;
 
-const KLReader = require('../lib/read_stream');
+const KLReadStream = require('../lib/read_stream');
 
 
-describe('KinesisLambda.reader', () => {
-  context('called function', () => {
-    it('OK', () => {
-      const event = require('./fixtures/events/data-0');
-      const readStream = KLReader(event);
-
-      readStream
-        .pipe(es.writeArray(function(err, array) {
-          assert.equal(array[0].toString(), 'data');
-        }))
-        .on('end', function() {
-          assert.isOk(true);
-          done();
-        });
-    });
-  });
-
+describe('KLReadStream', () => {
   context('aggregation mode', () => {
     context('data is invalid base64', () => {
       it('raises error event', (done) => {
         const event = require('./fixtures/events/agg-data-invalid');
-        const readStream = KLReader(event, { isAgg: true });
+        const readStream = new KLReadStream(event, { isAgg: true });
 
         readStream.on('error', function(err) {
             assert.instanceOf(err, Error);
