89 lines
1.8 KiB
Plaintext
89 lines
1.8 KiB
Plaintext
//#region src/utils/stream.ts
|
|
/**
|
|
* @module
|
|
* Stream utility.
|
|
*/
|
|
var StreamingApi = class {
|
|
writer;
|
|
encoder;
|
|
writable;
|
|
abortSubscribers = [];
|
|
responseReadable;
|
|
/**
|
|
* Whether the stream has been aborted.
|
|
*/
|
|
aborted = false;
|
|
/**
|
|
* Whether the stream has been closed normally.
|
|
*/
|
|
closed = false;
|
|
constructor(writable, _readable) {
|
|
this.writable = writable;
|
|
this.writer = writable.getWriter();
|
|
this.encoder = new TextEncoder();
|
|
const reader = _readable.getReader();
|
|
this.abortSubscribers.push(async () => {
|
|
await reader.cancel();
|
|
});
|
|
this.responseReadable = new ReadableStream({
|
|
async pull(controller) {
|
|
const { done, value } = await reader.read();
|
|
done ? controller.close() : controller.enqueue(value);
|
|
},
|
|
cancel: () => {
|
|
if (!this.closed) this.abort();
|
|
}
|
|
});
|
|
}
|
|
async write(input) {
|
|
try {
|
|
if (typeof input === "string") input = this.encoder.encode(input);
|
|
await this.writer.write(input);
|
|
} catch {}
|
|
return this;
|
|
}
|
|
async writeln(input) {
|
|
await this.write(input + "\n");
|
|
return this;
|
|
}
|
|
sleep(ms) {
|
|
return new Promise((res) => setTimeout(res, ms));
|
|
}
|
|
async close() {
|
|
this.closed = true;
|
|
try {
|
|
await this.writer.close();
|
|
} catch {}
|
|
}
|
|
async pipe(body) {
|
|
this.writer.releaseLock();
|
|
try {
|
|
await body.pipeTo(this.writable, {
|
|
preventClose: true,
|
|
preventAbort: true
|
|
});
|
|
} finally {
|
|
this.writer = this.writable.getWriter();
|
|
}
|
|
}
|
|
onAbort(listener) {
|
|
this.abortSubscribers.push(listener);
|
|
}
|
|
/**
|
|
* Abort the stream.
|
|
* You can call this method when stream is aborted by external event.
|
|
*/
|
|
abort() {
|
|
if (!this.aborted) {
|
|
this.aborted = true;
|
|
this.abortSubscribers.forEach((subscriber) => {
|
|
try {
|
|
Promise.resolve(subscriber()).catch(() => {});
|
|
} catch {}
|
|
});
|
|
}
|
|
}
|
|
};
|
|
//#endregion
|
|
export { StreamingApi };
|