lifion-kinesis

lifion-kinesis

Lifion's Node.js client for Amazon Kinesis Data Streams.

Getting Started

To install the module:

npm install lifion-kinesis --save

The main module export is a Kinesis class that instantiates as a readable stream.

const Kinesis = require('lifion-kinesis');

const kinesis = new Kinesis({
  streamName: 'sample-stream'
  /* other options from AWS.Kinesis */
});
kinesis.on('data', data => {
  console.log('Incoming data:', data);
});
kinesis.startConsumer();

To take advantage of back-pressure, the client can be piped to a writable stream:

const { promisify } = require('util');
const Kinesis = require('lifion-kinesis');
const stream = require('stream');

const asyncPipeline = promisify(stream.pipeline);
const kinesis = new Kinesis({
  streamName: 'sample-stream'
  /* other options from AWS.Kinesis */
});

asyncPipeline(
  kinesis,
  new stream.Writable({
    objectMode: true,
    write(data, encoding, callback) {
      console.log(data);
      callback();
    }
  })
).catch(console.error);
kinesis.startConsumer();

Features

• Standard Node.js stream abstraction of Kinesis streams.
• Node.js implementation of the new enhanced fan-out feature.
• Optional auto-creation, encryption, and tagging of Kinesis streams.
• Support for a polling mode, using the GetRecords API, with automatic checkpointing.
• Support for multiple concurrent consumers through automatic assignment of shards.
• Support for sending messages to streams, with auto-retries.

API Reference

• lifion-kinesis
  • Kinesis ⇐ PassThrough ⏏
    • new Kinesis(options)
    • instance
      • .startConsumer() ⇒ Promise
      • .stopConsumer()
      • .putRecord(params) ⇒ Promise
      • .listShards(params) ⇒ Promise
      • .putRecords(params) ⇒ Promise
      • .getStats() ⇒ Object
    • static
      • .getStats() ⇒ Object

Kinesis ⇐ PassThrough ⏏

A pass-through stream class specialization implementing a consumer of Kinesis Data Streams using the AWS SDK for JavaScript. Incoming data can be retrieved through either the data event or by piping the instance to other streams.

Kind: Exported class
Extends: PassThrough

new Kinesis(options)

Initializes a new instance of the Kinesis client.

Param	Type	Default	Description
options	Object		The initialization options. In addition to the below options, it can also contain any of the AWS.Kinesis options.
[options.compression]	string		The kind of data compression to use with records. The currently available compression options are either "LZ-UTF8" or none.
[options.consumerGroup]	string		The name of the group of consumers in which shards will be distributed and checkpoints will be shared. If not provided, it defaults to the name of the application/project using this module.
[options.createStreamIfNeeded]	boolean	true	Whether if the Kinesis stream should be automatically created if it doesn't exist upon connection
[options.dynamoDb]	Object	{}	The initialization options for the DynamoDB client used to store the state of the consumers. In addition to tableNames and tags, it can also contain any of the AWS.DynamoDB options.
[options.dynamoDb.tableName]	string		The name of the table in which to store the state of consumers. If not provided, it defaults to "lifion-kinesis-state".
[options.dynamoDb.tags]	Object		If provided, the client will ensure that the DynamoDB table where the state is stored is tagged with these tags. If the table already has tags, they will be merged.
[options.encryption]	Object		The encryption options to enforce in the stream.
[options.encryption.type]	string		The encryption type to use.
[options.encryption.keyId]	string		The GUID for the customer-managed AWS KMS key to use for encryption. This value can be a globally unique identifier, a fully specified ARN to either an alias or a key, or an alias name prefixed by "alias/".
[options.initialPositionInStream]	string	"LATEST"	The location in the shard from which the Consumer will start fetching records from when the application starts for the first time and there is no checkpoint for the shard. Set to LATEST to fetch new data only Set to TRIM_HORIZON to start from the oldest available data record.
[options.leaseAcquisitionInterval]	number	20000	The interval in milliseconds for how often to attempt lease acquisitions.
[options.leaseAcquisitionRecoveryInterval]	number	5000	The interval in milliseconds for how often to re-attempt lease acquisitions when an error is returned from aws.
[options.limit]	number	10000	The limit of records per get records call (only applicable with useEnhancedFanOut is set to false)
[options.logger]	Object		An object with the warn, debug, and error functions that will be used for logging purposes. If not provided, logging will be omitted.
[options.maxEnhancedConsumers]	number	5	An option to set the number of enhanced fan-out consumer ARNs that the module should initialize. Defaults to 5. Providing a number above the AWS limit (20) or below 1 will result in using the default.
[options.noRecordsPollDelay]	number	1000	The delay in milliseconds before attempting to get more records when there were none in the previous attempt (only applicable when useEnhancedFanOut is set to false)
[options.pollDelay]	number	250	When the usePausedPolling option is false, this option defines the delay in milliseconds in between poll requests for more records (only applicable when useEnhancedFanOut is set to false)
[options.s3]	Object	{}	The initialization options for the S3 client used to store large items in buckets. In addition to bucketName and endpoint, it can also contain any of the AWS.S3 options.
[options.s3.bucketName]	string		The name of the bucket in which to store large messages. If not provided, it defaults to the name of the Kinesis stream.
[options.s3.largeItemThreshold]	number	900	The size in KB above which an item should automatically be stored in s3.
[options.s3.nonS3Keys]	Array.<string>	[]	If the useS3ForLargeItems option is set to true, the nonS3Keys option lists the keys that will be sent normally on the kinesis record.
[options.s3.tags]	string		If provided, the client will ensure that the S3 bucket is tagged with these tags. If the bucket already has tags, they will be merged.
[options.shardCount]	number	1	The number of shards that the newly-created stream will use (if the createStreamIfNeeded option is set)
[options.shouldDeaggregate]	string | boolean	"auto"	Whether the method retrieving the records should expect aggregated records and deaggregate them appropriately.
[options.shouldParseJson]	string | boolean	"auto"	Whether if retrieved records' data should be parsed as JSON or not. Set to "auto" to only attempt parsing if data looks like JSON. Set to true to force data parse.
[options.statsInterval]	number	30000	The interval in milliseconds for how often to emit the "stats" event. The event is only available while the consumer is running.
options.streamName	string		The name of the stream to consume data from (required)
[options.supressThroughputWarnings]	boolean	false	Set to true to make the client log ProvisionedThroughputExceededException as debug rather than warning.
[options.tags]	Object		If provided, the client will ensure that the stream is tagged with these tags upon connection. If the stream is already tagged, the existing tags will be merged with the provided ones before updating them.
[options.useAutoCheckpoints]	boolean	true	Set to true to make the client automatically store shard checkpoints using the sequence number of the most-recently received record. If set to false consumers can use the setCheckpoint() function to store any sequence number as the checkpoint for the shard.
[options.useAutoShardAssignment]	boolean	true	Set to true to automatically assign the stream shards to the active consumers in the same group (so only one client reads from one shard at the same time). Set to false to make the client read from all shards.
[options.useEnhancedFanOut]	boolean	false	Set to true to make the client use enhanced fan-out consumers to read from shards.
[options.usePausedPolling]	boolean	false	Set to true to make the client not to poll for more records until the consumer calls continuePolling(). This option is useful when consumers want to make sure the records are fully processed before receiving more (only applicable when useEnhancedFanOut is set to false)
[options.useS3ForLargeItems]	boolean	false	Whether to automatically use an S3 bucket to store large items or not.

