npm package discovery and stats viewer.

Discover Tips

  • General search

    [free text search, go nuts!]

  • Package details

    pkg:[package-name]

  • User packages

    @[username]

Sponsor

Optimize Toolset

I’ve always been into building performant and accessible sites, but lately I’ve been taking it extremely seriously. So much so that I’ve been building a tool to help me optimize and monitor the sites that I build to make sure that I’m making an attempt to offer the best experience to those who visit them. If you’re into performant, accessible and SEO friendly sites, you might like it too! You can check it out at Optimize Toolset.

About

Hi, 👋, I’m Ryan Hefner  and I built this site for me, and you! The goal of this site was to provide an easy way for me to check the stats on my npm packages, both for prioritizing issues and updates, and to give me a little kick in the pants to keep up on stuff.

As I was building it, I realized that I was actually using the tool to build the tool, and figured I might as well put this out there and hopefully others will find it to be a fast and useful way to search and browse npm packages as I have.

If you’re interested in other things I’m working on, follow me on Twitter or check out the open source projects I’ve been publishing on GitHub.

I am also working on a Twitter bot for this site to tweet the most popular, newest, random packages from npm. Please follow that account now and it will start sending out packages soon–ish.

Open Software & Tools

This site wouldn’t be possible without the immense generosity and tireless efforts from the people who make contributions to the world and share their work via open source initiatives. Thank you 🙏

© 2024 – Pkg Stats / Ryan Hefner

lifion-kinesis

v1.3.3

Published

Lifion client for Amazon Kinesis Data streams

Downloads

11,562

Readme

lifion-kinesis

npm version

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

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