# Function pipeline

- pipeline<A, B>(source, destination, callback?): B extends WritableStream
  ? B
  : WritableStream

- A module method to pipe between streams and generators forwarding errors and properly cleaning up and provide a callback when the pipeline is complete.

      const { pipeline } = require('stream');
      const fs = require('fs');
      const zlib = require('zlib');

      // Use the pipeline API to easily pipe a series of streams
      // together and get notified when the pipeline is fully done.

      // A pipeline to gzip a potentially huge tar file efficiently:

      pipeline(
        fs.createReadStream('archive.tar'),
        zlib.createGzip(),
        fs.createWriteStream('archive.tar.gz'),
        (err) => {
          if (err) {
            console.error('Pipeline failed.', err);
          } else {
            console.log('Pipeline succeeded.');
          }
        }
      );

  The `pipeline` API provides a promise version, which can also receive an options argument as the last parameter with a`signal` `AbortSignal` property. When the signal is aborted,`destroy` will be called on the underlying pipeline, with an`AbortError`.

      const { pipeline } = require('stream/promises');

      async function run() {
        await pipeline(
          fs.createReadStream('archive.tar'),
          zlib.createGzip(),
          fs.createWriteStream('archive.tar.gz')
        );
        console.log('Pipeline succeeded.');
      }

      run().catch(console.error);

  To use an `AbortSignal`, pass it inside an options object, as the last argument:

      const { pipeline } = require('stream/promises');

      async function run() {
        const ac = new AbortController();
        const signal = ac.signal;

        setTimeout(() => ac.abort(), 1);
        await pipeline(
          fs.createReadStream('archive.tar'),
          zlib.createGzip(),
          fs.createWriteStream('archive.tar.gz'),
          { signal },
        );
      }

      run().catch(console.error); // AbortError

  The `pipeline` API also supports async generators:

      const { pipeline } = require('stream/promises');
      const fs = require('fs');

      async function run() {
        await pipeline(
          fs.createReadStream('lowercase.txt'),
          async function* (source, { signal }) {
            source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
            for await (const chunk of source) {
              yield await processChunk(chunk, { signal });
            }
          },
          fs.createWriteStream('uppercase.txt')
        );
        console.log('Pipeline succeeded.');
      }

      run().catch(console.error);

  Remember to handle the `signal` argument passed into the async generator. Especially in the case where the async generator is the source for the pipeline (i.e. first argument) or the pipeline will never complete.

      const { pipeline } = require('stream/promises');
      const fs = require('fs');

      async function run() {
        await pipeline(
          async function* ({ signal }) {
            await someLongRunningfn({ signal });
            yield 'asd';
          },
          fs.createWriteStream('uppercase.txt')
        );
        console.log('Pipeline succeeded.');
      }

      run().catch(console.error);

  `stream.pipeline()` will call `stream.destroy(err)` on all streams except:

  - `Readable` streams which have emitted `'end'` or `'close'`.
  - `Writable` streams which have emitted `'finish'` or `'close'`.

  `stream.pipeline()` leaves dangling event listeners on the streams after the `callback` has been invoked. In the case of reuse of streams after failure, this can cause event listener leaks and swallowed errors. If the last stream is readable, dangling event listeners will be removed so that the last stream can be consumed later.

  `stream.pipeline()` closes all the streams when an error is raised. The `IncomingRequest` usage with `pipeline` could lead to an unexpected behavior once it would destroy the socket without sending the expected response. See the example below:

      const fs = require('fs');
      const http = require('http');
      const { pipeline } = require('stream');

      const server = http.createServer((req, res) => {
        const fileStream = fs.createReadStream('./fileNotExist.txt');
        pipeline(fileStream, res, (err) => {
          if (err) {
            console.log(err); // No such file
            // this message can't be sent once `pipeline` already destroyed the socket
            return res.end('error!!!');
          }
        });
      });

  #### Type Parameters

  - A extends PipelineSource<any>
  - B extends
    | WritableStream
    | PipelineDestinationIterableFunction<string | Buffer>
    | PipelineDestinationPromiseFunction<string | Buffer, any>
    | PipelineDestinationIterableFunction<any>
    | PipelineDestinationPromiseFunction<any, any>

  #### Parameters

  - source: A
  - destination: B
  - `Optional` callback: PipelineCallback<B>

    Called when the pipeline is fully done.

  #### Returns B extends WritableStream
  ? B
  : WritableStream

  #### Since

  v10.0.0

  - Defined in node_modules/@types/node/stream.d.ts:1568
