refactor: use the block API from ipfs instead of ipld internals

packages/ipfs-unixfs-exporter/test/exporter-sharded.spec.js

@@ -18,13 +18,15 @@ const {
   DAGLink,
   DAGNode
 } = require('ipld-dag-pb')
+const blockApi = require('./helpers/block')
 
 const SHARD_SPLIT_THRESHOLD = 10
 
 describe('exporter sharded', function () {
   this.timeout(30000)
 
   let ipld
+  let block
 
   const createShard = (numFiles) => {
     return createShardWithFileNames(numFiles, (index) => `file-${index}`)
@@ -40,14 +42,15 @@ describe('exporter sharded', function () {
   }
 
   const createShardWithFiles = async (files) => {
-    return (await last(importer(files, ipld, {
+    return (await last(importer(files, block, {
       shardSplitThreshold: SHARD_SPLIT_THRESHOLD,
       wrapWithDirectory: true
     }))).cid
   }
 
   before(async () => {
     ipld = await inMemory(IPLD)
+    block = blockApi(ipld)
   })
 
   it('exports a sharded directory', async () => {
@@ -62,7 +65,7 @@ describe('exporter sharded', function () {
     const imported = await all(importer(Object.keys(files).map(path => ({
       path,
       content: files[path].content
-    })), ipld, {
+    })), block, {
       wrapWithDirectory: true,
       shardSplitThreshold: SHARD_SPLIT_THRESHOLD
     }))

packages/ipfs-unixfs-exporter/test/exporter-subtree.spec.js

@@ -11,16 +11,19 @@ const mc = require('multicodec')
 const all = require('async-iterator-all')
 const last = require('it-last')
 const randomBytes = require('async-iterator-buffer-stream')
+const blockApi = require('./helpers/block')
 
 const ONE_MEG = Math.pow(1024, 2)
 
 const exporter = require('./../src')
 
 describe('exporter subtree', () => {
   let ipld
+  let block
 
   before(async () => {
     ipld = await inMemory(IPLD)
+    block = blockApi(ipld)
   })
 
   it('exports a file 2 levels down', async () => {
@@ -32,7 +35,7 @@ describe('exporter subtree', () => {
     }, {
       path: './level-1/200Bytes.txt',
       content
-    }], ipld))
+    }], block))
 
     const exported = await exporter(`${imported.cid.toBaseEncodedString()}/level-1/200Bytes.txt`, ipld)
 
@@ -54,7 +57,7 @@ describe('exporter subtree', () => {
       content
     }, {
       path: './level-1/level-2'
-    }], ipld))
+    }], block))
 
     const exported = await exporter(`${imported.cid.toBaseEncodedString()}/level-1`, ipld)
     const files = await all(exported.content())
@@ -74,7 +77,7 @@ describe('exporter subtree', () => {
     const imported = await last(importer([{
       path: '/derp/200Bytes.txt',
       content: randomBytes(ONE_MEG)
-    }], ipld))
+    }], block))
 
     try {
       await exporter(`${imported.cid.toBaseEncodedString()}/doesnotexist`, ipld)
@@ -89,7 +92,7 @@ describe('exporter subtree', () => {
     const imported = await last(importer([{
       path: './level-1/200Bytes.txt',
       content
-    }], ipld, {
+    }], block, {
       wrapWithDirectory: true
     }))
 
@@ -122,7 +125,7 @@ describe('exporter subtree', () => {
     }, {
       path: './level-1/level-2/200Bytes.txt',
       content
-    }], ipld))
+    }], block))
 
     const exported = await all(exporter.path(`${imported.cid.toBaseEncodedString()}/level-1/level-2/200Bytes.txt`, ipld))
 

packages/ipfs-unixfs-exporter/test/exporter.spec.js

@@ -22,11 +22,13 @@ const last = require('it-last')
 const first = require('async-iterator-first')
 const randomBytes = require('async-iterator-buffer-stream')
 const AbortController = require('abort-controller')
