pipe"#

Stream values can be transformed and controlled by piping through operators. urx includes several operators like map, filter, scan, and throttleTime. The withLatestFrom operator allows the combination of values from other streams.

const foo = stream<number>()

// create an emitter that first adds 2 to the passed value, then multiplies it by * 2
const bar = pipe(foo, map(value => value + 2), map(value => value * 2))
subscribe(bar, value => console.log(value))
publish(foo, 2) // outputs 8

Implementing Custom Operators#

To implement your own operators, implement the Operator interface.

Index#

Interfaces#

Functions#

debounceTime#

▸ debounceTime<T>(interval: number): Operator<T>

Defined in urx/src/pipe.ts:289

Debounces flowing values at the provided interval in milliseconds. Throttle VS Debounce in SO.

 const foo = stream<number>()
 publish(foo, 1)

 setTimeout(() => publish(foo, 2), 20)
 setTimeout(() => publish(foo, 3), 20)

 subscribe(pipe(foo, debounceTime(50)), val => {
   console.log(value); // 3
 })

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`interval`	number

Returns: Operator<T>

defaultComparator#

▸ defaultComparator<T>(previous: T, next: T): boolean

Defined in urx/src/pipe.ts:112

The default Comparator for distinctUntilChanged and duc.

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`previous`	T
`next`	T

Returns: boolean

distinctUntilChanged#

▸ distinctUntilChanged<T>(comparator?: Comparator<T>): Operator<T>

Defined in urx/src/pipe.ts:130

Filters out identical values. Pass an optional Comparator if you need to filter non-primitive values.

const foo = stream<number>()

subscribe(
 pipe(foo, distinctUntilChanged()),
 console.log
) // will be called only once

publish(foo, 42)
publish(foo, 42)

Type parameters:#

Name
`T`

Parameters:#

Name	Type	Default value
`comparator`	Comparator<T>	defaultComparator

Returns: Operator<T>

filter#

▸ filter<T>(predicate: (value: T) => boolean): Operator<T>

Defined in urx/src/pipe.ts:156

Filters out values for which the predicator does not return true-ish.

const foo = stream<number>()

subscribe(
 pipe(foo, filter(value => value % 2 === 0)),
 console.log
) // will be called only with even values

publish(foo, 2)
publish(foo, 3)
publish(foo, 4)
publish(foo, 5)

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`predicate`	(value: T) => boolean

Returns: Operator<T>

map#

▸ map<T, K>(project: (value: T) => K): Operator<T, K>

Defined in urx/src/pipe.ts:176

Maps values using the provided project function.

const foo = stream<number>()

subscribe(
 pipe(foo, map(value => value * 2)),
 console.log
) // 4, 6

publish(foo, 2)
publish(foo, 3)

Type parameters:#

Name
`T`
`K`

Parameters:#

Name	Type
`project`	(value: T) => K

Returns: Operator<T, K>

mapTo#

▸ mapTo<T>(value: T): Operator<any, T>

Defined in urx/src/pipe.ts:194

Maps values to the hard-coded value.

const foo = stream<number>()

subscribe(
 pipe(foo, mapTo(3)),
 console.log
) // 3, 3

publish(foo, 1)
publish(foo, 2)

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`value`	T

Returns: Operator<any, T>

pipe#

▸ pipe<T>(s: Emitter<T>): Emitter<T>

Defined in urx/src/pipe.ts:76

Creates a new emitter from the passed one by piping its values through one or more operators. Operators can perform various actions like filter values, pull values from other emitters, or compute new values.

const foo = stream<number>()

// create an emitter that first adds 2 to the passed value, then multiplies it by * 2
const bar = pipe(foo, map(value => value + 2), map(value => value * 2))
subscribe(bar, value => console.log(value))
publish(foo, 2) // outputs 8

Sharing Subscription Calculations#

