123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309
// Copyright Node.js contributors. All rights reserved. MIT License.import { once } from "../_utils.ts";import { destroyer as implDestroyer } from "./destroy.ts";import eos from "./end_of_stream.ts";import createReadableStreamAsyncIterator from "./async_iterator.ts";import * as events from "../events.ts";import PassThrough from "./passthrough.ts";import {  ERR_INVALID_ARG_TYPE,  ERR_INVALID_CALLBACK,  ERR_INVALID_RETURN_VALUE,  ERR_MISSING_ARGS,  ERR_STREAM_DESTROYED,  NodeErrorAbstraction,} from "../_errors.ts";import type Duplex from "./duplex.ts";import type Readable from "./readable.ts";import type Stream from "./stream.ts";import type Transform from "./transform.ts";import type Writable from "./writable.ts";
type Streams = Duplex | Readable | Writable;// deno-lint-ignore no-explicit-anytype EndCallback = (err?: NodeErrorAbstraction | null, val?: any) => void;type TransformCallback =  // deno-lint-ignore no-explicit-any  | ((value?: any) => AsyncGenerator<any>)  // deno-lint-ignore no-explicit-any  | ((value?: any) => Promise<any>);/** * This type represents an array that contains a data source, * many Transform Streams, a writable stream destination * and end in an optional callback * */type DataSource =  // deno-lint-ignore no-explicit-any  | (() => AsyncGenerator<any>)  | // deno-lint-ignore no-explicit-any  AsyncIterable<any>  | Duplex  | // deno-lint-ignore no-explicit-any  Iterable<any>  | // deno-lint-ignore no-explicit-any  (() => Generator<any>)  | Readable;type Transformers = Duplex | Transform | TransformCallback | Writable;export type PipelineArguments = [  DataSource,  ...Array<Transformers | EndCallback>,];
function destroyer(  stream: Streams,  reading: boolean,  writing: boolean,  callback: EndCallback,) {  callback = once(callback);
  let finished = false;  stream.on("close", () => {    finished = true;  });
  eos(stream, { readable: reading, writable: writing }, (err) => {    finished = !err;
    // deno-lint-ignore no-explicit-any    const rState = (stream as any)?._readableState;    if (      err &&      err.code === "ERR_STREAM_PREMATURE_CLOSE" &&      reading &&      (rState?.ended && !rState?.errored && !rState?.errorEmitted)    ) {      stream        .once("end", callback)        .once("error", callback);    } else {      callback(err);    }  });
  return (err: NodeErrorAbstraction) => {    if (finished) return;    finished = true;    implDestroyer(stream, err);    callback(err || new ERR_STREAM_DESTROYED("pipe"));  };}
function popCallback(streams: PipelineArguments): EndCallback {  if (typeof streams[streams.length - 1] !== "function") {    throw new ERR_INVALID_CALLBACK(streams[streams.length - 1]);  }  return streams.pop() as EndCallback;}
// function isPromise(obj) {//   return !!(obj && typeof obj.then === "function");// }
// deno-lint-ignore no-explicit-anyfunction isReadable(obj: any): obj is Stream {  return !!(obj && typeof obj.pipe === "function");}
// deno-lint-ignore no-explicit-anyfunction isWritable(obj: any) {  return !!(obj && typeof obj.write === "function");}
// deno-lint-ignore no-explicit-anyfunction isStream(obj: any) {  return isReadable(obj) || isWritable(obj);}
// deno-lint-ignore no-explicit-anyfunction isIterable(obj: any, isAsync?: boolean) {  if (!obj) return false;  if (isAsync === true) return typeof obj[Symbol.asyncIterator] === "function";  if (isAsync === false) return typeof obj[Symbol.iterator] === "function";  return typeof obj[Symbol.asyncIterator] === "function" ||    typeof obj[Symbol.iterator] === "function";}
// deno-lint-ignore no-explicit-anyfunction makeAsyncIterable(val: Readable | Iterable<any> | AsyncIterable<any>) {  if (isIterable(val)) {    return val;  } else if (isReadable(val)) {    return fromReadable(val as Readable);  }  throw new ERR_INVALID_ARG_TYPE(    "val",    ["Readable", "Iterable", "AsyncIterable"],    val,  );}
async function* fromReadable(val: Readable) {  yield* createReadableStreamAsyncIterator(val);}
async function pump(  // deno-lint-ignore no-explicit-any  iterable: Iterable<any>,  writable: Duplex | Writable,  finish: (err?: NodeErrorAbstraction | null) => void,) {  let error;  try {    for await (const chunk of iterable) {      if (!writable.write(chunk)) {        if (writable.destroyed) return;        await events.once(writable, "drain");      }    }    writable.end();  } catch (err) {    error = err;  } finally {    finish(error);  }}
export default function pipeline(...args: PipelineArguments) {  const callback: EndCallback = once(popCallback(args));
  let streams: [DataSource, ...Transformers[]];  if (args.length > 1) {    streams = args as [DataSource, ...Transformers[]];  } else {    throw new ERR_MISSING_ARGS("streams");  }
  let error: NodeErrorAbstraction;  // deno-lint-ignore no-explicit-any  let value: any;  const destroys: Array<(err: NodeErrorAbstraction) => void> = [];
  let finishCount = 0;
  function finish(err?: NodeErrorAbstraction | null) {    const final = --finishCount === 0;
    if (err && (!error || error.code === "ERR_STREAM_PREMATURE_CLOSE")) {      error = err;    }
    if (!error && !final) {      return;    }
    while (destroys.length) {      (destroys.shift() as (err: NodeErrorAbstraction) => void)(error);    }
    if (final) {      callback(error, value);    }  }
  // TODO(Soremwar)  // Simplify the hell out of this  // deno-lint-ignore no-explicit-any  let ret: any;  for (let i = 0; i < streams.length; i++) {    const stream = streams[i];    const reading = i < streams.length - 1;    const writing = i > 0;
    if (isStream(stream)) {      finishCount++;      destroys.push(destroyer(stream as Streams, reading, writing, finish));    }
    if (i === 0) {      if (typeof stream === "function") {        ret = stream();        if (!isIterable(ret)) {          throw new ERR_INVALID_RETURN_VALUE(            "Iterable, AsyncIterable or Stream",            "source",            ret,          );        }      } else if (isIterable(stream) || isReadable(stream)) {        ret = stream;      } else {        throw new ERR_INVALID_ARG_TYPE(          "source",          ["Stream", "Iterable", "AsyncIterable", "Function"],          stream,        );      }    } else if (typeof stream === "function") {      ret = makeAsyncIterable(ret);      ret = stream(ret);
      if (reading) {        if (!isIterable(ret, true)) {          throw new ERR_INVALID_RETURN_VALUE(            "AsyncIterable",            `transform[${i - 1}]`,            ret,          );        }      } else {        // If the last argument to pipeline is not a stream        // we must create a proxy stream so that pipeline(...)        // always returns a stream which can be further        // composed through `.pipe(stream)`.        const pt = new PassThrough({          objectMode: true,        });        if (ret instanceof Promise) {          ret            .then((val) => {              value = val;              pt.end(val);            }, (err) => {              pt.destroy(err);            });        } else if (isIterable(ret, true)) {          finishCount++;          pump(ret, pt, finish);        } else {          throw new ERR_INVALID_RETURN_VALUE(            "AsyncIterable or Promise",            "destination",            ret,          );        }
        ret = pt;
        finishCount++;        destroys.push(destroyer(ret, false, true, finish));      }    } else if (isStream(stream)) {      if (isReadable(ret)) {        ret.pipe(stream as Readable);
        // TODO(Soremwar)        // Reimplement after stdout and stderr are implemented        // if (stream === process.stdout || stream === process.stderr) {        //   ret.on("end", () => stream.end());        // }      } else {        ret = makeAsyncIterable(ret);
        finishCount++;        pump(ret, stream as Writable, finish);      }      ret = stream;    } else {      const name = reading ? `transform[${i - 1}]` : "destination";      throw new ERR_INVALID_ARG_TYPE(        name,        ["Stream", "Function"],        ret,      );    }  }
  return ret as unknown as Readable;}