Merge pull request #120 from t-richard/support-delay-queues

Support delay on queues

Author: mnapoli

docs/queue.md

@@ -299,6 +299,24 @@ constructs:
             # The retry delay on the queue will be 10*6 => 60 seconds
 ```
 
+### Delivery delay
+
+When a message is sent to the queue, it will be available immediately to the worker.
+
+You can postpone the delivery of messages by a given amount of seconds using the `delay` option.
+
+The maximum value is 900 seconds (15 minutes).
+
+```yaml
+constructs:
+    my-queue:
+        # ...
+        worker:
+            handler: src/worker.handler
+            # Messages delivery will be delayed by 1 minute
+            delay: 60
+```
+
 ### Batch size
 
 ```yaml

src/constructs/aws/Queue.ts

@@ -38,6 +38,7 @@ const QUEUE_DEFINITION = {
             maximum: 10,
         },
         fifo: { type: "boolean" },
+        delay: { type: "number" },
     },
     additionalProperties: false,
     required: ["worker"],
@@ -128,6 +129,18 @@ export class Queue extends AwsConstruct {
 
         const maxRetries = configuration.maxRetries ?? 3;
 
+        let delay = undefined;
+        if (configuration.delay !== undefined) {
+            if (configuration.delay < 0 || configuration.delay > 900) {
+                throw new ServerlessError(
+                    `Invalid configuration in 'constructs.${this.id}': 'delay' must be between 0 and 900, '${configuration.delay}' given.`,
+                    "LIFT_INVALID_CONSTRUCT_CONFIGURATION"
+                );
+            }
+
+            delay = Duration.seconds(configuration.delay);
+        }
+
         const baseName = `${this.provider.stackName}-${id}`;
 
         const dlq = new CdkQueue(this, "Dlq", {
@@ -147,6 +160,7 @@ export class Queue extends AwsConstruct {
                 queue: dlq,
             },
             fifo: configuration.fifo,
+            deliveryDelay: delay,
             contentBasedDeduplication: configuration.fifo,
         });
 

test/unit/queues.test.ts

@@ -4,6 +4,7 @@ import type { DeleteMessageBatchResult, ReceiveMessageResult, SendMessageBatchRe
 import * as CloudFormationHelpers from "../../src/CloudFormation";
 import { pluginConfigExt, runServerless } from "../utils/runServerless";
 import { mockAws } from "../utils/mockAws";
+import ServerlessError from "../../src/utils/error";
 
 describe("queues", () => {
     afterEach(() => {
@@ -192,6 +193,49 @@ describe("queues", () => {
         });
     });
 
+    it("allows changing the delivery delay", async () => {
+        const { cfTemplate, computeLogicalId } = await runServerless({
+            fixture: "queues",
+            configExt: merge(pluginConfigExt, {
+                constructs: {
+                    emails: {
+                        delay: 10,
+                    },
+                },
+            }),
+            command: "package",
+        });
+        expect(cfTemplate.Resources[computeLogicalId("emails", "Queue")]).toMatchObject({
+            Properties: {
+                DelaySeconds: 10,
+            },
+        });
+    });
+
+    it("should throw an error if the delay is invalid", async () => {
+        expect.assertions(2);
+
+        try {
+            await runServerless({
+                fixture: "queues",
+                configExt: merge({}, pluginConfigExt, {
+                    constructs: {
+                        emails: {
+                            delay: 901,
+                        },
+                    },
+                }),
+                command: "package",
+            });
+        } catch (error) {
+            expect(error).toBeInstanceOf(ServerlessError);
+            expect(error).toHaveProperty(
+                "message",
+                "Invalid configuration in 'constructs.emails': 'delay' must be between 0 and 900, '901' given."
+            );
+        }
+    });
+
     it("allows defining a DLQ email alarm", async () => {
         const { cfTemplate, computeLogicalId } = await runServerless({
             fixture: "queues",