Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

psxcode/node-streams

Repository files navigation

Node Streams

Install

npm install node-streams

buffer

transform stream, buffers all chunks until wait function calls. Pushes buffered chunks array. (options: TransformOptions) => (wait: WaitFn) => Transform

WaitFn = (callback: () => void) => CancelFn
buffered stream will call wait function for timeout, so it is possible to provide different period each time

import { buffer } from 'node-streams'
// we have the streams with `objectMode` set to true
declare var inStream: ReadableStream
declare var outStream: WritableStream
const wait100ms = (callback: () => void) => {
 const id = setTimeout(callback, 100)
 return () => clearTimeout(id)
}
const buffered = buffer({
 objectMode: true // set up standard TranformOptions
})(
 wait100ms // buffer for 100ms then push accumulated data
)
inStream
 .pipe(buffered)
 .pipe(outStream)

bufferTime

buffers all chunks for provided time period. Pushes buffered chunks array. (options: TransformOptions) => (ms: number) => Transform

import { bufferTime } from 'node-streams'
// we have the streams with `objectMode` set to true
declare var inStream: ReadableStream
declare var outStream: WritableStream
const buffered = buffer({
 objectMode: true // set up standard TranformOptions
})(
 100 // buffer for 100ms then push accumulated data
)
inStream
 .pipe(buffered)
 .pipe(outStream)

combine

delivers latest values from all streams as an array, ends when any of streams does (options: ReadableOptions) => (...streams: ReadableStream[]): ReadableStream

import { combine } from 'node-streams'
// we have the streams
declare var stream0: ReadableStream [..1..2..3..4..]
declare var stream1: ReadableStream ['a'...'b']
// create combined readable
const combined = combine({
 objectMode: true // provide standard ReadableOptions
})(stream0, stream1)
combined
 .on('data', ([value0, value1]) => {
 // [undefined, 'a']..[1, 'a']..[2, 'a'].[2, 'b']
 })
 .on('end', () => {})

concat

concatenates all streams, subscribing to next on previous has ended (options: ReadableOptions) => (...streams: ReadableStream[]): ReadableStream

import { concat } from 'node-streams'
// we have the streams
declare var stream0: ReadableStream [..1..2]
declare var stream1: ReadableStream [..'a'..'b']
// create combined readable
const combined = concat({
 objectMode: true // provide standard ReadableOptions
})(stream0, stream1)
combined
 .on('data', ([value0, value1]) => {
 // [..1..2..'a'..'b']
 })
 .on('end', () => {})

debounce

skips all fast arriving data, until idle period, after that pushes last received chunk. (options: TransformOptions) => (wait: WaitFn) => Transform

debounced stream will cancel and rearm wait function repeatedly on every chunk received.
WaitFn = (cb: () => void) => CancelFn
provided wait function should properly release timeout resources on CancelFn call.

import { debounce } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
// we have `setTimeout` function with baked in 10ms time
// CAUTION: wait function should be able to cancel timeout
const wait10ms = (callback: () => void) => {
 const id = setTimeout(callback, 10)
 return () => clearTimeout(id)
}
const debounced = debounce({
 objectMode: true // set up standard TranformOptions
})(
 wait10ms // debounce stream for 10ms
)
inStream
 .pipe(debounced)
 .pipe(outStream)

debounceTime

skips all fast arriving data, until idle period, after that pushes last received chunk. (options: TransformOptions) => (ms: number) => Transform

import { debounceTime } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
const debounced = debounceTime({
 objectMode: true // set up standard TranformOptions
})(
 10 // debounce stream for 10ms
)
inStream
 .pipe(debounced)
 .pipe(outStream)

delay

(options: TransformOptions) => (ms: number) => Transform

import { delay } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
const delayed = delay({
 objectMode: true // set up standard TransformOptions
})(
 100 // delay stream events up to 100ms
)
inStream
 .pipe(delayed)
 .pipe(outStream)

distinct

(options: TransformOptions) => (isEqual: (a: T, b: T) => boolean) => Transform

