在上一篇文章中，我们分析了保存文件/内容的整体流程，基本上知道在这个过程中文件/内容是怎么处理的，但是还流下了一个疑问，就是文件是怎么分片的，又是怎么保存到本地系统，这篇文章我们就来解决这几个问题。通过上一篇文章，我们知道 ipfs-unixfs-importer 这个类库，它实现了 IPFS 用于处理文件的布局和分块机制，它的 index.js 文件内容只有一行代码 require('./importer')，接下来我们直接来看这个 importer/index.js 是怎么处理的。

1.    把参数传递的选项和默认选项进行合并，生成新的选项，然后检查选项的相关配置。

const options = Object.assign({}, defaultOptions, _options)

options.cidVersion = options.cidVersion || 0

if (options.cidVersion > 0 && _options.rawLeaves === undefined) {

    options.rawLeaves = true

}

if (_options && _options.hash !== undefined && _options.rawLeaves === undefined) {

    options.rawLeaves = true

}

默认选项即 defaultOptions 内容如下：

const defaultOptions = {

  chunker: 'fixed',

  rawLeaves: false,

  hashOnly: false,

  cidVersion: 0,

  hash: null,

  leafType: 'file',

  hashAlg: 'sha2-256'

}

2.    根据选项中指定的分割方式，从IPFS 中提供的所有分割方法找到对应的分割对象。

const Chunker = chunkers[options.chunker]

chunkers 表示系统提供的所有分割方法对象，在父目录下 chunker/index.js 文件中定义的，默认有 fixed、rabin 两种方法，默认使用的是的前者，即固定大小。

3.    生成一个 pull-through 的双向流，双向流的意思就是即可以从它读取数据，又可以提供数据让其它流读取。

  const entry = {

    sink: writable(

      (nodes, callback) => {

        pending += nodes.length

        nodes.forEach((node) => entry.source.push(node))

        setImmediate(callback)

      },

      null,

      1,

      (err) => entry.source.end(err)

    ),

    source: pushable()

  }

source 流是 pull-pushable 类库提供的一个可以其它流主动push 的 pull-stream 源流，它提供了一个 push 方法，当调用这个方法时，它开始调用回调函数，从而把数据传递给后续的 through 或 sink。当时，它还提供了一个 end 方法，当数据读取完成后，调用这个方法。

sink 流是 pull-write 类库提供的一个创建通用 pull-stream sinks 流的基础类。它的签名如下：(write,reduce, max, cb)， 因为它是一个 sinks 流，所以它会读取前一个流的数据，在读取到数据之后就调用它的 write 方法保存读取到的数据，如果数据读取完成就调用它的 cb 方法。

在这里 sink 函数从前一个流中读取数据，然后放入 source 中。同时，source 成为下一个流的读取函数。

4.    生成一个 dagStream 对象，这个对象也是一个 {source,sink} 对象。

const dagStream = DAGBuilder(Chunker, ipld, options)

DAGBuilder 函数定义于父目录下的 builder/index.js 中，接下来我们看下这个执行过程：

·    合并选项参数和默认选项

const options = Object.assign({}, defaultOptions, _options)

·    默认选项如下：

const defaultOptions = {

  strategy: 'balanced',

  highWaterMark: 100,

  reduceSingleLeafToSelf: true

}

·    根据选项指定的 reduce 策略，从系统提供的多个策略中选择指定的策略。

const strategyName = options.strategy

  const reducer = reducers[strategyName]

·    系统定义的的策略如下：

const reducers = {

  flat: require('./flat'),

  balanced: require('./balanced'),

  trickle: require('./trickle')

}

·    在用户不指定具体策略的默认情况下，根据前面执行过程，最终选定的策略为 balanced。

·    调用 Builder 方法创建最终的策略对象。

const createStrategy = Builder(Chunker, ipld, reducer, options)

·       Builder 方法位于 builder.js 文件中，它会创建一个pull-stream 的 through 流对象。在看它的内部之前，我们首先看下的 4个参数。

n  第 1 个参数 Chunker/createChunker，它表示具体分割内容的策略，默认情况下为 fixed，详见第一步中的 defaultOptions 变量内容；

n  第 2 个参数 ipld/ipld，这个是 IPFS 对象的 _ipld 属性，在 IPFS 对象创建时生成的，表示星际接续的数据，目前它可以连接比特币、以太坊、git、zcash 等，在 IPFS 体系中具有非常重要的位置；

n  第 3 个参数 reducer/createReducer 是具体的 reduce 策略，默认情况为 balanced，详见第四步中生成 reducer 变量的过程。

n  第 4 个参数 options/_options 为选项。

看完参数，接下来，我们看下它的执行逻辑。

o           合并指定的选项和自身默认的选项。

const options = extend({}, defaultOptions, _options)

o           默认选项如下：

const defaultOptions = {

  chunkerOptions: {

    maxChunkSize: 262144,

    avgChunkSize: 262144

  },

  rawLeaves: false,

  hashAlg: 'sha2-256',

  leafType: 'file',

  cidVersion: 0,

  progress: () => {}

}

