StreamEventSource

Source: src/AWS/Kinesis/StreamEventSource.ts

Event source connecting a Kinesis Stream to the hosting compute.

The contract is a Binding.Service; the Lambda implementation layer (AWS.Lambda.StreamEventSource) creates an event source mapping on the stream, grants the read IAM actions, and forwards aws:kinesis records into the handler’s Stream. Use the consumeStreamRecords helper rather than calling the service directly.

Consuming Records

export default MyFunction.make(
  { main: import.meta.url },
  Effect.gen(function* () {
    const stream = yield* AWS.Kinesis.Stream("OrdersStream");

    // init — registers the event source mapping and the record handler
    yield* AWS.Kinesis.consumeStreamRecords(
      stream,
      { startingPosition: "LATEST", batchSize: 10 },
      (records) =>
        records.pipe(
          Stream.runForEach((record) =>
            Effect.log(
              Buffer.from(record.kinesis.data, "base64").toString("utf8"),
            ),
          ),
        ),
    );

    return {};
  }).pipe(Effect.provide(AWS.Lambda.StreamEventSource)),
);