Handle TrimmedDataAccessException errors

.npmignore

@@ -0,0 +1,7 @@
+.git/
+.github/
+.gitmodules
+.idea/
+.travis.yml
+.vscode/
+test/

index.js

@@ -18,6 +18,7 @@ class DynamodDBSubscriber extends EventEmitter {
     this._region = params.region;
     this._table = params.table;
     this._streamArn = params.arn;
+    this._endpoint = params.endpoint;
 
     if (typeof params.interval === 'number') {
       this._interval = params.interval;
@@ -27,7 +28,12 @@ class DynamodDBSubscriber extends EventEmitter {
       this._interval = ms('10s');
     }
 
-    this._ddbStream = new aws.DynamoDBStreams({ region: params.region });
+    this._ddbStream = params.endpoint
+      ? new aws.DynamoDBStreams({
+        region: params.region,
+        endpoint: params.endpoint
+      })
+      : new aws.DynamoDBStreams({ region: params.region });
   }
 
   _getOpenShards (callback) {
@@ -88,6 +94,21 @@ class DynamodDBSubscriber extends EventEmitter {
       debug('stream.getRecords (start) Shard: %s', shard.ShardId);
       this._ddbStream.getRecords({ ShardIterator: shard.iterator }, (err, data) => {
         if (err) {
+          if (err.code === 'TrimmedDataAccessException') {
+            this._ddbStream.getShardIterator({
+              StreamArn: this._streamArn,
+              ShardId: shard.ShardId,
+              ShardIteratorType: 'TRIM_HORIZON'
+            }, (errr, dataa) => {
+              if (errr) {
+                return callback(errr);
+              }
+              debug('stream.getShardIterator (end) Has ShardIterator? %s', Boolean(dataa.ShardIterator));
+              shard.iterator = dataa.ShardIterator;
+              callback(null, true);
+            });
+            return;
+          }
           return callback(err);
         }
 
@@ -105,11 +126,14 @@ class DynamodDBSubscriber extends EventEmitter {
         shard.iterator = data.NextShardIterator;
         callback();
       });
-    }, (err) => {
+    }, (err, rerunJob) => {
       if (err) {
         this.emit('error', err);
         return job.done();
       }
+      if (rerunJob) {
+        return this._process(job);
+      }
       //if some shard does not longer has an iterator
       //we need to fetch the openshards again and
       //process again.
@@ -134,7 +158,9 @@ class DynamodDBSubscriber extends EventEmitter {
       //try get stream arn with dynamodb.describeTable
       cb => {
         if (this._streamArn) { return cb(); }
-        const dynamo = new aws.DynamoDB({ region: this._region });
+        const dynamo = this._endpoint 
+          ? new aws.DynamoDB({ region: this._region, endpoint: this._endpoint }) 
+          : new aws.DynamoDB({ region: this._region });
         dynamo.describeTable({ TableName: this._table }, (err, tableDescription) => {
           if (err) {
             return cb();

test/dsubscriber.tests.js

@@ -76,6 +76,7 @@ describe('DynamodbSubscriber', function () {
     const arn = 'urn:test:test';
     const ShardId = '456';
     const ShardIterator = 'iterator-123';
+    const ExpiredShardIterator = 'expired-iterator-012';
     const TableName = 'credentials';
 
     before(function() {
@@ -111,6 +112,9 @@ describe('DynamodbSubscriber', function () {
         if (params.ShardId !== ShardId) {
           return callback(new Error(`unknown ShardId ${params.ShardId}`));
         }
+        if (params.ShardIteratorType === 'LATEST') {
+          return callback(null, { ShardIterator: ExpiredShardIterator });
+        }
         callback(null, { ShardIterator });
       });
 
@@ -129,6 +133,9 @@ describe('DynamodbSubscriber', function () {
 
       AWS.mock('DynamoDBStreams', 'getRecords', (params, callback) => {
         if (params.ShardIterator !== ShardIterator) {
+          if (params.ShardIterator === ExpiredShardIterator) {
+            return callback({code: 'TrimmedDataAccessException'});
+          }
           return callback(new Error(`unknown ShardId ${params.ShardId}`));
         }
         callback(null, { Records: [ record ], NextShardIterator: 'iterator-456' });