o           返回一个函数对象。

return function (source) {

    return function (items, cb) {

      parallel(items.map((item) => (cb) => {

        if (!item.content) {

          return createAndStoreDir(item, (err, node) => {

            if (err) {

              return cb(err)

            }

            if (node) {

              source.push(node)

            }

            cb()

          })

        }

        createAndStoreFile(item, (err, node) => {

          if (err) {

            return cb(err)

          }

          if (node) {

            source.push(node)

          }

          cb()

        })

      }), cb)

    }

}

返回的这个函数，最终成为了一个 sink 流的 write 方法。

·    调用 createBuildStream 方法，生成一个双向流对象。

createBuildStream(createStrategy, ipld, options)

·       createBuildStream 方法位于 create-build-stream.js 文件中，代码如下：

const source = pullPushable()

const sink = pullWrite(

    createStrategy(source),

    null,

    options.highWaterMark,

    (err) => source.end(err)

)

return {

source: source,

sink: sink

}

·    在这段代码中，source 流是 pull-pushable 类库提供的一个可以主动 push 到其它流的 pull-stream 源流，这个类库在前面我们已经分析过，这里就直接略过。

sink 流是 pull-write 类库提供的一个创建通用 pull-stream sinks 流的基础类，这个类库也在前面分析过，这里也不细讲，我们只看下它的 write 方法，这里的 createStrategy 函数正是调用 Builder 方法返回的 createStrategy 函数，用 source 作为参数，调用它，用返回的第二层匿名函数作为 write 方法。

5.    生成一个树构建器流对象，并返回其双向流对象。

const treeBuilder = createTreeBuilder(ipld, options)

const treeBuilderStream = treeBuilder.stream()

createTreeBuilder 函数位于 tree-builder.js 文件中，我们来看它的执行逻辑。

·    首先，合并默认选项对象和指定的选项对象。

const options = Object.assign({}, defaultOptions, _options)

·    默认选择对象如下：

const defaultOptions = {

  wrap: false,

  shardSplitThreshold: 1000,

  onlyHash: false

}

·       onlyHash 表示是否不保存文件/内容，只计算其哈希。

·    创建一个队列对象。

const queue = createQueue(consumeQueue, 1)

·     

·    创建一个双向流对象

let stream = createStream()

·    其中 sink 对象是一个 pull-write 类库提供的流，这个已经见过多次了，它的 write 方法后面遇到时再来看，source 是一个 pull-pushable 类库提供的流，这个也见过多次。

·    创建一个 DirFlat 对象。

let tree = DirFlat({

    path: '',

    root: true,

    dir: true,

    dirty: false,

    flat: true

}, options) 

·    返回特权函数构成的对象。

return {

    flush: flushRoot,

    stream: getStream

}

6.    创建一个暂停流。这里什么也不做。

7.    调用 pull 方法，创建一个完整的流来保存文件内容。

pull(

    entry,

    pausable,

    dagStream,

    map((node) => {

      pending--

      if (!pending) {

        process.nextTick(() => {

          while (waitingPending.length) {

            waitingPending.shift()()

          }

        })

      }

      return node

    }),

    treeBuilderStream

)

pull 函数是 pull-stream 是类库中的核心函数，在它的执行过程中，最后的 sink 流通过依次调用前面的 through 流，最终从最前面的 source 流中拉取数据，除了最前面的 Source 流和最后面的 Sink 流，中间的都是 through 流，它们即可以被后面的流调用以提供数据，也可以调用前面的流来读取数据。

当 pull 函数在调用某个参数从前面读取数据时，如果当前参数是一个对象（即双向流）时，那么就会调用它的 sink 方法来读取，同时用它的 source 方法作为后面参数的读取方法。

下面我们分析这段代码中的几个流，它们太重要了。

首先是 entry 流，它是一个双向流，它的 sink 函数（类型为 pull-write 流）调用前一个流的 read 方法来读取数据，并把读取到的数据放在 source 中（类型为pull-pushable ）。

然后是 dagStream 流，它也是一个双向流，它的 sink 函数（类型为 pull-write 流）调用 entry 流的 source 方法来读取数据。sink 函数的异步写函数参数为 builder.js 中返回的第二层函数，当读取到数据之后，调用 builder.js 中返回的第二层函数进行处理，在第二层函数中，大致流程是把数据保存自身的 source 中（类型为pull-pushable ）。

dagStream 在 create-build-stream.js 中生成。为了方便理解，这里我们再次看下它的代码。

const source= pullPushable()

const sink =pullWrite( createStrategy(source), null, options.highWaterMark, (err) =>source.end(err) )

return {source: source, sink: sink }

最后是 treeBuilderStream 流，它也是一个双向流，它的 sink 函数（类型为 pull-write 流）调用 dagStream 流的 source 方法来读取数据，并把读取到的数据放在 source 中（类型为 pull-pushable ）。

其他两个流对流程没有任何影响，读者可以自行分析，这里略过不提。

