nodeguy / channel

Go-like channels for JavaScript

Introduction

This is an idiomatic, minimally-opinionated Channel type for JavaScript that's inspired by Go's channels. It works in browsers and in Node.js. If you know how to use an Array then you already know most of how to use a Channel.

Why

Go's use of channels for concurrency is amazing and with JavaScript's async/await feature we have the basis for it as well. All that's missing is a solid Channel type. There are existing libraries but I wanted an idiomatic Channel type that's simple and minimally-opinionated.

This document assumes you're familiar with Go's channels and why you'd want to use them. For explanatory background, see my blog article on the subject.

Installation

$ npm install @nodeguy/channel

Basic Use

Create a channel with Channel().

To send a value to a channel use push. To receive a value from a channel use shift. Always precede the method calls with await. Close the channel when there are no more values to push.

const assert = require(`assert`);
const Channel = require(`@nodeguy/channel`);

const channel = Channel();

const send = async () => {
  await channel.push(42);
  await channel.close();
};

const receive = async () => {
  assert.equal(await channel.shift(), 42);
  assert.equal(await channel.shift(), undefined);
};

await Promise.all([send(), receive()]);

The push and shift methods are usually called in different async functions. They represent the two different ends of the channel and act to synchronize the behavior of the async functions.

API

The API is in the API.md file.

Similar Projects

• Channel
• cochan
• js-csp
• node-csp

Copyright

Copyright 2017 David Braun

Licensed under the Apache License, Version 2.0 (the "License"); you may not use these files except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

channel's Issues

values() doesn't return an async iterator

The documentation says that values() returns an async iterator, but actually it just blocks until the channel is closed and then returns all the values at once in an array.

Make it possible to set fixed buffer size

Can I set fixed buffer size like 1? Or can I at least get the latest value in the channel and remove the rest?

If not, one solution that comes to my mind is to use shift() before each push(). Do you think it is a good idea? Like:

for (;;) {
  // ...
  channel.shift();
  channel.push(value);
  // ...
}

TypeScript usage?

I tried using channel with TypeScript and found it to fail type checking.

With the following code (mostly from the README):

import Channel = require("@nodeguy/channel");
import assert = require("assert");

const channel = Channel();

const send = async () => {
  await channel.push(42);
  await channel.close();
};

const receive = async () => {
  assert.equal(await channel.shift(), 42);
  assert.equal(await channel.shift(), undefined);
};

(async () => {
  await Promise.all([send(), receive()]);
})();

I ran the script with ts-node and got this type error:

TSError: ⨯ Unable to compile TypeScript:
foo.ts:5:17 - error TS2349: This expression is not callable.
  Type 'typeof import("redacted/path/node_modules/@nodeguy/channel/lib/index")' has no call signatures.

5 const channel = Channel();
                  ~~~~~~~

It runs fine, when I add // @ts-ignore before the offending line, so it seems the module itself is fine? Still I'd like not to use that hack and have it typecheck properly ;)

Am I just using this wrong? Perhaps @graup could weigh in, as they added the type definitions.

Suggestion on Handling errors in Channels

This departs a bit from the go spec, but I'm considering how to integrate error handling from my channel writers so my readers know something broke upstream.

My current workflow (pre-channels) is that my "getter" function returns a promise that either resolves with the entire get data set, or throws an error if it encounters something getting that data. The getter function can internally be doing multiple webservice/database calls, any of which may result in an distinct error. In pseudo-code (sorry, I think coffeescript), the workflow is like this:

something.getAll()
.each (obj) -> # one object of my dataset
.catch (err) ->
  console.log "Error fetching the data!"

Under channels, I can return a channel, and I can get data from the channel and know when it is closed, but not if the getter had an error or not.

something.getChannel()
.then (channel) ->
  channel.forEach (obj) ->
    # each object of my dataset
.catch (err) ->
  console.log "Exception from getChannel, but not from the thread pushing data to the channel

What I'm wondering is if it makes any sense for a channel writer to be able to throw an exception via the channel so the reader can know the datastream is incomplete:

channel.closeWithError( new Error "Something's Wrong")

and on the receiver, I can catch it:

something.getChannel()
.then (channel) ->
  channel.forEach (obj) ->
     # each object of my dataset
  .catch (err) ->
     console.log "Channel closed with error!"
.catch (err) -> 
   console.log "Error from getChannel"

The only other solution to this I can think of is if my getChannel returns both the channel and a separate promise that rejects if there is a processing issue. That seems clunkier however and I then have to process both the channel and the fetching process to ensure I have gotten all my data.

Mostly I'm just suggesting this as a possible feature and I'm interested in your take on it.

setImmediate is not defined in most browsers

The project description says the library can be used in browsers. Yet, unless I'm missing something, it uses the "setImmediate" function which is not defined in chromium 63 and, according to MDN [1], is hardly defined in any browser at all.

[1] https://developer.mozilla.org/en-US/docs/Web/API/Window/setImmediate

Possible heap overflow: Cancelled orders are not removed

Issue:
The following code generates a heap overflow:

import Channel from "@nodeguy/channel";

async function heapTest() {
    const closedDefault = new Channel<{}>();
    closedDefault.close();
    const randomOtherChannel = new Channel<{}>();

    while (true) {
        switch (await Channel.select([randomOtherChannel.push({}), closedDefault.shift()])) {
            case randomOtherChannel:
                // Will never happen
                throw new Error();
            default:
                // Will always happen
                break;
        }
    }
}

heapTest();

Why:
Every iteration of the while loop appends a new order object to the pushes array of randomOtherChannel. The order immediately get’s cancelled but is never removed from the array. It should normally be skipped inside matchPushesAndShifts() and then sliced off inside processOrders(). This never happens because the while loop condition inside matchPushesAndShifts() is never met (the shifts array of randomOtherChannel is always empty).

Solution: (Pull request submitted)
Filter out all cancelled orders from pushes and shifts at the end of every processOrders() call.

channel does not work with Bluebird promises

It seems like freezing the Promise object creates an issue when native promises are replaced with Bluebird promises.

Reproduction:
https://gist.github.com/druska/94fcf4bbdb8588391040b20915ed9bab

Result:

node --trace-warnings nodeguy_bluebird_issue.js

(node:86229) TypeError: Cannot assign to read only property '_promise0' of object '[object Object]'
    at Promise._addCallbacks (/private/tmp/node_modules/bluebird/js/release/promise.js:396:24)
    at Promise._then (/private/tmp/node_modules/bluebird/js/release/promise.js:270:16)
    at Promise.then (/private/tmp/node_modules/bluebird/js/release/promise.js:125:17)
    at process._tickCallback (internal/process/next_tick.js:188:7)
    at Function.Module.runMain (module.js:686:11)
    at startup (bootstrap_node.js:187:16)
    at bootstrap_node.js:608:3
(node:86229) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
    at emitWarning (internal/process/promises.js:78:15)
    at emitPendingUnhandledRejections (internal/process/promises.js:95:11)
    at process._tickCallback (internal/process/next_tick.js:189:7)
    at Function.Module.runMain (module.js:686:11)
    at startup (bootstrap_node.js:187:16)
    at bootstrap_node.js:608:3
/private/tmp/node_modules/bluebird/js/release/promise.js:294
    this._bitField = this._bitField | 33554432;
                   ^

TypeError: Cannot assign to read only property '_bitField' of object '[object Object]'
    at Promise._setFulfilled (/private/tmp/node_modules/bluebird/js/release/promise.js:294:20)
    at Promise._fulfill (/private/tmp/node_modules/bluebird/js/release/promise.js:633:10)
    at Promise._resolveCallback (/private/tmp/node_modules/bluebird/js/release/promise.js:432:57)
    at /private/tmp/node_modules/bluebird/js/release/promise.js:484:17
    at Object.resolve (/private/tmp/node_modules/@nodeguy/channel/lib/index.js:17:9)
    at Immediate.processOrders [as _onImmediate] (/private/tmp/node_modules/@nodeguy/channel/lib/index.js:64:15)
    at runCallback (timers.js:789:20)
    at tryOnImmediate (timers.js:751:5)
    at processImmediate [as _immediateCallback] (timers.js:722:5)