import { distinct } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..2..3..3..2..1]
declare var outStream: WritableStream
const unique = distinct({
 objectMode: true // set up standard TransformOptions
})(
 (a, b) => a === b // provide compare function
)
inStream
 .pipe(unique)
 .pipe(outStream) // [1..2....3....2..1]

distinctUntilChanged

(options: TransformOptions) => Transform

import { distinctUntilChanged } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..2..3..3..2..1]
declare var outStream: WritableStream
const unique = distinctUntilChanged({
 objectMode: true // set up standard TransformOptions
})
inStream
 .pipe(unique)
 .pipe(outStream) // [1..2....3....2..1]

empty

(options: ReadableOptions) => Readable

import { empty } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
(myCondition
 ? inStream
 : empty({})
).pipe(outStream)

filter

(options: TransformOptions) => (predicate: (arg: T) => boolean) => Transform

import { filter } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..3..4..5..6..7]
declare var outStream: WritableStream
const filtered = filter({
 objectMode: true // provide standard TransformOptions
})(
 a => a % 2 === 0 // is even
)
inStream
 .pipe(filtered)
 .pipe(outStream) // [..2....4....6..]

first

(options: TransformOptions) => Transform

import { first } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..3..4..5..6..7]
declare var outStream: WritableStream
const firstTransform = first({
 objectMode: true // provide standard TransformOptions
})
inStream
 .pipe(firstTransform)
 .pipe(outStream) // [1]

from

(options: ReadableOptions) => (iterable: Iterable<T>) => Readable

import { from } from 'node-streams'
const myStream = from({
 objectMode: true // provide standard ReadableOptions
})(
 [1, 2, 3, 4, 5] // provide Iterable
)
myStream
 .on('data', () => {}) // subscribe and get the data
 .on('end', () => {})

last

(options: TransformOptions) => Transform

import { last } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..3..4..5..6..7]
declare var outStream: WritableStream
const lastTransform = last({
 objectMode: true // provide standard TransformOptions
})
inStream
 .pipe(lastTransform)
 .pipe(outStream) // [7]

map

(options: TransformOptions) => (xf: (value: T) => R) => Transform

import { map } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
const mapped = map({
 objectMode: true // provide standard TransformOptions
})(
 x => x * 2
)
inStream
 .pipe(mapped)
 .pipe(outStream) // [2..4..6..8]

merge

(options: ReadableOptions) => (...streams: ReadableStreams[]) => Readable

import { merge } from 'node-streams'
// we have the streams in `objectMode`
declare var stream0: ReadableStream // [1...2...3...4]
declare var stream1: ReadableStream // [..'a'.....'b'...]
const merged = merge({
 objectMode: true // provide standard ReadableOptions
})(
 stream0,
 stream1
)
merged
 .on('data' () => {}) // [1..'a'..2..3..'b'..4..]
 .on('end', () => {})

of

(options: ReadableOptions) => (...values: T[]) => Readable

import { of } from 'node-streams'
const myStream = of({
 objectMode: true // provide standard ReadableOptions
})(
 1, 2, 3, 4
)
myStream
 .on('data', () => {}) // [1.2.3.4]
 .on('end', () => {})

ofAsync

(options: ReadableOptions) => (wait: WaitFn) => (...values: T[]) => Readable

WaitFn = (callback: () => void) => UnsubscribeFn

import { ofAsync } from 'node-streams'
// we have custom wait function
const wait100ms = (cb: () => void) => {
 const id = seTimeout(cb, 100)
 return () => clearTimeout(id)
}
const myStream = ofAsync({
 objectMode: true // provide standard ReadableOptions
})(
 wait100ms // set WaitFn
)(
 1, 2, 3, 4, 5 // provide values
)
myStream
 .on('data', () => {}) // [1...2...3...4...5]
 .on('end', () => {})

ofTime

(options: ReadableOptions) => (ms: number) => (...values: T[]) => Readable