在这一步中，通过 pull 函数把最重要的几个流连接到一起，并通过下面最后一步，把它们与外部的流联系到一起。

8.    最后，返回双向流对象。

{

    sink: entry.sink,

    source: treeBuilderStream.source,

    flush: flush

}

到这里，文件已经保存完成了。

啥？文件已经保存完成了？什么都没看到就把文件保存完了，不会骗我们的吧？哈，因为保存文件这个动作太复杂了，所以上面只是静态的从代码层面进行梳理，下面我们从头到尾从动态处理的过程来看下文件到底是怎么保存在本地的。

一切要从我们在上篇写的这个示例说起

const {createNode} = require('ipfs')

const node = createNode({

  libp2p:{

    config:{

      dht:{

        enabled:true

      }

    }

  }

})

node.on('ready', async () => {

    const content = `我爱黑萤`;

    const filesAdded = await node.add({

      content: Buffer.from(content)

    },{

      chunkerOptions:{

        maxChunkSize:1000,

        avgChunkSize:1000

      }

    })

    console.log('Added file:', filesAdded[0].path, filesAdded[0].hash)

})

上面这段代码，最终执行的是 core/components/files-regular/add-pull-stream.js 文件中的函数，它的主体就是下面的这段代码：

pull(

    pull.map(content => normalizeContent(content, opts)),

    pull.flatten(),

    importer(self._ipld, opts), 

    pull.asyncMap((file, cb) => prepareFile(file, self, opts, cb)),

    pull.map(file => preloadFile(file, self, opts)),

    pull.asyncMap((file, cb) => pinFile(file, self, opts, cb))

)

为了便于分析理解，我们在分析过程中仍然使用推的方式，从源流推到目的流中，注意这个仅是为了理解方便，真实的过程是目的流从源流中拉取数据。

下面代码简单解释如下：

Ø  首先，调用第一个 pull.map 流，对收到的文件或内容并进行一些必要的转换，

Ø  调用 pull.flatten 流，把前一步生成的数组进行扁平化处理。

Ø  调用 importer 流来保存内容。

Ø  调用 pull.asyncMap 方法，对已经保存的文件/内容进行预处理，生成用户看到的内容。

Ø  调用 pull.map 方法，把已经保存到本地的文件预加载到指定节点。

Ø  调用 pull.asyncMap 方法，把已经保存到本地的文件长期保存在本地，确保不被垃圾回收。

Ø  下面我们重点看下文件内容在 importer 流中的处理逻辑。

(1)    调用 entry.sink 函数从前面的 pull.flatten 流中读取保存的每一个文件/内容。

(2)  调用 dagStream.sink 函数从前面的流中读取数据，并在读取到数据之后，调用 builder.js 中定义的第二层匿名函数进行处理。在这个函数中，调用异步流程库 async 的 parallel 方法对收到的每个要处理的文件内容进行处理，具体处理如下：如果保存的是目录，那么调用 createAndStoreDir 方法，创建并保存目录；如果保存的是文件，那么调用 createAndStoreFile 方法，创建并保存主文件。因为我们保存的是文件，所以在这里详细看下 createAndStoreFile 方法，它的过程如下：

1.  如果保存的内容是 Buffer，那么调用pull-stream 的 values 方法，生成内容源流。

if (Buffer.isBuffer(file.content)) {

  file.content = values([file.content])

}

·    调用 createReducer 方法，创建 reducer 对象，默认为 balanced，所以这里创建的 reducer 对象类型为 balanced/balanced-reducer.js 文件中定义的函数。

const reducer = createReducer(reduce(file, ipld, options), options)

·    调用 createChunker 方法，创建 chunker 对象，默认为 fixed，所以这里创建的 chunker 对象类型为 chunker/fixed-size.js 主文件中定义的函数。

chunker = createChunker(options.chunkerOptions)

·    调用 pull 函数进行保存文件。

o           设置源流为 file.content。

o           调用 chunker 流，对保存的内容进行分块。

o           调用 paraMap 流（类型为 pull-paramap），对每一个分块进行处理。

o           调用 pullThrough 流（类型为 pull-through流），对收到的每个数据进行处理。

o           调用 reducer 流，把所有生成的分块进行reduce 处理。如果文件进行了多次分块，这里就会根据生成的分块生成一个父块。

o           调用 collect 流，调用回调函数即 createAndStoreFile ，把保存文件的结果传递到外部函数中。

这个 pull 函数会进行 IPFS 特有业务，涉及到 IPFS 保存文件核心逻辑，这块我们留在下一篇文章中进行分析。

(3)  调用 treeBuilderStream.sink 函数从前面的流中读取数据，在这里即为保存文件的结果，并在读取到保存文件结果之后，把结果保存在 source 中。当把保存文件的结果保存到 source 中之后，core/components/files-regular/add-pull-stream.js 文件中定义的 pull.asyncMap 就可以得到这个结果了。

作者：乔疯，加密货币爱好者，ipfs 爱好者，黑萤科技CTO。