Kinesis streams

Create a Kinesis stream, and subscribe to it with a function.

const stream = new sst.aws.KinesisStream("MyStream");

// Create a function subscribing to all events
stream.subscribe("AllSub", "subscriber.all");

// Create a function subscribing to events of `bar` type
stream.subscribe("FilteredSub", "subscriber.filtered", {
  filters: [
    {
      data: {
        type: ["bar"],
      },
    },
  ],
});

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

import { KinesisStreamHandler } from "aws-lambda";

export const all: KinesisStreamHandler = async (event) => {
  for (const record of event.Records) {
    const data = Buffer.from(record.kinesis.data, "base64").toString();
    console.log(`"All" subscriber received:`, data);
  }
};

export const filtered: KinesisStreamHandler = async (event) => {
  for (const record of event.Records) {
    const data = Buffer.from(record.kinesis.data, "base64").toString();
    console.log(`"Filtered" subscriber received:`, data);
  }
};

import { KinesisClient, PutRecordsCommand } from "@aws-sdk/client-kinesis";
import { Resource } from "sst";

export const handler = async (event) => {
  const client = new KinesisClient();

  await client.send(
    new PutRecordsCommand({
      Records: [
        {
          Data: JSON.stringify({ type: "foo" }),
          PartitionKey: "1",
        },
        {
          Data: JSON.stringify({ type: "bar" }),
          PartitionKey: "1",
        },
      ],
      StreamName: Resource.MyStream.name,
    })
  );

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

View the full example.