import { ofTime } from 'node-streams'
const myStream = ofTime({
 objectMode: true // provide standard ReadableOptions
})(
 100 // provide time in milliseconds
)(
 1, 2, 3, 4 // values to stream
)
myStream
 .on('data', () => {}) // [1...2...3...4]
 .on('end', () => {})

pipe

(...streams: Array<ReadWriteStream | ReadWriteStream[]>) => ReadWriteStream[]

import { pipe } from 'node-streams'
// we have the following streams
declare var inStream: ReadableStream
declare var outStream: WritableStream
declare var tripleValues: TransformStream
declare var isEvenValues: TransformStream
declare var takeFirstValue: TransformStream
const tripleEven = pipe(
 tripleValues,
 isEvenValues
)
const firstTripleEven = pipe(
 tripleEven,
 takeFirstValue
)
pipe(
 inStream,
 firstTripleEven,
 outStream
)

pluck

(opts: TransformOptions) => (propName: string) => Transform

import { pluck } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream
declare var outStream: WritableStream
const pluckMyProp = pluck({
 objectMode: true // provide standard TransformOptions
})(
 'my-prop' // property name
)
inStream
 .pipe(pluckMyProp)
 .pipe(outStream)

reduce

(options: TransformOptions) => (reducer: (state: S, value: T) => S) => Transform

import { reduce } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
// we have the following reducer
const addAll = (acc = 0, value: number) => acc + value
const reduceTransform = reduce({
 objectMode: true // provide standard TransformOptions
})(
 addAll // set the reducer
)
inStream
 .pipe(reduceTransform)
 .pipe(outStream) // [............10]

scan

(options: TransformOptions) => (reducer: (state: S, value: T) => S) => Transform

import { scan } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
// we have the following reducer
const addAll = (acc = 0, value: number) => acc + value
const scanTransform = reduce({
 objectMode: true //provide standard TransformOptions
})(
 addAll
)
inStream
 .pipe(scanTransform)
 .pipe(outStream) // [1..3..6..10]

side

(options: TransformOptions) => (sideEffect: (value: T) => void) => Transform

import { side } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
const sideEffect = side({
 objectMode: true // provide standard TransformOptions
})(
 console.log
)
inStream
 .pipe(sideEffect)
 .pipe(outStream)

skip

(options: TransformOptions) => (numSkip: number) => Transform

import { skip } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
const skipTransform = skip({
 objectMode: true // provide standard TransformOptions
})(
 2 // skip 2 chunks
)
inStream
 .pipe(skipTransform)
 .pipe(outStream) // [....3..4]

startWith

(options: ReadableOptions) => (...values: T[]) => (readable: ReadableStream) => Readable

import { startWith } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
const prependedReadable = startWith({
 objectMode: true // provide standard ReadableOptions
})(
 'a', 'b', 'c'
)(
 inStream // provide the stream to prepend
)
prependedReadable.pipe(outStream) // ['a'.'b'.'c'.1..2..3..4]

take

(options: TransformOptions) => (numTake: number) => Transform

import { take } from 'node-streams'
// we have the following streams in "objectMode"
declare var inStream: ReadableStream // [1..2..3..4]
declare var outStream: WritableStream
const takeTransform = take({
 objectMode: true // provide standard TransformOptions
})(
 2 // take 2 chunks
)
inStream
 .pipe(takeTransform)
 .pipe(outStream) // [1..2]

throttle

(options: TransformOptions) => (wait: WaitFn) => Transform

import { throttle } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
// we have `setTimeout` function with baked in 10ms time
// CAUTION: wait function should be able to cancel timeout
const wait10ms = (callback: () => void) => {
 const id = setTimeout(callback, 10)
 return () => clearTimeout(id)
}
const throttled = throttle({
 objectMode: true // set up standard TranformOptions
})(
 wait10ms // debounce stream for 10ms
)
inStream
 .pipe(throttled)
 .pipe(outStream)

throttleTime

(options: TransformOptions) => (ms: number) => Transform

