StreamSink

Source: src/AWS/Kinesis/StreamSink.ts

A partition-aware sink for batching PutRecords requests into a stream (500 records / 5 MiB per call).

Each input element is a raw PutRecordsRequestEntry, so callers stay in control of PartitionKey and optional ExplicitHashKey.

Records the API reports as failed (FailedRecordCount > 0, per-record ErrorCode — throughput exceeded or internal failure) are re-submitted on a bounded schedule; exhausting retries fails the sink with a typed BatchRetryExhaustedError carrying the stranded records.

Provide the implementation with Effect.provide(AWS.Kinesis.StreamSinkHttp).

Writing Streams of Records

// init — bind the sink to the stream
const sink = yield* AWS.Kinesis.StreamSink(stream);

// runtime — batches into PutRecords calls of up to 500 records / 5 MiB
yield* Stream.fromIterable(
  orders.map((order) => ({
    PartitionKey: order.id,
    Data: new TextEncoder().encode(JSON.stringify(order)),
  })),
).pipe(Stream.run(sink));