I am working with ReadableStreams in JS/Browser code and hit the following issue. When using a ReadableStreamBYOBReader to control the size of stream reads, if the stream is closed during a read request without enqueued data, the read never returns. See the first snippet below.
The last read returns as expected when ReadableStreamDefaultReader is used to read the stream. See the second snippet below (the stream implementations are identical in the snippets).
Given that this works the same in Chrome and FF, I’m assuming it expected behavior. But if so, how do you close a stream being read BYOD mode when you don’t know how much data is remaining until each read request is made? Raising an error closes the stream, but that’s rather ugly.
These snippets run in Chrome and FF, but not Safari because Safari doesn’t have the full stream API yet. Also, please ignore that byobRequest is not used and that reading isn’t optimized. My goal was to simplify the logic.
// Hangs if you try to read and the stream closes (when readBlocks > streamBlocks)
const readBlocks = 4;
const streamBlocks = 3;
const blockSize = 1024;
const stream = makeStream(streamBlocks);
const reader = stream.getReader({ mode: "byob" });
let buffer = new Uint8Array(blockSize * readBlocks);
readAllBYOB(reader, buffer).then(([blocks, done]) => {
reader.releaseLock();
let byteLen = 0;
for(const block of blocks) {
byteLen += block.byteLength;
}
console.log("all done, bytes read:", byteLen, done);
});
function makeStream(loops) {
let totalBytesOutput = 0;
console.log("creating stream size:", loops * blockSize);
return new ReadableStream({
type: "bytes",
async start(controller) {
console.log(
`stream start- ${controller.constructor.name}.byobRequest = ${controller.byobRequest}`,
);
try {
const data = new TextEncoder().encode("s".repeat(blockSize));
totalBytesOutput += data.byteLength;
console.log("stream start- enqueuing, total:", data.byteLength, totalBytesOutput);
controller.enqueue(data);
} catch (err) {
console.error("stream start- error, closing", err);
controller.error(err);
}
},
async pull(controller) {
// ignoring actual byobReuest object
console.log(
`stream pull- ${controller.constructor.name}.byobRequest = ${controller.byobRequest}`,
);
try {
// Pretend we don't know when data runs out until the request is made.
// In BYOD mode, the read never returns. Unless you do one of the following:
// 1. Enqueueing data before calling close (but we don't have any to enqueue)
// 2. Call controller.error() instead, but that's ugly
if (totalBytesOutput >= blockSize * loops) {
console.log("stream pull- closing");
controller.close();
return;
}
const data = new TextEncoder().encode("p".repeat(blockSize));
totalBytesOutput += data.byteLength;
console.log("stream pull- enqueuing, total:", data.byteLength, totalBytesOutput);
controller.enqueue(data);
} catch (err) {
console.error("stream pull- error, closing", err);
controller.error(err);
}
},
});
}
async function readAllBYOB(reader, output) {
let targetBytes = output.byteLength;
let readBytes = 0;
let blocks = [];
let streamDone = false;
console.log('readAllBYOB- start: ', targetBytes);
while (readBytes < targetBytes) {
console.log('readAllBYOB- try reading:', output.byteLength);
// This does not return on the final read, even when stream is closed
let { done, value } = await reader.read(output);
console.log('readAllBYOB- read, done:', value?.byteLength, done);
streamDone = done;
if (value) {
blocks.push(value);
readBytes += value.byteLength;
}
if (done || !value) {
break;
}
if (readBytes < targetBytes) {
output = new Uint8Array(targetBytes - readBytes);
}
}
console.log(
'readAllBYOB- blocks, remainingBytes, done:',
blocks.length,
targetBytes - readBytes,
streamDone
);
return [blocks, streamDone];
}// Works as expected
const streamBlocks = 3;
const blockSize = 1024;
const stream = makeStream(streamBlocks);
const reader = stream.getReader();
readAll(reader).then(([blocks, done]) => {
reader.releaseLock();
let byteLen = 0;
for(const block of blocks) {
byteLen += block.byteLength;
}
console.log("all done, bytes read:", byteLen, done);
});
function makeStream(loops) {
let totalBytesOutput = 0;
console.log("creating stream size:", loops * blockSize);
return new ReadableStream({
type: "bytes",
async start(controller) {
console.log(
`stream start- ${controller.constructor.name}.byobRequest = ${controller.byobRequest}`,
);
try {
const data = new TextEncoder().encode("s".repeat(blockSize));
totalBytesOutput += data.byteLength;
console.log("stream start- enqueuing, total:", data.byteLength, totalBytesOutput);
controller.enqueue(data);
} catch (err) {
console.error("stream start- error, closing", err);
controller.error(err);
}
},
async pull(controller) {
// ignoring actual byobReuest object
console.log(
`stream pull- ${controller.constructor.name}.byobRequest = ${controller.byobRequest}`,
);
try {
// Pretend we don't know when data runs out until the request is made.
// In BYOD mode, the read never returns. Unless you do one of the following:
// 1. Enqueueing data before calling close (but we don't have any to enqueue)
// 2. Call controller.error() instead, but that's ugly
if (totalBytesOutput >= blockSize * loops) {
console.log("stream pull- closing");
controller.close();
return;
}
const data = new TextEncoder().encode("p".repeat(blockSize));
totalBytesOutput += data.byteLength;
console.log("stream pull- enqueuing, total:", data.byteLength, totalBytesOutput);
controller.enqueue(data);
} catch (err) {
console.error("stream pull- error, closing", err);
controller.error(err);
}
},
});
}
async function readAll(reader) {
let readBytes = 0;
let blocks = [];
let streamDone = false;
console.log('readAll- start');
while (true) {
console.log('readAll- try reading');
// This always returns as expected
let { done, value } = await reader.read();
console.log('readAll- read, done:', value?.byteLength, done);
streamDone = done;
if (value) {
blocks.push(value);
readBytes += value.byteLength;
}
if (done || !value) {
break;
}
}
console.log(
'readAll- blocks, done:',
blocks.length,
streamDone
);
return [blocks, streamDone];
}