feat: introduce fetch pipe helper (google#1130)

* test: fetch > $

* feat: introduce pipe helper for `fetch`

closes google#977

* chore: code imprs

* docs: add `fetch.pipe` usage example

* docs: extend `fetch.pipe()` descr

Author: antongolub

.size-limit.json

@@ -9,14 +9,14 @@
   {
     "name": "zx/index",
     "path": "build/*.{js,cjs}",
-    "limit": "812 kB",
+    "limit": "813 kB",
     "brotli": false,
     "gzip": false
   },
   {
     "name": "dts libdefs",
     "path": "build/*.d.ts",
-    "limit": "39.3 kB",
+    "limit": "39.4 kB",
     "brotli": false,
     "gzip": false
   },
@@ -30,7 +30,7 @@
   {
     "name": "all",
     "path": "build/*",
-    "limit": "851.1 kB",
+    "limit": "852.5 kB",
     "brotli": false,
     "gzip": false
   }

docs/api.md

@@ -140,6 +140,15 @@ package.
 
 ```js
 const resp = await fetch('https://medv.io')
+const json = await resp.json()
+```
+
+For some cases, `text()` or `json()` can produce extremely large output that exceeds the string size limit.
+Streams are just for that, so we've attached a minor adjustment to the `fetch` API to make it more pipe friendly.
+
+```js
+const p1 = fetch('https://example.com').pipe($`cat`)
+const p2 = fetch('https://example.com').pipe`cat`
 ```
 
 ## `question()`

src/goods.ts

@@ -14,6 +14,7 @@
 
 import assert from 'node:assert'
 import { createInterface } from 'node:readline'
+import { Readable } from 'node:stream'
 import { $, within, ProcessOutput } from './core.ts'
 import {
   type Duration,
@@ -63,12 +64,43 @@ export function sleep(duration: Duration): Promise<void> {
   })
 }
 
-export async function fetch(
+const responseToReadable = (response: Response, rs: Readable) => {
+  const reader = response.body?.getReader()
+  if (!reader) {
+    rs.push(null)
+    return rs
+  }
+  rs._read = async () => {
+    const result = await reader.read()
+    if (!result.done) rs.push(Buffer.from(result.value))
+    else rs.push(null)
+  }
+  return rs
+}
+
+export function fetch(
   url: RequestInfo,
   init?: RequestInit
-): Promise<Response> {
+): Promise<Response> & { pipe: <D>(dest: D) => D } {
   $.log({ kind: 'fetch', url, init, verbose: !$.quiet && $.verbose })
-  return nodeFetch(url, init)
+  const p = nodeFetch(url, init)
+
+  return Object.assign(p, {
+    pipe(dest: any, ...args: any[]) {
+      const rs = new Readable()
+      const _dest = isStringLiteral(dest, ...args)
+        ? $({
+            halt: true,
+            signal: init?.signal as AbortSignal,
+          })(dest as TemplateStringsArray, ...args)
+        : dest
+      p.then(
+        (r) => responseToReadable(r, rs).pipe(_dest.run?.()),
+        (err) => _dest.abort?.(err)
+      )
+      return _dest
+    },
+  })
 }
 
 export function echo(...args: any[]): void

test/core.test.js

@@ -43,6 +43,7 @@ import {
   quiet,
   which,
   nothrow,
+  fetch,
 } from '../build/index.js'
 import { noop } from '../build/util.js'
 
@@ -716,6 +717,37 @@ describe('core', () => {
           assert.equal(stdout, 'TEST')
         })
 
+        test('fetch (stream) > $', async () => {
+          // stream.Readable.fromWeb requires Node.js 18+
+          const responseToReadable = (response) => {
+            const reader = response.body.getReader()
+            const rs = new Readable()
+            rs._read = async () => {
+              const result = await reader.read()
+              if (!result.done) rs.push(Buffer.from(result.value))
+              else rs.push(null)
+            }
+            return rs
+          }
+
+          const p = (
+            await fetch('https://example.com').then(responseToReadable)
+          ).pipe($`cat`)
+          const o = await p
+
+          assert.match(o.stdout, /Example Domain/)
+        })
+
+        test('fetch (pipe) > $', async () => {
+          const p1 = fetch('https://example.com').pipe($`cat`)
+          const p2 = fetch('https://example.com').pipe`cat`
+          const o1 = await p1
+          const o2 = await p2
+
+          assert.match(o1.stdout, /Example Domain/)
+          assert.equal(o1.stdout, o2.stdout)
+        })
+
         test('$ > stream > $', async () => {
           const p = $`echo "hello"`
           const { stdout } = await p.pipe(getUpperCaseTransform()).pipe($`cat`)