kinesis.startConsumer() ⇒ Promise

Starts the stream consumer, by ensuring that the stream exists, that it's ready, and configured as requested. The internal managers that deal with heartbeats, state, and consumers will also be started.

Kind: instance method of Kinesis
Fulfil: undefined - Once the consumer has successfully started.
Reject: Error - On any unexpected error while trying to start.

kinesis.stopConsumer()

Stops the stream consumer. The internal managers will also be stopped.

Kind: instance method of Kinesis

kinesis.putRecord(params) ⇒ Promise

Writes a single data record into a stream.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

Param	Type	Description
params	Object	The parameters.
params.data	*	The data to put into the record.
[params.explicitHashKey]	string	The hash value used to explicitly determine the shard the data record is assigned to by overriding the partition key hash.
[params.partitionKey]	string	Determines which shard in the stream the data record is assigned to. If omitted, it will be calculated based on a SHA-1 hash of the data.
[params.sequenceNumberForOrdering]	string	Set this to the sequence number obtained from the last put record operation to guarantee strictly increasing sequence numbers, for puts from the same client and to the same partition key. If omitted, records are coarsely ordered based on arrival time.
[params.streamName]	string	If provided, the record will be put into the specified stream instead of the stream name provided during the consumer instantiation.

kinesis.listShards(params) ⇒ Promise

List the shards of a stream.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

Param	Type	Description
params	Object	The parameters.
[params.streamName]	string	If provided, the method will list the shards of the specific stream instead of the stream name provided during the consumer instantiation.

kinesis.putRecords(params) ⇒ Promise

Writes multiple data records into a stream in a single call.

Kind: instance method of Kinesis
Fulfil: Object - The de-serialized data returned from the request.
Reject: Error - On any unexpected error while writing to the stream.

Param	Type	Description
params	Object	The parameters.
params.records	Array.<Object>	The records associated with the request.
params.records[].data	*	The record data.
[params.records[].explicitHashKey]	string	The hash value used to explicitly determine the shard the data record is assigned to by overriding the partition key hash.
[params.records[].partitionKey]	string	Determines which shard in the stream the data record is assigned to. If omitted, it will be calculated based on a SHA-1 hash of the data.
[params.streamName]	string	If provided, the record will be put into the specified stream instead of the stream name provided during the consumer instantiation.

kinesis.getStats() ⇒ Object

Returns statistics for the instance of the client.

Kind: instance method of Kinesis
Returns: Object - An object with the statistics.

Kinesis.getStats() ⇒ Object

Returns the aggregated statistics of all the instances of the client.

Kind: static method of Kinesis
Returns: Object - An object with the statistics.

License

MIT

Changelog

v1.3.3 (2023-08-02)

• #722: fix: package.json & package-lock.json to reduce vulnerabilities
• #721: [Snyk] Security upgrade aws-sdk from 2.1094.0 to 2.1354.0
• #625: [Snyk] Security upgrade got from 11.8.3 to 11.8.5
• #615: [Snyk] Security upgrade protobufjs from 6.11.2 to 6.11.3
• #578: Bump node-fetch from 2.6.5 to 2.6.7
• #574: Added timestamp to shard checkpoint
• 594870f: revert version bump
• 26b71ae: address comments
• d1dfaff: procced storing checkpoint if timestamp not exist
• 281c4b7: move approximateArrivalTimestamp to the end
• 5cfc79c: rename timestamp to approximateArrivalTimestamp
• 0a09f11: lint