+const blockApi = require('./helpers/block')
 
 const ONE_MEG = Math.pow(1024, 2)
 
 describe('exporter', () => {
   let ipld
+  let block
   let bigFile
   let smallFile
 
@@ -58,7 +60,7 @@ describe('exporter', () => {
     const result = await all(importer([{
       path,
       content: file
-    }], ipld, {
+    }], block, {
       strategy,
       rawLeaves,
       chunkerOptions: {
@@ -123,6 +125,7 @@ describe('exporter', () => {
 
   before(async () => {
     ipld = await inMemory(IPLD)
+    block = blockApi(ipld)
   })
 
   it('ensure hash inputs are sanitized', async () => {
@@ -148,7 +151,7 @@ describe('exporter', () => {
     const files = await all(importer([{
       path: filePath,
       content: smallFile
-    }], ipld))
+    }], block))
 
     const path = `/ipfs/${files[1].cid.toBaseEncodedString()}/${fileName}`
     const file = await exporter(path, ipld)
@@ -164,7 +167,7 @@ describe('exporter', () => {
     const files = await all(importer([{
       path: filePath,
       content: smallFile
-    }], ipld))
+    }], block))
 
     const path = `/ipfs/${files[1].cid.toBaseEncodedString()}/${fileName}`
     const file = await exporter(path, ipld)
@@ -333,7 +336,7 @@ describe('exporter', () => {
       content: randomBytes(ONE_MEG)
     }, {
       path: './level-1/level-2'
-    }], ipld))
+    }], block))
     const dir = await exporter(importedDir.cid, ipld)
     const files = await all(dir.content())
 
@@ -371,7 +374,7 @@ describe('exporter', () => {
       path: './dir-another'
     }, {
       path: './level-1'
-    }], ipld))
+    }], block))
 
     const dir = await exporter(importedDir.cid, ipld)
     const files = await all(dir.content())
@@ -516,7 +519,7 @@ describe('exporter', () => {
     const imported = await first(importer([{
       path: '1.2MiB.txt',
       content: bigFile
-    }], ipld, {
+    }], block, {
       rawLeaves: true
     }))
 
@@ -529,7 +532,7 @@ describe('exporter', () => {
   it('returns an empty stream for dir', async () => {
     const imported = await first(importer([{
       path: 'empty'
-    }], ipld))
+    }], block))
     const dir = await exporter(imported.cid, ipld)
     const files = await all(dir.content())
     expect(files.length).to.equal(0)
@@ -755,7 +758,7 @@ describe('exporter', () => {
     const imported = await first(importer([{
       path: '200Bytes.txt',
       content: bigFile
-    }], ipld, {
+    }], block, {
       rawLeaves: true
     }))
 
@@ -771,7 +774,7 @@ describe('exporter', () => {
     const imported = await first(importer([{
       path: '200Bytes.txt',
       content: smallFile
-    }], ipld, {
+    }], block, {
       rawLeaves: true
     }))
 
@@ -862,7 +865,7 @@ describe('exporter', () => {
     const imported = await all(importer([{
       path: '/foo/bar/baz.txt',
       content: Buffer.from('hello world')
-    }], ipld))
+    }], block))
 
     const exported = await exporter(imported[0].cid, ipld)
 
@@ -879,7 +882,7 @@ describe('exporter', () => {
     }, {
       path: '/foo/bar/quux.txt',
       content: Buffer.from('hello world')
-    }], ipld))
+    }], block))
 
     const exported = await all(exporter.recursive(dir.cid, ipld))
     const dirCid = dir.cid.toBaseEncodedString()

