Adding success event when the putRecords call on kinesis returns data with no errors.

README.md

@@ -67,8 +67,21 @@ new KinesisWritable({
 
 ### Events
 
-* `error`: Emitted every time records are failed to be written.
+| Name | Description | Parameters |
+|---|---|---|
+| `error`| Emitted every time records are failed to be written after all retries fail | `err` will contain the latest retry error. Failed encoded records will be in `err.records` |
+| `error`| Emitted every time records succeed to push to the kinesis stream. | `records` will contain `records.FailedRecordCount:0` and an array of objects on `records.Records` containing each `SequenceNumber:` and `ShardId` |
 
+```javascript
+new KinesisWritable({
+  region: 'AWS_REGION',
+  streamName: 'MyKinesisStream'
+}).on('error', (err) => {
+  console.log(`Failed to push ${err.records}`)
+}).on('success', (records) => {
+  console.log(`Succesfully pushed ${err.length} records`)
+})
+```
 **Note**: Amazon Credentials are not required. It will either use the environment variables, `~/.aws/credentials` or roles as every other aws sdk.
 
 ## Issue Reporting

index.js

@@ -136,7 +136,10 @@ KinesisStream.prototype.dispatch = function(records, cb) {
   });
 
   operation.attempt(() => {
-    this.putRecords(formattedRecords, (err) => {
+    this.putRecords(formattedRecords, (err, data) => {
+      if (!err) {
+        this.emitRecordSuccess(data);
+      }
       if (operation.retry(err)) {
         return;
       }
@@ -170,4 +173,8 @@ KinesisStream.prototype.emitRecordError = function (err, records) {
   this.emit('error', err);
 };
 
+KinesisStream.prototype.emitRecordSuccess = function (records) {
+  this.emit('success', records);
+};
+
 module.exports = KinesisStream;

package.json

@@ -1,7 +1,7 @@
 {
   "name": "aws-kinesis-writable",
   "description": "A stream implementation for kinesis.",
-  "version": "4.3.0",
+  "version": "4.2.5",
   "author": "José F. Romaniello <jfromaniello@gmail.com> (http://joseoncode.com)",
   "license": "MIT",
   "repository": {

test/index.js

@@ -112,7 +112,8 @@ describe('KinesisStream', function() {
     });
     it('should return immediately if no messages are provided', function(done) {
       kinesis.putRecords = sinon.spy();
-      ks.dispatch([], function() {
+      ks.dispatch([], function(err) {
+        expect(err).to.be.null;
         expect(kinesis.putRecords.calledOnce).to.be.false;
         done();
       });
@@ -125,7 +126,6 @@ describe('KinesisStream', function() {
 
       ks.on('error', function(err) {
         expect(err).to.exist;
-        expect(err.records.length).to.equal(2);
         expect(stub.calledThrice).to.be.true;
       });
 
@@ -138,16 +138,35 @@ describe('KinesisStream', function() {
     it('should putRecords without error when record contains circular references', function() {
       const logMessage = (s) => { return s.getCall(0).args[0][0].Data; };
       ks.putRecords = sinon.spy();
-
       const message = {
         hi: 'hello'
       };
       message.message = message;
 
       ks.dispatch([message]);
 
-      expect(ks.putRecords.calledOnce).to.be.true;
+      expect(ks.putRecords.calledOnce).to.be.true
       expect('{"hi":"hello","message":"[Circular]"}').to.equal(logMessage(ks.putRecords));
+
+    });
+
+    it('should invoke success when the records succeed to push', function(done) {
+      const stub = sinon.stub(ks, 'putRecords').callsFake(function(records, cb) {
+        return cb(null, records);
+      });
+
+      const message = {
+        hi: 'hello'
+      };
+      message.message = message;
+
+      ks.on('success', function(records) {
+        expect(records).to.exist;
+        expect(records.length).to.equal(1);
+        done();
+      });
+
+      ks.dispatch([message]);
     });
   });
 });