mirror of
https://github.com/denoland/deno.git
synced 2024-10-30 09:08:00 -04:00
241 lines
6.7 KiB
TypeScript
241 lines
6.7 KiB
TypeScript
// Copyright Node.js contributors. All rights reserved. MIT License.
|
|
import { once } from "../_utils.ts";
|
|
import type Duplex from "./duplex.ts";
|
|
import type Readable from "./readable.ts";
|
|
import type Stream from "./stream.ts";
|
|
import type { ReadableState } from "./readable.ts";
|
|
import type Writable from "./writable.ts";
|
|
import type { WritableState } from "./writable.ts";
|
|
import {
|
|
ERR_INVALID_ARG_TYPE,
|
|
ERR_STREAM_PREMATURE_CLOSE,
|
|
NodeErrorAbstraction,
|
|
} from "../_errors.ts";
|
|
|
|
export type StreamImplementations = Duplex | Readable | Stream | Writable;
|
|
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// function isRequest(stream: Stream) {
|
|
// return stream.setHeader && typeof stream.abort === "function";
|
|
// }
|
|
|
|
// deno-lint-ignore no-explicit-any
|
|
function isReadable(stream: any) {
|
|
return typeof stream.readable === "boolean" ||
|
|
typeof stream.readableEnded === "boolean" ||
|
|
!!stream._readableState;
|
|
}
|
|
|
|
// deno-lint-ignore no-explicit-any
|
|
function isWritable(stream: any) {
|
|
return typeof stream.writable === "boolean" ||
|
|
typeof stream.writableEnded === "boolean" ||
|
|
!!stream._writableState;
|
|
}
|
|
|
|
function isWritableFinished(stream: Writable) {
|
|
if (stream.writableFinished) return true;
|
|
const wState = stream._writableState;
|
|
if (!wState || wState.errored) return false;
|
|
return wState.finished || (wState.ended && wState.length === 0);
|
|
}
|
|
|
|
function nop() {}
|
|
|
|
function isReadableEnded(stream: Readable) {
|
|
if (stream.readableEnded) return true;
|
|
const rState = stream._readableState;
|
|
if (!rState || rState.errored) return false;
|
|
return rState.endEmitted || (rState.ended && rState.length === 0);
|
|
}
|
|
|
|
export interface FinishedOptions {
|
|
error?: boolean;
|
|
readable?: boolean;
|
|
writable?: boolean;
|
|
}
|
|
|
|
/**
|
|
* Appends an ending callback triggered when a stream is no longer readable,
|
|
* writable or has experienced an error or a premature close event
|
|
*/
|
|
export default function eos(
|
|
stream: StreamImplementations,
|
|
options: FinishedOptions | null,
|
|
callback: (err?: NodeErrorAbstraction | null) => void,
|
|
): () => void;
|
|
export default function eos(
|
|
stream: StreamImplementations,
|
|
callback: (err?: NodeErrorAbstraction | null) => void,
|
|
): () => void;
|
|
export default function eos(
|
|
stream: StreamImplementations,
|
|
x: FinishedOptions | ((err?: NodeErrorAbstraction | null) => void) | null,
|
|
y?: (err?: NodeErrorAbstraction | null) => void,
|
|
) {
|
|
let opts: FinishedOptions;
|
|
let callback: (err?: NodeErrorAbstraction | null) => void;
|
|
|
|
if (!y) {
|
|
if (typeof x !== "function") {
|
|
throw new ERR_INVALID_ARG_TYPE("callback", "function", x);
|
|
}
|
|
opts = {};
|
|
callback = x;
|
|
} else {
|
|
if (!x || Array.isArray(x) || typeof x !== "object") {
|
|
throw new ERR_INVALID_ARG_TYPE("opts", "object", x);
|
|
}
|
|
opts = x;
|
|
|
|
if (typeof y !== "function") {
|
|
throw new ERR_INVALID_ARG_TYPE("callback", "function", y);
|
|
}
|
|
callback = y;
|
|
}
|
|
|
|
callback = once(callback);
|
|
|
|
const readable = opts.readable ?? isReadable(stream);
|
|
const writable = opts.writable ?? isWritable(stream);
|
|
|
|
// deno-lint-ignore no-explicit-any
|
|
const wState: WritableState | undefined = (stream as any)._writableState;
|
|
// deno-lint-ignore no-explicit-any
|
|
const rState: ReadableState | undefined = (stream as any)._readableState;
|
|
const validState = wState || rState;
|
|
|
|
const onlegacyfinish = () => {
|
|
if (!(stream as Writable).writable) {
|
|
onfinish();
|
|
}
|
|
};
|
|
|
|
let willEmitClose = (
|
|
validState?.autoDestroy &&
|
|
validState?.emitClose &&
|
|
validState?.closed === false &&
|
|
isReadable(stream) === readable &&
|
|
isWritable(stream) === writable
|
|
);
|
|
|
|
let writableFinished = (stream as Writable).writableFinished ||
|
|
wState?.finished;
|
|
const onfinish = () => {
|
|
writableFinished = true;
|
|
// deno-lint-ignore no-explicit-any
|
|
if ((stream as any).destroyed) {
|
|
willEmitClose = false;
|
|
}
|
|
|
|
if (willEmitClose && (!(stream as Readable).readable || readable)) {
|
|
return;
|
|
}
|
|
if (!readable || readableEnded) {
|
|
callback.call(stream);
|
|
}
|
|
};
|
|
|
|
let readableEnded = (stream as Readable).readableEnded || rState?.endEmitted;
|
|
const onend = () => {
|
|
readableEnded = true;
|
|
// deno-lint-ignore no-explicit-any
|
|
if ((stream as any).destroyed) {
|
|
willEmitClose = false;
|
|
}
|
|
|
|
if (willEmitClose && (!(stream as Writable).writable || writable)) {
|
|
return;
|
|
}
|
|
if (!writable || writableFinished) {
|
|
callback.call(stream);
|
|
}
|
|
};
|
|
|
|
const onerror = (err: NodeErrorAbstraction) => {
|
|
callback.call(stream, err);
|
|
};
|
|
|
|
const onclose = () => {
|
|
if (readable && !readableEnded) {
|
|
if (!isReadableEnded(stream as Readable)) {
|
|
return callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
|
|
}
|
|
}
|
|
if (writable && !writableFinished) {
|
|
if (!isWritableFinished(stream as Writable)) {
|
|
return callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
|
|
}
|
|
}
|
|
callback.call(stream);
|
|
};
|
|
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// const onrequest = () => {
|
|
// stream.req.on("finish", onfinish);
|
|
// };
|
|
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// if (isRequest(stream)) {
|
|
// stream.on("complete", onfinish);
|
|
// stream.on("abort", onclose);
|
|
// if (stream.req) {
|
|
// onrequest();
|
|
// } else {
|
|
// stream.on("request", onrequest);
|
|
// }
|
|
// } else
|
|
if (writable && !wState) {
|
|
stream.on("end", onlegacyfinish);
|
|
stream.on("close", onlegacyfinish);
|
|
}
|
|
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// if (typeof stream.aborted === "boolean") {
|
|
// stream.on("aborted", onclose);
|
|
// }
|
|
|
|
stream.on("end", onend);
|
|
stream.on("finish", onfinish);
|
|
if (opts.error !== false) stream.on("error", onerror);
|
|
stream.on("close", onclose);
|
|
|
|
const closed = (
|
|
wState?.closed ||
|
|
rState?.closed ||
|
|
wState?.errorEmitted ||
|
|
rState?.errorEmitted ||
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// (rState && stream.req && stream.aborted) ||
|
|
(
|
|
(!writable || wState?.finished) &&
|
|
(!readable || rState?.endEmitted)
|
|
)
|
|
);
|
|
|
|
if (closed) {
|
|
queueMicrotask(callback);
|
|
}
|
|
|
|
return function () {
|
|
callback = nop;
|
|
stream.removeListener("aborted", onclose);
|
|
stream.removeListener("complete", onfinish);
|
|
stream.removeListener("abort", onclose);
|
|
// TODO(Soremwar)
|
|
// Bring back once requests are implemented
|
|
// stream.removeListener("request", onrequest);
|
|
// if (stream.req) stream.req.removeListener("finish", onfinish);
|
|
stream.removeListener("end", onlegacyfinish);
|
|
stream.removeListener("close", onlegacyfinish);
|
|
stream.removeListener("finish", onfinish);
|
|
stream.removeListener("end", onend);
|
|
stream.removeListener("error", onerror);
|
|
stream.removeListener("close", onclose);
|
|
};
|
|
}
|