import { throttleTime } from 'node-streams'
// we have the streams in `objectMode`
declare var inStream: ReadableStream
declare var outStream: WritableStream
const throttled = debounceTime({
 objectMode: true // set up standard TranformOptions
})(
 10 // debounce stream for 10ms
)
inStream
 .pipe(throttled)
 .pipe(outStream)

withLatest

(options: ReadableOptions) => (...streams: ReadableStream[]) => (mainStream: ReadableStream) => Readable

import { withLatest } from 'node-streams'
// we have the streams in `objectMode`
declare var mainStream: ReadableStream // [1..2..3]
declare var stream0: ReadableStream // ['a'..'b']
declare var stream1: ReadableStream // [true............false]
const combined = withLatest({
 objectMode: true // standard ReadableOptions
})(
 stream0,
 stream1 // streams to take the latest values
)(
 mainStream // mainStream to sync with
)
combined
 .on('data', () => {}) // [[1, 'a', true]..[2, 'a', true]..[3, 'b', true]]
 .on('end', () => {})

zip

(options: ReadableOptions) => (...streams: ReadableStream[]) => Readable

import { zip } from 'node-streams'
// we have the streams in `objectMode`
declare var stream0: ReadableStream // ['a'..'b']
declare var stream1: ReadableStream // [true............false]
const combined = zip({
 objectMode: true // standard ReadableOptions
})(
 stream0,
 stream1 // streams to combine
)
combined
 .on('data', () => {}) // [['a', true]..........['b', false]]
 .on('end', () => {})

subscribe

({ next, error?, complete? }: IObserver) => (...streams: ReadableStream[]) => UnsubscribeFn

type IObserver = {
 next: (value: T) => void,
 error?: (e: Error) => void,
 complete?: () => void
}
type UnsubscribeFn = () => void
import { subscribe } from 'node-streams'
// we have the following streams
declare var stream0: ReadableStream // [1..2..3]
declare var stream1: ReadableStream // [..'a'..'b'..]
const unsub = subscribe({
 next: console.log // [1..'a'..2..3..'b'..]
})(
 stream0,
 stream1
)

subscribeEx

({ next, error?, complete? }: IObserverEx) => (...streams: ReadableStream[]) => UnsubscribeFn

type EmitterValue = {
 value: T,
 index: number,
 emitter: EventEmitter,
 emitterIndex: number,
 event: string
}
type IObserverEx = {
 next: (value: EmitterValue) => void,
 error?: (e: Error) => void,
 complete: () => void
}
type UnsubscribeFn = () => void
import { subscribeEx } from 'node-streams'
// we have the following streams
declare var stream0: ReadableStream // [1..2..3]
declare var stream1: ReadableStream // [..'a'..'b'..]
const unsub = subscribeEx({
 next: ({value, index, emitter, emitterIndex, event}) => console.log(`value ${value} from stream ${emitterIndex}`)
})(
 stream0,
 stream1
)

subscribeReadable

({ next, error?, complete? }: IObserver) => (...streams: ReadableStream[]) => UnsubscribeFn

import { subscribeReadable } from 'node-streams'
// we have the following streams
declare var stream0: ReadableStream // [1..2..3]
declare var stream1: ReadableStream // [..'a'..'b'..]
const unsub = subscribeReadable({
 next: console.log // [1..'a'..2..3..'b'..]
})(
 stream0,
 stream1
)

subscribeReadableEx

({ next, error?, complete? }: IObserverEx) => (...streams: ReadableStream[]) => UnsubscribeFn

import { subscribeReadableEx } from 'node-streams'
// we have the following streams
declare var stream0: ReadableStream // [1..2..3]
declare var stream1: ReadableStream // [..'a'..'b'..]
const unsub = subscribeReadableEx({
 next: ({value, index, emitter, emitterIndex, event}) => console.log(`value ${value} from stream ${emitterIndex}`)
})(
 stream0,
 stream1
)

About

No description, website, or topics provided.

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Contributors 2

AltStyle によって変換されたページ (->オリジナル) /