feat: add, addPullStream, addFromFs and addFromStream

License: MIT
Signed-off-by: Alan Shaw <alan.shaw@protocol.ai>

package.json

@@ -19,7 +19,8 @@
     "fs": false,
     "stream": "readable-stream",
     "./src/lib/configure.js": "./src/lib/configure.browser.js",
-    "./src/lib/querystring.js": "./src/lib/querystring.browser.js"
+    "./src/lib/querystring.js": "./src/lib/querystring.browser.js",
+    "./src/add/form-data.js": "./src/add/form-data.browser.js"
   },
   "repository": "github:ipfs/js-ipfs-http-client",
   "scripts": {
@@ -37,6 +38,7 @@
   "dependencies": {
     "abort-controller": "^3.0.0",
     "async": "^2.6.1",
+    "async-iterator-to-pull-stream": "^1.3.0",
     "bignumber.js": "^9.0.0",
     "bl": "^3.0.0",
     "bs58": "^4.0.1",
@@ -78,6 +80,7 @@
     "promisify-es6": "^1.0.3",
     "pull-defer": "~0.2.3",
     "pull-stream": "^3.6.9",
+    "pull-stream-to-async-iterator": "^1.0.2",
     "pull-to-stream": "~0.1.1",
     "pump": "^3.0.0",
     "qs": "^6.5.2",

src/add-from-url.js

@@ -0,0 +1,25 @@
+'use strict'
+
+const configure = require('./lib/configure')
+const { ok, toIterable } = require('./lib/fetch')
+
+module.exports = configure(({ fetch, apiAddr, apiPath, headers }) => {
+  const add = require('./add')({ fetch, apiAddr, apiPath, headers })
+
+  return (url, options) => (async function * () {
+    options = options || {}
+    const res = await ok(fetch(url, {
+      signal: options.signal,
+      headers: options.headers || headers
+    }))
+
+    const input = {
+      path: decodeURIComponent(new URL(url).pathname.split('/').pop() || ''),
+      content: toIterable(res.body)
+    }
+
+    for await (const file of add(input, options)) {
+      yield file
+    }
+  })()
+})

src/add/form-data.browser.js

@@ -0,0 +1,30 @@
+'use strict'
+/* eslint-env browser */
+
+const normaliseInput = require('./normalise-input')
+
+exports.toFormData = async (input) => {
+  const files = normaliseInput(input)
+  const formData = new FormData()
+  let i = 0
+
+  for await (const file of files) {
+    if (file.content) {
+      // In the browser there's _currently_ no streaming upload, buffer up our
+      // async iterator chunks and append a big Blob :(
+      // One day, this will be browser streams
+      const bufs = []
+      for await (const chunk of file.content) {
+        bufs.push(Buffer.isBuffer(chunk) ? chunk.buffer : chunk)
+      }
+
+      formData.append(`file-${i}`, new Blob(bufs, { type: 'application/octet-stream' }), file.path)
+    } else {
+      formData.append(`dir-${i}`, new Blob([], { type: 'application/x-directory' }), file.path)
+    }
+
+    i++
+  }
+
+  return formData
+}

src/add/form-data.js

@@ -0,0 +1,42 @@
+'use strict'
+
+const FormData = require('form-data')
+const { Buffer } = require('buffer')
+const normaliseInput = require('./normalise-input')
+const toStream = require('../lib/iterable-to-readable-stream')
+
+exports.toFormData = async (input) => {
+  const files = normaliseInput(input)
+  const formData = new FormData()
+  let i = 0
+
+  for await (const file of files) {
+    if (file.content) {
+      // In Node.js, FormData can be passed a stream so no need to buffer
+      formData.append(
+        `file-${i}`,
+        // FIXME: add a `path` property to the stream so `form-data` doesn't set
+        // a Content-Length header that is only the sum of the size of the
+        // header/footer when knownLength option (below) is null.
+        Object.assign(
+          toStream(file.content),
+          { path: file.path || `file-${i}` }
+        ),
+        {
+          filepath: file.path,
+          contentType: 'application/octet-stream',
+          knownLength: file.content.length // Send Content-Length header if known
+        }
+      )
+    } else {
+      formData.append(`dir-${i}`, Buffer.alloc(0), {
+        filepath: file.path,
+        contentType: 'application/x-directory'
+      })
+    }
+
+    i++
+  }
+
+  return formData
+}

src/add/index.js

@@ -0,0 +1,56 @@
+'use strict'
+
+const ndjson = require('iterable-ndjson')
+const { objectToQuery } = require('../lib/querystring')
+const configure = require('../lib/configure')
+const { ok, toIterable } = require('../lib/fetch')
+const { toFormData } = require('./form-data')
+const toCamel = require('../lib/object-to-camel')
+
+module.exports = configure(({ fetch, apiAddr, apiPath, headers }) => {
+  return (input, options) => (async function * () {
+    options = options || {}
+
+    const qs = objectToQuery({
+      'stream-channels': true,
+      chunker: options.chunker,
+      'cid-version': options.cidVersion,
+      'cid-base': options.cidBase,
+      'enable-sharding-experiment': options.enableShardingExperiment,
+      hash: options.hashAlg,
+      'only-hash': options.onlyHash,
+      pin: options.pin,
+      progress: options.progress ? true : null,
+      quiet: options.quiet,
+      quieter: options.quieter,
+      'raw-leaves': options.rawLeaves,
+      'shard-split-threshold': options.shardSplitThreshold,
+      silent: options.silent,
+      trickle: options.trickle,
+      'wrap-with-directory': options.wrapWithDirectory,
+      ...(options.qs || {})
+    })
+
+    const url = `${apiAddr}${apiPath}/add${qs}`
+    const res = await ok(fetch(url, {
+      method: 'POST',
+      signal: options.signal,
+      headers: options.headers || headers,
+      body: await toFormData(input)
+    }))
+
+    for await (let file of ndjson(toIterable(res.body))) {
+      file = toCamel(file)
+      // console.log(file)
+      if (options.progress && file.bytes) {
+        options.progress(file.bytes)
+      } else {
+        yield toCoreInterface(file)
+      }
+    }
+  })()
+})
+
+function toCoreInterface ({ name, hash, size }) {
+  return { path: name, hash, size: parseInt(size) }
+}

src/add/normalise-input.js

@@ -0,0 +1,130 @@
+'use strict'
+/* eslint-env browser */
+
+const { Buffer } = require('buffer')
+const errCode = require('err-code')
+const toAsyncIterable = require('../lib/file-data-to-async-iterable')
+
+/*
+Transform one of:
+
+Buffer|ArrayBuffer|TypedArray
+Blob|File
+{ path, content: Buffer }
+{ path, content: Blob }
+{ path, content: Iterable<Buffer> }
+{ path, content: AsyncIterable<Buffer> }
+{ path, content: PullStream<Buffer> }
+Iterable<Number>
+Iterable<{ path, content: Buffer }>
+Iterable<{ path, content: Blob }>
+Iterable<{ path, content: Iterable<Number> }>
+Iterable<{ path, content: AsyncIterable<Buffer> }>
+Iterable<{ path, content: PullStream<Buffer> }>
+AsyncIterable<Buffer>
+AsyncIterable<{ path, content: Buffer }>
+AsyncIterable<{ path, content: Blob }>
+AsyncIterable<{ path, content: Iterable<Buffer> }>
+AsyncIterable<{ path, content: AsyncIterable<Buffer> }>
+AsyncIterable<{ path, content: PullStream<Buffer> }>
+PullStream<Buffer>
+
+Into:
+
+AsyncIterable<{ path, content: AsyncIterable<Buffer> }>
+*/
+
+module.exports = function normalizeInput (input) {
+  // Buffer|ArrayBuffer|TypedArray
+  if (Buffer.isBuffer(input) || ArrayBuffer.isView(input) || input instanceof ArrayBuffer) {
+    return (async function * () { // eslint-disable-line require-await
+      yield normalizeTuple({ path: '', content: input })
+    })()
+  }
+
+  // Blob|File
+  if (typeof Blob !== 'undefined' && input instanceof Blob) {
+    return (async function * () { // eslint-disable-line require-await
+      yield normalizeTuple({ path: '', content: input })
+    })()
+  }
+
+  // Iterable<Number>
+  // Iterable<{ path, content: Buffer }>
+  // Iterable<{ path, content: Blob }>
+  // Iterable<{ path, content: Iterable<Number> }>
+  // Iterable<{ path, content: AsyncIterable<Buffer> }>
+  // Iterable<{ path, content: PullStream<Buffer> }>
+  if (input[Symbol.iterator]) {
+    return (async function * () { // eslint-disable-line require-await
+      for (const chunk of input) {
+        if (typeof chunk === 'object' && (chunk.path || chunk.content)) {
+          yield normalizeTuple(chunk)
+        } else if (Number.isInteger(chunk)) { // Must be an Iterable<Number> i.e. Buffer/ArrayBuffer/Array of bytes
+          yield normalizeTuple({ path: '', content: input })
+          return
+        } else {
+          throw errCode(new Error('Unexpected input: ' + typeof chunk), 'ERR_UNEXPECTED_INPUT')
+        }
+      }
+    })()
+  }
+
+  // AsyncIterable<Buffer>
+  // AsyncIterable<{ path, content: Buffer }>
+  // AsyncIterable<{ path, content: Blob }>
+  // AsyncIterable<{ path, content: Iterable<Buffer> }>
+  // AsyncIterable<{ path, content: AsyncIterable<Buffer> }>
+  // AsyncIterable<{ path, content: PullStream<Buffer> }>
+  if (input[Symbol.asyncIterator]) {
+    return (async function * () {
+      for await (const chunk of input) {
+        if (typeof chunk === 'object' && (chunk.path || chunk.content)) {
+          yield normalizeTuple(chunk)
+        } else { // Must be an AsyncIterable<Buffer> i.e. a Stream
+          let path = ''
+
+          // fs.createReadStream will create a stream with a `path` prop
+          // If available, use it here!
+          if (input.path && input.path.split) {
+            path = input.path.split(/[/\\]/).pop() || ''
+          }
+
+          yield normalizeTuple({
+            path,
+            content: (async function * () {
+              yield chunk
+              for await (const restChunk of input) {
+                yield restChunk
+              }
+            })()
+          })
+          return
+        }
+      }
+    })()
+  }
+
+  // { path, content: Buffer }
+  // { path, content: Blob }
+  // { path, content: Iterable<Buffer> }
+  // { path, content: AsyncIterable<Buffer> }
+  // { path, content: PullStream<Buffer> }
+  if (typeof input === 'object' && (input.path || input.content)) {
+    // eslint-disable-next-line require-await
+    return (async function * () { yield normalizeTuple(input) })()
+  }
+
+  // PullStream
+  if (typeof input === 'function') {
+    return (async function * () { // eslint-disable-line require-await
+      yield normalizeTuple({ path: '', content: input })
+    })()
+  }
+
+  throw errCode(new Error('Unexpected input: ' + typeof input), 'ERR_UNEXPECTED_INPUT')
+}
+
+function normalizeTuple ({ path, content }) {
+  return { path: path || '', content: content ? toAsyncIterable(content) : null }
+}