chore: move withTimeoutOption to core-utils

packages/interface-ipfs-core/package.json

@@ -48,6 +48,7 @@
  "is-ipfs": "^2.0.0",
  "iso-random-stream": "^1.1.1",
  "it-all": "^1.0.4",
+ "it-buffer-stream": "^1.0.5",
  "it-concat": "^1.0.1",
  "it-drain": "^1.0.3",
  "it-last": "^1.0.4",

packages/interface-ipfs-core/src/add-all.js

@@ -15,6 +15,7 @@ const { isNode } = require('ipfs-utils/src/env')
 const { getDescribe, getIt, expect } = require('./utils/mocha')
 const testTimeout = require('./utils/test-timeout')
 const uint8ArrayFromString = require('uint8arrays/from-string')
+const bufferStream = require('it-buffer-stream')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
 /**
@@ -420,5 +421,84 @@ module.exports = (common, options) => {
  expect(files[0].cid.codec).to.equal('dag-pb')
  expect(files[0].size).to.equal(18)
  })
+
+ it('should support bidirectional streaming', async function () {
+ let progressInvoked
+
+ const handler = (bytes, path) => {
+ progressInvoked = true
+ }
+
+ const source = async function * () {
+ yield {
+ content: 'hello',
+ path: '/file'
+ }
+
+ await new Promise((resolve) => {
+ const interval = setInterval(() => {
+ // we've received a progress result, that means we've received some
+ // data from the server before we're done sending data to the server
+ // so the streaming is bidirectional and we can finish up
+ if (progressInvoked) {
+ clearInterval(interval)
+ resolve()
+ }
+ }, 10)
+ })
+ }
+
+ await drain(ipfs.addAll(source(), {
+ progress: handler,
+ fileImportConcurrency: 1
+ }))
+
+ expect(progressInvoked).to.be.true()
+ })
+
+ it('should error during add-all stream', async function () {
+ const source = async function * () {
+ yield {
+ content: 'hello',
+ path: '/file'
+ }
+
+ yield {
+ content: 'hello',
+ path: '/file'
+ }
+ }
+
+ await expect(drain(ipfs.addAll(source(), {
+ fileImportConcurrency: 1,
+ chunker: 'rabin-2048--50' // invalid chunker parameters, validated after the stream starts moving
+ }))).to.eventually.be.rejectedWith(/Chunker parameter avg must be an integer/)
+ })
+
+ it('should add big files', async function () {
+ const totalSize = 1024 * 1024 * 200
+ const chunkSize = 1024 * 1024 * 99
+
+ const source = async function * () {
+ yield {
+ path: '/dir/file-200mb-1',
+ content: bufferStream(totalSize, {
+ chunkSize
+ })
+ }
+
+ yield {
+ path: '/dir/file-200mb-2',
+ content: bufferStream(totalSize, {
+ chunkSize
+ })
+ }
+ }
+
+ const results = await all(ipfs.addAll(source()))
+
+ expect(await ipfs.files.stat(`/ipfs/${results[0].cid}`)).to.have.property('size', totalSize)
+ expect(await ipfs.files.stat(`/ipfs/${results[1].cid}`)).to.have.property('size', totalSize)
+ })
  })
 }

packages/ipfs-cli/tsconfig.json

@@ -9,14 +9,17 @@
  ],
  "references": [
  {
- "path": "../ipfs-core-utils"
+ "path": "../ipfs-core"
  },
  {
- "path": "../ipfs-core"
+ "path": "../ipfs-core-utils"
  },
  {
  "path": "../ipfs-http-client"
  },
+ {
+ "path": "../ipfs-http-gateway"
+ },
  {
  "path": "../ipfs-http-server"
  }

packages/ipfs-core-utils/package.json

@@ -38,6 +38,7 @@
  },
  "license": "MIT",
  "dependencies": {
+ "any-signal": "^2.0.0",
  "blob-to-it": "^1.0.1",
  "browser-readablestream-to-it": "^1.0.1",
  "cids": "^1.0.0",
@@ -48,6 +49,8 @@
  "it-peekable": "^1.0.1",
  "multiaddr": "^8.0.0",
  "multiaddr-to-uri": "^6.0.0",
+ "parse-duration": "^0.4.4",
+ "timeout-abort-controller": "^1.1.1",
  "uint8arrays": "^1.1.0"
  },
  "devDependencies": {

packages/ipfs-core-utils/src/errors.js

@@ -0,0 +1,12 @@
+'use strict'
+
+class TimeoutError extends Error {
+ constructor (message = 'request timed out') {
+ super(message)
+ this.name = 'TimeoutError'
+ this.code = TimeoutError.code
+ }
+}
+
+TimeoutError.code = 'ERR_TIMEOUT'
+exports.TimeoutError = TimeoutError

packages/ipfs-core-utils/src/files/normalise-input/normalise-input.js

@@ -8,7 +8,9 @@ const {
  isBytes,
  isBlob,
  isReadableStream,
- isFileObject
+ isFileObject,
+ mtimeToObject,
+ modeToNumber
 } = require('./utils')
 
 // eslint-disable-next-line complexity
@@ -105,7 +107,8 @@ module.exports = async function * normaliseInput (input, normaliseContent) {
 async function toFileObject (input, normaliseContent) {
  // @ts-ignore - Those properties don't exist on most input types
  const { path, mode, mtime, content } = input
- const file = { path: path || '', mode, mtime }
+
+ const file = { path: path || '', mode: modeToNumber(mode), mtime: mtimeToObject(mtime) }
 
  if (content) {
  file.content = await normaliseContent(content)

packages/ipfs-core-utils/src/files/normalise-input/utils.js

@@ -35,9 +35,94 @@ function isFileObject (obj) {
 const isReadableStream = (value) =>
  value && typeof value.getReader === 'function'
 
+/**
+ * @param {any} mtime
+ * @returns {{secs:number, nsecs:number}|undefined}
+ */
+function mtimeToObject (mtime) {
+ if (mtime == null) {
+ return undefined
+ }
+
+ // Javascript Date
+ if (mtime instanceof Date) {
+ const ms = mtime.getTime()
+ const secs = Math.floor(ms / 1000)
+
+ return {
+ secs: secs,
+ nsecs: (ms - (secs * 1000)) * 1000
+ }
+ }
+
+ // { secs, nsecs }
+ if (Object.prototype.hasOwnProperty.call(mtime, 'secs')) {
+ return {
+ secs: mtime.secs,
+ nsecs: mtime.nsecs
+ }
+ }
+
+ // UnixFS TimeSpec
+ if (Object.prototype.hasOwnProperty.call(mtime, 'Seconds')) {
+ return {
+ secs: mtime.Seconds,
+ nsecs: mtime.FractionalNanoseconds
+ }
+ }
+
+ // process.hrtime()
+ if (Array.isArray(mtime)) {
+ return {
+ secs: mtime[0],
+ nsecs: mtime[1]
+ }
+ }
+ /*
+ TODO: https://github.com/ipfs/aegir/issues/487
+
+ // process.hrtime.bigint()
+ if (typeof mtime === 'bigint') {
+ const secs = mtime / BigInt(1e9)
+ const nsecs = mtime - (secs * BigInt(1e9))
+
+ return {
+ secs: parseInt(secs),
+ nsecs: parseInt(nsecs)
+ }
+ }
+ */
+}
+
+/**
+ * @param {any} mode
+ * @returns {number|undefined}
+ */
+function modeToNumber (mode) {
+ if (mode == null) {
+ return undefined
+ }
+
+ if (typeof mode === 'number') {
+ return mode
+ }
+
+ mode = mode.toString()
+
+ if (mode.substring(0, 1) === '0') {
+ // octal string
+ return parseInt(mode, 8)
+ }
+
+ // decimal string
+ return parseInt(mode, 10)
+}
+
 module.exports = {
  isBytes,
  isBlob,
  isFileObject,
- isReadableStream
+ isReadableStream,
+ mtimeToObject,
+ modeToNumber
 }

