102 lines
2.7 KiB
Plaintext
102 lines
2.7 KiB
Plaintext
export const getStreamContents = async (stream, {init, convertChunk, getSize, truncateChunk, addChunk, getFinalChunk, finalize}, {maxBuffer = Number.POSITIVE_INFINITY} = {}) => {
|
|
if (!isAsyncIterable(stream)) {
|
|
throw new Error('The first argument must be a Readable, a ReadableStream, or an async iterable.');
|
|
}
|
|
|
|
const state = init();
|
|
state.length = 0;
|
|
|
|
try {
|
|
for await (const chunk of stream) {
|
|
const chunkType = getChunkType(chunk);
|
|
const convertedChunk = convertChunk[chunkType](chunk, state);
|
|
appendChunk({convertedChunk, state, getSize, truncateChunk, addChunk, maxBuffer});
|
|
}
|
|
|
|
appendFinalChunk({state, convertChunk, getSize, truncateChunk, addChunk, getFinalChunk, maxBuffer});
|
|
return finalize(state);
|
|
} catch (error) {
|
|
error.bufferedData = finalize(state);
|
|
throw error;
|
|
}
|
|
};
|
|
|
|
const appendFinalChunk = ({state, getSize, truncateChunk, addChunk, getFinalChunk, maxBuffer}) => {
|
|
const convertedChunk = getFinalChunk(state);
|
|
if (convertedChunk !== undefined) {
|
|
appendChunk({convertedChunk, state, getSize, truncateChunk, addChunk, maxBuffer});
|
|
}
|
|
};
|
|
|
|
const appendChunk = ({convertedChunk, state, getSize, truncateChunk, addChunk, maxBuffer}) => {
|
|
const chunkSize = getSize(convertedChunk);
|
|
const newLength = state.length + chunkSize;
|
|
|
|
if (newLength <= maxBuffer) {
|
|
addNewChunk(convertedChunk, state, addChunk, newLength);
|
|
return;
|
|
}
|
|
|
|
const truncatedChunk = truncateChunk(convertedChunk, maxBuffer - state.length);
|
|
|
|
if (truncatedChunk !== undefined) {
|
|
addNewChunk(truncatedChunk, state, addChunk, maxBuffer);
|
|
}
|
|
|
|
throw new MaxBufferError();
|
|
};
|
|
|
|
const addNewChunk = (convertedChunk, state, addChunk, newLength) => {
|
|
state.contents = addChunk(convertedChunk, state, newLength);
|
|
state.length = newLength;
|
|
};
|
|
|
|
const isAsyncIterable = stream => typeof stream === 'object' && stream !== null && typeof stream[Symbol.asyncIterator] === 'function';
|
|
|
|
const getChunkType = chunk => {
|
|
const typeOfChunk = typeof chunk;
|
|
|
|
if (typeOfChunk === 'string') {
|
|
return 'string';
|
|
}
|
|
|
|
if (typeOfChunk !== 'object' || chunk === null) {
|
|
return 'others';
|
|
}
|
|
|
|
// eslint-disable-next-line n/prefer-global/buffer
|
|
if (globalThis.Buffer?.isBuffer(chunk)) {
|
|
return 'buffer';
|
|
}
|
|
|
|
const prototypeName = objectToString.call(chunk);
|
|
|
|
if (prototypeName === '[object ArrayBuffer]') {
|
|
return 'arrayBuffer';
|
|
}
|
|
|
|
if (prototypeName === '[object DataView]') {
|
|
return 'dataView';
|
|
}
|
|
|
|
if (
|
|
Number.isInteger(chunk.byteLength)
|
|
&& Number.isInteger(chunk.byteOffset)
|
|
&& objectToString.call(chunk.buffer) === '[object ArrayBuffer]'
|
|
) {
|
|
return 'typedArray';
|
|
}
|
|
|
|
return 'others';
|
|
};
|
|
|
|
const {toString: objectToString} = Object.prototype;
|
|
|
|
export class MaxBufferError extends Error {
|
|
name = 'MaxBufferError';
|
|
|
|
constructor() {
|
|
super('maxBuffer exceeded');
|
|
}
|
|
}
|