refactor: convert block and pin APIs to async/await

src/cli/commands/block/rm.js

@@ -25,7 +25,7 @@ module.exports = {
  const ipfs = await getIpfs()
  let errored = false
 
- for await (const result of ipfs.block._rmAsyncIterator(hash, {
+ for await (const result of ipfs.block.rm(hash, {
  force,
  quiet
  })) {

src/cli/commands/dag/resolve.js

@@ -19,10 +19,9 @@ module.exports = {
  const options = {}
 
  try {
- const result = await ipfs.dag.resolve(ref, options)
  let lastCid
 
- for (const res of result) {
+ for await (const res of ipfs.dag.resolve(ref, options)) {
  if (CID.isCID(res.value)) {
  lastCid = res.value
  }

src/cli/commands/pin/ls.js

@@ -34,14 +34,13 @@ module.exports = {
  resolve((async () => {
  const paths = ipfsPath
  const ipfs = await getIpfs()
- const results = await ipfs.pin.ls(paths, { type })
- results.forEach((res) => {
+ for await (const res of ipfs.pin.ls(paths, { type })) {
  let line = cidToString(res.hash, { base: cidBase })
  if (!quiet) {
  line += ` ${res.type}`
  }
  print(line)
- })
+ }
  })())
  }
 }

src/core/components/block/get.js

@@ -0,0 +1,16 @@
+'use strict'
+
+const { cleanCid } = require('./utils')
+
+module.exports = ({ blockService, preload }) => {
+ return async function get (cid, options) { // eslint-disable-line require-await
+ options = options || {}
+ cid = cleanCid(cid)
+
+ if (options.preload !== false) {
+ preload(cid)
+ }
+
+ return blockService.get(cid)
+ }
+}

src/core/components/block/put.js

@@ -0,0 +1,51 @@
+'use strict'
+
+const Block = require('ipfs-block')
+const multihashing = require('multihashing-async')
+const CID = require('cids')
+
+module.exports = ({ blockService, gcLock, preload }) => {
+ return async function put (block, options) {
+ options = options || {}
+
+ if (Array.isArray(block)) {
+ throw new Error('Array is not supported')
+ }
+
+ if (!Block.isBlock(block)) {
+ if (options.cid && CID.isCID(options.cid)) {
+ block = new Block(block, options.cid)
+ } else {
+ const mhtype = options.mhtype || 'sha2-256'
+ const format = options.format || 'dag-pb'
+ let cidVersion
+
+ if (options.version == null) {
+ // Pick appropriate CID version
+ cidVersion = mhtype === 'sha2-256' && format === 'dag-pb' ? 0 : 1
+ } else {
+ cidVersion = options.version
+ }
+
+ const multihash = await multihashing(block, mhtype)
+ const cid = new CID(cidVersion, format, multihash)
+
+ block = new Block(block, cid)
+ }
+ }
+
+ const release = await gcLock.readLock()
+
+ try {
+ await blockService.put(block)
+
+ if (options.preload !== false) {
+ preload(block.cid)
+ }
+
+ return block
+ } finally {
+ release()
+ }
+ }
+}

src/core/components/block/rm.js

@@ -0,0 +1,65 @@
+'use strict'
+
+const CID = require('cids')
+const errCode = require('err-code')
+const { parallelMap, filter } = require('streaming-iterables')
+const pipe = require('it-pipe')
+const { PinTypes } = require('./pin/pin-manager')
+const { cleanCid } = require('./utils')
+
+const BLOCK_RM_CONCURRENCY = 8
+
+module.exports = ({ blockService, gcLock, pinManager }) => {
+ return async function * rm (cids, options) {
+ options = options || {}
+
+ if (!Array.isArray(cids)) {
+ cids = [cids]
+ }
+
+ // We need to take a write lock here to ensure that adding and removing
+ // blocks are exclusive operations
+ const release = await gcLock.writeLock()
+
+ try {
+ yield * pipe(
+ cids,
+ parallelMap(BLOCK_RM_CONCURRENCY, async cid => {
+ cid = cleanCid(cid)
+
+ const result = { hash: cid.toString() }
+
+ try {
+ const pinResult = await pinManager.isPinnedWithType(cid, PinTypes.all)
+
+ if (pinResult.pinned) {
+ if (CID.isCID(pinResult.reason)) { // eslint-disable-line max-depth
+ throw errCode(new Error(`pinned via ${pinResult.reason}`))
+ }
+
+ throw errCode(new Error(`pinned: ${pinResult.reason}`))
+ }
+
+ // remove has check when https:/ipfs/js-ipfs-block-service/pull/88 is merged
+ const has = await blockService._repo.blocks.has(cid)
+
+ if (!has) {
+ throw errCode(new Error('block not found'), 'ERR_BLOCK_NOT_FOUND')
+ }
+
+ await blockService.delete(cid)
+ } catch (err) {
+ if (!options.force) {
+ result.error = `cannot remove ${cid}: ${err.message}`
+ }
+ }
+
+ return result
+ }),
+ filter(() => !options.quiet)
+ )
+ } finally {
+ release()
+ }
+ }
+}

src/core/components/block/stat.js

@@ -0,0 +1,21 @@
+'use strict'
+
+const { cleanCid } = require('./utils')
+
+module.exports = ({ blockService, preload }) => {
+ return async function stat (cid, options) {
+ options = options || {}
+ cid = cleanCid(cid)
+
+ if (options.preload !== false) {
+ preload(cid)
+ }
+
+ const block = await blockService.get(cid)
+
+ return {
+ key: cid.toString(),
+ size: block.data.length
+ }
+ }
+}

src/core/components/block/utils.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const CID = require('cids')
+const errCode = require('err-code')
+
+exports.cleanCid = cid => {
+ if (CID.isCID(cid)) {
+ return cid
+ }
+
+ // CID constructor knows how to do the cleaning :)
+ try {
+ return new CID(cid)
+ } catch (err) {
+ throw errCode(err, 'ERR_INVALID_CID')
+ }
+}