- pipeline<A, T1, B>(source, transform1, destination, callback?): B extends WritableStream
  ? B
  : WritableStream

- #### Type Parameters

  - A extends PipelineSource<any>
  - T1 extends PipelineTransform<A, any>
  - B extends
    | WritableStream
    | PipelineDestinationIterableFunction<string | Buffer>
    | PipelineDestinationPromiseFunction<string | Buffer, any>
    | PipelineDestinationIterableFunction<any>
    | PipelineDestinationPromiseFunction<any, any>

  #### Parameters

  - source: A
  - transform1: T1
  - destination: B
  - `Optional` callback: PipelineCallback<B>

  #### Returns B extends WritableStream
  ? B
  : WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1573
- pipeline<A, T1, T2, B>(source, transform1, transform2, destination, callback?): B extends WritableStream
  ? B
  : WritableStream

- #### Type Parameters

  - A extends PipelineSource<any>
  - T1 extends PipelineTransform<A, any>
  - T2 extends PipelineTransform<T1, any>
  - B extends
    | WritableStream
    | PipelineDestinationIterableFunction<string | Buffer>
    | PipelineDestinationPromiseFunction<string | Buffer, any>
    | PipelineDestinationIterableFunction<any>
    | PipelineDestinationPromiseFunction<any, any>

  #### Parameters

  - source: A
  - transform1: T1
  - transform2: T2
  - destination: B
  - `Optional` callback: PipelineCallback<B>

  #### Returns B extends WritableStream
  ? B
  : WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1583
- pipeline<A, T1, T2, T3, B>(source, transform1, transform2, transform3, destination, callback?): B extends WritableStream
  ? B
  : WritableStream

- #### Type Parameters

  - A extends PipelineSource<any>
  - T1 extends PipelineTransform<A, any>
  - T2 extends PipelineTransform<T1, any>
  - T3 extends PipelineTransform<T2, any>
  - B extends
    | WritableStream
    | PipelineDestinationIterableFunction<string | Buffer>
    | PipelineDestinationPromiseFunction<string | Buffer, any>
    | PipelineDestinationIterableFunction<any>
    | PipelineDestinationPromiseFunction<any, any>

  #### Parameters

  - source: A
  - transform1: T1
  - transform2: T2
  - transform3: T3
  - destination: B
  - `Optional` callback: PipelineCallback<B>

  #### Returns B extends WritableStream
  ? B
  : WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1595
- pipeline<A, T1, T2, T3, T4, B>(source, transform1, transform2, transform3, transform4, destination, callback?): B extends WritableStream
  ? B
  : WritableStream

- #### Type Parameters

  - A extends PipelineSource<any>
  - T1 extends PipelineTransform<A, any>
  - T2 extends PipelineTransform<T1, any>
  - T3 extends PipelineTransform<T2, any>
  - T4 extends PipelineTransform<T3, any>
  - B extends
    | WritableStream
    | PipelineDestinationIterableFunction<string | Buffer>
    | PipelineDestinationPromiseFunction<string | Buffer, any>
    | PipelineDestinationIterableFunction<any>
    | PipelineDestinationPromiseFunction<any, any>

  #### Parameters

  - source: A
  - transform1: T1
  - transform2: T2
  - transform3: T3
  - transform4: T4
  - destination: B
  - `Optional` callback: PipelineCallback<B>

  #### Returns B extends WritableStream
  ? B
  : WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1609
- pipeline(streams, callback?): WritableStream

- #### Parameters

  - streams: readonly (WritableStream | ReadableStream | ReadWriteStream)[]
  - `Optional` callback: ((err: ErrnoException) => void)
    - - (err): void

      - #### Parameters

        - err: ErrnoException

        #### Returns void

  #### Returns WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1625
- pipeline(stream1, stream2, ...streams): WritableStream

- #### Parameters

  - stream1: ReadableStream
  - stream2: WritableStream | ReadWriteStream
  - `Rest`...streams: (WritableStream | ReadWriteStream | ((err: ErrnoException) => void))[]

  #### Returns WritableStream

  - Defined in node_modules/@types/node/stream.d.ts:1629