Subscribe to queues

Tip

This page is best viewed through the site search or through the AI.

Create an SQS queue, subscribe to it, and publish to it from a function.

sst.config.ts

const queue = new sst.aws.Queue("MyQueue");
queue.subscribe("subscriber.handler");

const app = new sst.aws.Function("MyApp", {
  handler: "publisher.handler",
  link: [queue],
  url: true,
});

The subscriber will read messages from the queue in batches. This array of messages exists on the Records property of the SQSEvent.

subscriber.ts

import type { SQSEvent, SQSHandler } from "aws-lambda";

export const handler: SQSHandler = async (event: SQSEvent) => {
  for (const record of event.Records){
    // Message bodies are always strings
    console.log(record.body)
  }
  return;
};

By default, all messages in the batch become visible in the queue again if an error occurs. This can lead to unnecessary extra processing and messages being processed more than once. The solution is to enable partial batch responsese and return which specific messages within the batch should be made visible again in the queue.

Update the queue subscriber.

sst.config.ts

queue.subscribe("subscriber.handler", {
  batch: {
    partialResponses: true,
  }
});

Then update the handler to return the failed items.

subscriber.ts

import type { SQSEvent, SQSHandler } from "aws-lambda";

export const handler: SQSHandler = async (event: SQSEvent) => {
  const batchItemFailures = []
  for (const record of event.Records){
    try {
      console.log(record.body)
      if (Math.random() < 0.1){
        throw new Error("An error occurred")
      }
    }
    catch (e) {
      batchItemFailures.push({ itemIdentifier: record.messageId });
    }
  }

  // Failed items will be made visible in the queue again
  return { batchItemFailures };
};

sst.config.ts

const queue = new sst.aws.Queue("MyQueue");
queue.subscribe("subscriber.handler", {
  batch: {
    partialResponses: true,
  }
});

const app = new sst.aws.Function("MyApp", {
  handler: "publisher.handler",
  link: [queue],
  url: true,
});

publisher.ts

import { Resource } from "sst";
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
const client = new SQSClient();

export const handler = async (event) => {
  // send a message
  await client.send(
    new SendMessageCommand({
      QueueUrl: Resource.MyQueue.url,
      MessageBody: "Hello from the subscriber",
    })
  );

  return {
    statusCode: 200,
    body: JSON.stringify({ status: "sent" }, null, 2),
  };
};

View the full example.