Use stream.pipeline()

Author: ehmicky

README.md

@@ -51,8 +51,10 @@ technical lead for 2.5 years. I am available for full-time remote positions.
 `gulpfile.js`:
 
 ```js
+import { pipeline } from 'node:stream/promises'
+
 import gulp from 'gulp'
-import { task, exec, stream } from 'gulp-execa'
+import { exec, stream, task } from 'gulp-execa'
 
 export const audit = task('npm audit')
 
@@ -61,10 +63,11 @@ export const outdated = async () => {
 }
 
 export const sort = () =>
-  gulp
-    .src('*.txt')
-    .pipe(stream(({ path }) => `sort ${path}`))
-    .pipe(gulp.dest('sorted'))
+  pipeline(
+    gulp.src('*.txt'),
+    stream(({ path }) => `sort ${path}`),
+    gulp.dest('sorted'),
+  )
 ```
 
 # Install
@@ -128,14 +131,17 @@ Returns a stream that executes a `command` on each input file.
   - `undefined`
 
 ```js
+import { pipeline } from 'node:stream/promises'
+
 import gulp from 'gulp'
 import { stream } from 'gulp-execa'
 
 export const sort = () =>
-  gulp
-    .src('*.txt')
-    .pipe(stream(({ path }) => `sort ${path}`))
-    .pipe(gulp.dest('sorted'))
+  pipeline(
+    gulp.src('*.txt'),
+    stream(({ path }) => `sort ${path}`),
+    gulp.dest('sorted'),
+  )
 ```
 
 Each file in the stream will spawn a separate process. This can consume lots of
