@n1ru4l/push-pull-async-iterable-iterator

Create an AsyncIterableIterator from anything (on any modern platform) while handling back-pressure!

1yarn install -E @n1ru4l/push-pull-async-iterable-iterator

Standalone Usage

1import { makePushPullAsyncIterableIterator } from "@n1ru4l/push-pull-async-iterable-iterator";
2
3const {
4  pushValue,
5  asyncIterableIterator
6} = makePushPullAsyncIterableIterator();
7pushValue(1);
8pushValue(2);
9pushValue(3);
10
11// prints 1, 2, 3
12for await (const value of asyncIterableIterator) {
13  console.log(value);
14}

Check if something is an AsyncIterable

1import { isAsyncIterable } from "@n1ru4l/push-pull-async-iterable-iterator";
2
3if (isAsyncIterable(something)) {
4  for await (const value of something) {
5    console.log(value);
6  }
7}

Note: On Safari iOS Symbol.asyncIterator is not available, therefore all async iterators used must be build using AsyncGenerators. If a AsyncIterable that is NO AsyncGenerator is passed to isAsyncIterable on the Safari iOS environment, it will return the value false.

Wrap a Sink

1import { makeAsyncIterableIteratorFromSink } from "@n1ru4l/push-pull-async-iterable-iterator";
2// let's use some GraphQL client :)
3import { createClient } from "graphql-ws/lib/use/ws";
4
5const client = createClient({
6  url: "ws://localhost:3000/graphql"
7});
8
9const asyncIterableIterator = makeAsyncIterableIteratorFromSink(sink => {
10  const dispose = client.subscribe(
11    {
12      query: "{ hello }"
13    },
14    {
15      next: sink.next,
16      error: sink.error,
17      complete: sink.complete
18    }
19  );
20  return () => dispose();
21});
22
23for await (const value of asyncIterableIterator) {
24  console.log(value);
25}

Apply an AsyncIterableIterator to a sink

1import Observable from "zen-observable";
2import {
3  makePushPullAsyncIterableIterator,
4  applyAsyncIterableIteratorToSink
5} from "@n1ru4l/push-pull-async-iterable-iterator";
6
7const { asyncIterableIterator } = makePushPullAsyncIterableIterator();
8
9const observable = new Observable(sink => {
10  const dispose = applyAsyncIterableIteratorToSink(asyncIterableIterator, sink);
11  // dispose will be called when the observable subscription got destroyed
12  // the dispose call will ensure that the async iterator is completed.
13  return () => dispose();
14});
15
16const subscription = observable.subscribe({
17  next: console.log,
18  complete: () => console.log("done."),
19  error: () => console.log("error.")
20});
21
22const interval = setInterval(() => {
23  iterator.push("hi");
24}, 1000);
25
26setTimeout(() => {
27  subscription.unsubscribe();
28  clearInterval(interval);
29}, 5000);

Put it all together

1import { Observable, RequestParameters, Variables } from "relay-runtime";
2import { createClient } from "graphql-ws/lib/use/ws";
3import {
4  makeAsyncIterableFromSink,
5  applyAsyncIterableIteratorToSink
6} from "@n1ru4l/push-pull-async-iterable-iterator";
7import { createApplyLiveQueryPatch } from "@n1ru4l/graphql-live-query-patch";
8
9const client = createClient({
10  url: "ws://localhost:3000/graphql"
11});
12
13export const execute = (request: RequestParameters, variables: Variables) => {
14  if (!request.text) {
15    throw new Error("Missing document.");
16  }
17  const query = request.text;
18
19  return Observable.create<GraphQLResponse>(sink => {
20    // Create our asyncIterator from a Sink
21    const executionResultIterator = makeAsyncIterableFromSink(wsSink => {
22      const dispose = client.subscribe({ query }, wsSink);
23      return () => dispose();
24    });
25
26    const applyLiveQueryPatch = createApplyLiveQueryPatch();
27
28    // apply some middleware to our asyncIterator
29    const compositeIterator = applyLiveQueryPatch(executionResultIterator);
30
31    // Apply our async iterable to the relay sink
32    // unfortunately relay cannot consume an async iterable right now.
33    const dispose = applyAsyncIterableIteratorToSink(compositeIterator, sink);
34    // dispose will be called by relay when the observable is disposed
35    // the dispose call will ensure that the async iterator is completed.
36    return () => dispose();
37  });
38};

Operators

This package also ships a few utilities that make your life easier!

map

Map a source

1import { map } from "@n1ru4l/push-pull-async-iterable-iterator";
2
3async function* source() {
4  yield 1;
5  yield 2;
6  yield 3;
7}
8
9const square = map((value: number): number => value * value);
10
11for await (const value of square(source())) {
12  console.log(value);
13}
14// logs 1, 4, 9

filter

Filter a source

1import { filter } from "@n1ru4l/push-pull-async-iterable-iterator";
2
3async function* source() {
4  yield 1;
5  yield 2;
6  yield 3;
7}
8
9const biggerThan1 = filter((value: number): number => value > 1);
10
11for await (const value of biggerThan1(source())) {
12  console.log(value);
13}
14// logs 2, 3

Other helpers

withHandlers

Attach a return and throw handler to a source.

1import { withReturn } from "@n1ru4l/push-pull-async-iterable-iterator";
2
3async function* source() {
4  yield 1;
5  yield 2;
6  yield 3;
7}
8
9const sourceInstance = source();
10
11const newSourceWithHandlers = withHandlers(
12  sourceInstance,
13  () => sourceInstance.return(),
14  err => sourceInstance.throw(err)
15);
16
17for await (const value of stream) {
18  // ...
19}