Security News
PyPI’s New Archival Feature Closes a Major Security Gap
PyPI now allows maintainers to archive projects, improving security and helping users make informed decisions about their dependencies.
kinesis-streams
Advanced tools
There once was a Kinesis readable stream without a home, and a Kinesis writable stream without a home, so now they're roommates.
NOTE: Kinesis was a bad idea, and we're switching to Kafka. So I won't be using my own library for much longer.
npm install kinesis-streams
const AWS = require('aws-sdk')
const { KinesisWritable } = require('kinesis-streams')
const client = new AWS.Kinesis()
client.config.update({ maxRetries: 10 })
const writable = new KinesisWritable(client, 'streamName', options)
inputStream.pipe(writable)
options.logger
(optional) bunyan, winston, or logger with debug
, error
and info
options.highWaterMark
(default: 16) Buffer this many records before writing to Kinesis. Equivalent to CollectionMaxCount
options.wait
(default: 500) How many milliseconds it should periodically flush. Equivalent to RecordMaxBufferedTime
Some of these options have equivalents in the official KPL.
These events are emitted:
kinesis.putRecords
Fires after records are put and the response is processed. You'll get the original response from AWS. See demo.js
for an example of how to interpret it
reader.on('kinesis.putRecords', (response: {FailedRecordCount: number, Records: Record[]}) => {})
By default, the partition key is to a dummy value, '0'
. If you have
multiple shards, you need to set a partition key in a way that makes sense
for your data. Here are two ways to do this:
getPartitionKey
method of the writable stream instance:const AWS = require('aws-sdk')
const { KinesisWritable } = require('kinesis-streams')
const client = new AWS.Kinesis()
const writable = new KinesisWritable(client, 'streamName', options)
writable.getPartitionKey = (data) => data.foo.substr(5)
inputStream.pipe(writable)
KinesisWritable
and provide your own getPartitionKey
. See the source for reference.const AWS = require('aws-sdk')
const { KinesisReadable } = require('kinesis-streams')
const client = new AWS.Kinesis()
const reader = new KinesisReadable(client, streamName, options)
reader.pipe(yourDestinationHere)
options.logger
(optional) bunyan, winston, or logger with debug
, error
and info
options.interval: number
(default: 2000
) Milliseconds between each Kinesis read. The AWS limit is 5 reads / second / shard
options.parser: Function
If this is set, this function is applied to the data. Example:
const reader = new KinesisReadable(client, streamName, {parser: JSON.parse})
reader.on('data', console.log(data.id))
options.restartOnClose: boolean
(default: false
) Rediscover new shards once all current shards have been closed
And any getShardIterator parameter
These events are emitted:
checkpoint
This fires when data is received so you can keep track of the last successful sequence read:
reader.on('checkpoint', (sequenceNumber: string) => {})
KinesisWritable
and KinesisReadable
both take an optional logger
option.
If this is omitted, the debug logger will be used instead. To see output, set
DEBUG=kinesis-streams:*
in your environment.
The writable stream is based on the interface of kinesis-write-stream. The
checkpoint
event in readable stream is based on kinesis-readable. The
readable stream was originally written as a proof of concept in
kinesis-console-consumer.
kinesis-write-stream was forked because at the time, it didn't support periodic flushes. Since then the configuration of the readable and writable streams have been rewritten to be consistent, and both emit lots of events now that consumers can use for instrumentation.
This package is licensed under Apache License 2.0, but the
tests/writable.spec.js
and test/fixture/*
are originally from
kinesis-write-stream MIT licensed from Espen Volden.
FAQs
Readable and writable streams for AWS Kinesis
The npm package kinesis-streams receives a total of 294 weekly downloads. As such, kinesis-streams popularity was classified as not popular.
We found that kinesis-streams demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI now allows maintainers to archive projects, improving security and helping users make informed decisions about their dependencies.
Research
Security News
Malicious npm package postcss-optimizer delivers BeaverTail malware, targeting developer systems; similarities to past campaigns suggest a North Korean connection.
Security News
CISA's KEV data is now on GitHub, offering easier access, API integration, commit history tracking, and automated updates for security teams and researchers.