Functional async generators
===========================

The base concept is that async iterators are streams, and that we should provide
tools to handle them.

This is similar to the concepts enabled by most.js or rxjs, but without a
proprietary loop: the scheduling is entirely done by the JavaScript engine using
async generators.

Since all operations are async, the streams are non-blocking.

Compatibility with Node.js streams
==================================

These streams are compatible with Node.js' Readable Stream API:

import {from} from '@shimaore/lake'
from( process.stdin )

and with Node.js' Writeable Stream API, using pipeline or Readable.from:

import {pipeline} from 'stream'
import fs from 'fs'
import {bigNaturals} from '@shimaore/lake'
pipeline(
bigNaturals().skip(1).map( x => x*x ).first(1000).map( x => ${x}\n ),
// This will create a file containing the first thousand squares
fs.createWriteStream('thousand-squares.txt'),
console.error
)

Fluent API, Merging streams
===========================

Some basic data streaming example, first one building integers (positive and
negative) from naturals (positive only):

import {merge} from '@shimaore/lake'

Merging streams is done in approximate round-robin fashion; this prevents one
stream from eagerly blocking other stream.

const BigIntegers = merge(
// will enumerate 1, 2, 3, 4, …
bigNaturals().skip(1),
// will enumerate 0, -1, -2, -3, …
bigNaturals().map( x => -x ),
)

const firstTwentyIntegers = from(
[0,-1,1,-2,2,-3,3,-4,4,-5,5,-6,6,-7,7,-8,8,-9,9,-10].map(BigInt)
)

import test from 'ava'
test('Compute first twenty integers', async t => {
t.true(await

BigIntegers
.first(20)

.equals( firstTwentyIntegers )

)
})

Notice how first and equals are methods on the stream.

Building the sum of the first 1000 squares in constant space:

const Sum = (a,v) => a+v

test('Compute the sum of the first 1000 squares', async t => {
t.is( 333_833_500n, await

bigNaturals()
.skip(1)
.map( x => x*x )
.first(1000n)
.reduce(Sum, 0n)

)
})

Concurrent Map
==============

Use Concurrent Map to process async operations concurrently while controlling the
maximum number of concurrent executions.

import {sleep} from '@shimaore/lake'

const squareMicroService = async (x) => {
await sleep(1)
return x*x
}

test('Compute the sum of the first 1000 squares using concurrentMap', async t => {
console.time('with concurrentMap')
t.is( 333_833_500n, await

bigNaturals()
.skip(1)
.concurrentMap(100,squareMicroService)
.first(1000n)
.reduce(Sum, 0n)

)
console.timeEnd('with concurrentMap')
})

test('Compute the sum of the first 1000 squares without concurrentMap', async t => {
console.time('without concurrentMap')
t.is( 333_833_500n, await

bigNaturals()
.skip(1)
.map( squareMicroService )
.first(1000n)
.reduce(Sum, 0n)

)
console.timeEnd('without concurrentMap')
})

Tests
=====

The examples in this README file are used as tests.

``
yarn install && yarn test
`

API
===

Lake(iterable|asynciterable)

An Async Iterator and Async Iterable that behaves as a Readable Stream and
supports Monadic Event Stream patterns, using only native operators.

It can be used as a proxy for the original (sync or async) iterable, turning it
into an async iterator.

It also is an async iterable, which means it can be turned back into a stream.

.map(transform)

Applies a (sync or async) transformation function to each element in the stream.

.concurrentMap(atmost,transform)

Applies an async transformation function to each element in the stream, running
at most atmost instances concurrently.

The ordering of the elements is not guaranteed, since it will depend on the
evaluation time of the async transform function.

.constant(value)

Transform this stream into a stream that produces the value for each element
the original stream produces.

.filter(fun)

Only forward stream values for which the (sync or async) fun function returns
(a Promise for) a truthy value.

.skipRepeats()

Optionally, a (sync or async) comparison function might be provided to compare
using a different criteria than ===; it should return true if its two
arguments are considered identical.

.first(n)

Only propagates the first n elements in the stream.

BigInt are used internally; n might be a integer or a BigInt.

.skip(n)

Skips the first n elements in the stream and only start propagating after the
n-th element has been received.

BigInt are used internally; n might be a integer or a BigInt.

.delay(timeout)

Insert a delay of timeout milliseconds between each received element.

.startWith(otherStream)

Concatenates the otherStream with this stream.

.continueWith(otherStream)

Contatenates this stream then the otherStream.

.forEach(func)

Executes the (sync or async) func function for each element in the stream.
The stream is unmodified but might be delayed by the execution time of func.
The stream will fail if func fails or rejects.

.ap(funs)

Apply a stream of (sync or async) functions to this stream.

Elements of this stream are dropped until funs provides a function.

.switchLatest()

Outputs the data from the latest stream in this stream-of-streams.

.reduce(reducer,accumulator)

Using a (sync or async) reducer function which accepts the latest value of the
accumulator and a new value, returns the final value of the accumulator.

.run()

Consumes this stream, throwing away values; returns a Promise.

The Promise will reject if the stream fails for any reason.

.last()

Consumes this stream, returning its last value (or undefined if no value was
produced) inside a Promise.

test('Retrieve the 14th BigInt', async t => {
t.is(14n, await

bigNaturals()
.skip(1) // skip 0
.first(14)
.last()

)
})

.equals(otherStream)

The optional isEqual (sync or async) comparison function should return true if
its two arguments are considered equal.

empty()

Builds a stream that finishes immediately.

always(v)

Builds a stream that continously generates the value.

bigNaturals()

Builds a stream that enumerates all positive BigInt values, starting at 0
and incrementing by 1 at each step.

periodic(period)

Builds a stream that generates a new element with the provided value (or
undefined if no value is provided) and generates similarly every period
milliseconds thereafter.

now(v)

Builds a stream that only produces once, with the value provided.

throwError(e)

Builds a stream that stops immediately with the provided error.

from(iterable|asynciterable)

From any iterable, generates a new "Lake" stream (described above).

The iterable might be an Array, an iterator, an AsyncIterator,
a ReadableStream, …

Use Node.js' events.on(emitter,eventName)
to create an AsyncIterator from an event-emitter.

equals(streamA,streamB)

The optional (sync or async) isEqual` function should return true to indicate
that its two arguments are considered identical.

import {combine} from '@shimaore/lake'

test('If you take the difference between consecutive square numbers, you generate the odd numbers.', async t => {
const squaresFromZero = bigNaturals().map( x => x*x )
const squaresFromOne = bigNaturals().skip(1).map( x => x*x )
const differences = combine(squaresFromZero,squaresFromOne).map( ([a,b]) => b-a )
const oddNaturals = bigNaturals().map( x => 2n*x+1n )
t.true(await

oddNaturals
.take(10000)
.equals(differences.take(10000))

)
})