Wait for (CoorpAcademy#140)

Author: godu

packages/serverless-offline-dynamodb-streams/src/dynamodb-streams.js

@@ -46,6 +46,19 @@ class DynamodbStreams {
     return this._dynamodbStreamsEvent(functionKey, dynamodbStreamsEvent);
   }
 
+  async _describeTable(tableName) {
+    try {
+      await this.client.waitFor('tableExists', {TableName: tableName}).promise();
+      return await this.client
+        .describeTable({
+          TableName: tableName
+        })
+        .promise();
+    } catch (err) {
+      return this._waitFor(tableName);
+    }
+  }
+
   async _dynamodbStreamsEvent(functionKey, dynamodbStreamsEvent) {
     const {
       enabled,
@@ -60,11 +73,7 @@ class DynamodbStreams {
 
     const {
       Table: {LatestStreamArn}
-    } = await this.client
-      .describeTable({
-        TableName: tableName
-      })
-      .promise();
+    } = await this._describeTable(tableName);
 
     const {
       StreamDescription: {Shards: shards}

packages/serverless-offline-kinesis/src/kinesis.js

@@ -41,18 +41,27 @@ class Kinesis {
     return this._kinesisEvent(functionKey, kinesisEvent);
   }
 
+  async _describeStream(streamName) {
+    try {
+      await this.client.waitFor('streamExists', {StreamName: streamName}).promise();
+      return await this.client
+        .describeStream({
+          StreamName: streamName
+        })
+        .promise();
+    } catch (err) {
+      return this._waitFor(streamName);
+    }
+  }
+
   async _kinesisEvent(functionKey, kinesisEvent) {
     const {enabled, streamName, arn, batchSize, startingPosition} = kinesisEvent;
 
     if (!enabled) return;
 
     const {
       StreamDescription: {Shards: shards}
-    } = await this.client
-      .describeStream({
-        StreamName: streamName
-      })
-      .promise();
+    } = await this._describeStream(streamName);
 
     shards.forEach(({ShardId: shardId}) => {
       const readable = KinesisReadable(

packages/serverless-offline-s3/src/s3.js

@@ -4,6 +4,8 @@ const {assign, toNumber} = require('lodash/fp');
 const S3EventDefinition = require('./s3-event-definition');
 const S3Event = require('./s3-event');
 
+const delay = timeout => new Promise(resolve => setTimeout(resolve, timeout));
+
 class S3 {
   constructor(lambda, resources, options) {
     this.lambda = null;
@@ -43,8 +45,17 @@ class S3 {
     return this._s3Event(functionKey, s3Event);
   }
 
-  _s3Event(functionKey, s3Event) {
+  async _waitFor(bucket) {
+    const exists = await this.client.bucketExists(bucket);
+    if (exists) return;
+
+    await delay(1000);
+    return this._waitFor(bucket);
+  }
+
+  async _s3Event(functionKey, s3Event) {
     const {event, bucket} = s3Event;
+    await this._waitFor(bucket);
 
     const listener = this.client.listenBucketNotification(bucket, '*', '*', [event]);
 

packages/serverless-offline-sqs/src/sqs.js

@@ -6,6 +6,8 @@ const {logWarning} = require('serverless-offline/dist/serverlessLog');
 const SQSEventDefinition = require('./sqs-event-definition');
 const SQSEvent = require('./sqs-event');
 
+const delay = timeout => new Promise(resolve => setTimeout(resolve, timeout));
+
 class SQS {
   constructor(lambda, resources, options) {
     this.lambda = null;
@@ -43,6 +45,15 @@ class SQS {
     return this._sqsEvent(functionKey, sqsEvent);
   }
 
+  async _getQueueUrl(queueName) {
+    try {
+      return await this.client.getQueueUrl({QueueName: queueName}).promise();
+    } catch (err) {
+      await delay(10000);
+      return this._getQueueUrl(queueName);
+    }
+  }
+
   async _sqsEvent(functionKey, sqsEvent) {
     const {enabled, arn, queueName, batchSize} = sqsEvent;