nodeguy/channel

Go-like channels for JavaScript

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.

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.

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)

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.

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);
  // ...
}

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