packages/ipfs-core-utils/src/index.js

@@ -1 +1,7 @@
 'use strict'
+
+/**
+ * @template {any[]} ARGS
+ * @template R
+ * @typedef {(...args: ARGS) => R} Fn
+ */

packages/ipfs-core-utils/src/with-timeout-option.js

@@ -0,0 +1,106 @@
+/* eslint-disable no-unreachable */
+'use strict'
+
+const TimeoutController = require('timeout-abort-controller')
+const { anySignal } = require('any-signal')
+const parseDuration = require('parse-duration').default
+const { TimeoutError } = require('./errors')
+
+/**
+ * @template {any[]} ARGS
+ * @template {Promise<any> | AsyncIterable<any>} R - The return type of `fn`
+ * @param {Fn<ARGS, R>} fn
+ * @param {number} [optionsArgIndex]
+ * @returns {Fn<ARGS, R>}
+ */
+function withTimeoutOption (fn, optionsArgIndex) {
+ // eslint-disable-next-line
+ return /** @returns {R} */(/** @type {ARGS} */...args) => {
+ const options = args[optionsArgIndex == null ? args.length - 1 : optionsArgIndex]
+ if (!options || !options.timeout) return fn(...args)
+
+ const timeout = typeof options.timeout === 'string'
+ ? parseDuration(options.timeout)
+ : options.timeout
+
+ const controller = new TimeoutController(timeout)
+
+ options.signal = anySignal([options.signal, controller.signal])
+
+ const fnRes = fn(...args)
+ // eslint-disable-next-line promise/param-names
+ const timeoutPromise = new Promise((_resolve, reject) => {
+ controller.signal.addEventListener('abort', () => {
+ reject(new TimeoutError())
+ })
+ })
+
+ const start = Date.now()
+
+ const maybeThrowTimeoutError = () => {
+ if (controller.signal.aborted) {
+ throw new TimeoutError()
+ }
+
+ const timeTaken = Date.now() - start
+
+ // if we have starved the event loop by adding microtasks, we could have
+ // timed out already but the TimeoutController will never know because it's
+ // setTimeout will not fire until we stop adding microtasks
+ if (timeTaken > timeout) {
+ controller.abort()
+ throw new TimeoutError()
+ }
+ }
+
+ if (fnRes[Symbol.asyncIterator]) {
+ // @ts-ignore
+ return (async function * () {
+ const it = fnRes[Symbol.asyncIterator]()
+
+ try {
+ while (true) {
+ const { value, done } = await Promise.race([it.next(), timeoutPromise])
+
+ if (done) {
+ break
+ }
+
+ maybeThrowTimeoutError()
+
+ yield value
+ }
+ } catch (err) {
+ maybeThrowTimeoutError()
+
+ throw err
+ } finally {
+ controller.clear()
+
+ if (it.return) {
+ it.return()
+ }
+ }
+ })()
+ }
+
+ // @ts-ignore
+ return (async () => {
+ try {
+ const res = await Promise.race([fnRes, timeoutPromise])
+
+ maybeThrowTimeoutError()
+
+ return res
+ } catch (err) {
+ maybeThrowTimeoutError()
+
+ throw err
+ } finally {
+ controller.clear()
+ }
+ })()
+ }
+}
+
+module.exports = withTimeoutOption

packages/ipfs-core/package.json

@@ -54,7 +54,6 @@
  "dep-check": "aegir dep-check -i typescript -i interface-ipfs-core"
  },
  "dependencies": {
- "any-signal": "^2.0.0",
  "array-shuffle": "^1.0.1",
  "bignumber.js": "^9.0.0",
  "cbor": "^5.1.0",
@@ -116,7 +115,6 @@
  "parse-duration": "^0.4.4",
  "peer-id": "^0.14.1",
  "streaming-iterables": "^5.0.2",
- "timeout-abort-controller": "^1.1.1",
  "uint8arrays": "^1.1.0"
  },
  "devDependencies": {