packages/ipfs-unixfs-exporter/test/helpers/block.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const DAG_PB = require('ipld-dag-pb')
+const multicodec = require('multicodec')
+const mh = require('multihashing-async').multihash
+
+module.exports = (ipld) => {
+  // make ipld behave like the block api, some tests need to pull
+  // data from ipld so can't use use a simple hash
+  return {
+    put: async (buf, { cid }) => {
+      const multihash = mh.decode(cid.multihash)
+
+      if (cid.codec === 'dag-pb') {
+        buf = DAG_PB.util.deserialize(buf)
+      }
+
+      await ipld.put(buf, cid.codec === 'dag-pb' ? multicodec.DAG_PB : multicodec.RAW, {
+        cidVersion: cid.version,
+        hashAlg: multihash.code
+      })
+
+      return { cid, data: buf }
+    },
+    get: async (cid, options) => {
+      const node = await ipld.get(cid, options)
+
+      if (cid.codec === 'dag-pb') {
+        return node.serialize()
+      }
+
+      return { cid, data: node }
+    }
+  }
+}

packages/ipfs-unixfs-importer/README.md

@@ -140,6 +140,8 @@ The input's file paths and directory structure will be preserved in the [`dag-pb
 - `leafType` (string, defaults to `'file'`) what type of UnixFS node leaves should be - can be `'file'` or `'raw'` (ignored when `rawLeaves` is `true`)
 - `blockWriteConcurrency` (positive integer, defaults to 10) How many blocks to hash and write to the block store concurrently. For small numbers of large files this should be high (e.g. 50).
 - `fileImportConcurrency` (number, defaults to 50) How many files to import concurrently. For large numbers of small files this should be high (e.g. 50).
+- `pin` (boolean, defaults to `false`) Whether to pin each block as it is created
+- `preload` (boolean, defaults to `false`) Whether to preload each block as it is created
 
 ## Overriding internals
 

packages/ipfs-unixfs-importer/package.json

@@ -46,6 +46,7 @@
     "ipld-in-memory": "^3.0.0",
     "it-buffer-stream": "^1.0.0",
     "it-last": "^1.0.1",
+    "multicodec": "^1.0.0",
     "nyc": "^15.0.0",
     "sinon": "^9.0.1"
   },
@@ -60,7 +61,6 @@
     "it-first": "^1.0.1",
     "it-parallel-batch": "^1.0.3",
     "merge-options": "^2.0.0",
-    "multicodec": "^1.0.0",
     "multihashing-async": "^0.8.0",
     "rabin-wasm": "^0.1.1"
   }

packages/ipfs-unixfs-importer/src/dag-builder/dir.js

@@ -6,22 +6,22 @@ const {
   DAGNode
 } = require('ipld-dag-pb')
 
-const dirBuilder = async (item, ipld, options) => {
+const dirBuilder = async (item, block, options) => {
   const unixfs = new UnixFS({
     type: 'directory',
     mtime: item.mtime,
     mode: item.mode
   })
 
-  const node = new DAGNode(unixfs.marshal(), [])
-  const cid = await persist(node, ipld, options)
+  const buffer = new DAGNode(unixfs.marshal()).serialize()
+  const cid = await persist(buffer, block, options)
   const path = item.path
 
   return {
     cid,
     path,
     unixfs,
-    size: node.size
+    size: buffer.length
   }
 }
 

packages/ipfs-unixfs-importer/src/dag-builder/file/buffer-importer.js

@@ -6,22 +6,17 @@ const {
   DAGNode
 } = require('ipld-dag-pb')
 
-async function * bufferImporter (file, source, ipld, options) {
-  for await (const buffer of source) {
+async function * bufferImporter (file, source, block, options) {
+  for await (let buffer of source) {
     yield async () => {
       options.progress(buffer.length)
-      let node
       let unixfs
-      let size
 
       const opts = {
         ...options
       }
 
       if (options.rawLeaves) {
-        node = buffer
-        size = buffer.length
-
         opts.codec = 'raw'
         opts.cidVersion = 1
       } else {
@@ -32,16 +27,13 @@ async function * bufferImporter (file, source, ipld, options) {
           mode: file.mode
         })
 
-        node = new DAGNode(unixfs.marshal())
-        size = node.size
+        buffer = new DAGNode(unixfs.marshal()).serialize()
       }
 
-      const cid = await persist(node, ipld, opts)
-
       return {
-        cid: cid,
+        cid: await persist(buffer, block, opts),
         unixfs,
-        size
+        size: buffer.length
       }
     }
   }