@@ -251,21 +257,22 @@ With [`stream()`](#streamfunction-options), whether the command result should:
 <!-- eslint-disable unicorn/no-null -->
 
 ```js
+import { pipeline } from 'node:stream/promises'
+
 import gulp from 'gulp'
 import { stream } from 'gulp-execa'
 import through from 'through2'
 
 export const task = () =>
-  gulp
-    .src('*.js')
+  pipeline(
+    gulp.src('*.js'),
     // Prints the number of lines of each file
-    .pipe(stream(({ path }) => `wc -l ${path}`, { result: 'save' }))
-    .pipe(
-      through.obj((file, encoding, func) => {
-        console.log(file.execa[0].stdout)
-        func(null, file)
-      }),
-    )
+    stream(({ path }) => `wc -l ${path}`, { result: 'save' }),
+    through.obj((file, encoding, func) => {
+      console.log(file.execa[0].stdout)
+      func(null, file)
+    }),
+  )
 ```
 
 ## from
@@ -279,23 +286,22 @@ Which output stream to use with [`result: 'replace'`](#result).
 <!-- eslint-disable unicorn/no-null -->
 
 ```js
+import { pipeline } from 'node:stream/promises'
+
 import gulp from 'gulp'
 import { stream } from 'gulp-execa'
 import through from 'through2'
 
 export const task = () =>
-  gulp
-    .src('*.js')
+  pipeline(
+    gulp.src('*.js'),
     // Prints the number of lines of each file, including `stderr`
-    .pipe(
-      stream(({ path }) => `wc -l ${path}`, { result: 'replace', from: 'all' }),
-    )
-    .pipe(
-      through.obj((file, encoding, func) => {
-        console.log(file.contents.toString())
-        func(null, file)
-      }),
-    )
+    stream(({ path }) => `wc -l ${path}`, { result: 'replace', from: 'all' }),
+    through.obj((file, encoding, func) => {
+      console.log(file.contents.toString())
+      func(null, file)
+    }),
+  )
 ```
 
 ## maxConcurrency

src/helpers/gulpfiles/stream.test.js

@@ -1,4 +1,5 @@
 import { Buffer } from 'node:buffer'
+import { pipeline } from 'node:stream/promises'
 import { fileURLToPath } from 'node:url'
 import { callbackify } from 'node:util'
 
@@ -20,66 +21,74 @@ const DUMMY_TWO = fileURLToPath(
 
 // Task used in most tests
 export const main = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(() => command, opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(() => command, opts),
+    through.obj(execVinyl),
+  )
 
 // `input` should be an async function
 export const inputAsync = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(() => Promise.resolve(command), opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(() => Promise.resolve(command), opts),
+    through.obj(execVinyl),
+  )
 
 // `input` should be fired with the Vinyl file
 export const inputFile = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(({ basename }) => `${command} ${basename}`, opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(({ basename }) => `${command} ${basename}`, opts),
+    through.obj(execVinyl),
+  )
 
 const noop = () => {}
 
 // File should be skipped when returning a non-string
 export const inputUndefined = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(noop, opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(noop, opts),
+    through.obj(execVinyl),
+  )
 
 // Should allow several files
 export const severalFiles = () =>
-  gulp
-    .src([DUMMY, DUMMY_TWO], { buffer })
-    .pipe(stream(() => command, opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src([DUMMY, DUMMY_TWO], { buffer }),
+    stream(() => command, opts),
+    through.obj(execVinyl),
+  )
 
 // Should allow doing several times
 export const severalTimes = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(() => command, opts))
-    .pipe(stream(() => command, opts))
-    .pipe(through.obj(execVinyl))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(() => command, opts),
+    stream(() => command, opts),
+    through.obj(execVinyl),
+  )
 
 // `input` should be a function
 export const inputNotFunc = () =>
-  gulp.src(DUMMY, { buffer }).pipe(stream(command, opts))
+  pipeline(gulp.src(DUMMY, { buffer }), stream(command, opts))
 
 // `input` exceptions should be propagated
 export const inputThrows = () =>
-  gulp.src(DUMMY, { buffer }).pipe(
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
     stream(() => {
       throw new Error('error')
     }, opts),
   )
 
 // `input` async exceptions should be propagated
 export const inputThrowsAsync = () =>
-  gulp
-    .src(DUMMY, { buffer })
-    .pipe(stream(() => Promise.reject(new Error('error')), opts))
+  pipeline(
+    gulp.src(DUMMY, { buffer }),
+    stream(() => Promise.reject(new Error('error')), opts),
+  )
 
 const cExecVinyl = async (file) => {
   // When `file.contents` is a stream and an `error` event should be emitted,

src/helpers/methods.test.js

@@ -10,11 +10,20 @@ export const TASK_METHODS = [
 ]
 
 export const STREAM_METHODS = [
-  // `gulp.src(...).pipe(stream(..., { result: 'replace' }))`
+  // `pipeline(
+  //    gulp.src(...),
+  //    stream(..., { result: 'replace' }),
+  //  )`
   { title: 'stream-buffer', method: 'stream' },
-  // `gulp.src(..., { buffer: false }).pipe(stream(..., { result: 'replace' }))`
+  // `pipeline(
+  //    gulp.src(..., { buffer: false }),
+  //    stream(..., { result: 'replace' }),
+  //  )`
   { title: 'stream-stream', method: 'stream', buffer: false },
-  // `gulp.src(...).pipe(stream(..., { result: 'save' }))`
+  // `pipeline(
+  //    gulp.src(...),
+  //    stream(..., { result: 'save' }),
+  //  )`
   { title: 'stream-save', method: 'stream', opts: { result: 'save' } },
 ]
 

src/main.d.ts

@@ -40,20 +40,22 @@ type StreamOptions = Omit<
        *
        * @example
        * ```js
+       * import { pipeline } from 'node:stream/promises'
+       *
        * import gulp from 'gulp'
+       * import { stream } from 'gulp-execa'
        * import through from 'through2'
        *
        * export const task = () =>
-       *   gulp
-       *     .src('*.js')
+       *   pipeline(
+       *     gulp.src('*.js'),
        *     // Prints the number of lines of each file
-       *     .pipe(stream(({ path }) => `wc -l ${path}`, { result: 'save' }))
-       *     .pipe(
-       *       through.obj((file, encoding, func) => {
-       *         console.log(file.execa[0].stdout)
-       *         func(null, file)
-       *       }),
-       *     )
+       *     stream(({ path }) => `wc -l ${path}`, { result: 'save' }),
+       *     through.obj((file, encoding, func) => {
+       *       console.log(file.execa[0].stdout)
+       *       func(null, file)
+       *     }),
+       *   )
        * ```
        */
       result: 'save' | 'replace'
@@ -65,22 +67,22 @@ type StreamOptions = Omit<
        *
        * @example
        * ```js
+       * import { pipeline } from 'node:stream/promises'
+       *
        * import gulp from 'gulp'
+       * import { stream } from 'gulp-execa'
        * import through from 'through2'
        *
        * export const task = () =>
-       *   gulp
-       *     .src('*.js')
+       *   pipeline(
+       *     gulp.src('*.js'),
        *     // Prints the number of lines of each file, including `stderr`
-       *     .pipe(
-       *       stream(({ path }) => `wc -l ${path}`, { result: 'replace', from: 'all' }),
-       *     )
-       *     .pipe(
-       *       through.obj((file, encoding, func) => {
-       *         console.log(file.contents.toString())
-       *         func(null, file)
-       *       }),
-       *     )
+       *     stream(({ path }) => `wc -l ${path}`, { result: 'replace', from: 'all' }),
+       *     through.obj((file, encoding, func) => {
+       *       console.log(file.contents.toString())
+       *       func(null, file)
+       *     }),
+       *   )
        * ```
        */
       from: 'stdout' | 'stderr' | 'all'
@@ -137,13 +139,17 @@ export function task<CallOptions extends NonStreamOptions = object>(
  *
  * @example
  * ```js
+ * import { pipeline } from 'node:stream/promises'
+ *
  * import gulp from 'gulp'
+ * import { stream } from 'gulp-execa'
  *
  * export const sort = () =>
- *   gulp
- *     .src('*.txt')
- *     .pipe(stream(({ path }) => `sort ${path}`))
- *     .pipe(gulp.dest('sorted'))
+ *   pipeline(
+ *     gulp.src('*.txt'),
+ *     stream(({ path }) => `sort ${path}`),
+ *     gulp.dest('sorted'),
+ *   )
  * ```
  */
 export function stream(

src/main.test-d.ts

@@ -1,7 +1,7 @@
 import type { Readable, Transform } from 'node:stream'
 
-import { exec, task, stream, type Options } from 'gulp-execa'
-import { expectType, expectAssignable, expectNotAssignable } from 'tsd'
+import { exec, stream, task, type Options } from 'gulp-execa'
+import { expectAssignable, expectNotAssignable, expectType } from 'tsd'
 // eslint-disable-next-line n/no-extraneous-import, @typescript-eslint/no-shadow
 import type File from 'vinyl'
 

src/stream/main.js

@@ -6,11 +6,11 @@ import through from 'through2-concurrent'
 import { throwError } from '../error.js'
 import { parseOpts } from '../options/main.js'
 
-import { getDefaultOpts, forcedOpts } from './options.js'
+import { forcedOpts, getDefaultOpts } from './options.js'
 import { setResult } from './result.js'
 
 // Creates a stream that fires child processes on each file:
-//   gulp.src(...).pipe(stream(({ path }) => `command ${path}`))
+//   pipeline(gulp.src(...), stream(({ path }) => `command ${path}`))
 export const stream = (getInput, opts) => {
   validateGetInput(getInput)