Handle case where ky responses have no body with a getter for a ReadableStream

.gitignore

@@ -12,3 +12,5 @@ coverage
 examples/sub-module/**/bundle.js
 examples/sub-module/**/*-minified.js
 examples/sub-module/*-bundle.js
+
+.vscode/

src/add-from-url.js

@@ -1,19 +1,19 @@
 'use strict'
 
 const kyDefault = require('ky-universal').default
-const toIterable = require('./lib/stream-to-iterable')
+const toAsyncIterable = require('./lib/stream-to-async-iterable')
 
-module.exports = (config) => {
+module.exports = config => {
   const add = require('./add')(config)
 
   return async function * addFromURL (url, options) {
     options = options || {}
 
-    const { body } = await kyDefault.get(url)
+    const res = await kyDefault.get(url)
 
     const input = {
       path: decodeURIComponent(new URL(url).pathname.split('/').pop() || ''),
-      content: toIterable(body)
+      content: toAsyncIterable(res)
     }
 
     yield * add(input, options)

src/add/index.js

@@ -2,7 +2,7 @@
 
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 const { toFormData } = require('./form-data')
 const toCamel = require('../lib/object-to-camel')
 
@@ -16,21 +16,36 @@ module.exports = configure(({ ky }) => {
     if (options.chunker) searchParams.set('chunker', options.chunker)
     if (options.cidVersion) searchParams.set('cid-version', options.cidVersion)
     if (options.cidBase) searchParams.set('cid-base', options.cidBase)
-    if (options.enableShardingExperiment != null) searchParams.set('enable-sharding-experiment', options.enableShardingExperiment)
+    if (options.enableShardingExperiment != null) {
+      searchParams.set(
+        'enable-sharding-experiment',
+        options.enableShardingExperiment
+      )
+    }
     if (options.hashAlg) searchParams.set('hash', options.hashAlg)
-    if (options.onlyHash != null) searchParams.set('only-hash', options.onlyHash)
+    if (options.onlyHash != null) { searchParams.set('only-hash', options.onlyHash) }
     if (options.pin != null) searchParams.set('pin', options.pin)
     if (options.progress) searchParams.set('progress', true)
     if (options.quiet != null) searchParams.set('quiet', options.quiet)
     if (options.quieter != null) searchParams.set('quieter', options.quieter)
-    if (options.rawLeaves != null) searchParams.set('raw-leaves', options.rawLeaves)
-    if (options.shardSplitThreshold) searchParams.set('shard-split-threshold', options.shardSplitThreshold)
+    if (options.rawLeaves != null) { searchParams.set('raw-leaves', options.rawLeaves) }
+    if (options.shardSplitThreshold) { searchParams.set('shard-split-threshold', options.shardSplitThreshold) }
     if (options.silent) searchParams.set('silent', options.silent)
     if (options.trickle != null) searchParams.set('trickle', options.trickle)
-    if (options.wrapWithDirectory != null) searchParams.set('wrap-with-directory', options.wrapWithDirectory)
+    if (options.wrapWithDirectory != null) { searchParams.set('wrap-with-directory', options.wrapWithDirectory) }
     if (options.preload != null) searchParams.set('preload', options.preload)
-    if (options.fileImportConcurrency != null) searchParams.set('file-import-concurrency', options.fileImportConcurrency)
-    if (options.blockWriteConcurrency != null) searchParams.set('block-write-concurrency', options.blockWriteConcurrency)
+    if (options.fileImportConcurrency != null) {
+      searchParams.set(
+        'file-import-concurrency',
+        options.fileImportConcurrency
+      )
+    }
+    if (options.blockWriteConcurrency != null) {
+      searchParams.set(
+        'block-write-concurrency',
+        options.blockWriteConcurrency
+      )
+    }
 
     const res = await ky.post('add', {
       timeout: options.timeout,
@@ -40,7 +55,7 @@ module.exports = configure(({ ky }) => {
       body: await toFormData(input)
     })
 
-    for await (let file of ndjson(toIterable(res.body))) {
+    for await (let file of ndjson(toAsyncIterable(res))) {
       file = toCamel(file)
       // console.log(file)
       if (options.progress && file.bytes) {

src/block/rm-async-iterator.js

@@ -3,7 +3,7 @@
 const CID = require('cids')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 const toCamel = require('../lib/object-to-camel')
 
 module.exports = configure(({ ky }) => {
@@ -30,7 +30,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const removed of ndjson(toIterable(res.body))) {
+    for await (const removed of ndjson(toAsyncIterable(res))) {
       yield toCamel(removed)
     }
   }

src/cat.js

@@ -3,7 +3,7 @@
 const CID = require('cids')
 const { Buffer } = require('buffer')
 const configure = require('./lib/configure')
-const toIterable = require('./lib/stream-to-iterable')
+const toAsyncIterable = require('./lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * cat (path, options) {
@@ -27,7 +27,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const chunk of toIterable(res.body)) {
+    for await (const chunk of toAsyncIterable(res)) {
       yield Buffer.from(chunk)
     }
   }

src/dht/find-peer.js

@@ -5,7 +5,7 @@ const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * findPeer (peerId, options) {
@@ -22,7 +22,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const message of ndjson(toIterable(res.body))) {
+    for await (const message of ndjson(toAsyncIterable(res))) {
       // 2 = FinalPeer
       // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
       if (message.Type === 2 && message.Responses) {

src/dht/find-provs.js

@@ -5,15 +5,15 @@ const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * findProvs (cid, options) {
     options = options || {}
 
     const searchParams = new URLSearchParams(options.searchParams)
     searchParams.set('arg', `${cid}`)
-    if (options.numProviders) searchParams.set('num-providers', options.numProviders)
+    if (options.numProviders) { searchParams.set('num-providers', options.numProviders) }
     if (options.verbose != null) searchParams.set('verbose', options.verbose)
 
     const res = await ky.post('dht/findprovs', {
@@ -23,7 +23,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const message of ndjson(toIterable(res.body))) {
+    for await (const message of ndjson(toAsyncIterable(res))) {
       // 4 = Provider
       // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L20
       if (message.Type === 4 && message.Responses) {

src/dht/get.js

@@ -2,7 +2,7 @@
 
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * get (key, options) {
@@ -19,7 +19,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const message of ndjson(toIterable(res.body))) {
+    for await (const message of ndjson(toAsyncIterable(res))) {
       // 5 = Value
       // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L21
       if (message.Type === 5) {

src/dht/provide.js

@@ -5,7 +5,7 @@ const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 const toCamel = require('../lib/object-to-camel')
 
 module.exports = configure(({ ky }) => {
@@ -15,7 +15,7 @@ module.exports = configure(({ ky }) => {
 
     const searchParams = new URLSearchParams(options.searchParams)
     cids.forEach(cid => searchParams.append('arg', `${cid}`))
-    if (options.recursive != null) searchParams.set('recursive', options.recursive)
+    if (options.recursive != null) { searchParams.set('recursive', options.recursive) }
     if (options.verbose != null) searchParams.set('verbose', options.verbose)
 
     const res = await ky.post('dht/provide', {
@@ -25,7 +25,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (let message of ndjson(toIterable(res.body))) {
+    for await (let message of ndjson(toAsyncIterable(res))) {
       message = toCamel(message)
       if (message.responses) {
         message.responses = message.responses.map(({ ID, Addrs }) => {

src/dht/put.js

@@ -5,7 +5,7 @@ const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 const encodeBufferURIComponent = require('../lib/encode-buffer-uri-component')
 const toCamel = require('../lib/object-to-camel')
 
@@ -16,8 +16,12 @@ module.exports = configure(({ ky }) => {
     const searchParams = new URLSearchParams(options.searchParams)
     if (options.verbose != null) searchParams.set('verbose', options.verbose)
 
-    key = Buffer.isBuffer(key) ? encodeBufferURIComponent(key) : encodeURIComponent(key)
-    value = Buffer.isBuffer(value) ? encodeBufferURIComponent(value) : encodeURIComponent(value)
+    key = Buffer.isBuffer(key)
+      ? encodeBufferURIComponent(key)
+      : encodeURIComponent(key)
+    value = Buffer.isBuffer(value)
+      ? encodeBufferURIComponent(value)
+      : encodeURIComponent(value)
 
     const url = `dht/put?arg=${key}&arg=${value}&${searchParams}`
     const res = await ky.post(url, {
@@ -26,7 +30,7 @@ module.exports = configure(({ ky }) => {
       headers: options.headers
     })
 
-    for await (let message of ndjson(toIterable(res.body))) {
+    for await (let message of ndjson(toAsyncIterable(res))) {
       message = toCamel(message)
       if (message.responses) {
         message.responses = message.responses.map(({ ID, Addrs }) => {

src/dht/query.js

@@ -4,7 +4,7 @@ const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * query (peerId, options) {
@@ -21,7 +21,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const message of ndjson(toIterable(res.body))) {
+    for await (const message of ndjson(toAsyncIterable(res))) {
       yield new PeerInfo(PeerId.createFromB58String(message.ID))
     }
   }

src/files/ls.js

@@ -2,7 +2,7 @@
 
 const CID = require('cids')
 const ndjson = require('iterable-ndjson')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 const configure = require('../lib/configure')
 const toCamelWithMetadata = require('../lib/object-to-camel-with-metadata')
 
@@ -28,7 +28,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const result of ndjson(toIterable(res.body))) {
+    for await (const result of ndjson(toAsyncIterable(res))) {
       // go-ipfs does not yet support the "stream" option
       if ('Entries' in result) {
         for (const entry of result.Entries || []) {

src/files/read.js

@@ -2,7 +2,7 @@
 
 const { Buffer } = require('buffer')
 const configure = require('../lib/configure')
-const toIterable = require('../lib/stream-to-iterable')
+const toAsyncIterable = require('../lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * read (path, options) {
@@ -20,7 +20,7 @@ module.exports = configure(({ ky }) => {
       searchParams
     })
 
-    for await (const chunk of toIterable(res.body)) {
+    for await (const chunk of toAsyncIterable(res)) {
       yield Buffer.from(chunk)
     }
   }

src/get.js

@@ -4,7 +4,7 @@ const configure = require('./lib/configure')
 const Tar = require('it-tar')
 const { Buffer } = require('buffer')
 const CID = require('cids')
-const toIterable = require('./lib/stream-to-iterable')
+const toAsyncIterable = require('./lib/stream-to-async-iterable')
 
 module.exports = configure(({ ky }) => {
   return async function * get (path, options) {
@@ -38,7 +38,7 @@ module.exports = configure(({ ky }) => {
 
     const extractor = Tar.extract()
 
-    for await (const { header, body } of extractor(toIterable(res.body))) {
+    for await (const { header, body } of extractor(toAsyncIterable(res))) {
       if (header.type === 'directory') {
         yield {
           path: header.name

src/lib/stream-to-async-iterable.js

@@ -0,0 +1,40 @@
+'use strict'
+
+module.exports = function toAsyncIterable (res) {
+  const { body } = res
+
+  // An env where res.body getter for ReadableStream with getReader
+  // is not supported, for example in React Native
+  if (!body) {
+    if (res.arrayBuffer) {
+      return (async function * () {
+        const arrayBuffer = await res.arrayBuffer()
+        yield arrayBuffer
+      })()
+    } else {
+      throw new Error('Neither Response.body nor Response.arrayBuffer is defined')
+    }
+  }
+
+  // Node.js stream
+  if (body[Symbol.asyncIterator]) return body
+
+  // Browser ReadableStream
+  if (body.getReader) {
+    return (async function * () {
+      const reader = body.getReader()
+
+      try {
+        while (true) {
+          const { done, value } = await reader.read()
+          if (done) return
+          yield value
+        }
+      } finally {
+        reader.releaseLock()
+      }
+    })()
+  }
+
+  throw new Error('unknown stream')
+}