chore: add mobile app

This commit is contained in:
2026-02-07 22:33:02 +01:00
parent db34640acc
commit e89384ad4e
29327 changed files with 3886944 additions and 12 deletions
+102
View File
@@ -0,0 +1,102 @@
# wonka
## 6.3.5
### Patch Changes
- Exclude `sourcesContent` from published sourcemaps
Submitted by [@kitten](https://github.com/kitten) (See [`68e608f`](https://github.com/0no-co/wonka/commit/68e608f46244e82d41c952ecfa1d7f0096e168f6))
## 6.3.4
### Patch Changes
- Add missing `Symbol.observable` global declaration back to typings
Submitted by [@kitten](https://github.com/kitten) (See [#168](https://github.com/0no-co/wonka/pull/168))
## 6.3.3
### Patch Changes
- Improve compatibility of `fromAsyncIterable` and `toAsyncIterable`. The `toAsyncIterable` will now output an object that's both an `AsyncIterator` and an `AsyncIterable`. Both helpers will now use a polyfill for `Symbol.asyncIterator` to improve compatibility with the Hermes engine and Babel transpilation
Submitted by [@kitten](https://github.com/kitten) (See [#165](https://github.com/0no-co/wonka/pull/165))
## 6.3.2
### Patch Changes
- Publish with npm provenance
Submitted by [@kitten](https://github.com/kitten) (See [#161](https://github.com/0no-co/wonka/pull/161))
## 6.3.1
### Patch Changes
- ⚠️ Fix missing `declare` keyword on internal ambient enums
Submitted by [@kitten](https://github.com/kitten) (See [#159](https://github.com/0no-co/wonka/pull/159))
## 6.3.0
### Minor Changes
- Add `addOne` argument to `takeWhile`, allowing an additional value to be issued
Submitted by [@kitten](https://github.com/kitten) (See [#156](https://github.com/0no-co/wonka/pull/156))
### Patch Changes
- Convert `Push<T>` and `Start<T>` signals to `{ tag, 0: value }` objects, which are sufficiently backwards compatible and result in slightly faster execution in v8
Submitted by [@kitten](https://github.com/kitten) (See [#155](https://github.com/0no-co/wonka/pull/155))
## 6.2.6
### Patch Changes
- ⚠️ Fix missing source contents in Wonka sourcemaps
Submitted by [@kitten](https://github.com/kitten) (See [`56d9708`](https://github.com/0no-co/wonka/commit/56d970861424fddd403262bf85d7e1e3572b15e2))
- ⚠️ Fix internal `SignalKind` and `TalkbackKind` enums not compiling away
Submitted by [@kitten](https://github.com/kitten) (See [#154](https://github.com/0no-co/wonka/pull/154))
## 6.2.5
### Patch Changes
- Make `closed: boolean` on `ObservableSubscription`s a required field to comply with the Observable proposal's type spec
Submitted by [@naporin0624](https://github.com/naporin0624) (See [#151](https://github.com/0no-co/wonka/pull/151))
## 6.2.4
### Patch Changes
- Add missing overload definition for `filter`, which allows types to be narrowed, e.g. by specifying a type predicate return type
Submitted by [@kitten](https://github.com/kitten) (See [#149](https://github.com/0no-co/wonka/pull/149))
## 6.2.3
### Patch Changes
- ⚠️ Fix overload of `pipe` type not being applied in bundled `d.ts` file, by [@kitten](https://github.com/kitten) (See [#144](https://github.com/0no-co/wonka/pull/144))
## 6.2.2
### Patch Changes
- ⚠️ Fix missing `Symbol.observable` typings and remove `const enum` exports, which aren't usable in isolated modules, by [@kitten](https://github.com/kitten) (See [#141](https://github.com/0no-co/wonka/pull/141))
## 6.2.1
### Patch Changes
- ⚠️ Fix accidental addition of `postinstall` script rather than `prepare` script, by [@kitten](https://github.com/kitten) (See [#138](https://github.com/0no-co/wonka/pull/138))
## 6.2.0
### Minor Changes
- Implement `toAsyncIterable`, converting a Wonka source to a JS Async Iterable, by [@kitten](https://github.com/kitten) (See [#133](https://github.com/0no-co/wonka/pull/133))
- Implement `d.ts` bundling. Only a single `wonka.d.ts` typings file will now be available to TypeScript, by [@kitten](https://github.com/kitten) (See [#135](https://github.com/0no-co/wonka/pull/135))
- 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, by [@kitten](https://github.com/kitten) (See [#136](https://github.com/0no-co/wonka/pull/136))
### Patch Changes
- ⚠️ Fix promise timing by adding missing `Promise.resolve()` tick to `toPromise` sink function, by [@kitten](https://github.com/kitten) (See [#131](https://github.com/0no-co/wonka/pull/131))
- ⚠️ Fix implementation of Observable spec as such that Observable.subscribe(onNext, onError, onComplete) becomes valid, by [@kitten](https://github.com/kitten) (See [#132](https://github.com/0no-co/wonka/pull/132))
+163
View File
@@ -0,0 +1,163 @@
# Development
Thanks for contributing! We want to ensure that `wonka` evolves
and fulfills its idea of being a tiny & capable push & pull stream library!
## How to contribute?
We follow fairly standard but lenient rules around pull requests and issues.
Please pick a title that describes your change briefly, optionally in the imperative
mood if possible.
If you have an idea for a feature or want to fix a bug, consider opening an issue
first. We're also happy to discuss and help you open a PR and get your changes
in!
- If you have a question, try [creating a GitHub Discussions thread.](https://github.com/0no-co/wonka/discussions/new/choose)
- If you think you've found a bug, [open a new issue.](https://github.com/0no-co/wonka/issues/new)
- or, if you found a bug you'd like to fix, [open a PR.](https://github.com/0no-co/wonka/compare)
- If you'd like to propose a change [open an RFC issue.](https://github.com/0no-co/wonka/issues/new?labels=future+%F0%9F%94%AE&template=RFC.md&title=RFC%3A+Your+Proposal) You can read more about the RFC process [below](#how-do-i-propose-changes).
### What are the issue conventions?
There are **no strict conventions**, but we do have two templates in place that will fit most
issues, since questions and other discussion start on GitHub Discussions. The bug template is fairly
standard and the rule of thumb is to try to explain **what you expected** and **what you got
instead.** Following this makes it very clear whether it's a known behavior, an unexpected issue,
or an undocumented quirk.
### How do I propose changes?
We follow an **RFC proposal process**. This allows anyone to propose a new feature or a change, and
allows us to communicate our current planned features or changes, so any technical discussion,
progress, or upcoming changes are always **documented transparently.** You can [find the RFC
template](https://github.com/0no-co/wonka/issues/new/choose) in our issue creator.
### What are the PR conventions?
This also comes with **no strict conventions**. We only ask you to follow the PR template we have
in place more strictly here than the templates for issues, since it asks you to list a summary
(maybe even with a short explanation) and a list of technical changes.
If you're **resolving** an issue please don't forget to add `Resolve #123` to the description so that
it's automatically linked, so that there's no ambiguity and which issue is being addressed (if any)
You'll find that a comment by the "Changeset" bot may pop up. If you don't know what a **changeset**
is and why it's asking you to document your changes, read on at ["How do I document a change for the
changelog"](#how-do-i-document-a-change-for-the-changelog)
We also typically **name** our PRs with a slightly descriptive title, e.g. `feat: Title`.
## How do I set up the project?
Luckily it's not hard to get started. You can install dependencie
[using `pnpm`](https://pnpm.io/installation#using-corepack).
Please don't use `npm` or `yarn` to respect the lockfile.
```sh
pnpm install
```
There are multiple commands you can run in the root folder to test your changes:
```sh
# TypeScript checks:
pnpm run check
# Linting (prettier & eslint):
pnpm run lint
# Unit Tests:
pnpm run test
# Builds:
pnpm run build
```
## How do I test my changes?
It's always good practice to run the tests when making changes. If you're unsure which packages
may be affected by your new tests or changes you may run `pnpm test` in the root of
the repository.
If your editor is not set up with type checks you may also want to run `pnpm run check` on your
changes.
Additionally you can head to any example in the `examples/` folder
and run them. There you'll also need to install their dependencies as they're isolated projects,
without a lockfile and without linking to packages in the monorepos.
All examples are started using the `package.json`'s `start` script.
## How do I lint my code?
We ensure consistency in this codebase using `eslint` and `prettier`.
They are run on a `precommit` hook, so if something's off they'll try
to automatically fix up your code, or display an error.
If you have them set up in your editor, even better!
## How do I document a change for the changelog?
This project uses [changesets](https://github.com/atlassian/changesets). This means that for
every PR there must be documentation for what has been changed and which package is affected.
You can document a change by running `changeset`, which will ask you which packages
have changed and whether the change is major/minor/patch. It will then ask you to write
a change entry as markdown.
```sh
# In the root of the urql repository call:
pnpm changeset
```
This will create a new "changeset file" in the `.changeset` folder, which you should commit and
push, so that it's added to your PR.
This will eventually end up in the package's `CHANGELOG.md` file when we do a release.
You won't need to add a changeset if you're simply making "non-visible" changes to the docs or other
files that aren't published to the npm registry.
[Read more about adding a `changeset` here.](https://github.com/atlassian/changesets/blob/master/docs/adding-a-changeset.md#i-am-in-a-multi-package-repository-a-mono-repo)
## How do I release new versions of our packages?
Hold up, that's **automated**! Since we use `changeset` to document our changes, which determines what
goes into the changelog and what kind of version bump a change should make, you can also use the
tool to check what's currently posed to change after a release batch using: `pnpm changeset status`.
We have a [GitHub Actions workflow](./.github/workflow/release.yml) which is triggered whenever new
changes are merged. It will always open a **"Version Packages" PR** which is kept up-to-date. This PR
documents all changes that are made and will show in its description what all new changelogs are
going to contain for their new entries.
Once a "Version Packages" PR is approved by a contributor and merged, the action will automatically
take care of creating the release, publishing all updated packages to the npm registry, and creating
appropriate tags on GitHub too.
This process is automated, but the changelog should be checked for errors.
As to **when** to merge the automated PR and publish? Maybe not after every change. Typically there
are two release batches: hotfixes and release batches. We expect that a hotfix for a single package
should go out as quickly as possible if it negatively affects users. For **release batches**
however, it's common to assume that if one change is made to a package that more will follow in the
same week. So waiting for **a day or two** when other changes are expected will make sense to keep the
fatigue as low as possible for downstream maintainers.
## How do I upgrade all dependencies?
It may be a good idea to keep all dependencies on this repository **up-to-date** every now and
then. Typically we do this by running `pnpm update --interactive --latest` and checking one-by-one
which dependencies will need to be bumped. In case of any security issues it may make sense to
just run `pnpm update [package]`.
While this is rare with `pnpm`, upgrading some transitive dependencies may accidentally duplicate
them if two packages depend on different compatible version ranges. This can be fixed by running:
```sh
pnpm dedupe
pnpm install
```
It's common to then **create a PR** (with a changeset documenting the packages that need to reflect
new changes if any `dependencies` have changed) with the name of
"(chore) - Upgrade direct and transitive dependencies" or something similar.
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 0no.co
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+48
View File
@@ -0,0 +1,48 @@
# Wonka
A tiny but capable push & pull stream library for TypeScript and Flow,
loosely following the [callbag spec](https://github.com/callbag/callbag)
> **NOTE:** The currently released version v6 is only compatible now with TypeScript, Flow, and JavaScript.
> If you're looking for Reason/OCaml/esy/dune support, please check v5, and if you're looking for the legacy version
> of this library check v4.
<br>
<a href="https://npmjs.com/package/wonka">
<img alt="NPM Version" src="https://img.shields.io/npm/v/wonka.svg" />
</a>
<a href="https://npmjs.com/package/wonka">
<img alt="License" src="https://img.shields.io/npm/l/wonka.svg" />
</a>
<a href="https://coveralls.io/github/kitten/wonka?branch=master">
<img src="https://coveralls.io/repos/github/kitten/wonka/badge.svg?branch=master" alt="Test Coverage" />
</a>
<a href="https://bundlephobia.com/result?p=wonka">
<img alt="Minified gzip size" src="https://img.shields.io/bundlephobia/minzip/wonka.svg?label=gzip%20size" />
</a>
<br>
> “Theres no earthly way of knowing<br>
> Which direction we are going<br>
> Theres no knowing where were rowing<br>
> Or which way the rivers flowing” **Willy Wonka**
<br>
![Wonka](/docs/wonka.jpg?raw=true)
Wonka is a lightweight iterable and observable library loosely based on
the [callbag spec](https://github.com/callbag/callbag). It exposes a set of helpers to create streams,
which are sources of multiple values, which allow you to create, transform
and consume event streams or iterable sets of data.
## [Documentation](https://wonka.kitten.sh/)
**See the documentation at [wonka.kitten.sh](https://wonka.kitten.sh)** for more information about using `wonka`!
- [Introduction](https://wonka.kitten.sh/)
- [**Getting started**](https://wonka.kitten.sh/getting-started)
- [Basics](https://wonka.kitten.sh/basics/)
- [API Reference](https://wonka.kitten.sh/api/)
The raw markdown files can be found [in this repository in the `docs` folder](https://github.com/kitten/wonka/tree/master/docs).
+1460
View File
File diff suppressed because it is too large Load Diff
+1395
View File
File diff suppressed because it is too large Load Diff
+389
View File
@@ -0,0 +1,389 @@
// @flow
// see https://gist.github.com/thecotne/6e5969f4aaf8f253985ed36b30ac9fe0
type $FlowGen$If<X: boolean, Then, Else = empty> = $Call<
((true, Then, Else) => Then) & ((false, Then, Else) => Else),
X,
Then,
Else
>;
type $FlowGen$Assignable<A, B> = $Call<
((...r: [B]) => true) & ((...r: [A]) => false),
A
>;
declare var TalkbackKind: {|
+Pull: 0, // 0
+Close: 1, // 1
|};
declare type TalkbackFn = (signal: $Values<typeof TalkbackKind>) => void;
declare type TeardownFn = () => void;
declare var SignalKind: {|
+Start: 0, // 0
+Push: 1, // 1
+End: 0, // 0
|};
declare interface Tag<T> {
tag: T;
}
declare type Start<_T> = { ...Tag<typeof SignalKind.Start>, ...[TalkbackFn] };
declare type Push<T> = { ...Tag<typeof SignalKind.Push>, ...[T] };
declare type Signal<T> = Start<T> | Push<T> | typeof SignalKind.End;
declare type Sink<T> = (signal: Signal<T>) => void;
declare type Source<T> = (sink: Sink<T>) => void;
declare type Operator<In, Out> = (a: Source<In>) => Source<Out>;
declare type TypeOfSource<T> = $FlowGen$If<
$FlowGen$Assignable<T, Source<U>>,
U,
empty
>;
declare interface Subscription {
unsubscribe(): void;
}
declare interface Observer<T> {
next(value: T): void;
complete(): void;
}
declare type Subject<T> = {
source: Source<T>,
...
} & Observer<T>;
declare type SourceIterable<T> = { ... } & AsyncIterator & AsyncIterable;
declare function lazy<T>(produce: () => Source<T>): Source<T>;
declare function fromAsyncIterable<T>(
iterable: AsyncIterable<T> | AsyncIterator<T>
): Source<T>;
declare function fromIterable<T>(
iterable: Iterable<T> | AsyncIterable<T>
): Source<T>;
declare var fromArray: <T>(array: T[]) => Source<T>;
declare function fromValue<T>(value: T): Source<T>;
declare function make<T>(
subscriber: (observer: Observer<T>) => TeardownFn
): Source<T>;
declare function makeSubject<T>(): Subject<T>;
declare var empty: Source<any>;
declare var never: Source<any>;
declare function interval(ms: number): Source<number>;
declare function fromDomEvent(
element: HTMLElement,
event: string
): Source<Event>;
declare function fromPromise<T>(promise: Promise<T>): Source<T>;
declare function buffer<S, T>(notifier: Source<S>): Operator<T, T[]>;
declare function concatMap<In, Out>(
map: (value: In) => Source<Out>
): Operator<In, Out>;
declare function concatAll<T>(source: Source<Source<T>>): Source<T>;
declare function concat<T>(sources: Source<T>[]): Source<T>;
declare function filter<In, Out: In>(
predicate: (value: In) => boolean
): Operator<In, Out>;
declare function filter<T>(predicate: (value: T) => boolean): Operator<T, T>;
declare function map<In, Out>(map: (value: In) => Out): Operator<In, Out>;
declare function mergeMap<In, Out>(
map: (value: In) => Source<Out>
): Operator<In, Out>;
declare function mergeAll<T>(source: Source<Source<T>>): Source<T>;
declare function merge<T>(sources: Source<T>[]): Source<T>;
declare function onEnd<T>(callback: () => void): Operator<T, T>;
declare function onPush<T>(callback: (value: T) => void): Operator<T, T>;
declare function onStart<T>(callback: () => void): Operator<T, T>;
declare function sample<S, T>(notifier: Source<S>): Operator<T, T>;
declare function scan<In, Out>(
reducer: (acc: Out, value: In) => Out,
seed: Out
): Operator<In, Out>;
declare function share<T>(source: Source<T>): Source<T>;
declare function skip<T>(wait: number): Operator<T, T>;
declare function skipUntil<S, T>(notifier: Source<S>): Operator<T, T>;
declare function skipWhile<T>(predicate: (value: T) => boolean): Operator<T, T>;
declare function switchMap<In, Out>(
map: (value: In) => Source<Out>
): Operator<In, Out>;
declare function switchAll<T>(source: Source<Source<T>>): Source<T>;
declare function take<T>(max: number): Operator<T, T>;
declare function takeLast<T>(max: number): Operator<T, T>;
declare function takeUntil<S, T>(notifier: Source<S>): Operator<T, T>;
declare function takeWhile<T>(
predicate: (value: T) => boolean,
addOne?: boolean
): Operator<T, T>;
declare function debounce<T>(timing: (value: T) => number): Operator<T, T>;
declare function delay<T>(wait: number): Operator<T, T>;
declare function throttle<T>(timing: (value: T) => number): Operator<T, T>;
declare function subscribe<T>(
subscriber: (value: T) => void
): (source: Source<T>) => Subscription;
declare function forEach<T>(
subscriber: (value: T) => void
): (source: Source<T>) => void;
declare function publish<T>(source: Source<T>): void;
declare var toAsyncIterable: <T>(source: Source<T>) => SourceIterable<T>;
declare function toArray<T>(source: Source<T>): T[];
declare function toPromise<T>(source: Source<T>): Promise<T>;
declare type TypeOfSourceArray<T: [] & any[]> = $FlowGen$If<
$FlowGen$Assignable<T, [Head] & Tail>,
[TypeOfSource<Head>] & TypeOfSourceArray<Tail>,
[]
>;
declare interface zip {
<Sources: [] & Source<any>[]>(
sources: [] & Sources
): Source<TypeOfSourceArray<Sources>>;
<
Sources: {
[prop: string]: Source<any>,
}
>(
sources: Sources
): Source<
$ObjMapi<
Sources,
<Property>(Property) => TypeOfSource<$ElementType<Sources, Property>>
>
>;
}
declare function zip<T>(
sources: Source<T>[] | { [key: string]: Source<T>, ... }
): Source<T[] | { [key: string]: T, ... }>;
declare function combine<Sources: Source<any>[]>(
...sources: Sources
): Source<TypeOfSourceArray<Sources>>;
declare module "global" {
declare interface SymbolConstructor {
+observable: Symbol;
}
}
declare interface ObservableSubscription {
closed: boolean;
unsubscribe(): void;
}
declare interface ObservableObserver<T> {
next(value: T): void;
error?: (error: any) => void;
complete?: () => void;
}
declare interface ObservableLike<T> {
subscribe(observer: ObservableObserver<T>): {
unsubscribe(): void,
...
};
}
declare interface Observable<T> {
subscribe(observer: ObservableObserver<T>): ObservableSubscription;
subscribe(
onNext: (value: T) => any,
onError?: (error: any) => any,
onComplete?: () => any
): ObservableSubscription;
}
declare function fromObservable<T>(input: ObservableLike<T>): Source<T>;
declare function toObservable<T>(source: Source<T>): Observable<T>;
declare interface Callbag<I, O> {
(t: 0, d: Callbag<O, I>): void;
(t: 1, d: I): void;
(t: 2, d?: any): void;
}
declare function fromCallbag<T>(callbag: Callbag<any, T>): Source<T>;
declare function toCallbag<T>(source: Source<T>): Callbag<any, T>;
declare interface UnaryFn<T, R> {
(source: T): R;
}
declare interface pipe {
<T, A>(source: Source<T>, op1: UnaryFn<Source<T>, Source<A>>): Source<A>;
<T, A, B>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>
): Source<B>;
<T, A, B, C>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>
): Source<C>;
<T, A, B, C, D>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>
): Source<D>;
<T, A, B, C, D, E>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>
): Source<E>;
<T, A, B, C, D, E, F>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>
): Source<F>;
<T, A, B, C, D, E, F, G>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>
): Source<G>;
<T, A, B, C, D, E, F, G, H>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
op8: UnaryFn<Source<G>, Source<H>>
): Source<H>;
<T, R>(source: Source<T>, consumer: UnaryFn<Source<T>, R>): R;
<T, A, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
consumer: UnaryFn<Source<A>, R>
): R;
<T, A, B, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
consumer: UnaryFn<Source<B>, R>
): R;
<T, A, B, C, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
consumer: UnaryFn<Source<C>, R>
): R;
<T, A, B, C, D, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
consumer: UnaryFn<Source<D>, R>
): R;
<T, A, B, C, D, E, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
consumer: UnaryFn<Source<E>, R>
): R;
<T, A, B, C, D, E, F, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
consumer: UnaryFn<Source<F>, R>
): R;
<T, A, B, C, D, E, F, G, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
consumer: UnaryFn<Source<G>, R>
): R;
<T, A, B, C, D, E, F, G, H, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
op8: UnaryFn<Source<G>, Source<H>>,
consumer: UnaryFn<Source<H>, R>
): R;
}
declare var pipe: pipe;
declare export {
Observer,
Operator,
Signal,
Sink,
Source,
Subject,
Subscription,
TeardownFn,
TypeOfSource,
buffer,
combine,
concat,
concatAll,
concatMap,
debounce,
delay,
empty,
filter,
mergeAll as flatten,
forEach,
fromArray,
fromAsyncIterable,
fromCallbag,
fromDomEvent,
fromIterable,
fromObservable,
fromPromise,
fromValue,
interval,
lazy,
make,
makeSubject,
map,
merge,
mergeAll,
mergeMap,
never,
onEnd,
onPush,
onStart,
pipe,
publish,
sample,
scan,
share,
skip,
skipUntil,
skipWhile,
subscribe,
switchAll,
switchMap,
take,
takeLast,
takeUntil,
takeWhile,
onPush as tap,
throttle,
toArray,
toAsyncIterable,
toCallbag,
toObservable,
toPromise,
zip,
};
+1
View File
@@ -0,0 +1 @@
{"version":3,"file":"wonka.js","sources":[],"sourcesContent":null,"names":[],"mappings":";;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;"}
+1363
View File
File diff suppressed because it is too large Load Diff
+1
View File
@@ -0,0 +1 @@
{"version":3,"file":"wonka.mjs","sources":[],"sourcesContent":null,"names":[],"mappings":";;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;"}
+136
View File
@@ -0,0 +1,136 @@
---
title: Getting Started
order: 1
---
This page will explain how to install the Wonka package and
its basic usage and helper functions.
## Installation
The `wonka` package from `npm` is all you need to install to use
Wonka. The process is the same with `yarn` and `esy`.
```bash
yarn add wonka
# or with npm:
npm install --save wonka
# or with esy:
esy add wonka
```
For **JavaScript projects**, the package contains both CommonJS and
ES Modules bundles. For Flow and TypeScript the package also contains
typings files already, so if you're using either you're already done and
ready to go.
If you're using **BuckleScript** or `bs-native` you will need to add `"wonka"`
to your `bs-dependencies` in your `bsconfig.json` configuration file:
```diff
{
"name": "<some_name>",
"version": "0.1.0",
"sources": ["src"],
"bsc-flags": ["-bs-super-errors"],
"bs-dependencies": [
+ "wonka"
]
}
```
If you're using **Dune** and **Esy** you will need to add `wonka` to
your `libraries` entry in the respective `dune` configuration file:
```diff
(library
(name some_name)
(public_name some_name)
+ (libraries wonka)
)
```
## Usage with JavaScript
In most cases you'll simply import or require `wonka` and use its exposed
methods and utilities. In both CommonJS and ES Modules the Wonka package
simply exposes all its utilities.
```js
// With CommonJS
const { fromArray } = require('wonka');
// With ES Modules
import { fromArray } from 'wonka';
```
There are also some special operators in Wonka that will only be exposed in
Web/JavaScript environments, like `fromPromise`, `toPromise`,
or `fromEvent`, or even `debounce` and `throttle`.
In TypeScript and Flow the typings also expose all types.
There's also a special utility in JavaScript environments to replace the pipeline
operator. This function is called `pipe` and simply calls functions that it's
being passed in order with the previous return value.
```js
import { pipe } from 'wonka';
const output = pipe(
'test',
(x) => x + ' this',
(x) => x.toUpperCase()
);
output; // "TEST THIS"
```
As shown above, the `pipe` function takes the first argument and passes it
in order to the other function arguments. The return value of one function will
be passed on to the next function.
In TypeScript and Flow the `pipe` function is also typed to handle all generics
in Wonka utilities correctly. Using it will ensure that most of the time you won't
have to specify the types of any generics manually.
If you're using Babel and the [pipeline proposal plugin](https://babeljs.io/docs/en/babel-plugin-proposal-pipeline-operator), you can just use
the pipeline operator to do the same and not use the `pipe` helper.
## Usage with Reason
Everything in the Wonka package is exposed under a single module called `Wonka`.
This module also contains `Wonka.Types`, which contains all internal types of the Wonka
library, but you will typically not need it.
In `BuckleScript` when you're compiling to JavaScript you will also have access to
more utilities like `fromPromise`, `toPromise`, `fromEvent`, or even `debounce` and `throttle`.
These utilities are missing in native compilation, like Dune or `bsb-native`, since they're
relying on JavaScript APIs like Promises, `window.addEventListener`, and `setTimeout`.
When using Wonka you'd simply either open the module and use its utilities or just
access them from the `Wonka` module:
```reason
Wonka.fromValue("test")
|> Wonka.map((.x) => x ++ " this")
|> Wonka.forEach((.x) => print_endline(x));
```
It's worth noting that most callbacks in Wonka need to be explicitly uncurried, since
this will help them compile cleanly to JavaScript.
## Interoperability
In JavaScript environments, Wonka comes with several utilities that make it easier
to interoperate with JavaScript primitives and other libraries:
- [`fromPromise`](./api/sources.md#frompromise) & [`toPromise`](./api/sinks.md#topromise) can be used to interoperate with Promises
- [`fromObservable`](./api/sources.md#fromobservable) & [`toObservable`](./api/sinks.md#toobservable) can be used to interoperate with spec-compliant Observables
- [`fromCallbag`](./api/sources.md#fromcallbag) & [`toCallbag`](./api/sinks.md#tocallbag) can be used to interoperate with spec-compliant Callbags
Furthermore there are a couple of operators that only work in JavaScript environments
since they need timing primitives, like `setTimeout` and `setInterval`:
- [`delay`](./api/operators.md#delay)
- [`debounce`](./api/operators.md#debounce)
- [`throttle`](./api/operators.md#throttle)
- [`interval`](./api/sources.md#interval)
+42
View File
@@ -0,0 +1,42 @@
---
title: Introduction
order: 0
---
Wonka is a lightweight iterable and observable library loosely based on
the [callbag spec](https://github.com/callbag/callbag). It exposes a set of helpers to create streams,
which are sources of multiple values, which allow you to create, transform
and consume event streams or iterable sets of data.
## What it is
Wonka is a library for streams _and_ iterables that behaves predictably
and can be used for many problems where you're dealing with streams of
values, asynchronous or not.
It's similar to [RxJS](https://github.com/ReactiveX/rxjs) in that it enables asynchronous programming with
observable streams, with an API that looks like functional programming on
iterables, but it's also similar to [IxJS](https://github.com/ReactiveX/IxJS) since Wonka streams will run
synchronously if an iterable source runs synchronously.
It also comes with many operators that users from [RxJS](https://github.com/ReactiveX/rxjs) will be used to.
## Reason Support
Wonka used to be written in [Reason](https://reasonml.github.io/),, a dialect of OCaml, and was usable
for native development and compileable with [BuckleScript](https://bucklescript.github.io).
Out of the box it supported usage with BuckleScript, `bs-native`, Dune, and Esy.
If you're looking for the legacy version that supported this, you may want to install v4 or v5 rather
than v6 onwards, which converted the project to TypeScript.
## About the docs
As mentioned in the prior section, Wonka supports not one but a couple of
environments and languages. To accommodate for this, most of the docs
are written with examples and sections for TypeScript and Reason.
We don't provide examples in most parts of the docs for Flow and OCaml because
their respective usage is almost identical to TypeScript and Reason, so for
the most part the examples mostly deal with the differences between a
TypeScript and a Reason project.
+127
View File
@@ -0,0 +1,127 @@
---
title: Migration
order: 2
---
This page lists breaking changes and migration guides for
various major releases of Wonka.
## v6.0.0
In `v6.0.0` of Wonka, we've migrated fully to TypeScript.
If you're using this project with Reason or OCaml before, we're sorry for
the inconvenience. However, v4 and v5-rc will remain usable for these
platforms and languages.
The internal API and data structures of Wonka haven't changed in `v6.0.0`
compared to the prior releases and are based on `v4.0.14`. This means that
from a TypeScript, Flow, and JS perspective, `v6.0.0` is backwards compatible
and continues to function as before.
However, the `fromList` API has been removed so far, and we reserve ourselves
room to make more breaking changes were behaviour before was broken.
We're also dropping IE11 support and are now bundling against an ES2015 target.
## v4.0.0
In `v4.0.0` of Wonka, we've migrated to BuckleScript v7 and
`genType` for automatic type generation for TypeScript. The
Flow types are derived from the automatic types and are generated
by `flowgen`.
This may mean that `bsb-native` and Dune/Esy builds are temporarily
broken, as they haven't been tested yet. If so, they will be fixed
in a future minor release. Please stick with `v3.2.2` if you're having
trouble.
This release has no breaking changes for Reason/OCaml in terms of
API changes. You can use the library exactly as you have before.
**For TypeScript and Flow some APIs have changed**.
### New TypeScript and Flow typings
The type for `Subscription`, `Observer`, and `Subject` have changed.
These used to be exposed as tuples (fixed-size arrays) in the past,
but are now compiled to objects, due to the upgrade to BuckleScript v7.
If you're using `subscribe`, `makeSubject`, or `make` you will have
to change some of your types. If you don't, you won't have to update
any of your code and can even mix Wonka `v4.0.0` with `v3.2.2` in the
same bundle.
The `Subscription` type has changed from `[() => void]` to
`{ unsubscribe: (_: void) => void }`:
```ts
import { subscribe } from 'wonka';
// Before:
const [unsubscribe] = subscribe(source);
// After:
const { unsubscribe } = subscribe(source);
```
The `Observer` type has changed similarly, so you'll have to
update your code if you're using `make`:
```ts
import { make } from 'wonka';
// Before:
const source = make(([next, complete]) => /* ... */);
// After:
const source = make(({ next, complete }) => /* ... */);
```
And lastly the `Subject` type has changed as well, so update
your usage of `makeSubject`:
```ts
import { makeSubject } from 'wonka';
// Before:
const [source, next, complete] = makeSubject();
// After:
const { source, next, complete } = makeSubject();
```
### Improvements
The test suite has been rewritten from scratch to improve our
testing of some tricky edge cases. In most cases operators have
been updated to behave more nicely and closer to the spec and
as expected. This is especially true if you're using synchronous
sources or iterables a lot.
Wonka has reached a much higher test coverage and operators like
`merge` and `switchMap` will now behave as expected with synchronous
sources.
This is the list of operators that have changed. If your code has
been working before, you _shouldn't see any different behaviour_.
The changed operators will simply have received bugfixes and will
behave more predictably (and hopefully correctly) in certain edge cases!
- [`buffer`](./api/operators.md#buffer)
- [`combine`](./api/operators.md#combine)
- [`debounce`](./api/operators.md#debounce)
- [`delay`](./api/operators.md#delay)
- [`sample`](./api/operators.md#sample)
- [`skipUntil`](./api/operators.md#skipuntil)
- [`take`](./api/operators.md#take)
- [`takeLast`](./api/operators.md#takelast)
- [`takeWhile`](./api/operators.md#takewhile)
- [`switchMap`](./api/operators.md#switchmap)
- [`mergeMap`](./api/operators.md#mergemap)
- [`concatMap`](./api/operators.md#concatmap)
- [`switchAll`](./api/operators.md#switchall)
- [`mergeAll`](./api/operators.md#mergeall)
- [`concatAll`](./api/operators.md#concatall)
- [`merge`](./api/operators.md#merge)
- [`concat`](./api/operators.md#concat)
The `take` operator is the only one that has been changed to fix
a notable new usage case. It can now accept a maximum of `0` or below,
to close the source immediately.
+3
View File
@@ -0,0 +1,3 @@
// @flow
declare export * from "./dist/wonka.js.flow"
+110
View File
@@ -0,0 +1,110 @@
{
"name": "wonka",
"description": "A tiny but capable push & pull stream library for TypeScript and Flow",
"version": "6.3.5",
"author": "0no.co <hi@0no.co>",
"source": "./src/index.ts",
"main": "./dist/wonka",
"module": "./dist/wonka.mjs",
"types": "./dist/wonka.d.ts",
"exports": {
".": {
"types": "./dist/wonka.d.ts",
"import": "./dist/wonka.mjs",
"require": "./dist/wonka.js",
"source": "./src/index.ts"
},
"./package.json": "./package.json"
},
"sideEffects": false,
"files": [
"src",
"dist",
"docs/*.md",
"index.js.flow",
"*.md"
],
"keywords": [
"wonka",
"typescript",
"events",
"callbag",
"callback",
"observable",
"iterable",
"stream"
],
"repository": "https://github.com/0no-co/wonka",
"bugs": {
"url": "https://github.com/0no-co/wonka/issues"
},
"license": "MIT",
"prettier": {
"singleQuote": true,
"tabWidth": 2,
"printWidth": 100
},
"lint-staged": {
"*.{ts,js}": "eslint -c scripts/eslint-preset.js --fix",
"*.json": "prettier --write",
"*.md": "prettier --write"
},
"husky": {
"hooks": {
"pre-commit": "lint-staged --quiet --relative"
}
},
"eslintConfig": {
"root": true,
"extends": [
"./scripts/eslint-preset.js"
]
},
"publishConfig": {
"provenance": true
},
"devDependencies": {
"@changesets/cli": "^2.27.1",
"@changesets/get-github-info": "^0.6.0",
"@rollup/plugin-buble": "^1.0.1",
"@rollup/plugin-commonjs": "^23.0.3",
"@rollup/plugin-node-resolve": "^15.0.1",
"@rollup/plugin-sucrase": "^5.0.1",
"@rollup/plugin-terser": "^0.1.0",
"@rollup/pluginutils": "^5.0.2",
"@types/zen-observable": "^0.8.3",
"@typescript-eslint/eslint-plugin": "^5.45.0",
"@typescript-eslint/parser": "^5.45.0",
"callbag-from-iter": "^1.3.0",
"callbag-iterate": "^1.0.0",
"callbag-take": "^1.5.0",
"dotenv": "^16.0.3",
"eslint": "^8.29.0",
"eslint-config-prettier": "^8.5.0",
"eslint-plugin-prettier": "^4.2.1",
"eslint-plugin-tsdoc": "^0.2.17",
"flowgen": "^1.21.0",
"glob": "^8.0.3",
"husky-v4": "^4.3.8",
"lint-staged": "^13.0.4",
"npm-run-all": "^4.1.5",
"prettier": "^2.8.0",
"rimraf": "^3.0.2",
"rollup": "^3.5.1",
"rollup-plugin-cjs-check": "^1.0.1",
"rollup-plugin-dts": "^5.1.1",
"tslib": "^2.4.1",
"typescript": "^4.9.5",
"vitest": "^1.2.2",
"zen-observable": "^0.10.0"
},
"scripts": {
"test": "vitest run",
"check": "tsc",
"lint": "eslint --ext=js,ts .",
"build": "rollup -c scripts/rollup.config.mjs",
"clean": "rimraf dist node_modules/.cache",
"changeset:version": "changeset version && pnpm install --lockfile-only",
"changeset:publish": "changeset publish"
}
}
+76
View File
@@ -0,0 +1,76 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { Source } from '../types';
import { fromValue, makeSubject } from '../sources';
import { forEach } from '../sinks';
import {
passesPassivePull,
passesActivePush,
passesSinkClose,
passesSourceEnd,
passesSingleStart,
passesStrictEnd,
} from './compliance';
import { combine, zip } from '../combine';
beforeEach(() => {
vi.useFakeTimers();
});
describe('zip', () => {
const noop = (source: Source<any>) => zip([fromValue(0), source]);
passesPassivePull(noop, [0, 0]);
passesActivePush(noop, [0, 0]);
passesSinkClose(noop);
passesSourceEnd(noop, [0, 0]);
passesSingleStart(noop);
passesStrictEnd(noop);
it('emits the zipped values of two sources', () => {
const { source: sourceA, next: nextA } = makeSubject<number>();
const { source: sourceB, next: nextB } = makeSubject<number>();
const fn = vi.fn();
const combined = combine(sourceA, sourceB);
forEach(fn)(combined);
nextA(1);
expect(fn).not.toHaveBeenCalled();
nextB(2);
expect(fn).toHaveBeenCalledWith([1, 2]);
});
it('emits the zipped values of three sources', () => {
const { source: sourceA, next: nextA } = makeSubject<number>();
const { source: sourceB, next: nextB } = makeSubject<number>();
const { source: sourceC, next: nextC } = makeSubject<number>();
const fn = vi.fn();
const combined = zip([sourceA, sourceB, sourceC]);
forEach(fn)(combined);
nextA(1);
expect(fn).not.toHaveBeenCalled();
nextB(2);
expect(fn).not.toHaveBeenCalled();
nextC(3);
expect(fn).toHaveBeenCalledWith([1, 2, 3]);
});
it('emits the zipped values of a dictionary of two sources', () => {
const { source: sourceA, next: nextA } = makeSubject<number>();
const { source: sourceB, next: nextB } = makeSubject<number>();
const fn = vi.fn();
const combined = zip({ a: sourceA, b: sourceB });
forEach(fn)(combined);
nextA(1);
expect(fn).not.toHaveBeenCalled();
nextB(2);
expect(fn).toHaveBeenCalledWith({ a: 1, b: 2 });
});
});
+406
View File
@@ -0,0 +1,406 @@
import { it, expect, vi } from 'vitest';
import { Source, Sink, Operator, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types';
import { push, start } from '../helpers';
/* This tests a noop operator for passive Pull talkback signals.
A Pull will be sent from the sink upwards and should pass through
the operator until the source receives it, which then pushes a
value down. */
export const passesPassivePull = (operator: Operator<any, any>, output: any = 0) => {
it('responds to Pull talkback signals (spec)', () => {
let talkback: TalkbackFn | null = null;
let pushes = 0;
const values: any[] = [];
const source: Source<any> = sink => {
sink(
start(signal => {
if (!pushes && signal === TalkbackKind.Pull) {
pushes++;
sink(push(0));
}
})
);
};
const sink: Sink<any> = signal => {
expect(signal).not.toBe(SignalKind.End);
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Push) {
values.push(signal[0]);
} else {
talkback = signal[0];
}
};
operator(source)(sink);
// The Start signal should always come in immediately
expect(talkback).not.toBe(null);
// No Push signals should be issued initially
expect(values).toEqual([]);
// When pulling a value we expect an immediate response
talkback!(TalkbackKind.Pull);
vi.runAllTimers();
expect(values).toEqual([output]);
});
};
/* This tests a noop operator for regular, active Push signals.
A Push will be sent downwards from the source, through the
operator to the sink. Pull events should be let through from
the sink after every Push event. */
export const passesActivePush = (operator: Operator<any, any>, result: any = 0) => {
it('responds to eager Push signals (spec)', () => {
const values: any[] = [];
let talkback: TalkbackFn | null = null;
let sink: Sink<any> | null = null;
let pulls = 0;
const source: Source<any> = _sink => {
(sink = _sink)(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
operator(source)(signal => {
expect(signal).not.toBe(SignalKind.End);
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Start) {
talkback = signal[0];
} else if (signal.tag === SignalKind.Push) {
values.push(signal[0]);
talkback!(TalkbackKind.Pull);
}
});
// No Pull signals should be issued initially
expect(pulls).toBe(0);
// When pushing a value we expect an immediate response
sink!(push(0));
vi.runAllTimers();
expect(values).toEqual([result]);
// Subsequently the Pull signal should have travelled upwards
expect(pulls).toBe(1);
});
};
/* This tests a noop operator for Close talkback signals from the sink.
A Close signal will be sent, which should be forwarded to the source,
which then ends the communication without sending an End signal. */
export const passesSinkClose = (operator: Operator<any, any>) => {
it('responds to Close signals from sink (spec)', () => {
let talkback: TalkbackFn | null = null;
let closing = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull && !closing) {
sink(push(0));
} else if (signal === TalkbackKind.Close) {
closing++;
}
})
);
};
const sink: Sink<any> = signal => {
expect(signal).not.toBe(SignalKind.End);
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Push) {
talkback!(TalkbackKind.Close);
} else {
talkback = signal[0];
}
};
operator(source)(sink);
// When pushing a value we expect an immediate close signal
talkback!(TalkbackKind.Pull);
vi.runAllTimers();
expect(closing).toBe(1);
});
};
/* This tests a noop operator for End signals from the source.
A Push and End signal will be sent after the first Pull talkback
signal from the sink, which shouldn't lead to any extra Close or Pull
talkback signals. */
export const passesSourceEnd = (operator: Operator<any, any>, result: any = 0) => {
it('passes on immediate Push then End signals from source (spec)', () => {
const signals: Signal<any>[] = [];
let talkback: TalkbackFn | null = null;
let pulls = 0;
let ending = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
expect(signal).not.toBe(TalkbackKind.Close);
if (signal === TalkbackKind.Pull) {
pulls++;
if (pulls === 1) {
sink(push(0));
sink(SignalKind.End);
}
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
ending++;
} else if (signal.tag === SignalKind.Push) {
signals.push(signal);
} else {
talkback = signal[0];
}
};
operator(source)(sink);
// When pushing a value we expect an immediate Push then End signal
talkback!(TalkbackKind.Pull);
vi.runAllTimers();
expect(ending).toBe(1);
expect(signals).toEqual([push(result), SignalKind.End]);
// Also no additional pull event should be created by the operator
expect(pulls).toBe(1);
});
};
/* This tests a noop operator for End signals from the source
after the first pull in response to another.
This is similar to passesSourceEnd but more well behaved since
mergeMap/switchMap/concatMap are eager operators. */
export const passesSourcePushThenEnd = (operator: Operator<any, any>, result: any = 0) => {
it('passes on End signals from source (spec)', () => {
const signals: Signal<any>[] = [];
let talkback: TalkbackFn | null = null;
let pulls = 0;
let ending = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
expect(signal).not.toBe(TalkbackKind.Close);
if (signal === TalkbackKind.Pull) {
pulls++;
if (pulls <= 2) {
sink(push(0));
} else {
sink(SignalKind.End);
}
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
ending++;
} else if (signal.tag === SignalKind.Push) {
signals.push(signal);
talkback!(TalkbackKind.Pull);
} else {
talkback = signal[0];
}
};
operator(source)(sink);
// When pushing a value we expect an immediate Push then End signal
talkback!(TalkbackKind.Pull);
vi.runAllTimers();
expect(ending).toBe(1);
expect(pulls).toBe(3);
expect(signals).toEqual([push(result), push(result), SignalKind.End]);
});
};
/* This tests a noop operator for Start signals from the source.
When the operator's sink is started by the source it'll receive
a Start event. As a response it should never send more than one
Start signals to the sink. */
export const passesSingleStart = (operator: Operator<any, any>) => {
it('sends a single Start event to the incoming sink (spec)', () => {
let starts = 0;
const source: Source<any> = sink => {
sink(start(() => {}));
};
const sink: Sink<any> = signal => {
if (signal !== SignalKind.End && signal.tag === SignalKind.Start) {
starts++;
}
};
// When starting the operator we expect a single start event on the sink
operator(source)(sink);
expect(starts).toBe(1);
});
};
/* This tests a noop operator for silence after End signals from the source.
When the operator receives the End signal it shouldn't forward any other
signals to the sink anymore.
This isn't a strict requirement, but some operators should ensure that
all sources are well behaved. This is particularly true for operators
that either Close sources themselves or may operate on multiple sources. */
export const passesStrictEnd = (operator: Operator<any, any>) => {
it('stops all signals after End has been received (spec: strict end)', () => {
let pulls = 0;
const signals: Signal<any>[] = [];
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
pulls++;
sink(SignalKind.End);
sink(push(123));
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
} else if (signal.tag === SignalKind.Push) {
signals.push(signal);
} else {
signal[0](TalkbackKind.Pull);
}
};
operator(source)(sink);
// The Push signal should've been dropped
vi.runAllTimers();
expect(signals).toEqual([SignalKind.End]);
expect(pulls).toBe(1);
});
it('stops all signals after Close has been received (spec: strict close)', () => {
const signals: Signal<any>[] = [];
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Close) {
sink(push(123));
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
} else if (signal.tag === SignalKind.Push) {
signals.push(signal);
} else {
signal[0](TalkbackKind.Close);
}
};
operator(source)(sink);
// The Push signal should've been dropped
vi.runAllTimers();
expect(signals).toEqual([]);
});
};
/* This tests an immediately closing operator for End signals to
the sink and Close signals to the source.
When an operator closes immediately we expect to see a Close
signal at the source and an End signal to the sink, since the
closing operator is expected to end the entire chain. */
export const passesCloseAndEnd = (closingOperator: Operator<any, any>) => {
it('closes the source and ends the sink correctly (spec: ending operator)', () => {
let closing = 0;
let ending = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
// For some operator tests we do need to send a single value
if (signal === TalkbackKind.Pull) {
sink(push(null));
} else {
closing++;
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
ending++;
} else if (signal.tag === SignalKind.Start) {
signal[0](TalkbackKind.Pull);
}
};
// We expect the operator to immediately end and close
closingOperator(source)(sink);
expect(closing).toBe(1);
expect(ending).toBe(1);
});
};
export const passesAsyncSequence = (operator: Operator<any, any>, result: any = 0) => {
it('passes an async push with an async end (spec)', () => {
let hasPushed = false;
const signals: Signal<any>[] = [];
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull && !hasPushed) {
hasPushed = true;
setTimeout(() => sink(push(0)), 10);
setTimeout(() => sink(SignalKind.End), 20);
}
})
);
};
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
} else if (signal.tag === SignalKind.Push) {
signals.push(signal);
} else {
setTimeout(() => {
signal[0](TalkbackKind.Pull);
}, 5);
}
};
// We initially expect to see the push signal
// Afterwards after all timers all other signals come in
operator(source)(sink);
expect(signals.length).toBe(0);
vi.advanceTimersByTime(5);
expect(hasPushed).toBeTruthy();
vi.runAllTimers();
expect(signals).toEqual([push(result), SignalKind.End]);
});
};
+854
View File
@@ -0,0 +1,854 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { Source, Sink, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types';
import { push, start } from '../helpers';
import {
passesPassivePull,
passesActivePush,
passesSinkClose,
passesSourceEnd,
passesSingleStart,
passesStrictEnd,
passesSourcePushThenEnd,
passesAsyncSequence,
passesCloseAndEnd,
} from './compliance';
import * as sources from '../sources';
import * as sinks from '../sinks';
import * as operators from '../operators';
beforeEach(() => {
vi.useFakeTimers();
});
describe('buffer', () => {
const valueThenNever: Source<any> = sink =>
sink(
start(signal => {
if (signal === TalkbackKind.Pull) sink(push(null));
})
);
const noop = operators.buffer(valueThenNever);
passesPassivePull(noop, [0]);
passesActivePush(noop, [0]);
passesSinkClose(noop);
passesSourcePushThenEnd(noop, [0]);
passesSingleStart(noop);
passesStrictEnd(noop);
it('emits batches of input values when a notifier emits', () => {
const { source: notifier$, next: notify } = sources.makeSubject();
const { source: input$, next } = sources.makeSubject();
const fn = vi.fn();
sinks.forEach(fn)(operators.buffer(notifier$)(input$));
next(1);
next(2);
expect(fn).not.toHaveBeenCalled();
notify(null);
expect(fn).toHaveBeenCalledWith([1, 2]);
next(3);
notify(null);
expect(fn).toHaveBeenCalledWith([3]);
});
});
describe('concatMap', () => {
const noop = operators.concatMap(x => sources.fromValue(x));
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourcePushThenEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
// This synchronous test for concatMap will behave the same as mergeMap & switchMap
it('emits values from each flattened synchronous source', () => {
const { source, next, complete } = sources.makeSubject<number>();
const fn = vi.fn();
operators.concatMap((x: number) => sources.fromArray([x, x + 1]))(source)(fn);
next(1);
next(3);
complete();
expect(fn).toHaveBeenCalledTimes(6);
expect(fn.mock.calls).toEqual([
[start(expect.any(Function))],
[push(1)],
[push(2)],
[push(3)],
[push(4)],
[SignalKind.End],
]);
});
// This synchronous test for concatMap will behave the same as mergeMap & switchMap
it('lets inner sources finish when outer source ends', () => {
const signals: Signal<any>[] = [];
const teardown = vi.fn();
const fn = (signal: Signal<any>) => {
signals.push(signal);
if (signal !== SignalKind.End && signal.tag === SignalKind.Start) {
signal[0](TalkbackKind.Pull);
signal[0](TalkbackKind.Close);
}
};
operators.concatMap(() => {
return sources.make(() => teardown);
})(sources.fromValue(null))(fn);
expect(teardown).toHaveBeenCalled();
expect(signals).toEqual([start(expect.any(Function))]);
});
// This asynchronous test for concatMap will behave differently than mergeMap & switchMap
it('emits values from each flattened asynchronous source, one at a time', () => {
const source = operators.delay<number>(4)(sources.fromArray([1, 10]));
const fn = vi.fn();
sinks.forEach(fn)(
operators.concatMap((x: number) => {
return operators.delay(5)(sources.fromArray([x, x * 2]));
})(source)
);
vi.advanceTimersByTime(14);
expect(fn.mock.calls).toEqual([[1], [2]]);
vi.runAllTimers();
expect(fn.mock.calls).toEqual([[1], [2], [10], [20]]);
});
it('works for fully asynchronous sources', () => {
const fn = vi.fn();
sinks.forEach(fn)(
operators.concatMap(() => {
return sources.make(observer => {
setTimeout(() => observer.next(1));
return () => {};
});
})(sources.fromValue(null))
);
vi.runAllTimers();
expect(fn).toHaveBeenCalledWith(1);
});
it('emits synchronous values in order', () => {
const values: any[] = [];
sinks.forEach(x => values.push(x))(
operators.concat([sources.fromArray([1, 2]), sources.fromArray([3, 4])])
);
expect(values).toEqual([1, 2, 3, 4]);
});
});
describe('debounce', () => {
const noop = operators.debounce(() => 0);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
it('waits for a specified amount of silence before emitting the last value', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.debounce(() => 100)(source));
next(1);
vi.advanceTimersByTime(50);
expect(fn).not.toHaveBeenCalled();
next(2);
vi.advanceTimersByTime(99);
expect(fn).not.toHaveBeenCalled();
vi.advanceTimersByTime(1);
expect(fn).toHaveBeenCalledWith(2);
});
it('emits debounced value with delayed End signal', () => {
const { source, next, complete } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.debounce(() => 100)(source));
next(1);
complete();
vi.advanceTimersByTime(100);
expect(fn).toHaveBeenCalled();
});
});
describe('delay', () => {
const noop = operators.delay(0);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('delays outputs by a specified delay timeout value', () => {
const { source, next } = sources.makeSubject();
const fn = vi.fn();
sinks.forEach(fn)(operators.delay(100)(source));
next(1);
expect(fn).not.toHaveBeenCalled();
vi.advanceTimersByTime(100);
expect(fn).toHaveBeenCalledWith(1);
});
});
describe('filter', () => {
const noop = operators.filter(() => true);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('prevents emissions for which a predicate fails', () => {
const { source, next } = sources.makeSubject<boolean>();
const fn = vi.fn();
sinks.forEach((x: true) => {
fn(x);
})(operators.filter((x): x is true => !!x)(source));
next(false);
expect(fn).not.toHaveBeenCalled();
next(true);
expect(fn).toHaveBeenCalledWith(true);
});
});
describe('map', () => {
const noop = operators.map(x => x);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('maps over values given a transform function', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.map((x: number) => x + 1)(source));
next(1);
expect(fn).toHaveBeenCalledWith(2);
});
});
describe('mergeMap', () => {
const noop = operators.mergeMap(x => sources.fromValue(x));
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourcePushThenEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
// This synchronous test for mergeMap will behave the same as concatMap & switchMap
it('emits values from each flattened synchronous source', () => {
const { source, next, complete } = sources.makeSubject<number>();
const fn = vi.fn();
operators.mergeMap((x: number) => sources.fromArray([x, x + 1]))(source)(fn);
next(1);
next(3);
complete();
expect(fn.mock.calls).toEqual([
[start(expect.any(Function))],
[push(1)],
[push(2)],
[push(3)],
[push(4)],
[SignalKind.End],
]);
});
// This synchronous test for mergeMap will behave the same as concatMap & switchMap
it('lets inner sources finish when outer source ends', () => {
const values: Signal<any>[] = [];
const teardown = vi.fn();
const fn = (signal: Signal<any>) => {
values.push(signal);
if (signal !== SignalKind.End && signal.tag === SignalKind.Start) {
signal[0](TalkbackKind.Pull);
signal[0](TalkbackKind.Close);
}
};
operators.mergeMap(() => {
return sources.make(() => teardown);
})(sources.fromValue(null))(fn);
expect(teardown).toHaveBeenCalled();
expect(values).toEqual([start(expect.any(Function))]);
});
// This asynchronous test for mergeMap will behave differently than concatMap & switchMap
it('emits values from each flattened asynchronous source simultaneously', () => {
const source = operators.delay<number>(4)(sources.fromArray([1, 10]));
const fn = vi.fn();
sinks.forEach(fn)(
operators.mergeMap((x: number) => {
return operators.delay(5)(sources.fromArray([x, x * 2]));
})(source)
);
vi.runAllTimers();
expect(fn.mock.calls).toEqual([[1], [10], [2], [20]]);
});
it('emits synchronous values in order', () => {
const values: any[] = [];
sinks.forEach(x => values.push(x))(
operators.merge([sources.fromArray([1, 2]), sources.fromArray([3, 4])])
);
expect(values).toEqual([1, 2, 3, 4]);
});
});
describe('onEnd', () => {
const noop = operators.onEnd(() => {});
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesStrictEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('calls a callback when the source ends', () => {
const { source, next, complete } = sources.makeSubject<any>();
const fn = vi.fn();
sinks.forEach(() => {})(operators.onEnd(fn)(source));
next(null);
expect(fn).not.toHaveBeenCalled();
complete();
expect(fn).toHaveBeenCalled();
});
});
describe('onPush', () => {
const noop = operators.onPush(() => {});
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesStrictEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('calls a callback when the source emits', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(() => {})(operators.onPush(fn)(source));
next(1);
expect(fn).toHaveBeenCalledWith(1);
next(2);
expect(fn).toHaveBeenCalledWith(2);
});
it('is the same as `tap`', () => {
expect(operators.onPush).toBe(operators.tap);
});
});
describe('onStart', () => {
const noop = operators.onStart(() => {});
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('is called when the source starts', () => {
let sink: Sink<any>;
const fn = vi.fn();
const source: Source<any> = _sink => {
sink = _sink;
};
sinks.forEach(() => {})(operators.onStart(fn)(source));
expect(fn).not.toHaveBeenCalled();
sink!(start(() => {}));
expect(fn).toHaveBeenCalled();
});
});
describe('sample', () => {
const valueThenNever: Source<any> = sink =>
sink(
start(signal => {
if (signal === TalkbackKind.Pull) sink(push(null));
})
);
const noop = operators.sample(valueThenNever);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourcePushThenEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
it('emits the latest value when a notifier source emits', () => {
const { source: notifier$, next: notify } = sources.makeSubject();
const { source: input$, next } = sources.makeSubject();
const fn = vi.fn();
sinks.forEach(fn)(operators.sample(notifier$)(input$));
next(1);
next(2);
expect(fn).not.toHaveBeenCalled();
notify(null);
expect(fn).toHaveBeenCalledWith(2);
});
});
describe('scan', () => {
const noop = operators.scan<any, any>((_acc, x) => x, null);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('folds values continuously with a reducer and initial value', () => {
const { source: input$, next } = sources.makeSubject<number>();
const fn = vi.fn();
const reducer = (acc: number, x: number) => acc + x;
sinks.forEach(fn)(operators.scan(reducer, 0)(input$));
next(1);
expect(fn).toHaveBeenCalledWith(1);
next(2);
expect(fn).toHaveBeenCalledWith(3);
});
});
describe('share', () => {
const noop = operators.share;
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
it('shares output values between sinks', () => {
let onPush = () => {};
const source: Source<any> = operators.share(sink => {
sink(start(() => {}));
onPush = () => {
sink(push([0]));
sink(SignalKind.End);
};
});
const fnA = vi.fn();
const fnB = vi.fn();
sinks.forEach(fnA)(source);
sinks.forEach(fnB)(source);
onPush();
expect(fnA).toHaveBeenCalledWith([0]);
expect(fnB).toHaveBeenCalledWith([0]);
expect(fnA.mock.calls[0][0]).toBe(fnB.mock.calls[0][0]);
});
it('completes the source when no more sink is listening', () => {
let onPush = () => {};
const talkback = vi.fn();
const source: Source<any> = operators.share(sink => {
sink(start(talkback));
onPush = () => {
sink(push([0]));
sink(push([1]));
sink(SignalKind.End);
};
});
const fnA = vi.fn();
const fnB = vi.fn();
sinks.forEach(fnA)(operators.take(1)(source));
sinks.forEach(fnB)(operators.take(1)(source));
onPush();
expect(fnA).toHaveBeenCalledWith([0]);
expect(fnB).toHaveBeenCalledWith([0]);
expect(fnA.mock.calls[0][0]).toBe(fnB.mock.calls[0][0]);
expect(talkback).toHaveBeenCalledWith(TalkbackKind.Close);
});
});
describe('skip', () => {
const noop = operators.skip(0);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('skips a number of values before emitting normally', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.skip(1)(source));
next(1);
expect(fn).not.toHaveBeenCalled();
next(2);
expect(fn).toHaveBeenCalledWith(2);
});
});
describe('skipUntil', () => {
const noop = operators.skipUntil(sources.fromValue(null));
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
passesStrictEnd(noop);
it('skips values until the notifier source emits', () => {
const { source: notifier$, next: notify } = sources.makeSubject();
const { source: input$, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.skipUntil(notifier$)(input$));
next(1);
expect(fn).not.toHaveBeenCalled();
notify(null);
next(2);
expect(fn).toHaveBeenCalledWith(2);
});
});
describe('skipWhile', () => {
const noop = operators.skipWhile(() => false);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('skips values until one fails a predicate', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.skipWhile((x: any) => x <= 1)(source));
next(1);
expect(fn).not.toHaveBeenCalled();
next(2);
expect(fn).toHaveBeenCalledWith(2);
});
});
describe('switchMap', () => {
const noop = operators.switchMap(x => sources.fromValue(x));
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourcePushThenEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
// This synchronous test for switchMap will behave the same as concatMap & mergeMap
it('emits values from each flattened synchronous source', () => {
const { source, next, complete } = sources.makeSubject<number>();
const fn = vi.fn();
operators.switchMap((x: number) => sources.fromArray([x, x + 1]))(source)(fn);
next(1);
next(3);
complete();
expect(fn).toHaveBeenCalledTimes(6);
expect(fn.mock.calls).toEqual([
[start(expect.any(Function))],
[push(1)],
[push(2)],
[push(3)],
[push(4)],
[SignalKind.End],
]);
});
// This synchronous test for switchMap will behave the same as concatMap & mergeMap
it('lets inner sources finish when outer source ends', () => {
const signals: Signal<any>[] = [];
const teardown = vi.fn();
const fn = (signal: Signal<any>) => {
signals.push(signal);
if (signal !== SignalKind.End && signal.tag === SignalKind.Start) {
signal[0](TalkbackKind.Pull);
signal[0](TalkbackKind.Close);
}
};
operators.switchMap(() => {
return sources.make(() => teardown);
})(sources.fromValue(null))(fn);
expect(teardown).toHaveBeenCalled();
expect(signals).toEqual([start(expect.any(Function))]);
});
// This asynchronous test for switchMap will behave differently than concatMap & mergeMap
it('emits values from each flattened asynchronous source, one at a time', () => {
const source = operators.delay<number>(4)(sources.fromArray([1, 10]));
const fn = vi.fn();
sinks.forEach(fn)(
operators.switchMap((x: number) =>
operators.take(2)(operators.map((y: number) => x * (y + 1))(sources.interval(5)))
)(source)
);
vi.runAllTimers();
expect(fn.mock.calls).toEqual([[1], [10], [20]]);
});
});
describe('take', () => {
const noop = operators.take(10);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
passesCloseAndEnd(operators.take(0));
it('emits values until a maximum is reached', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
operators.take(1)(source)(fn);
next(1);
expect(fn).toHaveBeenCalledTimes(3);
expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]);
});
});
describe('takeUntil', () => {
const noop = operators.takeUntil(sources.never);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourcePushThenEnd(noop);
passesSingleStart(noop);
passesStrictEnd(noop);
passesAsyncSequence(noop);
const ending = operators.takeUntil(sources.fromValue(null));
passesCloseAndEnd(ending);
it('emits values until a notifier emits', () => {
const { source: notifier$, next: notify } = sources.makeSubject<any>();
const { source: input$, next } = sources.makeSubject<number>();
const fn = vi.fn();
operators.takeUntil(notifier$)(input$)(fn);
next(1);
expect(fn).toHaveBeenCalledTimes(2);
expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)]]);
notify(null);
expect(fn).toHaveBeenCalledTimes(3);
expect(fn.mock.calls[2][0]).toEqual(SignalKind.End);
});
it('emits values until a notifier emits', () => {
const { source: input$, next } = sources.makeSubject<number>();
const fn = vi.fn();
let hasClosed = false;
operators.takeUntil(sink => {
sink(
start(talkback => {
if (talkback === TalkbackKind.Close) {
hasClosed = true;
} else if (talkback === TalkbackKind.Pull && !hasClosed) {
sink(push(1));
}
})
);
})(input$)(fn);
next(1);
expect(fn).toHaveBeenCalledTimes(2);
expect(fn.mock.calls).toEqual([[0], [start(expect.any(Function))]]);
expect(hasClosed).toBe(true);
});
});
describe('takeWhile', () => {
const noop = operators.takeWhile(() => true);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
const ending = operators.takeWhile(() => false);
passesCloseAndEnd(ending);
it('emits values while a predicate passes for all values', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
operators.takeWhile((x: any) => x < 2)(source)(fn);
next(1);
next(2);
next(3);
expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]);
});
it('emits values while a predicate passes for all values plus an additional one', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
operators.takeWhile((x: any) => x < 2, true)(source)(fn);
next(1);
next(2);
next(3);
expect(fn.mock.calls).toEqual([
[start(expect.any(Function))],
[push(1)],
[push(2)],
[SignalKind.End],
]);
});
});
describe('takeLast', () => {
passesCloseAndEnd(operators.takeLast(0));
it('emits the last max values of an ended source', () => {
const { source, next, complete } = sources.makeSubject<number>();
const signals: Signal<any>[] = [];
let talkback: TalkbackFn;
operators.takeLast(1)(source)(signal => {
signals.push(signal);
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
} else {
talkback!(TalkbackKind.Pull);
}
});
next(1);
next(2);
expect(signals.length).toBe(0);
complete();
expect(signals).toEqual([start(expect.any(Function)), push(2), SignalKind.End]);
});
});
describe('throttle', () => {
const noop = operators.throttle(() => 0);
passesPassivePull(noop);
passesActivePush(noop);
passesSinkClose(noop);
passesSourceEnd(noop);
passesSingleStart(noop);
passesAsyncSequence(noop);
it('should ignore emissions for a period of time after a value', () => {
const { source, next } = sources.makeSubject<number>();
const fn = vi.fn();
sinks.forEach(fn)(operators.throttle(() => 100)(source));
next(1);
expect(fn).toHaveBeenCalledWith(1);
vi.advanceTimersByTime(50);
next(2);
expect(fn).toHaveBeenCalledTimes(1);
vi.advanceTimersByTime(50);
next(3);
expect(fn).toHaveBeenCalledWith(3);
});
});
+487
View File
@@ -0,0 +1,487 @@
import { describe, it, expect, vi } from 'vitest';
import { Source, Sink, SignalKind, TalkbackKind } from '../types';
import { push, start } from '../helpers';
import * as sinks from '../sinks';
import * as sources from '../sources';
import * as callbag from '../callbag';
import * as observable from '../observable';
import Observable from 'zen-observable';
import callbagIterate from 'callbag-iterate';
import callbagTake from 'callbag-take';
describe('subscribe', () => {
it('sends Pull talkback signals every Push signal', () => {
let pulls = 0;
const fn = vi.fn();
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
if (pulls < 3) {
pulls++;
sink(push(0));
} else {
sink(SignalKind.End);
expect(pulls).toBe(3);
}
}
})
);
};
sinks.subscribe(fn)(source);
expect(fn).toHaveBeenCalledTimes(3);
expect(pulls).toBe(3);
});
it('cancels when unsubscribe is called', () => {
let pulls = 0;
let closing = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
if (!pulls) {
pulls++;
sink(push(0));
}
} else {
closing++;
}
})
);
};
const sub = sinks.subscribe(() => {})(source);
expect(pulls).toBe(1);
sub.unsubscribe();
expect(closing).toBe(1);
});
it('ignores cancellation when the source has already ended', () => {
let pulls = 0;
let closing = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
pulls++;
sink(SignalKind.End);
} else {
closing++;
}
})
);
};
const sub = sinks.subscribe(() => {})(source);
expect(pulls).toBe(1);
sub.unsubscribe();
expect(closing).toBe(0);
});
it('ignores Push signals after the source has ended', () => {
const fn = vi.fn();
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
sink(SignalKind.End);
sink(push(0));
}
})
);
};
sinks.subscribe(fn)(source);
expect(fn).not.toHaveBeenCalled();
});
it('ignores Push signals after cancellation', () => {
const fn = vi.fn();
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Close) {
sink(push(0));
}
})
);
};
sinks.subscribe(fn)(source).unsubscribe();
expect(fn).not.toHaveBeenCalled();
});
});
describe('publish', () => {
it('sends Pull talkback signals every Push signal', () => {
let pulls = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
if (pulls < 3) {
pulls++;
sink(push(0));
} else {
sink(SignalKind.End);
expect(pulls).toBe(3);
}
}
})
);
};
sinks.publish(source);
expect(pulls).toBe(3);
});
});
describe('toArray', () => {
it('sends Pull talkback signals every Push signal', () => {
let pulls = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
if (pulls < 3) {
pulls++;
sink(push(0));
} else {
sink(SignalKind.End);
expect(pulls).toBe(3);
}
}
})
);
};
const array = sinks.toArray(source);
expect(array).toEqual([0, 0, 0]);
expect(pulls).toBe(3);
});
it('sends a Close talkback signal after all synchronous values have been pulled', () => {
let pulls = 0;
let ending = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
if (!pulls) {
pulls++;
sink(push(0));
}
} else {
ending++;
}
})
);
};
const array = sinks.toArray(source);
expect(array).toEqual([0]);
expect(ending).toBe(1);
});
});
describe('toPromise', () => {
it('creates a Promise that resolves on the last value', async () => {
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
const fn = vi.fn();
const promise = sinks.toPromise(source).then(fn);
expect(pulls).toBe(1);
sink!(push(0));
expect(pulls).toBe(2);
sink!(push(1));
sink!(SignalKind.End);
expect(fn).not.toHaveBeenCalled();
await promise;
expect(fn).toHaveBeenCalledWith(1);
});
it('creates a Promise for synchronous sources', async () => {
const fn = vi.fn();
await sinks.toPromise(sources.fromArray([1, 2, 3])).then(fn);
expect(fn).toHaveBeenCalledWith(3);
});
});
describe('toAsyncIterable', () => {
it('creates an async iterable mirroring the Wonka source', async () => {
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
const next$ = asyncIterator.next();
sink!(push(0));
expect(await next$).toEqual({ value: 0, done: false });
expect(pulls).toBe(1);
sink!(push(1));
expect(await asyncIterator.next()).toEqual({ value: 1, done: false });
expect(pulls).toBe(2);
sink!(SignalKind.End);
expect(await asyncIterator.next()).toEqual({ done: true });
expect(pulls).toBe(2);
});
it('buffers actively pushed values', async () => {
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
const next$ = asyncIterator.next();
sink!(push(0));
sink!(push(1));
sink!(SignalKind.End);
expect(pulls).toBe(1);
expect(await next$).toEqual({ value: 0, done: false });
expect(await asyncIterator.next()).toEqual({ value: 1, done: false });
expect(await asyncIterator.next()).toEqual({ done: true });
});
it('asynchronously waits for pulled values', async () => {
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
asyncIterator.next();
expect(pulls).toBe(1);
let resolved = false;
const promise = asyncIterator.next().then(value => {
resolved = true;
return value;
});
await Promise.resolve();
expect(resolved).toBe(false);
sink!(push(0));
sink!(SignalKind.End);
expect(await promise).toEqual({ value: 0, done: false });
expect(await asyncIterator.next()).toEqual({ done: true });
});
it('supports cancellation via return', async () => {
let ended = false;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Close) ended = true;
})
);
};
const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
const next$ = asyncIterator.next();
sink!(push(0));
expect(await next$).toEqual({ value: 0, done: false });
expect(await asyncIterator.return!()).toEqual({ done: true });
sink!(push(1));
expect(await asyncIterator.next()).toEqual({ done: true });
expect(ended).toBeTruthy();
});
it('supports for-await-of', async () => {
let pulls = 0;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
sink(pulls < 3 ? push(pulls++) : SignalKind.End);
}
})
);
};
const iterable = sinks.toAsyncIterable(source);
const values: any[] = [];
for await (const value of iterable) {
values.push(value);
}
expect(values).toEqual([0, 1, 2]);
});
it('supports for-await-of with early break', async () => {
let pulls = 0;
let closed = false;
const source: Source<any> = sink => {
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
sink(pulls < 3 ? push(pulls++) : SignalKind.End);
} else {
closed = true;
}
})
);
};
const iterable = sinks.toAsyncIterable(source);
for await (const value of iterable) {
expect(value).toBe(0);
break;
}
expect(closed).toBe(true);
});
});
describe('toObservable', () => {
it('creates an Observable mirroring the Wonka source', () => {
const next = vi.fn();
const complete = vi.fn();
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
Observable.from(observable.toObservable(source) as any).subscribe({
next,
complete,
});
expect(pulls).toBe(1);
sink!(push(0));
expect(next).toHaveBeenCalledWith(0);
sink!(push(1));
expect(next).toHaveBeenCalledWith(1);
sink!(SignalKind.End);
expect(complete).toHaveBeenCalled();
});
it('forwards cancellations from the Observable as a talkback', () => {
let ending = 0;
const source: Source<any> = sink =>
sink(
start(signal => {
if (signal === TalkbackKind.Close) ending++;
})
);
const sub = Observable.from(observable.toObservable(source) as any).subscribe({});
expect(ending).toBe(0);
sub.unsubscribe();
expect(ending).toBe(1);
});
});
describe('toCallbag', () => {
it('creates a Callbag mirroring the Wonka source', () => {
const fn = vi.fn();
let pulls = 0;
let sink: Sink<any> | null = null;
const source: Source<any> = _sink => {
sink = _sink;
sink(
start(signal => {
if (signal === TalkbackKind.Pull) pulls++;
})
);
};
callbagIterate(fn)(callbag.toCallbag(source));
expect(pulls).toBe(1);
sink!(push(0));
expect(fn).toHaveBeenCalledWith(0);
sink!(push(1));
expect(fn).toHaveBeenCalledWith(1);
sink!(SignalKind.End);
});
it('forwards cancellations from the Callbag as a talkback', () => {
let ending = 0;
const fn = vi.fn();
const source: Source<any> = sink =>
sink(
start(signal => {
if (signal === TalkbackKind.Pull) {
sink(push(0));
} else {
ending++;
}
})
);
callbagIterate(fn)(callbagTake(1)(callbag.toCallbag(source) as any));
expect(fn.mock.calls).toEqual([[0]]);
expect(ending).toBe(1);
});
});
+386
View File
@@ -0,0 +1,386 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { Source, Sink, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types';
import { push, start, talkbackPlaceholder } from '../helpers';
import * as sources from '../sources';
import * as operators from '../operators';
import * as callbag from '../callbag';
import * as observable from '../observable';
import callbagFromArray from 'callbag-from-iter';
import Observable from 'zen-observable';
const collectSignals = (source: Source<any>, onStart?: (talkbackCb: TalkbackFn) => void) => {
let talkback = talkbackPlaceholder;
const signals: Signal<any>[] = [];
source(signal => {
signals.push(signal);
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Start) {
talkback = signal[0];
if (onStart) onStart(talkback);
talkback(TalkbackKind.Pull);
} else {
talkback(TalkbackKind.Pull);
}
});
return signals;
};
/* When a Close talkback signal is sent the source should immediately end */
const passesActiveClose = (source: Source<any>) => {
it('stops emitting when a Close talkback signal is received (spec)', () => {
let talkback: TalkbackFn | null = null;
const sink: Sink<any> = signal => {
expect(signal).not.toBe(SignalKind.End);
expect((signal as any).tag).not.toBe(SignalKind.Push);
if ((signal as any).tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Close);
}
};
source(sink);
expect(talkback).not.toBe(null);
});
};
/* All synchronous, cold sources won't send anything unless a Pull signal
has been received. */
const passesColdPull = (source: Source<any>) => {
it('sends nothing when no Pull talkback signal has been sent (spec)', () => {
let talkback: TalkbackFn | null = null;
let pushes = 0;
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Push) {
pushes++;
} else {
talkback = signal[0];
}
};
source(sink);
expect(talkback).not.toBe(null);
expect(pushes).toBe(0);
setTimeout(() => {
expect(pushes).toBe(0);
talkback!(TalkbackKind.Pull);
}, 10);
vi.runAllTimers();
expect(pushes).toBe(1);
});
};
/* All synchronous, cold sources need to use trampoline scheduling to avoid
recursively sending more and more Push signals which would eventually lead
to a call stack overflow when too many values are emitted. */
const passesTrampoline = (source: Source<any>) => {
it('uses trampoline scheduling instead of recursive push signals (spec)', () => {
let talkback: TalkbackFn | null = null;
let pushes = 0;
const signals: Signal<any>[] = [];
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
signals.push(signal);
expect(pushes).toBe(2);
} else if (signal.tag === SignalKind.Push) {
const lastPushes = ++pushes;
signals.push(signal);
talkback!(TalkbackKind.Pull);
expect(lastPushes).toBe(pushes);
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
expect(pushes).toBe(2);
}
};
source(sink);
expect(signals).toEqual([push(1), push(2), SignalKind.End]);
});
};
beforeEach(() => {
vi.useFakeTimers();
});
describe('fromArray', () => {
passesTrampoline(sources.fromArray([1, 2]));
passesColdPull(sources.fromArray([0]));
passesActiveClose(sources.fromArray([0]));
});
describe('fromValue', () => {
passesColdPull(sources.fromValue(0));
passesActiveClose(sources.fromValue(0));
it('sends a single value and ends', () => {
expect(collectSignals(sources.fromValue(1))).toEqual([
start(expect.any(Function)),
push(1),
SignalKind.End,
]);
});
});
describe('merge', () => {
const source = operators.merge<any>([sources.fromValue(0), sources.empty]);
passesColdPull(source);
passesActiveClose(source);
it('correctly merges two sources where the second is empty', () => {
const source = operators.merge<any>([sources.fromValue(0), sources.empty]);
expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]);
});
it('correctly merges hot sources', () => {
const onStart = vi.fn();
const source = operators.merge<any>([
operators.onStart(onStart)(sources.never),
operators.onStart(onStart)(sources.fromArray([1, 2])),
]);
const signals = collectSignals(source);
expect(onStart).toHaveBeenCalledTimes(2);
expect(signals).toEqual([start(expect.any(Function)), push(1), push(2)]);
});
it('correctly merges asynchronous sources', () => {
vi.useFakeTimers();
const onStart = vi.fn();
const source = operators.merge<any>([
operators.onStart(onStart)(sources.fromValue(-1)),
operators.onStart(onStart)(operators.take(2)(sources.interval(50))),
]);
const signals = collectSignals(source);
vi.advanceTimersByTime(100);
expect(onStart).toHaveBeenCalledTimes(2);
expect(signals).toEqual([
start(expect.any(Function)),
push(-1),
push(0),
push(1),
SignalKind.End,
]);
});
});
describe('concat', () => {
const source = operators.concat<any>([sources.fromValue(0), sources.empty]);
passesColdPull(source);
passesActiveClose(source);
it('correctly concats two sources where the second is empty', () => {
const source = operators.concat<any>([sources.fromValue(0), sources.empty]);
expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]);
});
});
describe('make', () => {
it('may be used to create async sources', () => {
const teardown = vi.fn();
const source = sources.make(observer => {
setTimeout(() => observer.next(1), 10);
setTimeout(() => observer.complete(), 20);
return teardown;
});
const signals = collectSignals(source);
expect(signals).toEqual([start(expect.any(Function))]);
vi.runAllTimers();
expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]);
});
it('supports active cancellation', () => {
const teardown = vi.fn();
const source = sources.make(() => teardown);
const sink: Sink<any> = signal => {
expect(signal).not.toBe(SignalKind.End);
expect((signal as any).tag).not.toBe(SignalKind.Push);
setTimeout(() => signal[0](TalkbackKind.Close));
};
source(sink);
expect(teardown).not.toHaveBeenCalled();
vi.runAllTimers();
expect(teardown).toHaveBeenCalled();
});
});
describe('makeSubject', () => {
it('may be used to emit signals programmatically', () => {
const { source, next, complete } = sources.makeSubject();
const signals = collectSignals(source);
expect(signals).toEqual([start(expect.any(Function))]);
next(1);
expect(signals).toEqual([start(expect.any(Function)), push(1)]);
complete();
expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]);
});
it('ignores signals after complete has been called', () => {
const { source, next, complete } = sources.makeSubject();
const signals = collectSignals(source);
complete();
expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]);
next(1);
complete();
expect(signals.length).toBe(2);
});
});
describe('never', () => {
it('emits nothing and ends immediately', () => {
const signals = collectSignals(sources.never);
expect(signals).toEqual([start(expect.any(Function))]);
});
});
describe('empty', () => {
it('emits nothing and ends immediately', () => {
const signals = collectSignals(sources.empty);
expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]);
});
});
describe('fromPromise', () => {
passesActiveClose(sources.fromPromise(Promise.resolve(null)));
it('emits a value when the promise resolves', async () => {
const promise = Promise.resolve(1);
const signals = collectSignals(sources.fromPromise(promise));
expect(signals).toEqual([start(expect.any(Function))]);
await Promise.resolve();
await promise;
await Promise.resolve();
expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]);
});
});
describe('fromObservable', () => {
beforeEach(() => {
vi.useRealTimers();
});
it('converts an Observable to a Wonka source', async () => {
const source = observable.fromObservable(Observable.from([1, 2]));
const signals = collectSignals(source);
await new Promise(resolve => setTimeout(resolve));
expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]);
});
it('supports cancellation on converted Observables', async () => {
const source = observable.fromObservable(Observable.from([1, 2]));
const signals = collectSignals(source, talkback => {
talkback(TalkbackKind.Close);
});
await new Promise(resolve => setTimeout(resolve));
expect(signals).toEqual([start(expect.any(Function))]);
});
});
describe('fromCallbag', () => {
it('converts a Callbag to a Wonka source', () => {
const source = callbag.fromCallbag(callbagFromArray([1, 2]) as any);
const signals = collectSignals(source);
expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]);
});
it('supports cancellation on converted Observables', () => {
const source = callbag.fromCallbag(callbagFromArray([1, 2]) as any);
const signals = collectSignals(source, talkback => {
talkback(TalkbackKind.Close);
});
expect(signals).toEqual([start(expect.any(Function))]);
});
});
describe('interval', () => {
it('emits Push signals until Cancel is sent', () => {
let pushes = 0;
let talkback: TalkbackFn | null = null;
const sink: Sink<any> = signal => {
if (signal === SignalKind.End) {
/*noop*/
} else if (signal.tag === SignalKind.Push) {
pushes++;
} else {
talkback = signal[0];
}
};
sources.interval(100)(sink);
expect(talkback).not.toBe(null);
expect(pushes).toBe(0);
vi.advanceTimersByTime(100);
expect(pushes).toBe(1);
vi.advanceTimersByTime(100);
expect(pushes).toBe(2);
talkback!(TalkbackKind.Close);
vi.advanceTimersByTime(100);
expect(pushes).toBe(2);
});
});
describe('fromDomEvent', () => {
it('emits Push signals for events on a DOM element', () => {
let talkback: TalkbackFn | null = null;
const element = {
addEventListener: vi.fn(),
removeEventListener: vi.fn(),
};
const sink: Sink<any> = signal => {
expect(signal).not.toBe(SignalKind.End);
if ((signal as any).tag === SignalKind.Start) talkback = signal[0];
};
sources.fromDomEvent(element as any, 'click')(sink);
expect(element.addEventListener).toHaveBeenCalledWith('click', expect.any(Function));
expect(element.removeEventListener).not.toHaveBeenCalled();
const listener = element.addEventListener.mock.calls[0][1];
listener(1);
listener(2);
talkback!(TalkbackKind.Close);
expect(element.removeEventListener).toHaveBeenCalledWith('click', listener);
});
});
+64
View File
@@ -0,0 +1,64 @@
import { Source, SignalKind } from './types';
import { push, start } from './helpers';
/** A definition of the Callbag type as per its specification.
* @see {@link https://github.com/callbag/callbag} for the Callbag specification.
*/
interface Callbag<I, O> {
(t: 0, d: Callbag<O, I>): void;
(t: 1, d: I): void;
(t: 2, d?: any): void;
}
/** Converts a Callbag to a {@link Source}.
* @param callbag - The {@link Callbag} object that will be converted.
* @returns A {@link Source} wrapping the passed Callbag.
*
* @remarks
* This converts a Callbag to a {@link Source}. When this Source receives a {@link Sink} and
* the subscription starts, internally, it'll subscribe to the passed Callbag, passing through
* all of its emitted values.
*/
export function fromCallbag<T>(callbag: Callbag<any, T>): Source<T> {
return sink => {
callbag(0, (signal: number, data: any) => {
if (signal === 0) {
sink(
start(signal => {
data(signal + 1);
})
);
} else if (signal === 1) {
sink(push(data));
} else {
sink(SignalKind.End);
}
});
};
}
/** Converts a {@link Source} to a Callbag.
* @param source - The {@link Source} that will be converted.
* @returns A {@link Callbag} wrapping the passed Source.
*
* @remarks
* This converts a {@link Source} to a {@link Callbag}. When this Callbag is subscribed to, it
* internally subscribes to the Wonka Source and pulls new values.
*/
export function toCallbag<T>(source: Source<T>): Callbag<any, T> {
return (signal: number, sink: any) => {
if (signal === 0) {
source(signal => {
if (signal === SignalKind.End) {
sink(2);
} else if (signal.tag === SignalKind.Start) {
sink(0, (num: number) => {
if (num < 3) signal[0](num - 1);
});
} else {
sink(1, signal[0]);
}
});
}
};
}
+137
View File
@@ -0,0 +1,137 @@
import { Source, TypeOfSource, SignalKind, TalkbackKind, TalkbackFn } from './types';
import { push, start, talkbackPlaceholder } from './helpers';
type TypeOfSourceArray<T extends readonly [...any[]]> = T extends [infer Head, ...infer Tail]
? [TypeOfSource<Head>, ...TypeOfSourceArray<Tail>]
: [];
/** Combines the latest values of several sources into a Source issuing either tuple or dictionary
* values.
*
* @param sources - Either an array or dictionary object of Sources.
* @returns A {@link Source} issuing a zipped value whenever any input Source updates.
*
* @remarks
* `zip` combines several {@link Source | Sources}. The resulting Source will issue its first value
* once all input Sources have at least issued one value, and will subsequently issue a new value
* each time any of the Sources emits a new value.
*
* Depending on whether an array or dictionary object of Sources is passed to `zip`, its emitted
* values will be arrays or dictionary objects of the Sources' values.
*
* @example
* An example of passing a dictionary object to `zip`. If an array is passed, the resulting
* values will output arrays of the sources' values instead.
*
* ```ts
* pipe(
* zip({
* x: fromValue(1),
* y: fromArray([2, 3]),
* }),
* subscribe(result => {
* // logs { x: 1, y: 2 } then { x: 1, y: 3 }
* console.log(result);
* })
* );
* ```
*/
interface zip {
<Sources extends readonly [...Source<any>[]]>(sources: [...Sources]): Source<
TypeOfSourceArray<Sources>
>;
<Sources extends { [prop: string]: Source<any> }>(sources: Sources): Source<{
[Property in keyof Sources]: TypeOfSource<Sources[Property]>;
}>;
}
function zip<T>(sources: Source<T>[] | Record<string, Source<T>>): Source<T[] | Record<string, T>> {
const size = Object.keys(sources).length;
return sink => {
const filled: Set<string | number> = new Set();
const talkbacks: TalkbackFn[] | Record<string, TalkbackFn | void> = Array.isArray(sources)
? new Array(size).fill(talkbackPlaceholder)
: {};
const buffer: T[] | Record<string, T> = Array.isArray(sources) ? new Array(size) : {};
let gotBuffer = false;
let gotSignal = false;
let ended = false;
let endCount = 0;
for (const key in sources) {
(sources[key] as Source<T>)(signal => {
if (signal === SignalKind.End) {
if (endCount >= size - 1) {
ended = true;
sink(SignalKind.End);
} else {
endCount++;
}
} else if (signal.tag === SignalKind.Start) {
talkbacks[key] = signal[0];
} else if (!ended) {
buffer[key] = signal[0];
filled.add(key);
if (!gotBuffer && filled.size < size) {
if (!gotSignal) {
for (const key in sources)
if (!filled.has(key)) (talkbacks[key] || talkbackPlaceholder)(TalkbackKind.Pull);
} else {
gotSignal = false;
}
} else {
gotBuffer = true;
gotSignal = false;
sink(push(Array.isArray(buffer) ? buffer.slice() : { ...buffer }));
}
}
});
}
sink(
start(signal => {
if (ended) {
/*noop*/
} else if (signal === TalkbackKind.Close) {
ended = true;
for (const key in talkbacks) talkbacks[key](TalkbackKind.Close);
} else if (!gotSignal) {
gotSignal = true;
for (const key in talkbacks) talkbacks[key](TalkbackKind.Pull);
}
})
);
};
}
export { zip };
/** Combines the latest values of all passed sources into a Source issuing tuple values.
*
* @see {@link zip | `zip`} which this helper wraps and uses.
* @param sources - A variadic list of {@link Source} parameters.
* @returns A {@link Source} issuing a zipped value whenever any input Source updates.
*
* @remarks
* `combine` takes one or more {@link Source | Sources} as arguments. Once all input Sources have at
* least issued one value it will issue an array of all of the Sources' values. Subsequently, it
* will issue a new array value whenever any of the Sources update.
*
* @example
*
* ```ts
* pipe(
* combine(fromValue(1), fromValue(2)),
* subscribe(result => {
* console.log(result); // logs [1, 2]
* })
* );
* ```
*/
export function combine<Sources extends Source<any>[]>(
...sources: Sources
): Source<TypeOfSourceArray<Sources>> {
return zip(sources) as Source<any>;
}
+62
View File
@@ -0,0 +1,62 @@
import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types';
/** Placeholder {@link TeardownFn | teardown functions} that's a no-op.
* @see {@link TeardownFn} for the definition and usage of teardowns.
* @internal
*/
export const teardownPlaceholder: TeardownFn = () => {
/*noop*/
};
/** Placeholder {@link TalkbackFn | talkback function} that's a no-op.
* @privateRemarks
* This is frequently used in the codebase as a no-op initializer value for talkback functions in
* the implementation of {@link Operator | Operators}. This is cheaper than initializing the
* variables of talkbacks to `undefined` or `null` and performing an extra check before calling
* them. Since the {@link Start | Start signal} is assumed to come first and carry a talkback, we can
* use this to our advantage and use a no-op placeholder before {@link Start} is received.
*
* @internal
*/
export const talkbackPlaceholder: TalkbackFn = teardownPlaceholder;
/** Wraps the passed {@link TalkbackFn | talkback function} in a {@link Start | Start signal}.
* @internal
*/
export function start<T>(talkback: TalkbackFn): Start<T> {
return {
tag: SignalKind.Start,
0: talkback,
} as Start<T>;
}
/** Wraps the passed value in a {@link Push | Push signal}.
* @internal
*/
export function push<T>(value: T): Push<T> {
return {
tag: SignalKind.Push,
0: value,
} as Push<T>;
}
/** Returns the well-known symbol specifying the default AsyncIterator.
* @internal
*/
export const asyncIteratorSymbol = (): typeof Symbol.asyncIterator =>
(typeof Symbol === 'function' && Symbol.asyncIterator) || ('@@asyncIterator' as any);
/** Returns the well-known symbol specifying the default ES Observable.
* @privateRemarks
* This symbol is used to mark an object as a default ES Observable. By the specification, an object
* that abides by the default Observable implementation must carry a method set to this well-known
* symbol that returns the Observable implementation. It's common for this object to be an
* Observable itself and return itself on this method.
*
* @see {@link https://github.com/0no-co/wonka/issues/122} for notes on the intercompatibility
* between Observable implementations.
*
* @internal
*/
export const observableSymbol = (): typeof Symbol.observable =>
(typeof Symbol === 'function' && Symbol.observable) || ('@@observable' as any);
+31
View File
@@ -0,0 +1,31 @@
/**
* A tiny but capable push & pull stream library for TypeScript and Flow.
*
* @remarks
* Wonka is a lightweight iterable and observable library and exposes a set of helpers to create
* streams, which are sources emitting multiple values, which allow you to create, transform, and
* consume event streams or iterable sets of data.
*
* It's loosely based on the Callbag spec: {@link https://github.com/callbag/callbag}
* @packageDocumentation
*/
export type {
TeardownFn,
Signal,
Sink,
Source,
Operator,
TypeOfSource,
Subscription,
Observer,
Subject,
} from './types';
export * from './sources';
export * from './operators';
export * from './sinks';
export * from './combine';
export * from './observable';
export * from './callbag';
export * from './pipe';
+207
View File
@@ -0,0 +1,207 @@
import { Source, SignalKind, TalkbackKind } from './types';
import { push, start, talkbackPlaceholder, observableSymbol } from './helpers';
// NOTE: This must be placed in an exported file for `rollup-plugin-dts`
// to include it in output typings files
declare global {
interface SymbolConstructor {
readonly observable: symbol;
}
}
/** A definition of the ES Observable Subscription type that is returned by
* {@link Observable.subscribe}
*
* @remarks
* The Subscription in ES Observables is a handle that is held while the Observable is actively
* streaming values. As such, it's used to indicate with {@link ObservableSubscription.closed}
* whether it's active, and {@link ObservableSubscription.unsubscribe} may be used to cancel the
* ongoing subscription and end the {@link Observable} early.
*
* @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification.
*/
interface ObservableSubscription {
/** A boolean flag indicating whether the subscription is closed.
* @remarks
* When `true`, the subscription will not issue new values to the {@link ObservableObserver} and
* has terminated. No new values are expected.
*
* @readonly
*/
closed: boolean;
/** Cancels the subscription.
* @remarks
* This cancels the ongoing subscription and the {@link ObservableObserver}'s callbacks will
* subsequently not be called at all. The subscription will be terminated and become inactive.
*/
unsubscribe(): void;
}
/** A definition of the ES Observable Observer type that is used to receive data from an
* {@link Observable}.
*
* @remarks
* The Observer in ES Observables is supplied to {@link Observable.subscribe} to receive events from
* an {@link Observable} as it issues them.
*
* @see {@link https://github.com/tc39/proposal-observable#observer} for the ES Observable
* specification of an Observer.
*/
interface ObservableObserver<T> {
/** Callback for the Observable issuing new values.
* @param value - The value that the {@link Observable} is sending.
*/
next(value: T): void;
/** Callback for the Observable encountering an error, terminating it.
* @param error - The error that the {@link Observable} has encountered.
*/
error?(error: any): void;
/** Callback for the Observable ending, after all values have been issued. */
complete?(): void;
}
/** A looser definition of ES Observable-like types that is used for interoperability.
* @remarks
* The Observable is often used by multiple libraries supporting or creating streams to provide
* interoperability for push-based streams. When converting from an Observable to a {@link Source},
* this looser type is accepted as an input.
*
* @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification.
* @see {@link Observable} for the full ES Observable type.
*/
interface ObservableLike<T> {
/**
* Subscribes to new signals from an {@link Observable} via callbacks.
* @param observer - An object containing callbacks for the various events of an Observable.
* @returns Subscription handle of type {@link ObservableSubscription}.
*
* @see {@link ObservableObserver} for the callbacks in an object that are called as Observables
* issue events.
*/
subscribe(observer: ObservableObserver<T>): { unsubscribe(): void };
/** The well-known symbol specifying the default ES Observable for an object. */
[Symbol.observable]?(): Observable<T>;
}
/** An ES Observable type that is a de-facto standard for push-based data sources across the JS
* ecosystem.
*
* @remarks
* The Observable is often used by multiple libraries supporting or creating streams to provide
* interoperability for push-based streams. As Wonka's {@link Source | Sources} are similar in
* functionality to Observables, it provides utilities to cleanly convert to and from Observables.
*
* @see {@link https://github.com/tc39/proposal-observable} for the ES Observable specification.
*/
interface Observable<T> {
/** Subscribes to new signals from an {@link Observable} via callbacks.
* @param observer - An object containing callbacks for the various events of an Observable.
* @returns Subscription handle of type {@link ObservableSubscription}.
*
* @see {@link ObservableObserver} for the callbacks in an object that are called as Observables
* issue events.
*/
subscribe(observer: ObservableObserver<T>): ObservableSubscription;
/** Subscribes to new signals from an {@link Observable} via callbacks.
* @param onNext - Callback for the Observable issuing new values.
* @param onError - Callback for the Observable encountering an error, terminating it.
* @param onComplete - Callback for the Observable ending, after all values have been issued.
* @returns Subscription handle of type {@link ObservableSubscription}.
*/
subscribe(
onNext: (value: T) => any,
onError?: (error: any) => any,
onComplete?: () => any
): ObservableSubscription;
/** The well-known symbol specifying the default ES Observable for an object. */
[Symbol.observable](): Observable<T>;
}
/** Converts an ES Observable to a {@link Source}.
* @param input - The {@link ObservableLike} object that will be converted.
* @returns A {@link Source} wrapping the passed Observable.
*
* @remarks
* This converts an ES Observable to a {@link Source}. When this Source receives a {@link Sink} and
* the subscription starts, internally, it'll subscribe to the passed Observable, passing through
* all of the Observable's values. As such, this utility provides intercompatibility converting from
* standard Observables to Wonka Sources.
*
* @throws
* When the passed ES Observable throws, the error is simply re-thrown as {@link Source} does
* not support or expect errors to be handled by streams.
*/
export function fromObservable<T>(input: ObservableLike<T>): Source<T> {
return sink => {
const subscription = (
input[observableSymbol()] ? input[observableSymbol()]!() : input
).subscribe({
next(value: T) {
sink(push(value));
},
complete() {
sink(SignalKind.End);
},
error(error) {
throw error;
},
});
sink(
start(signal => {
if (signal === TalkbackKind.Close) subscription.unsubscribe();
})
);
};
}
/** Converts a {@link Source} to an ES Observable.
* @param source - The {@link Source} that will be converted.
* @returns An {@link Observable} wrapping the passed Source.
*
* @remarks
* This converts a {@link Source} to an {@link Observable}. When this Observable is subscribed to, it
* internally subscribes to the Wonka Source and pulls new values. As such, this utility provides
* intercompatibility converting from Wonka Sources to standard ES Observables.
*/
export function toObservable<T>(source: Source<T>): Observable<T> {
return {
subscribe(
next: ObservableObserver<T> | ((value: T) => any),
error?: (error: any) => any | undefined,
complete?: () => any | undefined
) {
const observer: ObservableObserver<T> =
typeof next == 'object' ? next : { next, error, complete };
let talkback = talkbackPlaceholder;
let ended = false;
source(signal => {
if (ended) {
/*noop*/
} else if (signal === SignalKind.End) {
ended = true;
if (observer.complete) observer.complete();
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
} else {
observer.next(signal[0]);
talkback(TalkbackKind.Pull);
}
});
const subscription = {
closed: false,
unsubscribe() {
subscription.closed = true;
ended = true;
talkback(TalkbackKind.Close);
},
};
return subscription;
},
[observableSymbol()]() {
return this;
},
};
}
File diff suppressed because it is too large Load Diff
+185
View File
@@ -0,0 +1,185 @@
import { Source, Sink, Operator } from './types';
interface UnaryFn<T, R> {
(source: T): R;
}
/** Chain calls operators on a given source and returns the last result.
* @param args - A source, then a variable number of transform functions
*
* @remarks
* The `pipe` utility can be called with a {@link Source} then one or more unary transform functions.
* Each transform function will be called in turn with the last function's return value, starting
* with the source passed as the first argument to `pipe`.
*
* It's used to transform a source with a list of {@link Operator | Operators}. The last argument may
* also be a {@link Sink} that returns something else than a Source.
*
* @example
*
* ```ts
* pipe(
* fromArray([1, 2, 3]),
* map(x => x * 2),
* subscribe(console.log)
* );
* ```
*
* @see {@link https://github.com/tc39/proposal-pipeline-operator} for the JS Pipeline Operator spec, for which this is a replacement utility for.
*/
interface pipe {
/* pipe definitions for source + operators composition */
<T, A>(source: Source<T>, op1: UnaryFn<Source<T>, Source<A>>): Source<A>;
<T, A, B>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>
): Source<B>;
<T, A, B, C>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>
): Source<C>;
<T, A, B, C, D>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>
): Source<D>;
<T, A, B, C, D, E>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>
): Source<E>;
<T, A, B, C, D, E, F>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>
): Source<F>;
<T, A, B, C, D, E, F, G>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>
): Source<G>;
<T, A, B, C, D, E, F, G, H>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
op8: UnaryFn<Source<G>, Source<H>>
): Source<H>;
/* pipe definitions for source + operators + consumer composition */
<T, R>(source: Source<T>, consumer: UnaryFn<Source<T>, R>): R;
<T, A, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
consumer: UnaryFn<Source<A>, R>
): R;
<T, A, B, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
consumer: UnaryFn<Source<B>, R>
): R;
<T, A, B, C, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
consumer: UnaryFn<Source<C>, R>
): R;
<T, A, B, C, D, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
consumer: UnaryFn<Source<D>, R>
): R;
<T, A, B, C, D, E, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
consumer: UnaryFn<Source<E>, R>
): R;
<T, A, B, C, D, E, F, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
consumer: UnaryFn<Source<F>, R>
): R;
<T, A, B, C, D, E, F, G, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
consumer: UnaryFn<Source<G>, R>
): R;
<T, A, B, C, D, E, F, G, H, R>(
source: Source<T>,
op1: UnaryFn<Source<T>, Source<A>>,
op2: UnaryFn<Source<A>, Source<B>>,
op3: UnaryFn<Source<B>, Source<C>>,
op4: UnaryFn<Source<C>, Source<D>>,
op5: UnaryFn<Source<D>, Source<E>>,
op6: UnaryFn<Source<E>, Source<F>>,
op7: UnaryFn<Source<F>, Source<G>>,
op8: UnaryFn<Source<G>, Source<H>>,
consumer: UnaryFn<Source<H>, R>
): R;
}
const pipe: pipe = (...args: Function[]): any => {
let x = args[0];
for (let i = 1, l = args.length; i < l; i++) x = args[i](x);
return x;
};
export { pipe };
+253
View File
@@ -0,0 +1,253 @@
import { Source, Subscription, TalkbackKind, SignalKind, SourceIterable } from './types';
import { talkbackPlaceholder, asyncIteratorSymbol } from './helpers';
/** Creates a subscription to a given source and invokes a `subscriber` callback for each value.
* @param subscriber - A callback function called for each issued value.
* @returns A function accepting a {@link Source} and returning a {@link Subscription}.
*
* @remarks
* `subscribe` accepts a `subscriber` callback and returns a function accepting a {@link Source}.
* When a source is passed to the returned funtion, the subscription will start and `subscriber`
* will be called for each new value the Source issues. This will also return a {@link Subscription}
* object that can cancel the ongoing {@link Source} early.
*
* @example
* ```ts
* const subscription = pipe(
* fromValue('test'),
* subscribe(text => {
* console.log(text); // 'test'
* })
* );
* ```
*/
export function subscribe<T>(subscriber: (value: T) => void) {
return (source: Source<T>): Subscription => {
let talkback = talkbackPlaceholder;
let ended = false;
source(signal => {
if (signal === SignalKind.End) {
ended = true;
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
} else if (!ended) {
subscriber(signal[0]);
talkback(TalkbackKind.Pull);
}
});
return {
unsubscribe() {
if (!ended) {
ended = true;
talkback(TalkbackKind.Close);
}
},
};
};
}
/** Creates a subscription to a given source and invokes a `subscriber` callback for each value.
* @see {@link subscribe} which this helper aliases without returnin a {@link Subscription}.
* @param subscriber - A callback function called for each issued value.
* @returns A function accepting a {@link Source}.
*
* @remarks
* `forEach` accepts a `subscriber` callback and returns a function accepting a {@link Source}.
* When a source is passed to the returned funtion, the subscription will start and `subscriber`
* will be called for each new value the Source issues. Unlike `subscribe` it will not return a
* Subscription object and can't be cancelled early.
*
* @example
* ```ts
* pipe(
* fromValue('test'),
* forEach(text => {
* console.log(text); // 'test'
* })
* ); // undefined
* ```
*/
export function forEach<T>(subscriber: (value: T) => void) {
return (source: Source<T>): void => {
subscribe(subscriber)(source);
};
}
/** Creates a subscription to a given source and invokes a `subscriber` callback for each value.
* @see {@link subscribe} which this helper aliases without accepting parameters or returning a
* {@link Subscription | Subscription}.
*
* @param source - A {@link Source}.
*
* @remarks
* `publish` accepts a {@link Source} and subscribes to it, starting its values. The resulting
* values cannot be observed and the subscription can't be cancelled, as this helper is purely
* intended to start side-effects.
*
* @example
* ```ts
* pipe(
* lazy(() => {
* console.log('test'); // this is called
* return fromValue(123); // this is never used
* }),
* publish
* ); // undefined
* ```
*/
export function publish<T>(source: Source<T>): void {
subscribe(_value => {
/*noop*/
})(source);
}
const doneResult = { done: true } as IteratorReturnResult<void>;
/** Converts a Source to an AsyncIterable that pulls and issues values from the Source.
*
* @param source - A {@link Source}.
* @returns An {@link AsyncIterable | `AsyncIterable`} issuing values from the Source.
*
* @remarks
* `toAsyncIterable` will create an {@link AsyncIterable} that pulls and issues values from a given
* {@link Source}. This can be used in many interoperability situations, to provide an iterable when
* a consumer requires it.
*
* @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols}
* for the JS Iterable protocol.
*
* @example
* ```ts
* const iterable = toAsyncIterable(fromArray([1, 2, 3]));
* for await (const value of iterable) {
* console.log(value); // outputs: 1, 2, 3
* }
* ```
*/
export const toAsyncIterable = <T>(source: Source<T>): SourceIterable<T> => {
const buffer: T[] = [];
let ended = false;
let started = false;
let pulled = false;
let talkback = talkbackPlaceholder;
let next: ((value: IteratorResult<T>) => void) | void;
return {
async next(): Promise<IteratorResult<T>> {
if (!started) {
started = true;
source(signal => {
if (ended) {
/*noop*/
} else if (signal === SignalKind.End) {
if (next) next = next(doneResult);
ended = true;
} else if (signal.tag === SignalKind.Start) {
pulled = true;
(talkback = signal[0])(TalkbackKind.Pull);
} else {
pulled = false;
if (next) {
next = next({ value: signal[0], done: false });
} else {
buffer.push(signal[0]);
}
}
});
}
if (ended && !buffer.length) {
return doneResult;
} else if (!ended && !pulled && buffer.length <= 1) {
pulled = true;
talkback(TalkbackKind.Pull);
}
return buffer.length
? { value: buffer.shift()!, done: false }
: new Promise(resolve => (next = resolve));
},
async return(): Promise<IteratorReturnResult<void>> {
if (!ended) next = talkback(TalkbackKind.Close);
ended = true;
return doneResult;
},
[asyncIteratorSymbol()](): SourceIterable<T> {
return this;
},
};
};
/** Subscribes to a given source and collects all synchronous values into an array.
* @param source - A {@link Source}.
* @returns An array of values collected from the {@link Source}.
*
* @remarks
* `toArray` accepts a {@link Source} and returns an array of all synchronously issued values from
* this Source. It will issue {@link TalkbackKind.Pull | Pull signals} after every value it receives
* and expects the Source to recursively issue values.
*
* Any asynchronously issued values will not be
* added to the array and a {@link TalkbackKind.Close | Close signal} is issued by the sink before
* returning the array.
*
* @example
* ```ts
* toArray(fromArray([1, 2, 3])); // [1, 2, 3]
* ```
*/
export function toArray<T>(source: Source<T>): T[] {
const values: T[] = [];
let talkback = talkbackPlaceholder;
let ended = false;
source(signal => {
if (signal === SignalKind.End) {
ended = true;
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
} else {
values.push(signal[0]);
talkback(TalkbackKind.Pull);
}
});
if (!ended) talkback(TalkbackKind.Close);
return values;
}
/** Subscribes to a given source and returns a Promise that will resolve with the last value the
* source issues.
*
* @param source - A {@link Source}.
* @returns A {@link Promise} resolving to the last value of the {@link Source}.
*
* @remarks
* `toPromise` will subscribe to the passed {@link Source} and resolve to the last value of it once
* it receives the last value, as signaled by the {@link SignalKind.End | End signal}.
*
* To keep its implementation simple, padding sources that don't issue any values to `toPromise` is
* undefined behaviour and `toPromise` will issue `undefined` in that case.
*
* The returned {@link Promise} delays its value by a microtick, using `Promise.resolve`.
*
* @example
* ```ts
* toPromise(fromValue('test')); // resolves: 'test'
* ```
*/
export function toPromise<T>(source: Source<T>): Promise<T> {
return new Promise(resolve => {
let talkback = talkbackPlaceholder;
let value: T | void;
source(signal => {
if (signal === SignalKind.End) {
Promise.resolve(value!).then(resolve);
} else if (signal.tag === SignalKind.Start) {
(talkback = signal[0])(TalkbackKind.Pull);
} else {
value = signal[0];
talkback(TalkbackKind.Pull);
}
});
});
}
+407
View File
@@ -0,0 +1,407 @@
import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types';
import {
push,
start,
talkbackPlaceholder,
teardownPlaceholder,
asyncIteratorSymbol,
} from './helpers';
import { share } from './operators';
/** Helper creating a Source from a factory function when it's subscribed to.
* @param produce - A factory function returning a {@link Source}.
* @returns A {@link Source} lazyily subscribing to the Source returned by the given factory
* function.
*
* @remarks
* At times it's necessary to create a {@link Source} lazily. The time of a {@link Source} being
* created could be different from when it's subscribed to, and hence we may want to split the
* creation and subscription time. This is especially useful when the Source we wrap is "hot" and
* issues values as soon as it's created, which we may then not receive in a subscriber.
*
* @example An example of creating a {@link Source} that issues the timestamp of subscription. Here
* we effectively use `lazy` with the simple {@link fromValue | `fromValue`} source, to quickly
* create a Source that issues the time of its subscription, rather than the time of its creation
* that it would otherwise issue without `lazy`.
*
* ```ts
* lazy(() => fromValue(Date.now()));
* ```
*/
export function lazy<T>(produce: () => Source<T>): Source<T> {
return sink => produce()(sink);
}
/** Converts an AsyncIterable to a Source that pulls and issues values from it as requested.
*
* @see {@link fromIterable | `fromIterable`} for the non-async Iterable version of this helper,
* which calls this helper automatically as needed.
*
* @param iterable - An {@link AsyncIterable | `AsyncIterable`}.
* @returns A {@link Source} issuing values sourced from the Iterable.
*
* @remarks
* `fromAsyncIterable` will create a {@link Source} that pulls and issues values from a given
* {@link AsyncIterable}. This can be used in many interoperability situations, including to consume
* an async generator function.
*
* When the {@link Sink} throws an exception when a new value is pushed, this helper will rethrow it
* using {@link AsyncIterator.throw}, which allows an async generator to recover from the exception.
*
* @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols}
* for the JS Iterable protocol.
*/
export function fromAsyncIterable<T>(iterable: AsyncIterable<T> | AsyncIterator<T>): Source<T> {
return sink => {
const iterator: AsyncIterator<T> =
(iterable[asyncIteratorSymbol()] && iterable[asyncIteratorSymbol()]()) || iterable;
let ended = false;
let looping = false;
let pulled = false;
let next: IteratorResult<T>;
sink(
start(async signal => {
if (signal === TalkbackKind.Close) {
ended = true;
if (iterator.return) iterator.return();
} else if (looping) {
pulled = true;
} else {
for (pulled = looping = true; pulled && !ended; ) {
if ((next = await iterator.next()).done) {
ended = true;
if (iterator.return) await iterator.return();
sink(SignalKind.End);
} else {
try {
pulled = false;
sink(push(next.value));
} catch (error) {
if (iterator.throw) {
if ((ended = !!(await iterator.throw(error)).done)) sink(SignalKind.End);
} else {
throw error;
}
}
}
}
looping = false;
}
})
);
};
}
/** Converts an Iterable to a Source that pulls and issues values from it as requested.
* @see {@link fromAsyncIterable | `fromAsyncIterable`} for the AsyncIterable version of this helper.
* @param iterable - An {@link Iterable | `Iterable`} or an `AsyncIterable`
* @returns A {@link Source} issuing values sourced from the Iterable.
*
* @remarks
* `fromIterable` will create a {@link Source} that pulls and issues values from a given
* {@link Iterable | JS Iterable}. As iterables are the common standard for any lazily iterated list
* of values in JS it can be applied to many different JS data types, including a JS Generator
* function.
*
* This Source will only call {@link Iterator.next} on the iterator when the subscribing {@link Sink}
* has pulled a new value with the {@link TalkbackKind.Pull | Pull signal}. `fromIterable` can
* therefore also be applied to "infinite" iterables, without a predefined end.
*
* This helper will call {@link fromAsyncIterable | `fromAsyncIterable`} automatically when the
* passed object also implements the async iterator protocol.
*
* When the {@link Sink} throws an exception when a new value is pushed, this helper will rethrow it
* using {@link Iterator.throw}, which allows a generator to recover from the exception.
*
* @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_iterable_protocol}
* for the JS Iterable protocol.
*/
export function fromIterable<T>(iterable: Iterable<T> | AsyncIterable<T>): Source<T> {
if (iterable[Symbol.asyncIterator]) return fromAsyncIterable(iterable as AsyncIterable<T>);
return sink => {
const iterator = iterable[Symbol.iterator]();
let ended = false;
let looping = false;
let pulled = false;
let next: IteratorResult<T>;
sink(
start(signal => {
if (signal === TalkbackKind.Close) {
ended = true;
if (iterator.return) iterator.return();
} else if (looping) {
pulled = true;
} else {
for (pulled = looping = true; pulled && !ended; ) {
if ((next = iterator.next()).done) {
ended = true;
if (iterator.return) iterator.return();
sink(SignalKind.End);
} else {
try {
pulled = false;
sink(push(next.value));
} catch (error) {
if (iterator.throw) {
if ((ended = !!iterator.throw(error).done)) sink(SignalKind.End);
} else {
throw error;
}
}
}
}
looping = false;
}
})
);
};
}
/** Creates a Source that issues a each value of a given array synchronously.
* @see {@link fromIterable} which `fromArray` aliases.
* @param array - The array whose values will be issued one by one.
* @returns A {@link Source} issuing the array's values.
*
* @remarks
* `fromArray` will create a {@link Source} that issues the values of a given JS array one by one. It
* will issue values as they're pulled and is hence a "cold" source, not eagerly emitting values. It
* will end and issue the {@link SignalKind.End | End signal} when the array is exhausted of values.
*
* @example
* ```ts
* fromArray([1, 2, 3]);
* ```
*/
export const fromArray: <T>(array: T[]) => Source<T> = fromIterable;
/** Creates a Source that issues a single value and ends immediately after.
* @param value - The value that will be issued.
* @returns A {@link Source} issuing the single value.
*
* @example
* ```ts
* fromValue('test');
* ```
*/
export function fromValue<T>(value: T): Source<T> {
return sink => {
let ended = false;
sink(
start(signal => {
if (signal === TalkbackKind.Close) {
ended = true;
} else if (!ended) {
ended = true;
sink(push(value));
sink(SignalKind.End);
}
})
);
};
}
/** Creates a new Source from scratch from a passed `subscriber` function.
* @param subscriber - A callback that is called when the {@link Source} is subscribed to.
* @returns A {@link Source} created from the `subscriber` parameter.
*
* @remarks
* `make` is used to create a new, arbitrary {@link Source} from scratch. It calls the passed
* `subscriber` function when it's subscribed to.
*
* The `subscriber` function receives an {@link Observer}. You may call {@link Observer.next} to
* issue values on the Source, and {@link Observer.complete} to end the Source.
*
* Your `subscribr` function must return a {@link TeardownFn | teardown function} which is only
* called when your source is cancelled — not when you invoke `complete` yourself. As this creates a
* "cold" source, every time this source is subscribed to, it will invoke the `subscriber` function
* again and create a new source.
*
* @example
*
* ```ts
* make(observer => {
* const frame = requestAnimationFrame(() => {
* observer.next('animate!');
* });
* return () => {
* cancelAnimationFrame(frame);
* };
* });
* ```
*/
export function make<T>(subscriber: (observer: Observer<T>) => TeardownFn): Source<T> {
return sink => {
let ended = false;
const teardown = subscriber({
next(value: T) {
if (!ended) sink(push(value));
},
complete() {
if (!ended) {
ended = true;
sink(SignalKind.End);
}
},
});
sink(
start(signal => {
if (signal === TalkbackKind.Close && !ended) {
ended = true;
teardown();
}
})
);
};
}
/** Creates a new Subject which can be used as an IO event hub.
* @returns A new {@link Subject}.
*
* @remarks
* `makeSubject` creates a new {@link Subject}. A Subject is a {@link Source} and an {@link Observer}
* combined in one interface, as the Observer is used to send new signals to the Source. This means
* that it's "hot" and hence all subscriptions to {@link Subject.source} share the same underlying
* signals coming from {@link Subject.next} and {@link Subject.complete}.
*
* @example
* ```ts
* const subject = makeSubject();
* pipe(subject.source, subscribe(console.log));
* // This will log the string on the above subscription
* subject.next('hello subject!');
* ```
*/
export function makeSubject<T>(): Subject<T> {
let next: Subject<T>['next'] | void;
let complete: Subject<T>['complete'] | void;
return {
source: share(
make(observer => {
next = observer.next;
complete = observer.complete;
return teardownPlaceholder;
})
),
next(value: T) {
if (next) next(value);
},
complete() {
if (complete) complete();
},
};
}
/** A {@link Source} that immediately ends.
* @remarks
* `empty` is a {@link Source} that immediately issues an {@link SignalKind.End | End signal} when
* it's subscribed to, ending immediately.
*
* @see {@link never | `never`} for a source that instead never ends.
*/
export const empty: Source<any> = (sink: Sink<any>): void => {
let ended = false;
sink(
start(signal => {
if (signal === TalkbackKind.Close) {
ended = true;
} else if (!ended) {
ended = true;
sink(SignalKind.End);
}
})
);
};
/** A {@link Source} without values that never ends.
* @remarks
* `never` is a {@link Source} that never issues any signals and neither sends values nor ends.
*
* @see {@link empty | `empty`} for a source that instead ends immediately.
*/
export const never: Source<any> = (sink: Sink<any>): void => {
sink(start(talkbackPlaceholder));
};
/** Creates a Source that issues an incrementing integer in intervals.
* @param ms - The interval in milliseconds.
* @returns A {@link Source} issuing an incrementing count on each interval.
*
* @remarks
* `interval` will create a {@link Source} that issues an incrementing counter each time the `ms`
* interval expires.
*
* It'll only stop when it's cancelled by a {@link TalkbackKind.Close | Close signal}.
*
* @example
* An example printing `0`, then `1`, and so on, in intervals of 50ms.
*
* ```ts
* pipe(interval(50), subscribe(console.log));
* ```
*/
export function interval(ms: number): Source<number> {
return make(observer => {
let i = 0;
const id = setInterval(() => observer.next(i++), ms);
return () => clearInterval(id);
});
}
/** Converts DOM Events to a Source given an `HTMLElement` and an event's name.
* @param element - The {@link HTMLElement} to listen to.
* @param event - The DOM Event name to listen to.
* @returns A {@link Source} issuing the {@link Event | DOM Events} as they're issued by the DOM.
*
* @remarks
* `fromDomEvent` will create a {@link Source} that listens to the given element's events and issues
* them as values on the source. This source will only stop when it's cancelled by a
* {@link TalkbackKind.Close | Close signal}.
*
* @example
* An example printing `'clicked!'` when the given `#root` element is clicked.
*
* ```ts
* const element = document.getElementById('root');
* pipe(
* fromDomEvent(element, 'click'),
* subscribe(() => console.log('clicked!'))
* );
* ```
*/
export function fromDomEvent(element: HTMLElement, event: string): Source<Event> {
return make(observer => {
element.addEventListener(event, observer.next);
return () => element.removeEventListener(event, observer.next);
});
}
/** Converts a Promise to a Source that issues the resolving Promise's value and then ends.
* @param promise - The promise that will be wrapped.
* @returns A {@link Source} issuing the promise's value when it resolves.
*
* @remarks
* `fromPromise` will create a {@link Source} that issues the {@link Promise}'s resolving value
* asynchronously and ends immediately after resolving.
*
* This helper will not handle the promise's exceptions, and will cause uncaught errors if the
* promise rejects without a value.
*
* @example
* An example printing `'resolved!'` when the given promise resolves after a tick.
*
* ```ts
* pipe(fromPromise(Promise.resolve('resolved!')), subscribe(console.log));
* ```
*/
export function fromPromise<T>(promise: Promise<T>): Source<T> {
return make(observer => {
promise.then(value => {
Promise.resolve(value).then(() => {
observer.next(value);
observer.complete();
});
});
return teardownPlaceholder;
});
}
+207
View File
@@ -0,0 +1,207 @@
/**
* Talkback signal that sends instructions from a sink to a source.
*
* @remarks
* This signal is issued via {@link TalkbackFn | talkback functions} that a {@link Sink} receives via
* the {@link Start} signal, to tell a {@link Source} to either send a new value (pulling) or stop
* sending values altogether (cancellation).
*/
export declare enum TalkbackKind {
/** Instructs the {@link Source} to send the next value. */
Pull = 0,
/** Instructs the {@link Source} to stop sending values and cancels it. */
Close = 1,
}
/**
* Talkback callback that sends instructions to a source.
*
* @remarks
* This function sends a {@link TalkbackKind} signal to the source to instruct it to send a new value
* (pulling) or to be cancelled and stop sending values altogether.
*/
export type TalkbackFn = (signal: TalkbackKind) => void;
/**
* Callback that is called when a source is cancelled.
*
* @remarks
* This is used, in particular, in the {@link make | make Source} and is a returned function that is
* called when the {@link TalkbackKind.Close} signal is received by the source.
*/
export type TeardownFn = () => void;
/**
* Tag enum that is used to on signals that are sent from a source to a sink.
*
* @remarks
* This signal is issued by a {@link Source} and {@link Sink | Sinks} are called with it. The signals
* carrying values ({@link Start} and {@link Push}) are sent as a unary `[T]` tuple tagged with
* {@link Tag}. The {@link End} signal carries no value and is sent as a raw `0` value.
* @see {@link Start} for the data structure of the start signal.
* @see {@link Push} for the data structure of the push signal, carrying values.
*/
export declare enum SignalKind {
/**
* Informs the {@link Sink} that it's being called by a {@link Source}.
*
* @remarks
* This starts the stream of values and carries a {@link TalkbackFn | talkback function} with it
* that is used by the {@link Sink} to communicate back to the {@link Source}.
* @see {@link Start} for the data structure of the signal.
*/
Start = 0,
/**
* Informs the {@link Sink} of a new values that's incoming from the {@link Source}.
*
* @remarks
* This informs the {@link Sink} of new values that are sent by the {@link Source}.
* @see {@link Push} for the data structure of the signal.
*/
Push = 1,
/**
* Informs the {@link Sink} that the {@link Source} has ended and that it won't send more values.
*
* @remarks
* This signal signifies that the stream has stopped and that no more values are expected. Some
* sources don't have a set end or limit on how many values will be sent. This signal is not sent
* when the {@link Source} is cancelled with a {@link TalkbackKind.Close | Close talkback signal}.
*/
End = 0,
}
/**
* The tag property that's put on unary `[T]` tuple to turn them into signals carrying values.
*
* @internal
*/
export interface Tag<T> {
tag: T;
}
/**
* Indicates the start of a stream to a {@link Sink}.
*
* @remarks
* This signal is sent from a {@link Source} to a {@link Sink} at the start of a stream to inform it
* that values can be pulled and/or will be sent. This signal carries a
* {@link TalkbackFn | talkback function} that is used by the {@link Sink} to communicate back to the
* {@link Source} as a callback. The talkback accepts {@link TalkbackKind.Pull | Pull} and
* {@link TalkbackKind.Close | Close} signals.
*/
export type Start<_T> = Tag<SignalKind.Start> & [TalkbackFn];
/**
* Sends a new value to a {@link Sink}.
*
* @remarks
* This signal is sent from a {@link Source} to a {@link Sink} to send a new value to it. This is
* essentially the signal that wraps new values coming in, like an event. Values are carried on
* unary tuples and can be accessed using `signal[0]`.
*/
export type Push<T> = Tag<SignalKind.Push> & [T];
/**
* Signals are sent from {@link Source | Sources} to {@link Sink | Sinks} to inform them of changes.
*
* @remarks
* A {@link Source}, when consumed, sends a sequence of events to {@link Sink | Sinks}. In order, a
* {@link SignalKind.Start | Start} signal will always be sent first, followed optionally by one or
* more {@link SignalKind.Push | Push signals}, carrying values and representing the stream. A
* {@link Source} will send the {@link SignalKind.End | End signal} when it runs out of values. The
* End signal will be omitted if the Source is cancelled by a
* {@link TalkbackKind.Close | Close signal}, sent back from the {@link Sink}.
* @see {@link SignalKind} for the kinds signals sent by {@link Source | Sources}.
* @see {@link Start} for the data structure of the start signal.
* @see {@link Push} for the data structure of the push signal.
*/
export type Signal<T> = Start<T> | Push<T> | SignalKind.End;
/**
* Callback function that is called by a {@link Source} with {@link Signal | Signals}.
*
* @remarks
* A Sink is a function that is called repeatedly with signals from a {@link Source}. It represents
* the receiver of the stream of signals/events coming from a {@link Source}.
* @see {@link Signal} for the data structure of signals.
*/
export type Sink<T> = (signal: Signal<T>) => void;
/** Factory function that calls {@link Sink | Sinks} with {@link Signal | Signals} when invoked.
* @remarks
* A Source is a factory function that when invoked with a {@link Sink}, calls it with
* {@link Signal | Signals} to create a stream of events, informing it of new values and the
* potential end of the stream of values. The first signal a Source sends is always a
* {@link Start | Start signal} that sends a talkback function to the {@link Sink}, so it may request
* new values or cancel the source.
*
* @see {@link Signal} for the data structure of signals.
* @see {@link Sink} for the data structure of sinks.
*/
export type Source<T> = (sink: Sink<T>) => void;
/** Transform function that accepts a {@link Source} and returns a new one.
* @remarks
* Wonka comes with several helper operators that transform a given {@link Source} into a new one,
* potentially changing its outputs, or the outputs' timing. An "operator" in Wonka typically
* accepts arguments and then returns this kind of function, so they can be chained and composed.
*
* @see {@link pipe | `pipe`} for the helper used to compose operators.
*/
export type Operator<In, Out> = (a: Source<In>) => Source<Out>;
/** Type utility to determine the type of a {@link Source}. */
export type TypeOfSource<T> = T extends Source<infer U> ? U : never;
/** Subscription object that can be used to cancel a {@link Source}.
* @see {@link subscribe | subscribe sink} for a helper that returns this structure.
*/
export interface Subscription {
/**
* Cancels a {@link Source} to stop the subscription from receiving new values.
*
* @see {@link TalkbackKind.Close | Close signal} This uses the {@link TalkbackFn | talkback function} to send a {@link TalkbackKind.Close | Close signal}
* to the subscribed-to {@link Source} to stop it from sending new values. This cleans up the subscription
* and ends it immediately.
*/
unsubscribe(): void;
}
/** An Observer represents sending signals manually to a {@link Sink}.
* @remarks
* The Observer is used whenever a utility allows for signals to be sent manually as a {@link Source}
* would send them.
*
* @see {@link make | `make` source} for a helper that uses this structure.
*/
export interface Observer<T> {
/** Sends a new value to the receiving Sink.
* @remarks
* This creates a {@link Push | Push signal} that is sent to a {@link Sink}.
*/
next(value: T): void;
/** Indicates to the receiving Sink that no more values will be sent.
* @remarks
* This creates an {@link SignalKind.End | End signal} that is sent to a {@link Sink}. The Observer
* will accept no more values via {@link Observer.next | `next` calls} once this method has been
* invoked.
*/
complete(): void;
}
/** Subjects combine a {@link Source} with the {@link Observer} that is used to send values on said Source.
* @remarks
* A Subject is used whenever an event hub-like structure is needed, as it both provides the
* {@link Observer}'s methods to send signals, as well as the `source` to receive said signals.
*
* @see {@link makeSubject | `makeSubject` source} for a helper that creates this structure.
*/
export interface Subject<T> extends Observer<T> {
/** The {@link Source} that issues the signals as the {@link Observer} methods are called. */
source: Source<T>;
}
/** Async Iterable/Iterator after having converted a {@link Source}.
* @see {@link toAsyncIterable} for a helper that creates this structure.
*/
export interface SourceIterable<T> extends AsyncIterator<T>, AsyncIterable<T> {}
+10
View File
@@ -0,0 +1,10 @@
export const TalkbackKind = {
Pull: 0,
Close: 1,
};
export const SignalKind = {
Start: 0,
Push: 1,
End: 0,
};