npm package: @fine-js/channels
Description: Bits of Clojure's `core.async` ported to JS
Author: Aleksei Zabrodskii
Latest version: 0.0.2

# Channels.js

Library for asynchronous in-process communication
based on [Clojure's](https://clojure.org)
excellent [`core.async`](https://github.com/clojure/core.async).
Read through [introductory blog post](https://clojure.org/news/2013/06/28/clojure-clore-async-channels)
or watch [Strange Loop presentation](https://www.youtube.com/watch?v=drmNlZVkUeE)
for more details and inspiration.

It's queues, essentially, except there are `Promise`s on top because JavaScript.

<a href="https://www.npmjs.com/package/@fine-js/channels">

```
npm install @fine-js/channels
```

</a>

## Supported Platforms

Library's source code is plain JavaScript with no runtime dependencies that works in Node.js
and modern browsers.


### Node.js

After installing a dependency, `require` the module as usual:

```js
const {chan} = require('@fine-js/channels')
console.dir(chan())
```

### Browsers

A build for browsers is available as a [single-file download](browser.js)
and via [UNPKG](https://unpkg.com/browse/@fine-js/channels/) ([docs](https://unpkg.com)).
Here is an example of embedding version `0.0.2` in particular:

```html
<script src="https://unpkg.com/@fine-js/channels@0.0.2/browser.js"
        integrity="sha384-aLUgfMcOf6P0qxZ4k0e084VdlxfruOqU0zXhYBSZS28Y07u7Zoo1NBbYnNpNynck"
        crossorigin="anonymous"></script>

<script>
  const {chan} = window.finejs.channels
  console.dir(chan())
</script>
```

Alternatively, you can build your own version with [Browserify](http://browserify.org)
or any other bundler, as well as serve `require.resolve('@fine-js/channels/browser.js')`
from your server directly.

## Overview

*TODO…*

### Terminology

**Channel** is a plain object with two operations:

 - `put(ch, val)` puts a value onto the channel
 - `take(ch)` takes a value from the channel

**Port** is a description of either operation:

 - `ch` is a port representing taking a value from channel `ch`
 - `[ch, val]` is a port representing putting `val` onto the channel `ch`

Each channel can be backed by a **buffer** of given capacity.

**Parking** is creating and returning a promise that will possibly never resolve,
or resolve at indefinite time in the future. This is in contrast
to **immediately** which is used in this document to roughly mean guaranteed resolution
in the near future (often, after a couple of microtask ticks).

## API

###### Basics

- [`chan()`](#chan) creates a channel
- [`put()`](#put) puts a value onto the channel
- [`take()`](#take) takes a value from the channel
- [`close()`](#close) closes a channel

###### Core

- [`alt()`](#alt) runs at most one of the given operations returning result of handler
- [`alts()`](#alts) runs at most one of the given ports
- [`poll()`](#poll) takes a value from channel, when immediately possible
- [`offer()`](#offer) puts a value onto channel, when immediately possible
- [`timeout()`](#timeout) creates a self-closing channel

###### Buffers

- [`unbuffered()`](#unbuffered) for directly connecting writers to readers
- [`buffer()`](#buffer) up to given capacity
- [`dropping()`](#dropping) ignores writes over capacity
- [`sliding()`](#sliding) writes over capacity remove oldest written value

###### JavaScript Extras

  - [`ch[Symbol.asyncIterator]`](#iter) makes channel iterable with [`for await…of`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of)

<a id="chan"></a>
### `chan()` <small>[⬏](#api)</small>

Creates a channel with an optional buffer:

- `chan()` unbuffered channel
- `chan(capacity)` blocking buffer of given capacity
- `chan(buffer)` passed buffer

```js
chan()
chan(10)
chan(sliding(1))
```

<a id="put"></a>
### `put()` <small>[⬏](#api)</small>

Puts a value onto the channel. `null` or `undefined` are not allowed.
Will park if no buffer space is available.
Resolves to `true` unless channel is already closed.

```js
put(ch, 42)
ch.put(42)
```

<a id="take"></a>
### `take()` <small>[⬏](#api)</small>

Takes a value from the channel. Will park if nothing is available.
Resolves to a value taken or `null` if channel is already closed.

```js
take(ch)
ch.take()
```

<a id="close"></a>
### `close()` <small>[⬏](#api)</small>

Closes given channel. The channel will no longer accept any puts (they will be ignored).
Data in the channel remains available for taking, until exhausted,
after which takes will return `null`.

If there are any pending takes, they will be dispatched with `null`.
Any parked puts will remain so until a taker releases them.
Closing a closed channel is a no-op. Returns `null`.

```js
close(ch)
ch.close()
```

<a id="timeout"></a>
### `timeout()` <small>[⬏](#api)</small>

Returns a channel that will close after given number of millisecods.

```js
timeout(100)
```

<a id="alt"></a>
### `alt()` <small>[⬏](#api)</small>

Executes one of several channel operations via `alts()` and passess its result to a handler.
Accepts a list of operations follwed by corresponding handler and options to pass to `alts()`, where:

- each operation is either
  - single channel to take from
  - array of ports
- each handler is either
  - a function accepting `(val, channel)`
  - anything else

Each channel may only appear once.

Resolves to handler's result if it's a function, handler itself otherwise.

```js
alt([
  a,
  (val) => console.log('read %o from a', val),

  [b, c],
  (val, ch) => console.log('read %o from %o', val, ch),

  [[ch, 42]],
  'wrote 42 to ch',
])

alt([
  [[worker1, task], [worker2, task], [worker3, task]],
  `task ${task.id} queued`,
], {default: () => `try again later in ${waittime()} seconds`})

alt([
  results,
  (val) => ({status: 200, text: val}),

  timeout(150),
  {status: 504},
], {priority: true})
```

<a id="alts"></a>
### `alts()` <small>[⬏](#api)</small>

Completes at most one of several ports.

Unless the `priority` option is `true`, if more than one port is ready,
a non-deterministic choice will be made. If no operation is
ready and a `default` value is supplied, `[default-val, alts.default]` will
be returned. Otherwise `alts` will park until the first operation to
become ready completes.

Resolves to `[val port]` of the completed operation,
where `val` is the value taken for takes,
and a boolean (`true` unless already closed, as per `ch.put`) for puts.

```js
alts([ch1, ch2, ch3])

alts([
  [worker1, task],
  [worker2, task],
  [worker3, task],
], {default: 'try again later'})

alts([
  results,
  timeout(150),
], {priority: true})
```


<a id="poll"></a>
### `poll()` <small>[⬏](#api)</small>

Takes a value from the channel, when immediately possible.
Resolves a value taken if successful, `null` otherwise.

```js
poll(ch)
```

<a id="offer"></a>
### `offer()` <small>[⬏](#api)</small>

Puts a value into the channel, when immediately possible.
Resolves to `true` if offer succeeds.

```js
offer(ch, 42)
```

<a id="unbuffered"></a>
### `unbuffered()` <small>[⬏](#api)</small>

Giving this buffer to a channel will make it act as a synchronisation point
between readers and writers: puts are parked until the value is taken by consumer.

This is a buffer channel will get when you call `chan()` or `chan(0)`.

<a id="buffer"></a>
### `buffer()` <small>[⬏](#api)</small>

Creates a blocking buffer:

- `buffer()` with capacity 1
- `buffer(capacity)` with given capacity.

When full, puts will be parked.

```js
ch(16)
ch(buffer(16))
```

<a id="dropping"></a>
### `dropping()` <small>[⬏](#api)</small>

Creates a dropping non-blocking buffer:

- `dropping()` with capacity 1
- `dropping(capacity)` with given capacity

When full, puts will complete puts, but values discarded.

```js
ch(dropping(3000))
```

<a id="sliding"></a>
### `sliding()` <small>[⬏](#api)</small>

Creates a sliding non-blocking buffer:

- `sliding()` with capacity 1
- `sliding(capacity)` with given capacity

When full, puts will complete puts, thier values buffered, but oldest
elements in buffer will be dropped.

```js
ch(sliding(1))
```

<a id="iter"></a>
### `ch[Symbol.asyncIterator]()` <small>[⬏](#api)</small>

Returns Async Iterator over values being put onto the channel.
Not meant to be used directly, instead:

```js
// Simply consuming everything until the channel closes.
for await (const message of ch)
  console.log(message)

// Note that the loop's body will NOT see every single message being put onto the channel,
// since, effectively, this calls `take()` on each iteration and somewhat equivalent to:
while (!closed(ch))
  console.log(await ch.take())

// Reading until we see a sentinel value
for await (const byte of ch)
  if (byte === 0)
    break;
```

Read more over at [MDN](https://developer.mozilla.org/en-US/):

- [`for await…of`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of)
- [`Symbol.asyncIterator`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Symbol/asyncIterator)
- [Iteration protocols](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols)