KinesisStore — Amazon Kinesis Data Streams

KinesisStore — Amazon Kinesis Data Streams Overview Amazon Kinesis Data Streams is a managed, sharded, append-only record-streaming service — a cloud sibling of Kafka. There is no query language and no SQL-shaped access pattern: producers put records onto a named stream, and consumers read them back per shard via shard iterators. Because none of that fits the SQL-shaped Dialect interface, KinesisStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes Kinesis's own streaming operations (produce, consume, administer) directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | --------------------------- | name | 'kinesis' | library | '@aws-sdk/client-kinesis' | The store is built on AWS SDK v3 (@aws-sdk/client-kinesis), which speaks the Kinesis wire protocol via the command pattern: you construct a command object (new PutRecordCommand(input)) and hand it to client.send(command). Lazy loading — not a hard dependency @aws-sdk/client-kinesis is not a hard dependency of this package. The SDK is loaded lazily via require() inside connect() (and on demand when a command is first built), rather than a top-level import. Importing this module therefore does not require the SDK to be installed — it is only needed when a Kinesis store is actually connected and driving real commands. Injected client KinesisStoreOptions accepts a pre-built client (any object exposing a compatible send(command) method, typed as KinesisClientLike). When provided, connect() uses it directly (and best-effort-loads the SDK only for its command constructors). This is how the test suite injects a mock client (no SDK, no network), and how callers can supply a custom-configured KinesisClient. Message encoding publish() accepts a string (UTF-8 encoded), a Buffer / Uint8Array (sent as-is), or any other JSON-serializable value (JSON.stringifyd, then UTF-8 encoded). The resulting bytes become the record's Data. Connection Build a store from connection options and call connect(): All connection options are optional: Option | Type | Purpose | ------------- | ------------------------------------------------- | ------------------------------------------------------------------------------------ | region | string | AWS region, e.g. 'us-east-1'. | credentials | { accessKeyId; secretAccessKey; sessionToken? } | Static credentials. Omit to fall back to the default AWS credential provider chain. | endpoint | string | Override the service endpoint, e.g. 'http://localhost:4566' for LocalStack. | client | KinesisClientLike | A pre-built client to use directly (mock or custom config). When set, the SDK is not required to build a client. | Injected-client form Supply your own client (or a mock) to bypass SDK-based client construction: Methods Internally, every operation follows the AWS v3 command pattern: the store resolves a command constructor by name from the lazily loaded SDK, builds new XxxCommand(input), and calls client.send(command). SDK failures are wrapped in a DatabaseError (preserving the original error); using any method before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Uses an injected client if provided, otherwise lazy-requires @aws-sdk/client-kinesis and builds a KinesisClient from the options. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | Stops all tracked subscriptions, calls the client's optional destroy(), and clears connection state. | isConnected | isConnected(): boolean | true only when connected and a client is present. | getClient | getClient(): KinesisClientLike| Returns the underlying client for operations not wrapped here. Throws ConnectionError if not connected. | Producing Method | Signature | AWS command | Behavior | --------- | ------------------------------------------------------------------------------------------------- | ----------------- | ----------------------------------------------------------------------------------------------------------- | publish | publish(stream: string, message: KinesisMessage, options?: PutRecordOptions): Promise<PutRecordResult> | PutRecordCommand | Encodes message to bytes and puts one record to stream. Defaults PartitionKey to a generated key when omitted. Returns { shardId?, sequenceNumber? }. | Stream administration Method | Signature | AWS command | Behavior | -------------- | ----------------------------------------------------------------- | -------------------- | ---------------------------------------------------------------------------------------------- | createStream | createStream(name: string, options?: CreateStreamOptions): Promise<void> | CreateStreamCommand | Provisions a new stream. Defaults ShardCount to 1; sets StreamModeDetails only when streamMode is given. | createTopic | createTopic(name: string, options?: CreateStreamOptions): Promise<void> | (delegates) | Alias for createStream(), matching the common store surface. | Consuming Method | Signature | AWS commands | Behavior | ----------- | ----------------------------------------------------------------------------------------------------------------- | ----------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- | subscribe | subscribe(stream: string, handler: (record: any) => void \| Promise<void>, options?: KinesisSubscribeOptions): Promise<KinesisSubscription> | GetShardIteratorCommand, GetRecordsCommand | Acquires a shard iterator, then polls GetRecords in a loop, invoking handler once per record. Resolves once the initial iterator is acquired; the poll loop runs detached (not awaited). Returns a subscription whose stop() ends the loop and untracks it. | Subscriptions are tracked internally so disconnect() stops each poll loop. GetShardIterator failures are wrapped in DatabaseError; the background poll loop itself is best-effort and swallows errors. Example Verification status Unit / mock-verified only. The tests in tests/nosql/kinesis.test.ts are fully mock-driven: a fake client whose send(command) records the command it was given and returns canned output is injected via KinesisStoreOptions.client. The real @aws-sdk/client-kinesis package is not installed, and there is no live Kinesis access and no network in the test run — the command constructors are stubbed on the store's aws handle so command-type assertions are deterministic. What this proves: Each method builds and sends the correct command type (PutRecordCommand, CreateStreamCommand, GetShardIteratorCommand, GetRecordsCommand) with the expected .input. Message encoding: strings become UTF-8 record bytes; objects are JSON-encoded. createStream / createTopic send CreateStreamCommand with the shard count; createTopic aliases createStream. subscribe acquires a shard iterator with the documented defaults, then polls GetRecords and invokes the handler per record; stop() ends the loop and untracks the subscription. Lifecycle behavior: connecting via an injected client, disconnect() destroying the client, and ConnectionError before connect(). Error handling: SDK failures (PutRecord, GetShardIterator) wrapped in DatabaseError. What this does not prove: live execution against real Amazon Kinesis (or LocalStack). The command/response shape is verified against the SDK's documented contract, but end-to-end execution over the wire has not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories