Commonly used stream patterns for Web Streams API and NodeJS Stream.
If you're iterating over an array more than once, it's time to use streams.
@datastream/core- pipeline
- pipejoin
- result
- streamToArray
- streamToObject
- streamToString
- streamToBuffer
- isReadable
- isWritable
- makeOptions
- createReadableStream
- createPassThroughStream
- createTransformStream
- createWritableStream
- resolveLazy
- shallowClone
- deepClone
- shallowEqual
- deepEqual
- timeout
- createReadableStreamFromString (Node only)
- createReadableStreamFromArrayBuffer (Node only)
- backpressureGauge (Node only)
- Readable: The start of a pipeline of streams that injects data into a stream.
- PassThrough: Does not modify the data, but listens to the data and prepares a result that can be retrieved.
- Transform: Modifies data as it passes through.
- Writable: The end of a pipeline of streams that stores data from the stream.
@datastream/string- stringReadableStream [Readable]
- stringLengthStream [PassThrough]
- stringCountStream [PassThrough]
- stringMinimumFirstChunkSize [Transform]
- stringMinimumChunkSize [Transform]
- stringSkipConsecutiveDuplicates [Transform]
- stringReplaceStream [Transform]
- stringSplitStream [Transform]
@datastream/object- objectReadableStream [Readable]
- objectCountStream [PassThrough]
- objectBatchStream [Transform]
- objectPivotLongToWideStream [Transform]
- objectPivotWideToLongStream [Transform]
- objectKeyValueStream [Transform]
- objectKeyValuesStream [Transform]
- objectKeyJoinStream [Transform]
- objectKeyMapStream [Transform]
- objectValueMapStream [Transform]
- objectPickStream [Transform]
- objectOmitStream [Transform]
- objectFromEntriesStream [Transform]
- objectToEntriesStream [Transform]
- objectSkipConsecutiveDuplicatesStream [Transform]
@datastream/file- fileReadStream [Readable]
- fileWriteStream [Writable]
@datastream/fetch- fetchResponseStream [Readable]
@datastream/base64- base64EncodeStream [Transform]
- base64DecodeStream [Transform]
@datastream/charset[/{detect,decode,encode}]- charsetDetectStream [PassThrough]
- charsetDecodeStream [Transform]
- charsetEncodeStream [Transform]
@datastream/compress[/{gzip,deflate,brotli,zstd}]- gzipCompressStream [Transform]
- gzipDecompressStream [Transform]
- deflateCompressStream [Transform]
- deflateDecompressStream [Transform]
- brotliCompressStream [Transform]
- brotliDecompressStream [Transform]
- zstdCompressStream [Transform]
- zstdDecompressStream [Transform]
@datastream/digest- digestStream [PassThrough]
@datastream/csv[/{parse,format}]- csvParseStream [Transform]
- csvFormatStream [Transform]
@datastream/json- jsonParseStream [Transform]
- jsonFormatStream [Transform]
- ndjsonParseStream [Transform]
- ndjsonFormatStream [Transform]
@datastream/encrypt- encryptStream [Transform]
- decryptStream [Transform]
- generateEncryptionKey
@datastream/validate- validateStream [Transform]
@datastream/arrow- arrowDetectSchemaStream [PassThrough]
- arrowBatchFromArrayStream [Transform]
- arrowBatchFromObjectStream [Transform]
- arrowToArrayStream [Transform]
- arrowToObjectStream [Transform]
@datastream/duckdb- duckdbConnect (connection helper)
- duckdbAppenderStream [Writable]
- duckdbArrowInsertStream [Writable]
@datastream/indexeddb- indexedDBReadStream [Readable]
- indexedDBWriteStream [Writable]
@datastream/ipfs- ipfsGetStream [Readable]
- ipfsAddStream [Writable]
@datastream/protobuf- protobufEncodeStream [Transform]
- protobufDecodeStream [Transform]
- protobufLengthPrefixFrameStream [Transform]
- protobufLengthPrefixUnframeStream [Transform]
@datastream/schema-registry- confluentFrameStream [Transform]
- confluentUnframeStream [Transform]
- glueFrameStream [Transform]
- glueUnframeStream [Transform]
@datastream/kafka- kafkaConnect (producer/consumer connection helper)
- kafkaConsumeStream [Readable]
- kafkaProduceStream [Writable]
@datastream/aws/msk-iam- awsMskIamMechanism (kafkajs OAUTHBEARER SASL config)
@datastream/aws/glue-schema-registry- awsGlueSchemaRegistryResolver (cached GetSchemaVersion lookup)
npm install @datastream/core @datastream/{module}stateDiagram-v2
[*] --> fileRead: path
[*] --> fetchResponse: URL
[*] --> sqlCopyTo*: SQL
[*] --> stringReadable: string
[*] --> stringReadable: string[]
[*] --> objectReadable: object[]
[*] --> createReadable: blob
readable --> charsetDetect: binary
charsetDetect --> [*]
readable --> decryption
decryption --> passThroughBuffer: buffer
readable --> decompression
decompression --> passThroughBuffer: buffer
passThroughBuffer --> charsetDecode: buffer
charsetDecode --> passThroughString: string
passThroughString --> parse: string
parse --> validate: object
validate --> passThroughObject: object
passThroughObject --> transform: object
transform --> format: object
format --> charsetEncode: string
charsetEncode --> compression: buffer
compression --> writable: buffer
charsetEncode --> encryption: buffer
encryption --> writable: buffer
state readable {
fileRead
fetchResponse
sqlCopyTo*
createReadable
stringReadable
objectReadable
awsS3Get
awsDynamoDBQuery
awsDynamoDBScan
awsDynamoDBGet
}
state decompression {
brotliDecompress
gzipDecompress
deflateDecompress
zstdDecompress
}
state decryption {
decrypt
}
state parse {
csvParse
jsonParse
xmlParse*
}
state passThroughBuffer {
digest
}
state passThroughString {
stringLength
stringCount
}
state passThroughObject {
objectCount
objectBatch
}
state transform {
objectBatch
objectPivotLongToWide
objectPivotWideToLong
objectKeyValue
objectKeyValue
}
state format {
csvFormat
jsonFormat
xmlFormat*
}
state compression {
brotliCompress
gzipCompress
deflateCompress
zstdCompress
}
state encryption {
encrypt
}
state writable {
fileWrite
fetchRequest*
sqlCopyFrom
awsS3Put
awsDynamoDBPut
awsDynamoDBDelete
}
writable --> [*]
* possible future package
Read a CSV file, validate the structure, pivot data, then save compressed.
- fs.creatReadStream
- gzip
- cryptoDigest
- charsetDecode
- csvParse
- countChunks
- validate
- changeCase (pascal to snake)
- parquet?
- csvFormat
- postgesCopyFrom
- cryptoDigest
- charsetDetect
- jsonParse?
- validate
Upload file with brotli compression?
Fetch a protobuf-encoded file, decode the protobuf framing, then parse JSON
filter
file (docs only?)
- fetch
- node:fs
- input type=file
- readable string/array/etc
Licensed under MIT License. Copyright (c) 2026 will Farrell and contributors.