value: Convert from a getter to a method.
This commit is contained in:
David Braun
parent
8ec8fdacb2
commit
5470201c79
3 changed files with 196 additions and 195 deletions
351
lib/index.js
351
lib/index.js
|
|
@ -36,7 +36,7 @@ const Order = channel => {
|
|||
|
||||
const prototype = {};
|
||||
|
||||
const Channel = function(bufferLength = 0) {
|
||||
const Channel = function(length = 0) {
|
||||
let buffered = 0;
|
||||
let closed = false;
|
||||
let lastValue;
|
||||
|
|
@ -70,7 +70,7 @@ const Channel = function(bufferLength = 0) {
|
|||
for (
|
||||
;
|
||||
resolvedIndex < index.push ||
|
||||
(resolvedIndex < pushes.length && buffered < bufferLength);
|
||||
(resolvedIndex < pushes.length && buffered < length);
|
||||
resolvedIndex++
|
||||
) {
|
||||
const { cancelled, resolve } = pushes[resolvedIndex];
|
||||
|
|
@ -80,7 +80,7 @@ const Channel = function(bufferLength = 0) {
|
|||
buffered++;
|
||||
}
|
||||
|
||||
resolve(bufferLength);
|
||||
resolve(length);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -101,205 +101,209 @@ const Channel = function(bufferLength = 0) {
|
|||
resolvedIndex -= index.push;
|
||||
};
|
||||
|
||||
const readOnly = Object.assign(Object.create(prototype), {
|
||||
every: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
return true;
|
||||
} else {
|
||||
if (!callbackfn.call(thisArg, value)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
filter: (callbackfn, thisArg) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
const readOnly = Object.freeze(
|
||||
Object.assign(Object.create(prototype), {
|
||||
every: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
await output.close();
|
||||
break;
|
||||
} else if (callbackfn.call(thisArg, value)) {
|
||||
await output.push(value);
|
||||
return true;
|
||||
} else {
|
||||
if (!callbackfn.call(thisArg, value)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
})();
|
||||
},
|
||||
|
||||
return output;
|
||||
},
|
||||
filter: (callbackfn, thisArg) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
forEach: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
break;
|
||||
} else {
|
||||
await callbackfn.call(thisArg, value);
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
join: async separator => {
|
||||
const elements = [];
|
||||
|
||||
await readOnly.forEach(element => {
|
||||
elements.push(element);
|
||||
});
|
||||
|
||||
return elements.join(separator);
|
||||
},
|
||||
|
||||
map: (callbackfn, thisArg) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
await readOnly.forEach(value =>
|
||||
output.push(callbackfn.call(thisArg, value))
|
||||
);
|
||||
|
||||
output.close();
|
||||
})();
|
||||
|
||||
return output;
|
||||
},
|
||||
|
||||
readOnly: () => readOnly,
|
||||
|
||||
reduce: async (callbackfn, ...initialValue) => {
|
||||
let previousValue = initialValue[0];
|
||||
let previousValueDefined = initialValue.length > 0;
|
||||
|
||||
await readOnly.forEach(currentValue => {
|
||||
if (previousValueDefined) {
|
||||
previousValue = callbackfn(previousValue, currentValue);
|
||||
} else {
|
||||
previousValue = currentValue;
|
||||
previousValueDefined = true;
|
||||
}
|
||||
});
|
||||
|
||||
if (previousValueDefined) {
|
||||
return previousValue;
|
||||
} else {
|
||||
throw new TypeError(
|
||||
`No values in channel and initialValue wasn't provided.`
|
||||
);
|
||||
}
|
||||
},
|
||||
|
||||
shift: function() {
|
||||
const { order, promise } = Order(this);
|
||||
shifts.push(order);
|
||||
setImmediate(processOrders);
|
||||
return Object.freeze(promise);
|
||||
},
|
||||
|
||||
slice: (start, end = Infinity) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
for (let index = 0; index < start; index++) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
break;
|
||||
if (value === undefined) {
|
||||
await output.close();
|
||||
break;
|
||||
} else if (callbackfn.call(thisArg, value)) {
|
||||
await output.push(value);
|
||||
}
|
||||
}
|
||||
}
|
||||
})();
|
||||
|
||||
for (let index = start; index < end; index++) {
|
||||
return output;
|
||||
},
|
||||
|
||||
forEach: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
break;
|
||||
} else {
|
||||
await output.push(value);
|
||||
await callbackfn.call(thisArg, value);
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
await output.close();
|
||||
})();
|
||||
join: async separator => {
|
||||
const elements = [];
|
||||
|
||||
return output;
|
||||
},
|
||||
await readOnly.forEach(element => {
|
||||
elements.push(element);
|
||||
});
|
||||
|
||||
some: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
return elements.join(separator);
|
||||
},
|
||||
|
||||
if (value === undefined) {
|
||||
return false;
|
||||
} else {
|
||||
if (callbackfn.call(thisArg, value)) {
|
||||
return true;
|
||||
map: (callbackfn, thisArg) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
await readOnly.forEach(value =>
|
||||
output.push(callbackfn.call(thisArg, value))
|
||||
);
|
||||
|
||||
output.close();
|
||||
})();
|
||||
|
||||
return output;
|
||||
},
|
||||
|
||||
readOnly: () => readOnly,
|
||||
|
||||
reduce: async (callbackfn, ...initialValue) => {
|
||||
let previousValue = initialValue[0];
|
||||
let previousValueDefined = initialValue.length > 0;
|
||||
|
||||
await readOnly.forEach(currentValue => {
|
||||
if (previousValueDefined) {
|
||||
previousValue = callbackfn(previousValue, currentValue);
|
||||
} else {
|
||||
previousValue = currentValue;
|
||||
previousValueDefined = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
toString: () => `Channel(${bufferLength})`,
|
||||
|
||||
values: async () => {
|
||||
const array = [];
|
||||
|
||||
await readOnly.forEach(item => {
|
||||
array.push(item);
|
||||
});
|
||||
|
||||
return array;
|
||||
}
|
||||
});
|
||||
|
||||
Object.defineProperties(readOnly, {
|
||||
length: { get: () => bufferLength },
|
||||
value: { get: () => lastValue }
|
||||
});
|
||||
|
||||
Object.freeze(readOnly);
|
||||
|
||||
const writeOnly = Object.freeze({
|
||||
close: () =>
|
||||
new Promise((resolve, reject) => {
|
||||
if (closed) {
|
||||
reject(new Error(`Can't close an already-closed channel.`));
|
||||
if (previousValueDefined) {
|
||||
return previousValue;
|
||||
} else {
|
||||
closed = true;
|
||||
processOrders();
|
||||
|
||||
// Give remaining orders in flight time to resolve before returning.
|
||||
setImmediate(resolve);
|
||||
throw new TypeError(
|
||||
`No values in channel and initialValue wasn't provided.`
|
||||
);
|
||||
}
|
||||
}),
|
||||
},
|
||||
|
||||
push: function(value) {
|
||||
const { order, promise } = Order(this);
|
||||
order.value = value;
|
||||
|
||||
if (closed) {
|
||||
order.reject(new Error(`Can't push to closed channel.`));
|
||||
} else if (value === undefined) {
|
||||
order.reject(
|
||||
new TypeError(`Can't push 'undefined' to channel, use close instead.`)
|
||||
);
|
||||
} else if (arguments.length > 1) {
|
||||
order.reject(new Error(`Can't push more than one value at a time.`));
|
||||
} else {
|
||||
pushes.push(order);
|
||||
shift: function() {
|
||||
const { order, promise } = Order(this);
|
||||
shifts.push(order);
|
||||
setImmediate(processOrders);
|
||||
return Object.freeze(promise);
|
||||
},
|
||||
|
||||
slice: (start, end = Infinity) => {
|
||||
const output = Channel();
|
||||
(async () => {
|
||||
for (let index = 0; index < start; index++) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
for (let index = start; index < end; index++) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
break;
|
||||
} else {
|
||||
await output.push(value);
|
||||
}
|
||||
}
|
||||
|
||||
await output.close();
|
||||
})();
|
||||
|
||||
return output;
|
||||
},
|
||||
|
||||
some: async (callbackfn, thisArg) => {
|
||||
for (;;) {
|
||||
const value = await readOnly.shift();
|
||||
|
||||
if (value === undefined) {
|
||||
return false;
|
||||
} else {
|
||||
if (callbackfn.call(thisArg, value)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
toString: () => `Channel(${length})`,
|
||||
|
||||
value: () => lastValue,
|
||||
|
||||
values: async () => {
|
||||
const array = [];
|
||||
|
||||
await readOnly.forEach(item => {
|
||||
array.push(item);
|
||||
});
|
||||
|
||||
return array;
|
||||
}
|
||||
})
|
||||
);
|
||||
|
||||
return Object.freeze(promise);
|
||||
},
|
||||
const writeOnly = Object.freeze(
|
||||
Object.assign(Object.create(prototype), {
|
||||
close: () =>
|
||||
new Promise((resolve, reject) => {
|
||||
if (closed) {
|
||||
reject(new Error(`Can't close an already-closed channel.`));
|
||||
} else {
|
||||
closed = true;
|
||||
processOrders();
|
||||
|
||||
writeOnly: () => writeOnly
|
||||
});
|
||||
// Give remaining orders in flight time to resolve before returning.
|
||||
setImmediate(resolve);
|
||||
}
|
||||
}),
|
||||
|
||||
// Use Object.create because readOnly has a getter.
|
||||
return Object.freeze(Object.assign(Object.create(readOnly), writeOnly));
|
||||
length,
|
||||
|
||||
push: function(value) {
|
||||
const { order, promise } = Order(this);
|
||||
order.value = value;
|
||||
|
||||
if (closed) {
|
||||
order.reject(new Error(`Can't push to closed channel.`));
|
||||
} else if (value === undefined) {
|
||||
order.reject(
|
||||
new TypeError(
|
||||
`Can't push 'undefined' to channel, use close instead.`
|
||||
)
|
||||
);
|
||||
} else if (arguments.length > 1) {
|
||||
order.reject(new Error(`Can't push more than one value at a time.`));
|
||||
} else {
|
||||
pushes.push(order);
|
||||
setImmediate(processOrders);
|
||||
}
|
||||
|
||||
return Object.freeze(promise);
|
||||
},
|
||||
|
||||
writeOnly: () => writeOnly
|
||||
})
|
||||
);
|
||||
|
||||
return Object.freeze(
|
||||
Object.assign(Object.create(prototype), readOnly, writeOnly)
|
||||
);
|
||||
};
|
||||
|
||||
Channel.from = values => {
|
||||
|
|
@ -384,7 +388,10 @@ Channel.select = methodPromises =>
|
|||
// Channel.slice(10)(Infinity)(channel)
|
||||
|
||||
const channel = Channel();
|
||||
const methods = Object.keys(channel).concat(Object.keys(channel.readOnly()));
|
||||
|
||||
const methods = Object.keys(channel).filter(
|
||||
method => typeof channel[method] === `function`
|
||||
);
|
||||
|
||||
methods.forEach(method => {
|
||||
const bound = function(...args) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue