Make sure the shouldParseJson flag is passed down

lifion · Oct 23, 2019 · 64d3a28 · 64d3a28
1 parent 9e65110
commit 64d3a28
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 4 deletions.
diff --git a/lib/consumers-manager.js b/lib/consumers-manager.js
@@ -45,6 +45,8 @@ class ConsumersManager {
    *        option defines the delay in milliseconds in between poll requests for more records.
    * @param {Function} options.pushToStream - A function to push incoming records to the consumer.
    * @param {object} options.s3 - The S3 options in the current kinesis client.
+   * @param {string|boolean} [options.shouldParseJson] - Whether if retrieved records' data should
+   *        be parsed as JSON or not.
    * @param {object} options.stateStore - An instance of the state store.
    * @param {string} options.streamName - The name of the Kinesis stream.
    * @param {boolean} options.useAutoCheckpoints - Whether to automatically store shard checkpoints
@@ -68,6 +70,7 @@ class ConsumersManager {
       pushToStream,
       s3,
       s3Client,
+      shouldParseJson,
       stateStore,
       streamName,
       useAutoCheckpoints,
@@ -88,6 +91,7 @@ class ConsumersManager {
       pushToStream,
       s3,
       s3Client,
+      shouldParseJson,
       stateStore,
       streamName,
       useAutoCheckpoints,
@@ -118,6 +122,7 @@ class ConsumersManager {
       pushToStream,
       s3,
       s3Client,
+      shouldParseJson,
       stateStore,
       streamName,
       useAutoCheckpoints,
@@ -177,6 +182,7 @@ class ConsumersManager {
                   stateStore,
                   stopConsumer,
                   streamName,
+                  shouldParseJson,
                   useAutoCheckpoints,
                   usePausedPolling,
                   useS3ForLargeItems,
@@ -191,6 +197,7 @@ class ConsumersManager {
                   pushToStream,
                   s3,
                   s3Client,
+                  shouldParseJson,
                   shardId,
                   stateStore,
                   stopConsumer,

diff --git a/lib/fan-out-consumer.js b/lib/fan-out-consumer.js
@@ -202,6 +202,7 @@ class FanOutConsumer {
       pushToStream,
       s3,
       shardId,
+      shouldParseJson,
       stateStore,
       stopConsumer,
       streamName,
@@ -241,6 +242,7 @@ class FanOutConsumer {
       retryPipeline: true,
       s3,
       shardId,
+      shouldParseJson,
       stateStore,
       stopConsumer,
       streamName,
@@ -267,6 +269,7 @@ class FanOutConsumer {
       s3,
       s3Client,
       shardId,
+      shouldParseJson,
       stateStore,
       stopConsumer,
       streamName,
@@ -356,7 +359,14 @@ class FanOutConsumer {
           stream,
           new PreProcess({ requestFlags }),
           new Parser(),
-          new RecordsDecoder({ compression, logger, s3, s3Client, useS3ForLargeItems }),
+          new RecordsDecoder({
+            compression,
+            logger,
+            s3,
+            s3Client,
+            shouldParseJson,
+            useS3ForLargeItems
+          }),
           new PostProcess({
             abort,
             logger,

diff --git a/lib/index.js b/lib/index.js
@@ -269,7 +269,7 @@ class Kinesis extends PassThrough {
       pollDelay = 250,
       s3 = {},
       shardCount = 1,
-      shouldParseJson,
+      shouldParseJson = 'auto',
       statsInterval = 30000,
       streamName,
       tags,
@@ -332,7 +332,7 @@ class Kinesis extends PassThrough {
       },
       s3Client: null,
       shardCount: shardCountNumber >= 1 ? shardCountNumber : 1,
-      shouldParseJson: 'auto',
+      shouldParseJson,
       statsInterval: statsIntervalNumber >= 1000 ? statsIntervalNumber : 30000,
       streamName,
       tags,

diff --git a/lib/polling-consumer.js b/lib/polling-consumer.js
@@ -220,6 +220,7 @@ class PollingConsumer {
       s3,
       s3Client,
       shardId,
+      shouldParseJson,
       stateStore,
       stopConsumer,
       streamName,
@@ -245,6 +246,7 @@ class PollingConsumer {
         inputEncoding: 'Buffer',
         logger,
         s3Client,
+        shouldParseJson,
         useS3ForLargeItems
       }),
       s3,

diff --git a/lib/records.js b/lib/records.js
@@ -89,7 +89,11 @@ function getRecordsDecoder({
         data = Buffer.from(data, 'base64').toString('utf8');
       }
 
-      if ((shouldParseJson === 'auto' && IS_JSON_REGEX.test(data)) || shouldParseJson === true) {
+      if (
+        ((shouldParseJson === undefined || shouldParseJson === 'auto') &&
+          IS_JSON_REGEX.test(data)) ||
+        shouldParseJson === true
+      ) {
         data = JSON.parse(data);
       }