pipe acts as a proxy for the source emitter, and re-runs the operators for each subscription to the derived emitter. Use streamFromEmitter or statefulStreamFromEmitter to avoid that.

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`s`	Emitter<T>

Returns: Emitter<T>

▸ pipe<T, O1>(s: Emitter<T>, o1: O<T, O1>): Emitter<O1>

Defined in urx/src/pipe.ts:77

Type parameters:#

Name
`T`
`O1`

Parameters:#

Name	Type
`s`	Emitter<T>
`o1`	O<T, O1>

Returns: Emitter<O1>

▸ pipe<T, O1, O2>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>]): Emitter<O2>

Defined in urx/src/pipe.ts:78

Type parameters:#

Name
`T`
`O1`
`O2`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>]

Returns: Emitter<O2>

▸ pipe<T, O1, O2, O3>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>, O<O2, O3>]): Emitter<O3>

Defined in urx/src/pipe.ts:79

Type parameters:#

Name
`T`
`O1`
`O2`
`O3`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>, O<O2, O3>]

Returns: Emitter<O3>

▸ pipe<T, O1, O2, O3, O4>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>]): Emitter<O4>

Defined in urx/src/pipe.ts:80

Type parameters:#

Name
`T`
`O1`
`O2`
`O3`
`O4`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>]

Returns: Emitter<O4>

▸ pipe<T, O1, O2, O3, O4, O5>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>]): Emitter<O5>

Defined in urx/src/pipe.ts:81

Type parameters:#

Name
`T`
`O1`
`O2`
`O3`
`O4`
`O5`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>]

Returns: Emitter<O5>

▸ pipe<T, O1, O2, O3, O4, O5, O6>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>, O<O5, O6>]): Emitter<O6>

Defined in urx/src/pipe.ts:82

Type parameters:#

Name
`T`
`O1`
`O2`
`O3`
`O4`
`O5`
`O6`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>, O<O5, O6>]

Returns: Emitter<O6>

▸ pipe<T, O1, O2, O3, O4, O5, O6, O7>(s: Emitter<T>, ...o: [O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>, O<O5, O6>, O<O6, O7>]): Emitter<O7>

Defined in urx/src/pipe.ts:83

Type parameters:#

Name
`T`
`O1`
`O2`
`O3`
`O4`
`O5`
`O6`
`O7`

Parameters:#

Name	Type
`s`	Emitter<T>
`...o`	[O<T, O1>, O<O1, O2>, O<O2, O3>, O<O3, O4>, O<O4, O5>, O<O5, O6>, O<O6, O7>]

Returns: Emitter<O7>

scan#

▸ scan<T, K>(scanner: (current: T, value: K) => T, initial: T): Operator<K, T>

Defined in urx/src/pipe.ts:213

Works like Array#reduce. Applies an accumulator function on the emitter, and outputs intermediate result. Starts with the initial value.

const foo = stream<number>()

