feat: Add TSDocs to all internal and external utilities and types (#136)

+5

.changeset/violet-suits-yawn.md

··· 1 + --- 2 + 'wonka': minor 3 + --- 4 + 5 + Add extensive TSDoc documentation for all `wonka` internals and exports. This will replace the documentation and give consumers more guidance on each of the library's extensive utilities.

+3

.editorconfig

··· 7 7 indent_style = space 8 8 insert_final_newline = true 9 9 trim_trailing_whitespace = true 10 + 11 + [*.ts] 12 + max_line_length = 100

+1

package.json

··· 87 87 "eslint": "^8.29.0", 88 88 "eslint-config-prettier": "^8.5.0", 89 89 "eslint-plugin-prettier": "^4.2.1", 90 + "eslint-plugin-tsdoc": "^0.2.17", 90 91 "flowgen": "^1.21.0", 91 92 "glob": "^8.0.3", 92 93 "husky-v4": "^4.3.8",

+33

pnpm-lock.yaml

··· 18 18 eslint: ^8.29.0 19 19 eslint-config-prettier: ^8.5.0 20 20 eslint-plugin-prettier: ^4.2.1 21 + eslint-plugin-tsdoc: ^0.2.17 21 22 flowgen: ^1.21.0 22 23 glob: ^8.0.3 23 24 husky-v4: ^4.3.8 ··· 51 52 eslint: 8.29.0 52 53 eslint-config-prettier: 8.5.0_eslint@8.29.0 53 54 eslint-plugin-prettier: 4.2.1_nrhoyyjffvfyk4vtlt5destxgm 55 + eslint-plugin-tsdoc: 0.2.17 54 56 flowgen: 1.21.0 55 57 glob: 8.0.3 56 58 husky-v4: 4.3.8 ··· 398 400 fs-extra: 8.1.0 399 401 globby: 11.1.0 400 402 read-yaml-file: 1.1.0 403 + dev: true 404 + 405 + /@microsoft/tsdoc-config/0.16.2: 406 + resolution: {integrity: sha512-OGiIzzoBLgWWR0UdRJX98oYO+XKGf7tiK4Zk6tQ/E4IJqGCe7dvkTvgDZV5cFJUzLGDOjeAXrnZoA6QkVySuxw==} 407 + dependencies: 408 + '@microsoft/tsdoc': 0.14.2 409 + ajv: 6.12.6 410 + jju: 1.4.0 411 + resolve: 1.19.0 412 + dev: true 413 + 414 + /@microsoft/tsdoc/0.14.2: 415 + resolution: {integrity: sha512-9b8mPpKrfeGRuhFH5iO1iwCLeIIsV6+H1sRfxbkoGXIyQE2BTsPd9zqSqQJ+pv5sJ/hT5M1zvOFL02MnEezFug==} 401 416 dev: true 402 417 403 418 /@nodelib/fs.scandir/2.1.5: ··· 1571 1586 prettier-linter-helpers: 1.0.0 1572 1587 dev: true 1573 1588 1589 + /eslint-plugin-tsdoc/0.2.17: 1590 + resolution: {integrity: sha512-xRmVi7Zx44lOBuYqG8vzTXuL6IdGOeF9nHX17bjJ8+VE6fsxpdGem0/SBTmAwgYMKYB1WBkqRJVQ+n8GK041pA==} 1591 + dependencies: 1592 + '@microsoft/tsdoc': 0.14.2 1593 + '@microsoft/tsdoc-config': 0.16.2 1594 + dev: true 1595 + 1574 1596 /eslint-scope/5.1.1: 1575 1597 resolution: {integrity: sha512-2NxwbF/hZ0KpepYN0cNbo+FN6XoK7GaHlQhgx/hIZl6Va0bF45RQOOwhLIy8lQDbuCiadSLCBnH2CFYquit5bw==} 1576 1598 engines: {node: '>=8.0.0'} ··· 2267 2289 2268 2290 /isexe/2.0.0: 2269 2291 resolution: {integrity: sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==} 2292 + dev: true 2293 + 2294 + /jju/1.4.0: 2295 + resolution: {integrity: sha512-8wb9Yw966OSxApiCt0K3yNJL8pnNeIv+OEq2YMidz4FKP6nonSRoOXc80iXY4JaN2FC11B9qsNmDsm+ZOfMROA==} 2270 2296 dev: true 2271 2297 2272 2298 /js-sdsl/4.2.0: ··· 3070 3096 /resolve-from/5.0.0: 3071 3097 resolution: {integrity: sha512-qYg9KP24dD5qka9J47d0aVky0N+b4fTU89LN9iDnjB5waksiC49rvMB0PrUJQGoTmH50XPiqOvAjDfaijGxYZw==} 3072 3098 engines: {node: '>=8'} 3099 + dev: true 3100 + 3101 + /resolve/1.19.0: 3102 + resolution: {integrity: sha512-rArEXAgsBG4UgRGcynxWIWKFvh/XZCcS8UJdHhwy91zwAvCZIbcs+vAbflgBnNjYMs/i/i+/Ux6IZhML1yPvxg==} 3103 + dependencies: 3104 + is-core-module: 2.11.0 3105 + path-parse: 1.0.7 3073 3106 dev: true 3074 3107 3075 3108 /resolve/1.22.1:

+2 -1

scripts/eslint-preset.js

··· 7 7 }, 8 8 }, 9 9 extends: ['prettier'], 10 - plugins: ['prettier'], 10 + plugins: ['prettier', 'tsdoc'], 11 11 ignorePatterns: ['node_modules/', 'dist/', 'coverage/', 'perf/'], 12 12 rules: { 13 13 'sort-keys': 'off', ··· 47 47 '@typescript-eslint/no-empty-function': 'off', 48 48 '@typescript-eslint/no-unused-vars': 'off', 49 49 'prefer-rest-params': 'off', 50 + 'tsdoc/syntax': 'error', 50 51 }, 51 52 }, 52 53 ],

+3 -1

scripts/flow-typings-plugin.mjs

··· 8 8 // NOTE: Computed property names will be omitted 9 9 code = code.replace(/\[Symbol\.\w+\][?()]*:(?:.*);\n?/g, ''); 10 10 11 - let flowdef = compiler.compileDefinitionString(code); 11 + let flowdef = compiler.compileDefinitionString(code, { 12 + jsdoc: false, 13 + }); 12 14 13 15 flowdef = beautify(flowdef); 14 16 flowdef = flowdef.replace(/import/g, 'import type');

+20

src/callbag.ts

