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://github.com/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')
+  }
+}