Connect the parsed records into the client

lifion · Oct 11, 2018 · cd5d904 · cd5d904
1 parent d30f709
commit cd5d904
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 24 deletions.
diff --git a/lib/index.js b/lib/index.js
@@ -1,6 +1,7 @@
 'use strict';
 
 const { Kinesis: AwsKinesis } = require('aws-sdk');
+const { PassThrough } = require('stream');
 const ShardSubscriber = require('./shard-subscriber');
 const consumer = require('./consumer');
 const stream = require('./stream');
@@ -13,8 +14,9 @@ function internal(instance) {
   return privateData.get(instance);
 }
 
-class Kinesis {
+class Kinesis extends PassThrough {
   constructor(options = {}) {
+    super({ objectMode: true });
     const {
       compression,
       consumerName,
@@ -71,9 +73,11 @@ class Kinesis {
     ctx.shards = (await stream.getShards(ctx)) || [];
 
     logger.debug(`Creating subscribers for the stream shards using "${consumerName}"…`);
-    ctx.shardSubscribers = ctx.shards.map(shard =>
-      new ShardSubscriber({ ...ctx, emitter: this, shard }).start()
-    );
+    ctx.shards.forEach(shard => {
+      const subscriber = new ShardSubscriber({ ...ctx, emitter: this, shard });
+      subscriber.start();
+      subscriber.pipe(this);
+    });
 
     logger.debug('The client is now connected.');
   }

diff --git a/lib/shard-subscriber.js b/lib/shard-subscriber.js
@@ -4,7 +4,7 @@ const AWS = require('aws-sdk');
 const aws4 = require('aws4');
 const got = require('got');
 const { Parser } = require('lifion-aws-event-stream');
-const { pipeline, Writable } = require('stream');
+const { pipeline, PassThrough } = require('stream');
 const Decoder = require('./records-decoder');
 
 const AWS_API_TARGET = 'Kinesis_20131202.SubscribeToShard';
@@ -18,8 +18,9 @@ function internal(instance) {
   return privateData.get(instance);
 }
 
-class ShardSubscriber {
+class ShardSubscriber extends PassThrough {
   constructor(ctx) {
+    super({ objectMode: true });
     const { options } = ctx;
     const { endpoint = 'https://kinesis.us-east-1.amazonaws.com', region } = options;
     const credentialsChain = new AWS.CredentialProviderChain();
@@ -67,24 +68,12 @@ class ShardSubscriber {
       } else logger.debug('Subscription to shard is successful.');
     });
 
-    pipeline(
-      stream,
-      new Parser(),
-      new Decoder(ctx),
-      new Writable({
-        objectMode: true,
-        write(chunk, encoding, callback) {
-          logger.debug(chunk);
-          callback();
-        }
-      }),
-      err => {
-        if (err) {
-          logger.error(err);
-          request.abort();
-        } else logger.debug('Subscription pipeline completed.');
-      }
-    );
+    pipeline([stream, new Parser(), new Decoder(ctx), this], err => {
+      if (err) {
+        logger.error(err.stack);
+        request.abort();
+      } else logger.debug('Subscription pipeline completed.');
+    });
 
     return this;
   }