Merge e781829 into 728fc0a

README.md

@@ -61,8 +61,12 @@ Reader readable stream too!
 
 ### Options
 
-* `interval` (default: `2000`) Milliseconds between each Kinesis read. Remember limit is 5 reads / second / shard
-* Any [getShardIterator] param
+* `interval: number` (default: `2000`) Milliseconds between each Kinesis read. Remember limit is 5 reads / second / shard
+* `parser: Function` If this is set, this function is applied to the data. Example:
+      const client = AWS.Kinesis()
+      const reader = new KinesisStreamReader(client, streamName, {parser: JSON.parse})
+      reader.on('data', console.log(data.id))
+* And any [getShardIterator] parameter
 
 ### Custom events
 

index.js

@@ -38,13 +38,15 @@ function getShardIterator (client, streamName, shardId, options) {
 
 class KinesisStreamReader extends Readable {
   constructor (client, streamName, options) {
-    // Use objectMode since we get whole objects at a time? Maybe have an
-    // `options.json` that sets object mode and automagically does JSON.parse
-    // or `options.parse` that can be set to `JSON.parse`
-    super({})
+    super({
+      objectMode: !!options.parser,  // Should this always be true?
+    })
     this.client = client
     this.streamName = streamName
-    this.options = Object.assign({interval: 2000}, options)
+    this.options = Object.assign({
+      interval: 2000,
+      parser: (x) => x,
+    }, options)
     this._started = false  // TODO this is probably built into Streams
     this.iterators = new Set()
   }
@@ -85,7 +87,7 @@ class KinesisStreamReader extends Readable {
       if (data.MillisBehindLatest > 60 * 1000) {
         debug('warning: behind by %d milliseconds', data.MillisBehindLatest)
       }
-      data.Records.forEach((x) => this.push(x.Data))
+      data.Records.forEach((x) => this.push(this.options.parser(x.Data)))
       if (data.Records.length) {
         this.emit('checkpoint', data.Records[data.Records.length - 1].SequenceNumber)
       }

test/index.spec.js

@@ -191,7 +191,7 @@ describe('main', () => {
       })
 
       it('continues to read open shard', () => {
-        const clock = sinon.useFakeTimers()
+        const clock = sandbox.useFakeTimers()
         const getNextIterator = sinon.stub()
         const record = {
           Data: '',
@@ -215,7 +215,21 @@ describe('main', () => {
         assert.strictEqual(getNextIterator.callCount, 1)
         clock.tick(10000)  // A number bigger than the idle time
         assert.strictEqual(getNextIterator.callCount, 2)
-        clock.restore()
+      })
+
+      it('parses incoming records', () => {
+        const record = {
+          Data: '{}',
+          SequenceNumber: 'seq-1',
+        }
+        const getNextIterator = sinon.stub().returns(undefined)
+        client.getRecords = (params, cb) =>
+          cb(undefined, {Records: [record], NextShardIterator: getNextIterator()})
+        const reader = new main.KinesisStreamReader(client, 'stream name', {
+          parser: JSON.parse,
+        })
+
+        reader.readShard('shard-iterator-3')
       })
     })
   })