··· 1 1 import { Source, SignalKind } from './types'; 2 2 import { push, start } from './helpers'; 3 3 4 + /** A definition of the Callbag type as per its specification. 5 + * @see {@link https://github.com/callbag/callbag} for the Callbag specification. 6 + */ 4 7 interface Callbag<I, O> { 5 8 (t: 0, d: Callbag<O, I>): void; 6 9 (t: 1, d: I): void; 7 10 (t: 2, d?: any): void; 8 11 } 9 12 13 + /** Converts a Callbag to a {@link Source}. 14 + * @param callbag - The {@link Callbag} object that will be converted. 15 + * @returns A {@link Source} wrapping the passed Callbag. 16 + * 17 + * @remarks 18 + * This converts a Callbag to a {@link Source}. When this Source receives a {@link Sink} and 19 + * the subscription starts, internally, it'll subscribe to the passed Callbag, passing through 20 + * all of its emitted values. 21 + */ 10 22 export function fromCallbag<T>(callbag: Callbag<any, T>): Source<T> { 11 23 return sink => { 12 24 callbag(0, (signal: number, data: any) => { ··· 25 37 }; 26 38 } 27 39 40 + /** Converts a {@link Source} to a Callbag. 41 + * @param source - The {@link Source} that will be converted. 42 + * @returns A {@link Callbag} wrapping the passed Source. 43 + * 44 + * @remarks 45 + * This converts a {@link Source} to a {@link Callbag}. When this Callbag is subscribed to, it 46 + * internally subscribes to the Wonka Source and pulls new values. 47 + */ 28 48 export function toCallbag<T>(source: Source<T>): Callbag<any, T> { 29 49 return (signal: number, sink: any) => { 30 50 if (signal === 0) {

+64 -9

src/combine.ts

··· 5 5 ? [TypeOfSource<Head>, ...TypeOfSourceArray<Tail>] 6 6 : []; 7 7 8 - export function zip<Sources extends readonly [...Source<any>[]]>( 9 - sources: [...Sources] 10 - ): Source<TypeOfSourceArray<Sources>>; 8 + /** Combines the latest values of several sources into a Source issuing either tuple or dictionary 9 + * values. 10 + * 11 + * @param sources - Either an array or dictionary object of Sources. 12 + * @returns A {@link Source} issuing a zipped value whenever any input Source updates. 13 + * 14 + * @remarks 15 + * `zip` combines several {@link Source | Sources}. The resulting Source will issue its first value 16 + * once all input Sources have at least issued one value, and will subsequently issue a new value 17 + * each time any of the Sources emits a new value. 18 + * 19 + * Depending on whether an array or dictionary object of Sources is passed to `zip`, its emitted 20 + * values will be arrays or dictionary objects of the Sources' values. 21 + * 22 + * @example 23 + * An example of passing a dictionary object to `zip`. If an array is passed, the resulting 24 + * values will output arrays of the sources' values instead. 25 + * 26 + * ```ts 27 + * pipe( 28 + * zip({ 29 + * x: fromValue(1), 30 + * y: fromArray([2, 3]), 31 + * }), 32 + * subscribe(result => { 33 + * // logs { x: 1, y: 2 } then { x: 1, y: 3 } 34 + * console.log(result); 35 + * }) 36 + * ); 37 + * ``` 38 + */ 39 + interface zip { 40 + <Sources extends readonly [...Source<any>[]]>(sources: [...Sources]): Source< 41 + TypeOfSourceArray<Sources> 42 + >; 11 43 12 - export function zip<Sources extends { [prop: string]: Source<any> }>( 13 - sources: Sources 14 - ): Source<{ [Property in keyof Sources]: TypeOfSource<Sources[Property]> }>; 44 + <Sources extends { [prop: string]: Source<any> }>(sources: Sources): Source<{ 45 + [Property in keyof Sources]: TypeOfSource<Sources[Property]>; 46 + }>; 47 + } 15 48 16 - export function zip<T>( 17 - sources: Source<T>[] | Record<string, Source<T>> 18 - ): Source<T[] | Record<string, T>> { 49 + function zip<T>(sources: Source<T>[] | Record<string, Source<T>>): Source<T[] | Record<string, T>> { 19 50 const size = Object.keys(sources).length; 20 51 return sink => { 21 52 const filled: Set<string | number> = new Set(); ··· 75 106 }; 76 107 } 77 108 109 + export { zip }; 110 + 111 + /** Combines the latest values of all passed sources into a Source issuing tuple values. 112 + * 113 + * @see {@link zip | `zip`} which this helper wraps and uses. 114 + * @param sources - A variadic list of {@link Source} parameters. 115 + * @returns A {@link Source} issuing a zipped value whenever any input Source updates. 116 + * 117 + * @remarks 118 + * `combine` takes one or more {@link Source | Sources} as arguments. Once all input Sources have at 119 + * least issued one value it will issue an array of all of the Sources' values. Subsequently, it 120 + * will issue a new array value whenever any of the Sources update. 121 + * 122 + * @example 123 + * 124 + * ```ts 125 + * pipe( 126 + * combine(fromValue(1), fromValue(2)), 127 + * subscribe(result => { 128 + * console.log(result); // logs [1, 2] 129 + * }) 130 + * ); 131 + * ``` 132 + */ 78 133 export function combine<Sources extends Source<any>[]>( 79 134 ...sources: Sources 80 135 ): Source<TypeOfSourceArray<Sources>> {

+21

src/helpers.ts

··· 1 1 import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types'; 2 2 3 + /** Placeholder {@link TeardownFn | teardown functions} that's a no-op. 4 + * @see {@link TeardownFn} for the definition and usage of teardowns. 5 + * @internal 6 + */ 3 7 export const teardownPlaceholder: TeardownFn = () => { 4 8 /*noop*/ 5 9 }; 10 + 11 + /** Placeholder {@link TalkbackFn | talkback function} that's a no-op. 12 + * @privateRemarks 13 + * This is frequently used in the codebase as a no-op initializer value for talkback functions in 14 + * the implementation of {@link Operator | Operators}. This is cheaper than initializing the 15 + * variables of talkbacks to `undefined` or `null` and performing an extra check before calling 16 + * them. Since the {@link Start | Start signal} is assumed to come first and carry a talkback, we can 17 + * use this to our advantage and use a no-op placeholder before {@link Start} is received. 18 + * 19 + * @internal 20 + */ 6 21 export const talkbackPlaceholder: TalkbackFn = teardownPlaceholder; 7 22 23 + /** Wraps the passed {@link TalkbackFn | talkback function} in a {@link Start | Start signal}. 24 + * @internal 25 + */ 8 26 export function start<T>(talkback: TalkbackFn): Start<T> { 9 27 const box: any = [talkback]; 10 28 box.tag = SignalKind.Start; 11 29 return box; 12 30 } 13 31 32 + /** Wraps the passed value in a {@link Push | Push signal}. 33 + * @internal 34 + */ 14 35 export function push<T>(value: T): Push<T> { 15 36 const box: any = [value]; 16 37 box.tag = SignalKind.Push;

+12

src/index.ts

··· 1 + /** 2 + * A tiny but capable push & pull stream library for TypeScript and Flow. 3 + * 4 + * @remarks 5 + * Wonka is a lightweight iterable and observable library and exposes a set of helpers to create 6 + * streams, which are sources emitting multiple values, which allow you to create, transform, and 7 + * consume event streams or iterable sets of data. 8 + * 9 + * It's loosely based on the Callbag spec: {@link https://github.com/callbag/callbag} 10 + * @packageDocumentation 11 + */ 12 + 1 13 export * from './types'; 2 14 export * from './sources'; 3 15 export * from './operators';

+121 -2

src/observable.ts

··· 1 1 import { Source, SignalKind, TalkbackKind } from './types'; 2 2 import { push, start, talkbackPlaceholder } from './helpers'; 3 3 4 + /** A definition of the ES Observable Subscription type that is returned by 5 + * {@link Observable.subscribe} 6 + * 7 + * @remarks 8 + * The Subscription in ES Observables is a handle that is held while the Observable is actively 9 + * streaming values. As such, it's used to indicate with {@link ObservableSubscription.closed} 10 + * whether it's active, and {@link ObservableSubscription.unsubscribe} may be used to cancel the 11 + * ongoing subscription and end the {@link Observable} early. 12 + * 13 + * @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification. 14 + */ 4 15 interface ObservableSubscription { 16 + /** A boolean flag indicating whether the subscription is closed. 17 + * @remarks 18 + * When `true`, the subscription will not issue new values to the {@link ObservableObserver} and 19 + * has terminated. No new values are expected. 20 + * 21 + * @readonly 22 + */ 5 23 closed?: boolean; 24 + /** Cancels the subscription. 25 + * @remarks 26 + * This cancels the ongoing subscription and the {@link ObservableObserver}'s callbacks will 27 + * subsequently not be called at all. The subscription will be terminated and become inactive. 28 + */ 6 29 unsubscribe(): void; 7 30 } 8 31 32 + /** A definition of the ES Observable Observer type that is used to receive data from an 33 + * {@link Observable}. 34 + * 35 + * @remarks 36 + * The Observer in ES Observables is supplied to {@link Observable.subscribe} to receive events from 37 + * an {@link Observable} as it issues them. 38 + * 39 + * @see {@link https://github.com/tc39/proposal-observable#observer} for the ES Observable 40 + * specification of an Observer. 41 + */ 9 42 interface ObservableObserver<T> { 43 + /** Callback for the Observable issuing new values. 44 + * @param value - The value that the {@link Observable} is sending. 45 + */ 10 46 next(value: T): void; 47 + /** Callback for the Observable encountering an error, terminating it. 48 + * @param error - The error that the {@link Observable} has encountered. 49 + */ 11 50 error?(error: any): void; 51 + /** Callback for the Observable ending, after all values have been issued. */ 12 52 complete?(): void; 13 53 } 14 54 55 + /** A looser definition of ES Observable-like types that is used for interoperability. 56 + * @remarks 57 + * The Observable is often used by multiple libraries supporting or creating streams to provide 58 + * interoperability for push-based streams. When converting from an Observable to a {@link Source}, 59 + * this looser type is accepted as an input. 60 + * 61 + * @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification. 62 + * @see {@link Observable} for the full ES Observable type. 63 + */ 15 64 interface ObservableLike<T> { 65 + /** 66 + * Subscribes to new signals from an {@link Observable} via callbacks. 67 + * @param observer - An object containing callbacks for the various events of an Observable. 68 + * @returns Subscription handle of type {@link ObservableSubscription}. 69 + * 70 + * @see {@link ObservableObserver} for the callbacks in an object that are called as Observables 71 + * issue events. 72 + */ 16 73 subscribe(observer: ObservableObserver<T>): ObservableSubscription; 74 + 75 + /** The well-known symbol specifying the default ES Observable for an object. */ 17 76 [Symbol.observable]?(): Observable<T>; 18 77 } 19 78 79 + /** An ES Observable type that is a de-facto standard for push-based data sources across the JS 80 + * ecosystem. 81 + * 82 + * @remarks 83 + * The Observable is often used by multiple libraries supporting or creating streams to provide 84 + * interoperability for push-based streams. As Wonka's {@link Source | Sources} are similar in 85 + * functionality to Observables, it provides utilities to cleanly convert to and from Observables. 86 + * 87 + * @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification. 88 + */ 20 89 interface Observable<T> { 90 + /** Subscribes to new signals from an {@link Observable} via callbacks. 91 + * @param observer - An object containing callbacks for the various events of an Observable. 92 + * @returns Subscription handle of type {@link ObservableSubscription}. 93 + * 94 + * @see {@link ObservableObserver} for the callbacks in an object that are called as Observables 95 + * issue events. 96 + */ 21 97 subscribe(observer: ObservableObserver<T>): ObservableSubscription; 22 98 99 + /** Subscribes to new signals from an {@link Observable} via callbacks. 100 + * @param onNext - Callback for the Observable issuing new values. 101 + * @param onError - Callback for the Observable encountering an error, terminating it. 102 + * @param onComplete - Callback for the Observable ending, after all values have been issued. 103 + * @returns Subscription handle of type {@link ObservableSubscription}. 104 + */ 23 105 subscribe( 24 106 onNext: (value: T) => any, 25 107 onError?: (error: any) => any, 26 108 onComplete?: () => any 27 109 ): ObservableSubscription; 28 110 111 + /** The well-known symbol specifying the default ES Observable for an object. */ 29 112 [Symbol.observable](): Observable<T>; 30 113 } 31 114 115 + /** Returns the well-known symbol specifying the default ES Observable. 116 + * @privateRemarks 117 + * This symbol is used to mark an object as a default ES Observable. By the specification, an object 118 + * that abides by the default Observable implementation must carry a method set to this well-known 119 + * symbol that returns the Observable implementation. It's common for this object to be an 120 + * Observable itself and return itself on this method. 121 + * 122 + * @see {@link https://github.com/0no-co/wonka/issues/122} for notes on the intercompatibility 123 + * between Observable implementations. 124 + * 125 + * @internal 126 + */ 32 127 const observableSymbol = (): typeof Symbol.observable => Symbol.observable || '@@observable'; 33 128 129 + /** Converts an ES Observable to a {@link Source}. 130 + * @param input - The {@link ObservableLike} object that will be converted. 131 + * @returns A {@link Source} wrapping the passed Observable. 132 + * 133 + * @remarks 134 + * This converts an ES Observable to a {@link Source}. When this Source receives a {@link Sink} and 135 + * the subscription starts, internally, it'll subscribe to the passed Observable, passing through 136 + * all of the Observable's values. As such, this utility provides intercompatibility converting from 137 + * standard Observables to Wonka Sources. 138 + * 139 + * @throws 140 + * When the passed ES Observable throws, the error is simply re-thrown as {@link Source} does 141 + * not support or expect errors to be handled by streams. 142 + */ 34 143 export function fromObservable<T>(input: ObservableLike<T>): Source<T> { 35 - input = input[observableSymbol()] ? (input as any)[observableSymbol()]() : input; 36 144 return sink => { 37 - const subscription = input.subscribe({ 145 + const subscription = ( 146 + input[observableSymbol()] ? input[observableSymbol()]!() : input 147 + ).subscribe({ 38 148 next(value: T) { 39 149 sink(push(value)); 40 150 }, ··· 53 163 }; 54 164 } 55 165 166 + /** Converts a {@link Source} to an ES Observable. 167 + * @param source - The {@link Source} that will be converted. 168 + * @returns An {@link Observable} wrapping the passed Source. 169 + * 170 + * @remarks 171 + * This converts a {@link Source} to an {@link Observable}. When this Observable is subscribed to, it 172 + * internally subscribes to the Wonka Source and pulls new values. As such, this utility provides 173 + * intercompatibility converting from Wonka Sources to standard ES Observables. 174 + */ 56 175 export function toObservable<T>(source: Source<T>): Observable<T> { 57 176 return { 58 177 subscribe(

+599

src/operators.ts

··· 4 4 5 5 const identity = <T>(x: T): T => x; 6 6 7 + /** Buffers values and emits the array of bufferd values each time a `notifier` Source emits. 8 + * 9 + * @param notifier - A {@link Source} that releases the current buffer. 10 + * @returns An {@link Operator}. 11 + * 12 + * @remarks 13 + * `buffer` will buffer values from the input {@link Source}. When the passed `notifier` Source 14 + * emits, it will emit an array of all buffered values. 15 + * 16 + * This can be used to group values over time. A buffer will only be emitted when it contains any 17 + * values. 18 + * 19 + * @example 20 + * ```ts 21 + * pipe( 22 + * interval(50), 23 + * buffer(interval(100)), 24 + * subscribe(x => { 25 + * console.log(text); // logs: [0], [1, 2], [3, 4]... 26 + * }) 27 + * ); 28 + * ``` 29 + */ 7 30 export function buffer<S, T>(notifier: Source<S>): Operator<T, T[]> { 8 31 return source => sink => { 9 32 let buffer: T[] = []; ··· 64 87 }; 65 88 } 66 89 90 + /** Emits in order from the Sources returned by a mapping function per value of the Source. 91 + * 92 + * @param map - A function returning a {@link Source} per value. 93 + * @returns An {@link Operator}. 94 + * 95 + * @remarks 96 + * `concatMap` accepts a mapping function which must return a {@link Source} per value. 97 + * The output {@link Source} will emit values from each Source the function returned, in order, 98 + * queuing sources that aren't yet active. 99 + * 100 + * This can be used to issue multiple values per emission of an input {@link Source}, while keeping 101 + * the order of their outputs consistent. 102 + * 103 + * @example 104 + * ```ts 105 + * pipe( 106 + * fromArray([1, 2]), 107 + * concatMap(x => fromArray([x, x * 2])), 108 + * subscribe(x => { 109 + * console.log(text); // logs: 1, 2, 2, 4 110 + * }) 111 + * ); 112 + * ``` 113 + */ 67 114 export function concatMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 68 115 return source => sink => { 69 116 const inputQueue: In[] = []; ··· 144 191 }; 145 192 } 146 193 194 + /** Flattens a Source emitting Sources into a single Source emitting the inner values in order. 195 + * 196 + * @see {@link concatMap} which this helper uses and instead accept a mapping function. 197 + * @param source - An {@link Source} emitting {@link Source | Sources}. 198 + * @returns A {@link Source} emitting values from the inner Sources. 199 + * 200 + * @remarks 201 + * `concatAll` accepts a {@link Source} emitting {@link Source | Sources}. 202 + * The output {@link Source} will emit values from each Source, in order, queuing sources that 203 + * aren't yet active. 204 + * 205 + * @example 206 + * ```ts 207 + * pipe( 208 + * fromArray([ 209 + * fromArray([1, 2]), 210 + * fromArray([3, 4]), 211 + * ]), 212 + * concatAll, 213 + * subscribe(x => { 214 + * console.log(text); // logs: 1, 2, 3, 4 215 + * }) 216 + * ); 217 + * ``` 218 + */ 147 219 export function concatAll<T>(source: Source<Source<T>>): Source<T> { 148 220 return concatMap<Source<T>, T>(identity)(source); 149 221 } 150 222 223 + /** Emits values from the passed sources in order. 224 + * 225 + * @param sources - An array of {@link Source | Sources}. 226 + * @returns A {@link Source} emitting values from the input Sources. 227 + * 228 + * @remarks 229 + * `concat` accepts an array of {@link Source | Sources} and will emit values from them, starting 230 + * with the first one and continuing to the next only when the prior source ended. 231 + * 232 + * This can be used to issue combine sources while keeping the order of their values intact. 233 + * 234 + * @example 235 + * ```ts 236 + * pipe( 237 + * concat([ 238 + * fromArray([1, 2]), 239 + * fromArray([3, 4]), 240 + * ]), 241 + * subscribe(x => { 242 + * console.log(text); // logs: 1, 2, 3, 4 243 + * }) 244 + * ); 245 + * ``` 246 + */ 151 247 export function concat<T>(sources: Source<T>[]): Source<T> { 152 248 return concatAll(fromArray(sources)); 153 249 } 154 250 251 + /** Filters out emitted values for which the passed predicate function returns `false`. 252 + * 253 + * @param predicate - A function returning a boolean per value. 254 + * @returns An {@link Operator}. 255 + * 256 + * @remarks 257 + * `filter` will omit values from the {@link Source} for which the passed `predicate` function 258 + * returns `false`. 259 + * 260 + * @example 261 + * ```ts 262 + * pipe( 263 + * fromArray([1, 2, 3]), 264 + * filter(x => x % 2 === 0), 265 + * subscribe(x => { 266 + * console.log(text); // logs: 2 267 + * }) 268 + * ); 269 + * ``` 270 + */ 155 271 export function filter<T>(predicate: (value: T) => boolean): Operator<T, T> { 156 272 return source => sink => { 157 273 let talkback = talkbackPlaceholder; ··· 170 286 }; 171 287 } 172 288 289 + /** Maps emitted values using the passed mapping function. 290 + * 291 + * @param map - A function returning transforming the {@link Source | Source's} values. 292 + * @returns An {@link Operator}. 293 + * 294 + * @remarks 295 + * `map` accepts a transform function and calls it on each emitted value. It then emits 296 + * the values returned by the transform function instead. 297 + * 298 + * @example 299 + * ```ts 300 + * pipe( 301 + * fromArray([1, 2, 3]), 302 + * map(x => x * 2), 303 + * subscribe(x => { 304 + * console.log(text); // logs: 2, 4, 6 305 + * }) 306 + * ); 307 + * ``` 308 + */ 173 309 export function map<In, Out>(map: (value: In) => Out): Operator<In, Out> { 174 310 return source => sink => 175 311 source(signal => { ··· 181 317 }); 182 318 } 183 319 320 + /** Emits from the Sources returned by a mapping function per value of the Source. 321 + * 322 + * @param map - A function returning a {@link Source} per value. 323 + * @returns An {@link Operator}. 324 + * 325 + * @remarks 326 + * `mergeMap` accepts a mapping function which must return a {@link Source} per value. 327 + * The output {@link Source} will emit values from all {@link Source | Sources} the mapping function 328 + * returned. 329 + * 330 + * This can be used to issue multiple values per emission of an input {@link Source}, essentially 331 + * multiplexing all values to multiple Sources. 332 + * 333 + * @example 334 + * ```ts 335 + * pipe( 336 + * interval(50), 337 + * mergeMap(x => pipe( 338 + * fromValue(x), 339 + * delay(100) 340 + * )), 341 + * subscribe(x => { 342 + * console.log(text); // logs: 0, 1, 2... 343 + * }) 344 + * ); 345 + * ``` 346 + */ 184 347 export function mergeMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 185 348 return source => sink => { 186 349 let innerTalkbacks: TalkbackFn[] = []; ··· 254 417 }; 255 418 } 256 419 420 + /** Flattens a Source emitting Sources into a single Source emitting the inner values. 421 + * 422 + * @see {@link mergeMap} which this helper uses and instead accept a mapping function. 423 + * @param source - An {@link Source} emitting {@link Source | Sources}. 424 + * @returns A {@link Source} emitting values from the inner Sources. 425 + * 426 + * @remarks 427 + * `mergeAll` accepts a {@link Source} which must emit {@link Source | Sources}. It will subscribe 428 + * to each incoming source immediately and start passing its emitted values through. 429 + * 430 + * @example 431 + * ```ts 432 + * pipe( 433 + * fromArray([ 434 + * interval(50), 435 + * interval(100), 436 + * ]), 437 + * mergeAll, 438 + * subscribe(x => { 439 + * console.log(text); // logs: 0, 0, 1, 2, 1, 3, 4, 2 440 + * }) 441 + * ); 442 + * ``` 443 + */ 257 444 export function mergeAll<T>(source: Source<Source<T>>): Source<T> { 258 445 return mergeMap<Source<T>, T>(identity)(source); 259 446 } 260 447 448 + /** Emits values from the passed sources simultaneously. 449 + * 450 + * @param sources - An array of {@link Source | Sources}. 451 + * @returns A {@link Source} emitting values from the input Sources. 452 + * 453 + * @remarks 454 + * `merge` accepts an array of {@link Source | Sources} and will subscribe to all of them, passing 455 + * through all their emitted values simultaneously. 456 + * 457 + * This can be used to interleave the values of multiple sources. 458 + * 459 + * @example 460 + * ```ts 461 + * pipe( 462 + * merge([ 463 + * interval(50), 464 + * interval(100), 465 + * ]), 466 + * subscribe(x => { 467 + * console.log(text); // logs: 0, 0, 1, 2, 1, 3, 4, 2 468 + * }) 469 + * ); 470 + * ``` 471 + */ 261 472 export function merge<T>(sources: Source<T>[]): Source<T> { 262 473 return mergeAll(fromArray(sources)); 263 474 } 264 475 476 + /** Calls the passed callback function when the Source ends or is closed. 477 + * 478 + * @param callback - A function that is called when the {@link Source} ends. 479 + * @returns An {@link Operator}. 480 + * 481 + * @remarks 482 + * `onEnd` accepts a callback which is called when the {@link Source} either ends 483 + * or is closed. 484 + * 485 + * This operator can be used to add side-effects to a Source. 486 + * 487 + * @example 488 + * ```ts 489 + * pipe( 490 + * fromArray([1, 2, 3]), 491 + * take(1), 492 + * onEnd(() => { 493 + * console.log('end'); 494 + * }), 495 + * publish 496 + * ); 497 + * ``` 498 + */ 265 499 export function onEnd<T>(callback: () => void): Operator<T, T> { 266 500 return source => sink => { 267 501 let ended = false; ··· 292 526 }; 293 527 } 294 528 529 + /** Calls the passed callback function when the Source emits a value. 530 + * 531 + * @param callback - A function that is called with each value the {@link Source} emits. 532 + * @returns An {@link Operator}. 533 + * 534 + * @remarks 535 + * `onPush` accepts a callback which is called for every emitted value of 536 + * the {@link Source}. 537 + * 538 + * This operator can be used to add side-effects to a Source. 539 + * 540 + * @example 541 + * ```ts 542 + * pipe( 543 + * fromArray([1, 2, 3]), 544 + * onPush(value => { 545 + * console.log(value); // logs: 1, 2, 3 546 + * }), 547 + * publish 548 + * ); 549 + * ``` 550 + */ 295 551 export function onPush<T>(callback: (value: T) => void): Operator<T, T> { 296 552 return source => sink => { 297 553 let ended = false; ··· 317 573 }; 318 574 } 319 575 576 + /** Calls the passed callback function when the Source starts. 577 + * 578 + * @param callback - A function that is called when the {@link Source} is started. 579 + * @returns An {@link Operator}. 580 + * 581 + * @remarks 582 + * `onPush` accepts a callback which is called for every emitted value of 583 + * the {@link Source}. 584 + * 585 + * This operator can be used to add side-effects to a Source. 586 + * Specifically, it's useful to add a side-effect for a Source that triggers only once 587 + * the {@link Source} is used and started. 588 + * 589 + * @example 590 + * ```ts 591 + * pipe( 592 + * fromArray([1, 2, 3]), 593 + * onStart(() => { 594 + * console.log('start'); 595 + * }), 596 + * publish 597 + * ); 598 + * ``` 599 + */ 320 600 export function onStart<T>(callback: () => void): Operator<T, T> { 321 601 return source => sink => 322 602 source(signal => { ··· 331 611 }); 332 612 } 333 613 614 + /** Emits the last value the {@link Source} emitted, whenever the notifier Source emits a value. 615 + * 616 + * @param notifier - A {@link Source} that triggers the last value to be emitted. 617 + * @returns An {@link Operator}. 618 + * 619 + * @remarks 620 + * `sample` will store the latest value the {@link Source} emitted. Every time the `notifier` Source 621 + * emits, it will emit the latest value. 622 + * 623 + * This is a back pressure operator that can be used to omit values from a {@link Source} coming in 624 + * too frequently. 625 + * 626 + * {@link Source | Sources} emitting `undefined` are undefined behaviour and these values will be 627 + * ignored. 628 + * 629 + * @example 630 + * ```ts 631 + * pipe( 632 + * interval(50), 633 + * sample(interval(100)), 634 + * subscribe(x => { 635 + * console.log(text); // logs: 0, 2, 4... 636 + * }) 637 + * ); 638 + * ``` 639 + */ 334 640 export function sample<S, T>(notifier: Source<S>): Operator<T, T> { 335 641 return source => sink => { 336 642 let sourceTalkback = talkbackPlaceholder; ··· 389 695 }; 390 696 } 391 697 698 + /** Maps emitted values using the passed reducer function. 699 + * 700 + * @param reducer - A function called with the last value by the `reducer` and the emitted value. 701 + * @param seed - The initial value that is passed to the `reducer`. 702 + * @returns An {@link Operator}. 703 + * 704 + * @remarks 705 + * `scan` accepts a reducer function and a seed value. The reducer will be called initially with the 706 + * seed value and the first emitted value. The {@link Source} will then emit the value returned by 707 + * the reducer function. Subsequently, the `reducer` is called with the last value the `reducer` 708 + * returned and the emitted value. 709 + * 710 + * This operator is similar to `Array.prototype.reduce`, but instead is called over time and emits 711 + * each value of the reducer. 712 + * 713 + * @example 714 + * ```ts 715 + * pipe( 716 + * fromArray([1, 2, 3]), 717 + * scan((acc, x) => acc + x, 0), 718 + * subscribe(x => { 719 + * console.log(text); // logs: 1, 3, 6 720 + * }) 721 + * ); 722 + * ``` 723 + */ 392 724 export function scan<In, Out>(reducer: (acc: Out, value: In) => Out, seed: Out): Operator<In, Out> { 393 725 return source => sink => { 394 726 let acc = seed; ··· 404 736 }; 405 737 } 406 738 739 + /** Shares one underlying subscription to the Source between all Sinks. 740 + * 741 + * @param source - A {@link Source} that should be shared. 742 + * @returns A shared {@link Source}. 743 + * 744 + * @remarks 745 + * `share` accepts a {@link Source} and returns one. It will emit all values as normal, however, it 746 + * will share one subscription to the input source. This allows side-effects on the input 747 + * {@link Source} to only be triggerd once. 748 + */ 407 749 export function share<T>(source: Source<T>): Source<T> { 408 750 let sinks: Sink<T>[] = []; 409 751 let talkback = talkbackPlaceholder; ··· 438 780 }; 439 781 } 440 782 783 + /** Omits `wait` amount of values from the Source and then runs as usual. 784 + * 785 + * @param wait - The number of values to be omitted. 786 + * @returns An {@link Operator}. 787 + * 788 + * @remarks 789 + * `skip` will skip `wait` number of emitted values, then issue all values as normal afterwards. 790 + * This essentially skips a given number of values on the input {@link Source}. 791 + * 792 + * @example 793 + * ```ts 794 + * pipe( 795 + * fromArray([1, 2, 3]), 796 + * skip(2), 797 + * subscribe(x => { 798 + * console.log(text); // logs: 3 799 + * }) 800 + * ); 801 + * ``` 802 + */ 441 803 export function skip<T>(wait: number): Operator<T, T> { 442 804 return source => sink => { 443 805 let talkback = talkbackPlaceholder; ··· 457 819 }; 458 820 } 459 821 822 + /** Omits values from an input Source until a notifier Source emits a value. 823 + * 824 + * @param notifier - A {@link Source} that starts the operator's sent values. 825 + * @returns An {@link Operator}. 826 + * 827 + * @remarks 828 + * `skipUntil` will omit all values from the input {@link Source} until the `notifier` 829 + * Source emits a value of its own. It'll then start passing values from the Source through. 830 + * 831 + * @example 832 + * ```ts 833 + * pipe( 834 + * interval(50), 835 + * skipUntil(interval(150)), 836 + * subscribe(x => { 837 + * console.log(text); // logs: 2, 3... 838 + * }) 839 + * ); 840 + * ``` 841 + */ 460 842 export function skipUntil<S, T>(notifier: Source<S>): Operator<T, T> { 461 843 return source => sink => { 462 844 let sourceTalkback = talkbackPlaceholder; ··· 513 895 }; 514 896 } 515 897 898 + /** Omits values from an input Source until a predicate function returns `false`. 899 + * 900 + * @param predicate - A function returning a boolean per value. 901 + * @returns An {@link Operator}. 902 + * 903 + * @remarks 904 + * `skipWhile` will omit all values from the input {@link Source} until the `predicate` 905 + * function returns `false`. When the `predicate` function returns `false`, the Source's values will 906 + * be passed through. 907 + * 908 + * @example 909 + * ```ts 910 + * pipe( 911 + * fromArray([1, 2, 3]), 912 + * skipWhile(x => x < 2), 913 + * subscribe(x => { 914 + * console.log(text); // logs: 2, 3 915 + * }) 916 + * ); 917 + * ``` 918 + */ 516 919 export function skipWhile<T>(predicate: (value: T) => boolean): Operator<T, T> { 517 920 return source => sink => { 518 921 let talkback = talkbackPlaceholder; ··· 537 940 }; 538 941 } 539 942 943 + /** Emits from the latest Source returned by a mapping function per value of the Source. 944 + * 945 + * @param map - A function returning a {@link Source} per value. 946 + * @returns An {@link Operator}. 947 + * 948 + * @remarks 949 + * `switchMap` accepts a mapping function which must return a {@link Source} per value. 950 + * The output {@link Source} will emit values from the latest Source the mapping function 951 + * returned. If a value is emitted while the last returned Source is still active, the prior Source 952 + * will be closed. 953 + * 954 + * This can be used to issue multiple values per emission of an input {@link Source}, while only 955 + * letting one of these sub-Sources be active at a time. 956 + * 957 + * @example 958 + * ```ts 959 + * pipe( 960 + * interval(100), 961 + * switchMap(() => interval(50)), 962 + * subscribe(x => { 963 + * console.log(text); // logs: 0, 0, 0... 964 + * }) 965 + * ); 966 + * ``` 967 + */ 540 968 export function switchMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 541 969 return source => sink => { 542 970 let outerTalkback = talkbackPlaceholder; ··· 619 1047 }; 620 1048 } 621 1049 1050 + /** Flattens a Source emitting Sources into a single Source emitting the inner values. 1051 + * 1052 + * @see {@link switchMap} which this helper uses and instead accept a mapping function. 1053 + * @param source - An {@link Source} emitting {@link Source | Sources}. 1054 + * @returns A {@link Source} emitting values from the inner Sources. 1055 + * 1056 + * @remarks 1057 + * `switchAll` accepts a {@link Source} which must emit {@link Source | Sources}. Each time it 1058 + * receives a {@link Source} it will close its prior subscription and subscribe to the new Source 1059 + * instead, passing through its values. 1060 + * 1061 + * @example 1062 + * ```ts 1063 + * pipe( 1064 + * interval(100), 1065 + * map(() => interval(50)), 1066 + * switchAll, 1067 + * subscribe(x => { 1068 + * console.log(text); // logs: 0, 0, 0... 1069 + * }) 1070 + * ); 1071 + * ``` 1072 + */ 622 1073 export function switchAll<T>(source: Source<Source<T>>): Source<T> { 623 1074 return switchMap<Source<T>, T>(identity)(source); 624 1075 } 625 1076 1077 + /** Emits `max` values from the Source and then ends. 1078 + * 1079 + * @param max - The maximum number of values emitted. 1080 + * @returns An {@link Operator}. 1081 + * 1082 + * @remarks 1083 + * `take` will issue all values as normal until the `max` number of emitted values has been reached. 1084 + * It will then end and close the {@link Source}. 1085 + * 1086 + * @example 1087 + * ```ts 1088 + * pipe( 1089 + * fromArray([1, 2, 3]), 1090 + * take(2), 1091 + * subscribe(x => { 1092 + * console.log(text); // logs: 1, 2 1093 + * }) 1094 + * ); 1095 + * ``` 1096 + */ 626 1097 export function take<T>(max: number): Operator<T, T> { 627 1098 return source => sink => { 628 1099 let talkback = talkbackPlaceholder; ··· 666 1137 }; 667 1138 } 668 1139 1140 + /** Buffers the `max` last values of the Source and emits them once the Source ends. 1141 + * 1142 + * @param max - The maximum number of values buffered. 1143 + * @returns An {@link Operator}. 1144 + * 1145 + * @remarks 1146 + * `takeLast` will buffer values from the input {@link Source} up until the given `max` number. It 1147 + * will only emit values stored in the buffer once the {@link Source} ends. 1148 + * 1149 + * All values in the buffer are emitted like the {@link fromArray | `fromArray`} source would 1150 + * synchronously. 1151 + * 1152 + * @example 1153 + * ```ts 1154 + * pipe( 1155 + * fromArray([1, 2, 3]), 1156 + * takeLast(1), 1157 + * subscribe(x => { 1158 + * console.log(text); // logs: 3 1159 + * }) 1160 + * ); 1161 + * ``` 1162 + */ 669 1163 export function takeLast<T>(max: number): Operator<T, T> { 670 1164 return source => sink => { 671 1165 const queue: T[] = []; ··· 689 1183 }; 690 1184 } 691 1185 1186 + /** Takes values from an input Source until a notifier Source emits a value. 1187 + * 1188 + * @param notifier - A {@link Source} that stops the operator's sent values. 1189 + * @returns An {@link Operator}. 1190 + * 1191 + * @remarks 1192 + * `takeUntil` will issue all values as normal from the input {@link Source} until the `notifier` 1193 + * Source emits a value of its own. It'll then close the {@link Source}. 1194 + * 1195 + * @example 1196 + * ```ts 1197 + * pipe( 1198 + * interval(50), 1199 + * takeUntil(interval(150)), 1200 + * subscribe(x => { 1201 + * console.log(text); // logs: 0, 1 1202 + * }) 1203 + * ); 1204 + * ``` 1205 + */ 692 1206 export function takeUntil<S, T>(notifier: Source<S>): Operator<T, T> { 693 1207 return source => sink => { 694 1208 let sourceTalkback = talkbackPlaceholder; ··· 733 1247 }; 734 1248 } 735 1249 1250 + /** Takes values from an input Source until a predicate function returns `false`. 1251 + * 1252 + * @param predicate - A function returning a boolean per value. 1253 + * @returns An {@link Operator}. 1254 + * 1255 + * @remarks 1256 + * `takeWhile` will issue all values as normal from the input {@link Source} until the `predicate` 1257 + * function returns `false`. When the `predicate` function returns `false`, the current value is 1258 + * omitted and the {@link Source} is closed. 1259 + * 1260 + * @example 1261 + * ```ts 1262 + * pipe( 1263 + * fromArray([1, 2, 3]), 1264 + * takeWhile(x => x < 2), 1265 + * subscribe(x => { 1266 + * console.log(text); // logs: 1 1267 + * }) 1268 + * ); 1269 + * ``` 1270 + */ 736 1271 export function takeWhile<T>(predicate: (value: T) => boolean): Operator<T, T> { 737 1272 return source => sink => { 738 1273 let talkback = talkbackPlaceholder; ··· 757 1292 }; 758 1293 } 759 1294 1295 + /** Debounces a Source by omitting values until a given timeframe has passed. 1296 + * 1297 + * @param timing - A function returning a debounce time (ms) per emitted value. 1298 + * @returns An {@link Operator}. 1299 + * 1300 + * @remarks 1301 + * `debounce` accepts a mapping function that can be used to return a time (in ms) per emitted 1302 + * value. All emitted values issued by the {@link Source} during the returned time will be omitted 1303 + * until the time has passed. 1304 + * 1305 + * Debouncing means that the returned {@link Source} will wait for a minimum time of silence until a 1306 + * value is let through. 1307 + * 1308 + * This is a back pressure operator that can be used to omit values from a {@link Source} coming in 1309 + * too frequently. 1310 + * 1311 + * @example 1312 + * ```ts 1313 + * pipe( 1314 + * interval(50), 1315 + * debounce(() => 100), 1316 + * subscribe(x => { 1317 + * console.log(text); // never logs any value 1318 + * }) 1319 + * ); 1320 + * ``` 1321 + */ 760 1322 export function debounce<T>(timing: (value: T) => number): Operator<T, T> { 761 1323 return source => sink => { 762 1324 let id: any | void; ··· 798 1360 }; 799 1361 } 800 1362 1363 + /** Delays each signal emitted by a Source by given time (ms). 1364 + * 1365 + * @param wait - A time (in ms) by which each {@link SignalKind | signal} is delayed. 1366 + * @returns An {@link Operator}. 1367 + * 1368 + * @remarks 1369 + * `delay` accepts a time (in ms) by which each {@link SignalKind | signal} will be delayed by. 1370 + * This will create a timeout per received signal and delay the emitted values accordingly. 1371 + * 1372 + * Since the operator only calls `setTimeout` per signal, it relies on the timeout implementation to 1373 + * be ordered. Otherwise, signals will arrive in the wrong order at the sink. 1374 + */ 801 1375 export function delay<T>(wait: number): Operator<T, T> { 802 1376 return source => sink => { 803 1377 let active = 0; ··· 817 1391 }; 818 1392 } 819 1393 1394 + /** Throttles a Source by omitting values that are emitted before a given timeout. 1395 + * 1396 + * @param timing - A function returning a throttle time (ms) per emitted value. 1397 + * @returns An {@link Operator}. 1398 + * 1399 + * @remarks 1400 + * `throttle` accepts a mapping function that can be used to return a time (in ms) per emitted 1401 + * value. During the returned timeframe all values issued by the {@link Source} will be omitted and 1402 + * dropped. 1403 + * 1404 + * This is a back pressure operator that can be used to omit values from a {@link Source} coming in 1405 + * too frequently. 1406 + * 1407 + * @example 1408 + * ```ts 1409 + * pipe( 1410 + * interval(50), 1411 + * throttle(() => 100), 1412 + * subscribe(x => { 1413 + * // omits every second value: 0, 2, 4... 1414 + * console.log(text); 1415 + * }) 1416 + * ); 1417 + * ``` 1418 + */ 820 1419 export function throttle<T>(timing: (value: T) => number): Operator<T, T> { 821 1420 return source => sink => { 822 1421 let skip = false;

+155 -130

src/pipe.ts

··· 1 - import { Source } from './types'; 1 + import { Source, Sink, Operator } from './types'; 2 2 3 3 interface UnaryFn<T, R> { 4 4 (source: T): R; 5 5 } 6 6 7 - /* pipe definitions for source + operators composition */ 7 + /** Chain calls operators on a given source and returns the last result. 8 + * @param args - A source, then a variable number of transform functions 9 + * 10 + * @remarks 11 + * The `pipe` utility can be called with a {@link Source} then one or more unary transform functions. 12 + * Each transform function will be called in turn with the last function's return value, starting 13 + * with the source passed as the first argument to `pipe`. 14 + * 15 + * It's used to transform a source with a list of {@link Operator | Operators}. The last argument may 16 + * also be a {@link Sink} that returns something else than a Source. 17 + * 18 + * @example 19 + * 20 + * ```ts 21 + * pipe( 22 + * fromArray([1, 2, 3]), 23 + * map(x => x * 2), 24 + * subscribe(console.log) 25 + * ); 26 + * ``` 27 + * 28 + * @see {@link https://github.com/tc39/proposal-pipeline-operator} for the JS Pipeline Operator spec, for which this is a replacement utility for. 29 + */ 30 + interface pipe { 31 + /* pipe definitions for source + operators composition */ 8 32 9 - function pipe<T, A>(source: Source<T>, op1: UnaryFn<Source<T>, Source<A>>): Source<A>; 33 + <T, A>(source: Source<T>, op1: UnaryFn<Source<T>, Source<A>>): Source<A>; 10 34 11 - function pipe<T, A, B>( 12 - source: Source<T>, 13 - op1: UnaryFn<Source<T>, Source<A>>, 14 - op2: UnaryFn<Source<A>, Source> 15 - ): Source; 35 + <T, A, B>( 36 + source: Source<T>, 37 + op1: UnaryFn<Source<T>, Source<A>>, 38 + op2: UnaryFn<Source<A>, Source> 39 + ): Source; 16 40 17 - function pipe<T, A, B, C>( 18 - source: Source<T>, 19 - op1: UnaryFn<Source<T>, Source<A>>, 20 - op2: UnaryFn<Source<A>, Source>, 21 - op3: UnaryFn<Source, Source<C>> 22 - ): Source<C>; 41 + <T, A, B, C>( 42 + source: Source<T>, 43 + op1: UnaryFn<Source<T>, Source<A>>, 44 + op2: UnaryFn<Source<A>, Source>, 45 + op3: UnaryFn<Source, Source<C>> 46 + ): Source<C>; 23 47 24 - function pipe<T, A, B, C, D>( 25 - source: Source<T>, 26 - op1: UnaryFn<Source<T>, Source<A>>, 27 - op2: UnaryFn<Source<A>, Source>, 28 - op3: UnaryFn<Source, Source<C>>, 29 - op4: UnaryFn<Source<C>, Source<D>> 30 - ): Source<D>; 48 + <T, A, B, C, D>( 49 + source: Source<T>, 50 + op1: UnaryFn<Source<T>, Source<A>>, 51 + op2: UnaryFn<Source<A>, Source>, 52 + op3: UnaryFn<Source, Source<C>>, 53 + op4: UnaryFn<Source<C>, Source<D>> 54 + ): Source<D>; 31 55 32 - function pipe<T, A, B, C, D, E>( 33 - source: Source<T>, 34 - op1: UnaryFn<Source<T>, Source<A>>, 35 - op2: UnaryFn<Source<A>, Source>, 36 - op3: UnaryFn<Source, Source<C>>, 37 - op4: UnaryFn<Source<C>, Source<D>>, 38 - op5: UnaryFn<Source<D>, Source<E>> 39 - ): Source<E>; 56 + <T, A, B, C, D, E>( 57 + source: Source<T>, 58 + op1: UnaryFn<Source<T>, Source<A>>, 59 + op2: UnaryFn<Source<A>, Source>, 60 + op3: UnaryFn<Source, Source<C>>, 61 + op4: UnaryFn<Source<C>, Source<D>>, 62 + op5: UnaryFn<Source<D>, Source<E>> 63 + ): Source<E>; 40 64 41 - function pipe<T, A, B, C, D, E, F>( 42 - source: Source<T>, 43 - op1: UnaryFn<Source<T>, Source<A>>, 44 - op2: UnaryFn<Source<A>, Source>, 45 - op3: UnaryFn<Source, Source<C>>, 46 - op4: UnaryFn<Source<C>, Source<D>>, 47 - op5: UnaryFn<Source<D>, Source<E>>, 48 - op6: UnaryFn<Source<E>, Source<F>> 49 - ): Source<F>; 65 + <T, A, B, C, D, E, F>( 66 + source: Source<T>, 67 + op1: UnaryFn<Source<T>, Source<A>>, 68 + op2: UnaryFn<Source<A>, Source>, 69 + op3: UnaryFn<Source, Source<C>>, 70 + op4: UnaryFn<Source<C>, Source<D>>, 71 + op5: UnaryFn<Source<D>, Source<E>>, 72 + op6: UnaryFn<Source<E>, Source<F>> 73 + ): Source<F>; 50 74 51 - function pipe<T, A, B, C, D, E, F, G>( 52 - source: Source<T>, 53 - op1: UnaryFn<Source<T>, Source<A>>, 54 - op2: UnaryFn<Source<A>, Source>, 55 - op3: UnaryFn<Source, Source<C>>, 56 - op4: UnaryFn<Source<C>, Source<D>>, 57 - op5: UnaryFn<Source<D>, Source<E>>, 58 - op6: UnaryFn<Source<E>, Source<F>>, 59 - op7: UnaryFn<Source<F>, Source<G>> 60 - ): Source<G>; 75 + <T, A, B, C, D, E, F, G>( 76 + source: Source<T>, 77 + op1: UnaryFn<Source<T>, Source<A>>, 78 + op2: UnaryFn<Source<A>, Source>, 79 + op3: UnaryFn<Source, Source<C>>, 80 + op4: UnaryFn<Source<C>, Source<D>>, 81 + op5: UnaryFn<Source<D>, Source<E>>, 82 + op6: UnaryFn<Source<E>, Source<F>>, 83 + op7: UnaryFn<Source<F>, Source<G>> 84 + ): Source<G>; 61 85 62 - function pipe<T, A, B, C, D, E, F, G, H>( 63 - source: Source<T>, 64 - op1: UnaryFn<Source<T>, Source<A>>, 65 - op2: UnaryFn<Source<A>, Source>, 66 - op3: UnaryFn<Source, Source<C>>, 67 - op4: UnaryFn<Source<C>, Source<D>>, 68 - op5: UnaryFn<Source<D>, Source<E>>, 69 - op6: UnaryFn<Source<E>, Source<F>>, 70 - op7: UnaryFn<Source<F>, Source<G>>, 71 - op8: UnaryFn<Source<G>, Source<H>> 72 - ): Source<H>; 86 + <T, A, B, C, D, E, F, G, H>( 87 + source: Source<T>, 88 + op1: UnaryFn<Source<T>, Source<A>>, 89 + op2: UnaryFn<Source<A>, Source>, 90 + op3: UnaryFn<Source, Source<C>>, 91 + op4: UnaryFn<Source<C>, Source<D>>, 92 + op5: UnaryFn<Source<D>, Source<E>>, 93 + op6: UnaryFn<Source<E>, Source<F>>, 94 + op7: UnaryFn<Source<F>, Source<G>>, 95 + op8: UnaryFn<Source<G>, Source<H>> 96 + ): Source<H>; 73 97 74 - /* pipe definitions for source + operators + consumer composition */ 98 + /* pipe definitions for source + operators + consumer composition */ 75 99 76 - function pipe<T, R>(source: Source<T>, consumer: UnaryFn<Source<T>, R>): R; 100 + <T, R>(source: Source<T>, consumer: UnaryFn<Source<T>, R>): R; 77 101 78 - function pipe<T, A, R>( 79 - source: Source<T>, 80 - op1: UnaryFn<Source<T>, Source<A>>, 81 - consumer: UnaryFn<Source<A>, R> 82 - ): R; 102 + <T, A, R>( 103 + source: Source<T>, 104 + op1: UnaryFn<Source<T>, Source<A>>, 105 + consumer: UnaryFn<Source<A>, R> 106 + ): R; 83 107 84 - function pipe<T, A, B, R>( 85 - source: Source<T>, 86 - op1: UnaryFn<Source<T>, Source<A>>, 87 - op2: UnaryFn<Source<A>, Source>, 88 - consumer: UnaryFn<Source, R> 89 - ): R; 108 + <T, A, B, R>( 109 + source: Source<T>, 110 + op1: UnaryFn<Source<T>, Source<A>>, 111 + op2: UnaryFn<Source<A>, Source>, 112 + consumer: UnaryFn<Source, R> 113 + ): R; 90 114 91 - function pipe<T, A, B, C, R>( 92 - source: Source<T>, 93 - op1: UnaryFn<Source<T>, Source<A>>, 94 - op2: UnaryFn<Source<A>, Source>, 95 - op3: UnaryFn<Source, Source<C>>, 96 - consumer: UnaryFn<Source<C>, R> 97 - ): R; 115 + <T, A, B, C, R>( 116 + source: Source<T>, 117 + op1: UnaryFn<Source<T>, Source<A>>, 118 + op2: UnaryFn<Source<A>, Source>, 119 + op3: UnaryFn<Source, Source<C>>, 120 + consumer: UnaryFn<Source<C>, R> 121 + ): R; 98 122 99 - function pipe<T, A, B, C, D, R>( 100 - source: Source<T>, 101 - op1: UnaryFn<Source<T>, Source<A>>, 102 - op2: UnaryFn<Source<A>, Source>, 103 - op3: UnaryFn<Source, Source<C>>, 104 - op4: UnaryFn<Source<C>, Source<D>>, 105 - consumer: UnaryFn<Source<D>, R> 106 - ): R; 123 + <T, A, B, C, D, R>( 124 + source: Source<T>, 125 + op1: UnaryFn<Source<T>, Source<A>>, 126 + op2: UnaryFn<Source<A>, Source>, 127 + op3: UnaryFn<Source, Source<C>>, 128 + op4: UnaryFn<Source<C>, Source<D>>, 129 + consumer: UnaryFn<Source<D>, R> 130 + ): R; 107 131 108 - function pipe<T, A, B, C, D, E, R>( 109 - source: Source<T>, 110 - op1: UnaryFn<Source<T>, Source<A>>, 111 - op2: UnaryFn<Source<A>, Source>, 112 - op3: UnaryFn<Source, Source<C>>, 113 - op4: UnaryFn<Source<C>, Source<D>>, 114 - op5: UnaryFn<Source<D>, Source<E>>, 115 - consumer: UnaryFn<Source<E>, R> 116 - ): R; 132 + <T, A, B, C, D, E, R>( 133 + source: Source<T>, 134 + op1: UnaryFn<Source<T>, Source<A>>, 135 + op2: UnaryFn<Source<A>, Source>, 136 + op3: UnaryFn<Source, Source<C>>, 137 + op4: UnaryFn<Source<C>, Source<D>>, 138 + op5: UnaryFn<Source<D>, Source<E>>, 139 + consumer: UnaryFn<Source<E>, R> 140 + ): R; 117 141 118 - function pipe<T, A, B, C, D, E, F, R>( 119 - source: Source<T>, 120 - op1: UnaryFn<Source<T>, Source<A>>, 121 - op2: UnaryFn<Source<A>, Source>, 122 - op3: UnaryFn<Source, Source<C>>, 123 - op4: UnaryFn<Source<C>, Source<D>>, 124 - op5: UnaryFn<Source<D>, Source<E>>, 125 - op6: UnaryFn<Source<E>, Source<F>>, 126 - consumer: UnaryFn<Source<F>, R> 127 - ): R; 142 + <T, A, B, C, D, E, F, R>( 143 + source: Source<T>, 144 + op1: UnaryFn<Source<T>, Source<A>>, 145 + op2: UnaryFn<Source<A>, Source>, 146 + op3: UnaryFn<Source, Source<C>>, 147 + op4: UnaryFn<Source<C>, Source<D>>, 148 + op5: UnaryFn<Source<D>, Source<E>>, 149 + op6: UnaryFn<Source<E>, Source<F>>, 150 + consumer: UnaryFn<Source<F>, R> 151 + ): R; 128 152 129 - function pipe<T, A, B, C, D, E, F, G, R>( 130 - source: Source<T>, 131 - op1: UnaryFn<Source<T>, Source<A>>, 132 - op2: UnaryFn<Source<A>, Source>, 133 - op3: UnaryFn<Source, Source<C>>, 134 - op4: UnaryFn<Source<C>, Source<D>>, 135 - op5: UnaryFn<Source<D>, Source<E>>, 136 - op6: UnaryFn<Source<E>, Source<F>>, 137 - op7: UnaryFn<Source<F>, Source<G>>, 138 - consumer: UnaryFn<Source<G>, R> 139 - ): R; 153 + <T, A, B, C, D, E, F, G, R>( 154 + source: Source<T>, 155 + op1: UnaryFn<Source<T>, Source<A>>, 156 + op2: UnaryFn<Source<A>, Source>, 157 + op3: UnaryFn<Source, Source<C>>, 158 + op4: UnaryFn<Source<C>, Source<D>>, 159 + op5: UnaryFn<Source<D>, Source<E>>, 160 + op6: UnaryFn<Source<E>, Source<F>>, 161 + op7: UnaryFn<Source<F>, Source<G>>, 162 + consumer: UnaryFn<Source<G>, R> 163 + ): R; 140 164 141 - function pipe<T, A, B, C, D, E, F, G, H, R>( 142 - source: Source<T>, 143 - op1: UnaryFn<Source<T>, Source<A>>, 144 - op2: UnaryFn<Source<A>, Source>, 145 - op3: UnaryFn<Source, Source<C>>, 146 - op4: UnaryFn<Source<C>, Source<D>>, 147 - op5: UnaryFn<Source<D>, Source<E>>, 148 - op6: UnaryFn<Source<E>, Source<F>>, 149 - op7: UnaryFn<Source<F>, Source<G>>, 150 - op8: UnaryFn<Source<G>, Source<H>>, 151 - consumer: UnaryFn<Source<H>, R> 152 - ): R; 165 + <T, A, B, C, D, E, F, G, H, R>( 166 + source: Source<T>, 167 + op1: UnaryFn<Source<T>, Source<A>>, 168 + op2: UnaryFn<Source<A>, Source>, 169 + op3: UnaryFn<Source, Source<C>>, 170 + op4: UnaryFn<Source<C>, Source<D>>, 171 + op5: UnaryFn<Source<D>, Source<E>>, 172 + op6: UnaryFn<Source<E>, Source<F>>, 173 + op7: UnaryFn<Source<F>, Source<G>>, 174 + op8: UnaryFn<Source<G>, Source<H>>, 175 + consumer: UnaryFn<Source<H>, R> 176 + ): R; 177 + } 153 178 154 - function pipe(...args: any[]) { 179 + function pipe(...args: Function[]): any { 155 180 let x = args[0]; 156 181 for (let i = 1, l = args.length; i < l; i++) x = args[i](x); 157 182 return x;

+122

src/sinks.ts

··· 1 1 import { Source, Subscription, TalkbackKind, SignalKind } from './types'; 2 2 import { talkbackPlaceholder } from './helpers'; 3 3 4 + /** Creates a subscription to a given source and invokes a `subscriber` callback for each value. 5 + * @param subscriber - A callback function called for each issued value. 6 + * @returns A function accepting a {@link Source} and returning a {@link Subscription}. 7 + * 8 + * @remarks 9 + * `subscribe` accepts a `subscriber` callback and returns a function accepting a {@link Source}. 10 + * When a source is passed to the returned funtion, the subscription will start and `subscriber` 11 + * will be called for each new value the Source issues. This will also return a {@link Subscription} 12 + * object that can cancel the ongoing {@link Source} early. 13 + * 14 + * @example 15 + * ```ts 16 + * const subscription = pipe( 17 + * fromValue('test'), 18 + * subscribe(text => { 19 + * console.log(text); // 'test' 20 + * }) 21 + * ); 22 + * ``` 23 + */ 4 24 export function subscribe<T>(subscriber: (value: T) => void) { 5 25 return (source: Source<T>): Subscription => { 6 26 let talkback = talkbackPlaceholder; ··· 26 46 }; 27 47 } 28 48 49 + /** Creates a subscription to a given source and invokes a `subscriber` callback for each value. 50 + * @see {@link subscribe} which this helper aliases without returnin a {@link Subscription}. 51 + * @param subscriber - A callback function called for each issued value. 52 + * @returns A function accepting a {@link Source}. 53 + * 54 + * @remarks 55 + * `forEach` accepts a `subscriber` callback and returns a function accepting a {@link Source}. 56 + * When a source is passed to the returned funtion, the subscription will start and `subscriber` 57 + * will be called for each new value the Source issues. Unlike `subscribe` it will not return a 58 + * Subscription object and can't be cancelled early. 59 + * 60 + * @example 61 + * ```ts 62 + * pipe( 63 + * fromValue('test'), 64 + * forEach(text => { 65 + * console.log(text); // 'test' 66 + * }) 67 + * ); // undefined 68 + * ``` 69 + */ 29 70 export function forEach<T>(subscriber: (value: T) => void) { 30 71 return (source: Source<T>): void => { 31 72 subscribe(subscriber)(source); 32 73 }; 33 74 } 34 75 76 + /** Creates a subscription to a given source and invokes a `subscriber` callback for each value. 77 + * @see {@link subscribe} which this helper aliases without accepting parameters or returning a 78 + * {@link Subscription | Subscription}. 79 + * 80 + * @param source - A {@link Source}. 81 + * 82 + * @remarks 83 + * `publish` accepts a {@link Source} and subscribes to it, starting its values. The resulting 84 + * values cannot be observed and the subscription can't be cancelled, as this helper is purely 85 + * intended to start side-effects. 86 + * 87 + * @example 88 + * ```ts 89 + * pipe( 90 + * lazy(() => { 91 + * console.log('test'); // this is called 92 + * return fromValue(123); // this is never used 93 + * }), 94 + * publish 95 + * ); // undefined 96 + * ``` 97 + */ 35 98 export function publish<T>(source: Source<T>): void { 36 99 subscribe(_value => { 37 100 /*noop*/ ··· 40 103 41 104 const doneResult = { done: true } as IteratorReturnResult<void>; 42 105 106 + /** Converts a Source to an AsyncIterable that pulls and issues values from the Source. 107 + * 108 + * @param source - A {@link Source}. 109 + * @returns An {@link AsyncIterable | `AsyncIterable`} issuing values from the Source. 110 + * 111 + * @remarks 112 + * `toAsyncIterable` will create an {@link AsyncIterable} that pulls and issues values from a given 113 + * {@link Source}. This can be used in many interoperability situations, to provide an iterable when 114 + * a consumer requires it. 115 + * 116 + * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols} 117 + * for the JS Iterable protocol. 118 + * 119 + * @example 120 + * ```ts 121 + * const iterable = toAsyncIterable(fromArray([1, 2, 3])); 122 + * for await (const value of iterable) { 123 + * console.log(value); // outputs: 1, 2, 3 124 + * } 125 + * ``` 126 + */ 43 127 export const toAsyncIterable = <T>(source: Source<T>): AsyncIterable<T> => ({ 44 128 [Symbol.asyncIterator](): AsyncIterator<T> { 45 129 const buffer: T[] = []; ··· 84 168 }, 85 169 }); 86 170 171 + /** Subscribes to a given source and collects all synchronous values into an array. 172 + * @param source - A {@link Source}. 173 + * @returns An array of values collected from the {@link Source}. 174 + * 175 + * @remarks 176 + * `toArray` accepts a {@link Source} and returns an array of all synchronously issued values from 177 + * this Source. It will issue {@link TalkbackKind.Pull | Pull signals} after every value it receives 178 + * and expects the Source to recursively issue values. 179 + * 180 + * Any asynchronously issued values will not be 181 + * added to the array and a {@link TalkbackKind.Close | Close signal} is issued by the sink before 182 + * returning the array. 183 + * 184 + * @example 185 + * ```ts 186 + * toArray(fromArray([1, 2, 3])); // [1, 2, 3] 187 + * ``` 188 + */ 87 189 export function toArray<T>(source: Source<T>): T[] { 88 190 const values: T[] = []; 89 191 let talkback = talkbackPlaceholder; ··· 102 204 return values; 103 205 } 104 206 207 + /** Subscribes to a given source and returns a Promise that will resolve with the last value the 208 + * source issues. 209 + * 210 + * @param source - A {@link Source}. 211 + * @returns A {@link Promise} resolving to the last value of the {@link Source}. 212 + * 213 + * @remarks 214 + * `toPromise` will subscribe to the passed {@link Source} and resolve to the last value of it once 215 + * it receives the last value, as signaled by the {@link SignalKind.End | End signal}. 216 + * 217 + * To keep its implementation simple, padding sources that don't issue any values to `toPromise` is 218 + * undefined behaviour and `toPromise` will issue `undefined` in that case. 219 + * 220 + * The returned {@link Promise} delays its value by a microtick, using `Promise.resolve`. 221 + * 222 + * @example 223 + * ```ts 224 + * toPromise(fromValue('test')); // resolves: 'test' 225 + * ``` 226 + */ 105 227 export function toPromise<T>(source: Source<T>): Promise<T> { 106 228 return new Promise(resolve => { 107 229 let talkback = talkbackPlaceholder;

+206 -4

src/sources.ts

··· 2 2 import { push, start, talkbackPlaceholder, teardownPlaceholder } from './helpers'; 3 3 import { share } from './operators'; 4 4 5 - export function lazy<T>(make: () => Source<T>): Source<T> { 6 - return sink => make()(sink); 5 + /** Helper creating a Source from a factory function when it's subscribed to. 6 + * @param produce - A factory function returning a {@link Source}. 7 + * @returns A {@link Source} lazyily subscribing to the Source returned by the given factory 8 + * function. 9 + * 10 + * @remarks 11 + * At times it's necessary to create a {@link Source} lazily. The time of a {@link Source} being 12 + * created could be different from when it's subscribed to, and hence we may want to split the 13 + * creation and subscription time. This is especially useful when the Source we wrap is "hot" and 14 + * issues values as soon as it's created, which we may then not receive in a subscriber. 15 + * 16 + * @example An example of creating a {@link Source} that issues the timestamp of subscription. Here 17 + * we effectively use `lazy` with the simple {@link fromValue | `fromValue`} source, to quickly 18 + * create a Source that issues the time of its subscription, rather than the time of its creation 19 + * that it would otherwise issue without `lazy`. 20 + * 21 + * ```ts 22 + * lazy(() => fromValue(Date.now())); 23 + * ``` 24 + */ 25 + export function lazy<T>(produce: () => Source<T>): Source<T> { 26 + return sink => produce()(sink); 7 27 } 8 28 29 + /** Converts an AsyncIterable to a Source that pulls and issues values from it as requested. 30 + * 31 + * @see {@link fromIterable | `fromIterable`} for the non-async Iterable version of this helper, 32 + * which calls this helper automatically as needed. 33 + * 34 + * @param iterable - An {@link AsyncIterable | `AsyncIterable`}. 35 + * @returns A {@link Source} issuing values sourced from the Iterable. 36 + * 37 + * @remarks 38 + * `fromAsyncIterable` will create a {@link Source} that pulls and issues values from a given 39 + * {@link AsyncIterable}. This can be used in many interoperability situations, including to consume 40 + * an async generator function. 41 + * 42 + * When the {@link Sink} throws an exception when a new value is pushed, this helper will rethrow it 43 + * using {@link AsyncIterator.throw}, which allows an async generator to recover from the exception. 44 + * 45 + * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols} 46 + * for the JS Iterable protocol. 47 + */ 9 48 export function fromAsyncIterable<T>(iterable: AsyncIterable<T>): Source<T> { 10 49 return sink => { 11 50 const iterator = iterable[Symbol.asyncIterator](); ··· 46 85 }; 47 86 } 48 87 88 + /** Converts an Iterable to a Source that pulls and issues values from it as requested. 89 + * @see {@link fromAsyncIterable | `fromAsyncIterable`} for the AsyncIterable version of this helper. 90 + * @param iterable - An {@link Iterable | `Iterable`} or an `AsyncIterable` 91 + * @returns A {@link Source} issuing values sourced from the Iterable. 92 + * 93 + * @remarks 94 + * `fromIterable` will create a {@link Source} that pulls and issues values from a given 95 + * {@link Iterable | JS Iterable}. As iterables are the common standard for any lazily iterated list 96 + * of values in JS it can be applied to many different JS data types, including a JS Generator 97 + * function. 98 + * 99 + * This Source will only call {@link Iterator.next} on the iterator when the subscribing {@link Sink} 100 + * has pulled a new value with the {@link TalkbackKind.Pull | Pull signal}. `fromIterable` can 101 + * therefore also be applied to "infinite" iterables, without a predefined end. 102 + * 103 + * This helper will call {@link fromAsyncIterable | `fromAsyncIterable`} automatically when the 104 + * passed object also implements the async iterator protocol. 105 + * 106 + * When the {@link Sink} throws an exception when a new value is pushed, this helper will rethrow it 107 + * using {@link Iterator.throw}, which allows a generator to recover from the exception. 108 + * 109 + * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_iterable_protocol} 110 + * for the JS Iterable protocol. 111 + */ 49 112 export function fromIterable<T>(iterable: Iterable<T> | AsyncIterable<T>): Source<T> { 50 113 if (iterable[Symbol.asyncIterator]) return fromAsyncIterable(iterable as AsyncIterable<T>); 51 114 return sink => { ··· 87 150 }; 88 151 } 89 152 153 + /** Creates a Source that issues a each value of a given array synchronously. 154 + * @see {@link fromIterable} which `fromArray` aliases. 155 + * @param array - The array whose values will be issued one by one. 156 + * @returns A {@link Source} issuing the array's values. 157 + * 158 + * @remarks 159 + * `fromArray` will create a {@link Source} that issues the values of a given JS array one by one. It 160 + * will issue values as they're pulled and is hence a "cold" source, not eagerly emitting values. It 161 + * will end and issue the {@link SignalKind.End | End signal} when the array is exhausted of values. 162 + * 163 + * @example 164 + * ```ts 165 + * fromArray([1, 2, 3]); 166 + * ``` 167 + */ 90 168 export const fromArray: <T>(array: T[]) => Source<T> = fromIterable; 91 169 170 + /** Creates a Source that issues a single value and ends immediately after. 171 + * @param value - The value that will be issued. 172 + * @returns A {@link Source} issuing the single value. 173 + * 174 + * @example 175 + * ```ts 176 + * fromValue('test'); 177 + * ``` 178 + */ 92 179 export function fromValue<T>(value: T): Source<T> { 93 180 return sink => { 94 181 let ended = false; ··· 106 193 }; 107 194 } 108 195 109 - export function make<T>(produce: (observer: Observer<T>) => TeardownFn): Source<T> { 196 + /** Creates a new Source from scratch from a passed `subscriber` function. 197 + * @param subscriber - A callback that is called when the {@link Source} is subscribed to. 198 + * @returns A {@link Source} created from the `subscriber` parameter. 199 + * 200 + * @remarks 201 + * `make` is used to create a new, arbitrary {@link Source} from scratch. It calls the passed 202 + * `subscriber` function when it's subscribed to. 203 + * 204 + * The `subscriber` function receives an {@link Observer}. You may call {@link Observer.next} to 205 + * issue values on the Source, and {@link Observer.complete} to end the Source. 206 + * 207 + * Your `subscribr` function must return a {@link TeardownFn | teardown function} which is only 208 + * called when your source is cancelled — not when you invoke `complete` yourself. As this creates a 209 + * "cold" source, every time this source is subscribed to, it will invoke the `subscriber` function 210 + * again and create a new source. 211 + * 212 + * @example 213 + * 214 + * ```ts 215 + * make(observer => { 216 + * const frame = requestAnimationFrame(() => { 217 + * observer.next('animate!'); 218 + * }); 219 + * return () => { 220 + * cancelAnimationFrame(frame); 221 + * }; 222 + * }); 223 + * ``` 224 + */ 225 + export function make<T>(subscriber: (observer: Observer<T>) => TeardownFn): Source<T> { 110 226 return sink => { 111 227 let ended = false; 112 - const teardown = produce({ 228 + const teardown = subscriber({ 113 229 next(value: T) { 114 230 if (!ended) sink(push(value)); 115 231 }, ··· 131 247 }; 132 248 } 133 249 250 + /** Creates a new Subject which can be used as an IO event hub. 251 + * @returns A new {@link Subject}. 252 + * 253 + * @remarks 254 + * `makeSubject` creates a new {@link Subject}. A Subject is a {@link Source} and an {@link Observer} 255 + * combined in one interface, as the Observer is used to send new signals to the Source. This means 256 + * that it's "hot" and hence all subscriptions to {@link Subject.source} share the same underlying 257 + * signals coming from {@link Subject.next} and {@link Subject.complete}. 258 + * 259 + * @example 260 + * ```ts 261 + * const subject = makeSubject(); 262 + * pipe(subject.source, subscribe(console.log)); 263 + * // This will log the string on the above subscription 264 + * subject.next('hello subject!'); 265 + * ``` 266 + */ 134 267 export function makeSubject<T>(): Subject<T> { 135 268 let next: Subject<T>['next'] | void; 136 269 let complete: Subject<T>['complete'] | void; ··· 151 284 }; 152 285 } 153 286 287 + /** A {@link Source} that immediately ends. 288 + * @remarks 289 + * `empty` is a {@link Source} that immediately issues an {@link SignalKind.End | End signal} when 290 + * it's subscribed to, ending immediately. 291 + * 292 + * @see {@link never | `never`} for a source that instead never ends. 293 + */ 154 294 export const empty: Source<any> = (sink: Sink<any>): void => { 155 295 let ended = false; 156 296 sink( ··· 165 305 ); 166 306 }; 167 307 308 + /** A {@link Source} without values that never ends. 309 + * @remarks 310 + * `never` is a {@link Source} that never issues any signals and neither sends values nor ends. 311 + * 312 + * @see {@link empty | `empty`} for a source that instead ends immediately. 313 + */ 168 314 export const never: Source<any> = (sink: Sink<any>): void => { 169 315 sink(start(talkbackPlaceholder)); 170 316 }; 171 317 318 + /** Creates a Source that issues an incrementing integer in intervals. 319 + * @param ms - The interval in milliseconds. 320 + * @returns A {@link Source} issuing an incrementing count on each interval. 321 + * 322 + * @remarks 323 + * `interval` will create a {@link Source} that issues an incrementing counter each time the `ms` 324 + * interval expires. 325 + * 326 + * It'll only stop when it's cancelled by a {@link TalkbackKind.Close | Close signal}. 327 + * 328 + * @example 329 + * An example printing `0`, then `1`, and so on, in intervals of 50ms. 330 + * 331 + * ```ts 332 + * pipe(interval(50), subscribe(console.log)); 333 + * ``` 334 + */ 172 335 export function interval(ms: number): Source<number> { 173 336 return make(observer => { 174 337 let i = 0; ··· 177 340 }); 178 341 } 179 342 343 + /** Converts DOM Events to a Source given an `HTMLElement` and an event's name. 344 + * @param element - The {@link HTMLElement} to listen to. 345 + * @param event - The DOM Event name to listen to. 346 + * @returns A {@link Source} issuing the {@link Event | DOM Events} as they're issued by the DOM. 347 + * 348 + * @remarks 349 + * `fromDomEvent` will create a {@link Source} that listens to the given element's events and issues 350 + * them as values on the source. This source will only stop when it's cancelled by a 351 + * {@link TalkbackKind.Close | Close signal}. 352 + * 353 + * @example 354 + * An example printing `'clicked!'` when the given `#root` element is clicked. 355 + * 356 + * ```ts 357 + * const element = document.getElementById('root'); 358 + * pipe( 359 + * fromDomEvent(element, 'click'), 360 + * subscribe(() => console.log('clicked!')) 361 + * ); 362 + * ``` 363 + */ 180 364 export function fromDomEvent(element: HTMLElement, event: string): Source<Event> { 181 365 return make(observer => { 182 366 element.addEventListener(event, observer.next); ··· 184 368 }); 185 369 } 186 370 371 + /** Converts a Promise to a Source that issues the resolving Promise's value and then ends. 372 + * @param promise - The promise that will be wrapped. 373 + * @returns A {@link Source} issuing the promise's value when it resolves. 374 + * 375 + * @remarks 376 + * `fromPromise` will create a {@link Source} that issues the {@link Promise}'s resolving value 377 + * asynchronously and ends immediately after resolving. 378 + * 379 + * This helper will not handle the promise's exceptions, and will cause uncaught errors if the 380 + * promise rejects without a value. 381 + * 382 + * @example 383 + * An example printing `'resolved!'` when the given promise resolves after a tick. 384 + * 385 + * ```ts 386 + * pipe(fromPromise(Promise.resolve('resolved!')), subscribe(console.log)); 387 + * ``` 388 + */ 187 389 export function fromPromise<T>(promise: Promise<T>): Source<T> { 188 390 return make(observer => { 189 391 promise.then(value => {

+161 -9

src/types.ts

··· 1 - /** A talkback signal is used to tell a [Source] that either the [Sink] is ready for new values or that the stream should be cancelled */ 1 + /** 2 + * Talkback signal that sends instructions from a sink to a source. 3 + * 4 + * @remarks 5 + * This signal is issued via {@link TalkbackFn | talkback functions} that a {@link Sink} receives via 6 + * the {@link Start} signal, to tell a {@link Source} to either send a new value (pulling) or stop 7 + * sending values altogether (cancellation). 8 + */ 2 9 export const enum TalkbackKind { 10 + /** Instructs the {@link Source} to send the next value. */ 3 11 Pull = 0, 12 + /** Instructs the {@link Source} to stop sending values and cancels it. */ 4 13 Close = 1, 5 14 } 6 15 7 - /** A talkback callback is sent to the sink with the [Start] signal to communicate signals back to the source. */ 16 + /** 17 + * Talkback callback that sends instructions to a source. 18 + * 19 + * @remarks 20 + * This function sends a {@link TalkbackKind} signal to the source to instruct it to send a new value 21 + * (pulling) or to be cancelled and stop sending values altogether. 22 + */ 8 23 export type TalkbackFn = (signal: TalkbackKind) => void; 24 + 25 + /** 26 + * Callback that is called when a source is cancelled. 27 + * 28 + * @remarks 29 + * This is used, in particular, in the {@link make | make Source} and is a returned function that is 30 + * called when the {@link TalkbackKind.Close} signal is received by the source. 31 + */ 9 32 export type TeardownFn = () => void; 10 33 34 + /** 35 + * Tag enum that is used to on signals that are sent from a source to a sink. 36 + * 37 + * @remarks 38 + * This signal is issued by a {@link Source} and {@link Sink | Sinks} are called with it. The signals 39 + * carrying values ({@link Start} and {@link Push}) are sent as a unary `[T]` tuple tagged with 40 + * {@link Tag}. The {@link End} signal carries no value and is sent as a raw `0` value. 41 + * @see {@link Start} for the data structure of the start signal. 42 + * @see {@link Push} for the data structure of the push signal, carrying values. 43 + */ 11 44 export const enum SignalKind { 45 + /** 46 + * Informs the {@link Sink} that it's being called by a {@link Source}. 47 + * 48 + * @remarks 49 + * This starts the stream of values and carries a {@link TalkbackFn | talkback function} with it 50 + * that is used by the {@link Sink} to communicate back to the {@link Source}. 51 + * @see {@link Start} for the data structure of the signal. 52 + */ 12 53 Start = 0, 54 + /** 55 + * Informs the {@link Sink} of a new values that's incoming from the {@link Source}. 56 + * 57 + * @remarks 58 + * This informs the {@link Sink} of new values that are sent by the {@link Source}. 59 + * @see {@link Push} for the data structure of the signal. 60 + */ 13 61 Push = 1, 62 + /** 63 + * Informs the {@link Sink} that the {@link Source} has ended and that it won't send more values. 64 + * 65 + * @remarks 66 + * This signal signifies that the stream has stopped and that no more values are expected. Some 67 + * sources don't have a set end or limit on how many values will be sent. This signal is not sent 68 + * when the {@link Source} is cancelled with a {@link TalkbackKind.Close | Close talkback signal}. 69 + */ 14 70 End = 0, 15 71 } 16 72 73 + /** 74 + * The tag property that's put on unary `[T]` tuple to turn them into signals carrying values. 75 + * 76 + * @internal 77 + */ 17 78 export interface Tag<T> { 18 79 tag: T; 19 80 } 20 81 21 - /** The start [Signal] is the first signal and carries a callback (talkback) so the sink can send signals to the source */ 82 + /** 83 + * Indicates the start of a stream to a {@link Sink}. 84 + * 85 + * @remarks 86 + * This signal is sent from a {@link Source} to a {@link Sink} at the start of a stream to inform it 87 + * that values can be pulled and/or will be sent. This signal carries a 88 + * {@link TalkbackFn | talkback function} that is used by the {@link Sink} to communicate back to the 89 + * {@link Source} as a callback. The talkback accepts {@link TalkbackKind.Pull | Pull} and 90 + * {@link TalkbackKind.Close | Close} signals. 91 + */ 22 92 export type Start<_T> = Tag<SignalKind.Start> & [TalkbackFn]; 23 - /** The Push [Signal] carries new values to the sink, like in an event emitter */ 93 + 94 + /** 95 + * Sends a new value to a {@link Sink}. 96 + * 97 + * @remarks 98 + * This signal is sent from a {@link Source} to a {@link Sink} to send a new value to it. This is 99 + * essentially the signal that wraps new values coming in, like an event. Values are carried on 100 + * unary tuples and can be accessed using `signal[0]`. 101 + */ 24 102 export type Push<T> = Tag<SignalKind.Push> & [T]; 25 103 26 - /** A signal that communicates new events to a sink. */ 104 + /** 105 + * Signals are sent from {@link Source | Sources} to {@link Sink | Sinks} to inform them of changes. 106 + * 107 + * @remarks 108 + * A {@link Source}, when consumed, sends a sequence of events to {@link Sink | Sinks}. In order, a 109 + * {@link SignalKind.Start | Start} signal will always be sent first, followed optionally by one or 110 + * more {@link SignalKind.Push | Push signals}, carrying values and representing the stream. A 111 + * {@link Source} will send the {@link SignalKind.End | End signal} when it runs out of values. The 112 + * End signal will be omitted if the Source is cancelled by a 113 + * {@link TalkbackKind.Close | Close signal}, sent back from the {@link Sink}. 114 + * @see {@link SignalKind} for the kinds signals sent by {@link Source | Sources}. 115 + * @see {@link Start} for the data structure of the start signal. 116 + * @see {@link Push} for the data structure of the push signal. 117 + */ 27 118 export type Signal<T> = Start<T> | Push<T> | SignalKind.End; 28 119 29 - /** A sink accepts new values from a [Source], like [Push], [Start], and an end signal. The [Start] is used to receive a callback to send talkback signals back to the source. */ 120 + /** 121 + * Callback function that is called by a {@link Source} with {@link Signal | Signals}. 122 + * 123 + * @remarks 124 + * A Sink is a function that is called repeatedly with signals from a {@link Source}. It represents 125 + * the receiver of the stream of signals/events coming from a {@link Source}. 126 + * @see {@link Signal} for the data structure of signals. 127 + */ 30 128 export type Sink<T> = (signal: Signal<T>) => void; 31 - /** A source is a function that accepts a [Sink] and then starts sending [Signal]s to it. */ 129 + 130 + /** Factory function that calls {@link Sink | Sinks} with {@link Signal | Signals} when invoked. 131 + * @remarks 132 + * A Source is a factory function that when invoked with a {@link Sink}, calls it with 133 + * {@link Signal | Signals} to create a stream of events, informing it of new values and the 134 + * potential end of the stream of values. The first signal a Source sends is always a 135 + * {@link Start | Start signal} that sends a talkback function to the {@link Sink}, so it may request 136 + * new values or cancel the source. 137 + * 138 + * @see {@link Signal} for the data structure of signals. 139 + * @see {@link Sink} for the data structure of sinks. 140 + */ 32 141 export type Source<T> = (sink: Sink<T>) => void; 33 - /** An operator transforms a [Source] and returns a new [Source], potentially with different timings or output types. */ 142 + 143 + /** Transform function that accepts a {@link Source} and returns a new one. 144 + * @remarks 145 + * Wonka comes with several helper operators that transform a given {@link Source} into a new one, 146 + * potentially changing its outputs, or the outputs' timing. An "operator" in Wonka typically 147 + * accepts arguments and then returns this kind of function, so they can be chained and composed. 148 + * 149 + * @see {@link pipe | `pipe`} for the helper used to compose operators. 150 + */ 34 151 export type Operator<In, Out> = (a: Source<In>) => Source<Out>; 35 152 36 - /** Extracts the type of a given Source */ 153 + /** Type utility to determine the type of a {@link Source}. */ 37 154 export type TypeOfSource<T> = T extends Source<infer U> ? U : never; 38 155 156 + /** Subscription object that can be used to cancel a {@link Source}. 157 + * @see {@link subscribe | subscribe sink} for a helper that returns this structure. 158 + */ 39 159 export interface Subscription { 160 + /** 161 + * Cancels a {@link Source} to stop the subscription from receiving new values. 162 + * 163 + * @see {@link TalkbackKind.Close | Close signal} This uses the {@link TalkbackFn | talkback function} to send a {@link TalkbackKind.Close | Close signal} 164 + * to the subscribed-to {@link Source} to stop it from sending new values. This cleans up the subscription 165 + * and ends it immediately. 166 + */ 40 167 unsubscribe(): void; 41 168 } 42 169 170 + /** An Observer represents sending signals manually to a {@link Sink}. 171 + * @remarks 172 + * The Observer is used whenever a utility allows for signals to be sent manually as a {@link Source} 173 + * would send them. 174 + * 175 + * @see {@link make | `make` source} for a helper that uses this structure. 176 + */ 43 177 export interface Observer<T> { 178 + /** Sends a new value to the receiving Sink. 179 + * @remarks 180 + * This creates a {@link Push | Push signal} that is sent to a {@link Sink}. 181 + */ 44 182 next(value: T): void; 183 + /** Indicates to the receiving Sink that no more values will be sent. 184 + * @remarks 185 + * This creates an {@link SignalKind.End | End signal} that is sent to a {@link Sink}. The Observer 186 + * will accept no more values via {@link Observer.next | `next` calls} once this method has been 187 + * invoked. 188 + */ 45 189 complete(): void; 46 190 } 47 191 192 + /** Subjects combine a {@link Source} with the {@link Observer} that is used to send values on said Source. 193 + * @remarks 194 + * A Subject is used whenever an event hub-like structure is needed, as it both provides the 195 + * {@link Observer}'s methods to send signals, as well as the `source` to receive said signals. 196 + * 197 + * @see {@link makeSubject | `makeSubject` source} for a helper that creates this structure. 198 + */ 48 199 export interface Subject<T> extends Observer<T> { 200 + /** The {@link Source} that issues the signals as the {@link Observer} methods are called. */ 49 201 source: Source<T>; 50 202 }

Configure Feed

Configure Feed