subscribe(
 pipe(foo, scan((acc, value) => acc + value, 2),
 console.log
) // 3, 5

publish(foo, 1)
publish(foo, 2)

Type parameters:#

Name
`T`
`K`

Parameters:#

Name	Type
`scanner`	(current: T, value: K) => T
`initial`	T

Returns: Operator<K, T>

skip#

▸ skip<T>(times: number): Operator<T>

Defined in urx/src/pipe.ts:233

Skips the specified amount of values from the emitter.

const foo = stream<number>()

subscribe(
 pipe(foo, skip(2)),
 console.log
) // 3, 4

publish(foo, 1) // skipped
publish(foo, 2) // skipped
publish(foo, 3)
publish(foo, 4)

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`times`	number

Returns: Operator<T>

throttleTime#

▸ throttleTime<T>(interval: number): Operator<T>

Defined in urx/src/pipe.ts:255

Throttles flowing values at the provided interval in milliseconds. Throttle VS Debounce in SO.

 const foo = stream<number>()
 publish(foo, 1)

 setTimeout(() => publish(foo, 2), 20)
 setTimeout(() => publish(foo, 3), 20)

 subscribe(pipe(foo, throttleTime(50)), val => {
   console.log(value); // 3
 })

Type parameters:#

Name
`T`

Parameters:#

Name	Type
`interval`	number

Returns: Operator<T>

withLatestFrom#

▸ withLatestFrom<T, R1>(...s: [Emitter<R1>]): Operator<T, [T, R1]>

Defined in urx/src/pipe.ts:327

Combines the source Emitter with the latest values from the specified Emitters into an array. Outputs only when the source Emitter emits. See combineLatest for a transformer that outputs when any of the emitters emit.

 const foo = stream<number>()
 const bar = stream<number>()
 subscribe(
   pipe(
     foo,
     withLatestFrom(bar)
   ),
   (([foo, bar]) => console.log({ foo, bar }))
 )

 publish(foo, 1) // nothing happens, bar has not emitted yet
 publish(bar, 1) // still nothing
 publish(foo, 2) // logs { foo: 2, bar: 1 }
 publish(bar, 2)
 publish(foo, 3) // logs { foo: 3, bar: 2 }

Type parameters:#

Name
`T`
`R1`

Parameters:#

Name	Type
`...s`	[Emitter<R1>]

Returns: Operator<T, [T, R1]>

▸ withLatestFrom<T, R1, R2>(...s: [Emitter<R1>, Emitter<R2>]): Operator<T, [T, R1, R2]>

Defined in urx/src/pipe.ts:328

Type parameters:#

Name
`T`
`R1`
`R2`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>]

Returns: Operator<T, [T, R1, R2]>

▸ withLatestFrom<T, R1, R2, R3>(...s: [Emitter<R1>, Emitter<R2>, Emitter<R3>]): Operator<T, [T, R1, R2, R3]>

Defined in urx/src/pipe.ts:329

Type parameters:#

Name
`T`
`R1`
`R2`
`R3`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>, Emitter<R3>]

Returns: Operator<T, [T, R1, R2, R3]>

▸ withLatestFrom<T, R1, R2, R3, R4>(...s: [Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>]): Operator<T, [T, R1, R2, R3, R4]>

Defined in urx/src/pipe.ts:330

Type parameters:#

Name
`T`
`R1`
`R2`
`R3`
`R4`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>]

Returns: Operator<T, [T, R1, R2, R3, R4]>

▸ withLatestFrom<T, R1, R2, R3, R4, R5>(...s: [Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>]): Operator<T, [T, R1, R2, R3, R4, R5]>

Defined in urx/src/pipe.ts:331

Type parameters:#

Name
`T`
`R1`
`R2`
`R3`
`R4`
`R5`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>]

Returns: Operator<T, [T, R1, R2, R3, R4, R5]>

▸ withLatestFrom<T, R1, R2, R3, R4, R5, R6>(...s: [Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>, Emitter<R6>]): Operator<T, [T, R1, R2, R3, R4, R5, R6]>

Defined in urx/src/pipe.ts:332

Type parameters:#

Name
`T`
`R1`
`R2`
`R3`
`R4`
`R5`
`R6`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>, Emitter<R6>]

Returns: Operator<T, [T, R1, R2, R3, R4, R5, R6]>

▸ withLatestFrom<T, R1, R2, R3, R4, R5, R6, R7>(...s: [Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>, Emitter<R6>, Emitter<R7>]): Operator<T, [T, R1, R2, R3, R4, R5, R6, R7]>

Defined in urx/src/pipe.ts:333

Type parameters:#

Name
`T`
`R1`
`R2`
`R3`
`R4`
`R5`
`R6`
`R7`

Parameters:#

Name	Type
`...s`	[Emitter<R1>, Emitter<R2>, Emitter<R3>, Emitter<R4>, Emitter<R5>, Emitter<R6>, Emitter<R7>]

Returns: Operator<T, [T, R1, R2, R3, R4, R5, R6, R7]>