Allow promises to be passed through a channel.
This commit is contained in:
parent
965d14d82c
commit
57247c6d10
2 changed files with 36 additions and 2 deletions
16
lib/index.js
16
lib/index.js
|
@ -60,8 +60,13 @@ const Channel = function(length = 0) {
|
||||||
} else if (shift.cancelled) {
|
} else if (shift.cancelled) {
|
||||||
index.shift++;
|
index.shift++;
|
||||||
} else {
|
} else {
|
||||||
lastValue = push.value;
|
if (`reason` in push) {
|
||||||
shift.resolve(lastValue);
|
shift.reject(push.reason);
|
||||||
|
} else {
|
||||||
|
lastValue = push.value;
|
||||||
|
shift.resolve(lastValue);
|
||||||
|
}
|
||||||
|
|
||||||
buffered = Math.max(0, buffered - 1);
|
buffered = Math.max(0, buffered - 1);
|
||||||
index.push++;
|
index.push++;
|
||||||
index.shift++;
|
index.shift++;
|
||||||
|
@ -320,6 +325,13 @@ const Channel = function(length = 0) {
|
||||||
const { order, promise } = Order(this);
|
const { order, promise } = Order(this);
|
||||||
order.value = value;
|
order.value = value;
|
||||||
|
|
||||||
|
// If value is a promise that rejects, catch it in case there hasn't
|
||||||
|
// been a matching shift yet in order to prevent an unhandledRejection
|
||||||
|
// error. Reject it again when there's a shift.
|
||||||
|
Promise.resolve(value).catch(reason => {
|
||||||
|
order.reason = reason;
|
||||||
|
});
|
||||||
|
|
||||||
if (closed) {
|
if (closed) {
|
||||||
order.reject(new Error(`Can't push to closed channel.`));
|
order.reject(new Error(`Can't push to closed channel.`));
|
||||||
} else if (value === undefined) {
|
} else if (value === undefined) {
|
||||||
|
|
|
@ -453,3 +453,25 @@ describe(`Channel object`, function() {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it(`allows promises to be sent through a channel`, function() {
|
||||||
|
return new Promise(async (resolve, reject) => {
|
||||||
|
process.once(`unhandledRejection`, reject);
|
||||||
|
|
||||||
|
const channel = Channel.of(
|
||||||
|
Promise.resolve(`resolved`),
|
||||||
|
new Promise((resolve, reject) => {
|
||||||
|
setImmediate(reject, new Error(`rejected`));
|
||||||
|
})
|
||||||
|
);
|
||||||
|
|
||||||
|
assert.equal(await channel.shift(), `resolved`);
|
||||||
|
|
||||||
|
try {
|
||||||
|
await channel.shift();
|
||||||
|
} catch (exception) {
|
||||||
|
assert.equal(exception.message, `rejected`);
|
||||||
|
resolve();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue