René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"pipe() allow destroying source!","articleBody":"### What is the problem this feature will solve?\r\n\r\nnode 20.15.1. The idea is to pass around a stream without having to track the sources and being able to destroy it. So the transformer kinda starts owning control of end of life of it's source. Example use case: Implemet tail -f and abort after 5 lines. Then reading bytes is the source, and turning into lines is a transformer. The consumer (who wants to find a mysqld ready for connections in the log) should not care about the tail -f implementation.\r\n\r\n```\r\nstream = createSource().pipe(transformer)\r\nstream.destroy()\r\n\r\n```\r\nshould destroy the source, too.\r\nThis doesn't happen.\r\n\r\ncomplete test case\r\n\r\n```\r\n// this file tests that using pipeDestroySource actually destroys\r\n// the reader while standard .pipe() doesn't ?\r\nimport {Readable, Writable, Transform} from \"node:stream\";\r\n\r\nconst sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))\r\n\r\nexport const pipeDestroySource = (source:  Readable,  destination: Writable) => {\r\n    // unpipe or close might work\r\n    destination.on('unpipe', (e) => { if (e == source) source.destroy()})\r\n    return source.pipe(destination)\r\n}\r\n\r\nexport type OnDestroy = () => void\r\n\r\nclass MyReadStream extends Readable {\r\n\r\n  constructor(private ondestroy: OnDestroy) {\r\n    super({autoDestroy: true});\r\n  }\r\n\r\n  _read() {\r\n        this.push(Buffer.from(\"A\\n\"))\r\n  }\r\n  _destroy(err: any, callback: any) {\r\n        console.log(\"MyReadStream _destroy\");\r\n    this.ondestroy()\r\n    callback(err);\r\n  }\r\n} \r\n\r\n\r\nexport class MyTransform extends Transform {\r\n\r\n  constructor(private ondestroy: OnDestroy) {\r\n    super({objectMode: true, autoDestroy: true});\r\n  }\r\n\r\n  _transform(chunk: Buffer, encoding: string, callback: any) {\r\n        this.push(chunk)\r\n        callback();\r\n  }\r\n\r\n  _destroy(err: any, callback: any){\r\n        console.log('MyTransform _destroy');\r\n        this.ondestroy()\r\n        super._destroy(err, callback)\r\n  }\r\n\r\n}\r\nconst main = async () => {\r\n    const test = async (title: string, pipe: (source: Readable, destination: Writable) => Writable, expectations: boolean[] ) => {\r\n        console.log(`=== test ${title} started`);\r\n        let r_destroy_called = false\r\n        let t_destroy_called = false\r\n        const r = new MyReadStream(() => { r_destroy_called = true})\r\n        const t = new MyTransform(()  => { t_destroy_called = true })\r\n        const s = pipe(r, t)\r\n        s.destroy()\r\n\r\n        // logging won't work because the events actions will be run after this method returned!\r\n        // console.log({r_destroy_called, t_destroy_called});\r\n        console.log(`test ${title} done`);\r\n    }\r\n\r\n    await test(   \"pipe\",\r\n            (s, d) => s.pipe(d),\r\n            [false, true]\r\n        )\r\n\r\n    await sleep(1000)\r\n\r\n    await test(\r\n           \"pipeDestroySource\",\r\n           pipeDestroySource,\r\n           [true, true]\r\n       )\r\n}\r\n\r\nmain().catch((e) => {\r\n    console.log(\"== ERROR uncaught exception  ==\");\r\n    console.log(e);\r\n    process.exit(1)\r\n})\r\n\r\n```\r\n\r\n\r\n### What is the feature you are proposing to solve the problem?\r\n\r\n```\r\nstream = createSource().pipe(transformer, { destroySource: true })\r\nstream.destroy()\r\n```\r\n\r\n### What alternatives have you considered?\r\n\r\nusing this code instead:\r\n```\r\nexport const pipeDestroySource = (source:  Readable,  destination: Writable) => {\r\n    // unpipe or close might work\r\n    destination.on('unpipe', (e) => { if (e == source) source.destroy()})\r\n    return source.pipe(destination)\r\n}\r\n```","author":{"url":"https://github.com/MarcWeber","@type":"Person","name":"MarcWeber"},"datePublished":"2024-09-19T12:38:21Z","url":"https://github.com/nodejs/node/issues/55010"}