精通IPFS：IPFS 保存内容之上篇

经过前面的分析，我们已经明白了 IPFS 启动过程，从今天起，我会分析一些常见的命令或动作，希望大家喜欢。

在开始真正分析这些命令/动作之前，先要对 pull-stream 类库进行简单介绍，如果不熟悉这个类库，接下来就没办法进行。

pull-stream 是一个新型的流库，数据被从源中拉取到目的中，它有两种基本类型的流：Source 源和 Sink 接收器。除此之外，有两种复合类型的流：Through 通道流（比如转换）和 Duplex 双向流。

source 流，这类流返回一个匿名函数，这个匿名函数被称为 read 函数，它被后续的 sink 流函数或 through 流函数调用，从而读取 source 流中的内容。

sink 流，这类流最终都返回内部 drain.js 中的 sink 函数。这类流主要是读取数据，并且对每一个读取到的数据进行处理，如果流已经结束，则调用用户指定结束函数进行处理。

through 流，这类流的函数会返回嵌套的匿名函数，第一层函数接收一个 source 流的 read 函数或其他 through 函数返回的第一层函数为参数，第二层函数接收最终 sink 提供的写函数或其他 through 返回的第二层函数，第二层函数内部调用 read 函数，从而直接或间接从 source 中取得数据，获取数据后直接或间接调用 sink 函数，从而把数据写入到目的地址。

在 pull-streams 中，数据在流动之前，必须有一个完整的管道，这意味着一个源、零个或多个通道、一个接收器。但是仍然可以创建一个部分化的管道，这非常有用。也就是说可以创建一个完整的管道，比如pull(source, sink) => undefined，也可以部分化的管道，比如pull(through, sink) => sink，或者pull(through1, through2) => through，我们在下面会大量遇到这种部分化的管道。

今天，我们看下第一个最常用的add命令/动作，我们使用 IPFS 就是为了把文件保存到 IPFS，自然少不了保存操作，add命令就是干这个的，闲话少数，我们来看一段代码。

const{createNode}=require('ipfs')

constnode=createNode({ libp2p:{ config:{ dht:{ enabled:true } } } })

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

constcontent=`我爱黑萤`;

constfilesAdded=awaitnode.add({ content:Buffer.from(content) },{ chunkerOptions:{ maxChunkSize:1000, avgChunkSize:1000 } })

console.log('Addedfile:',filesAdded[0].path,filesAdded[0].hash) })

这次我们没有完全使用默认配置，开启了 DHT，看过我文章的读者都知道 DHT 是什么东东，这里不详细解释。在程序中，通过调用 IPFS 节点的add方法来上传内容，内容可以是文件，也可以是直接的内容，两者有稍微的区别，在讲到相关代码时，我们指出这种区别的，这里我们为了简单直接上传内容为例来说明。

add方法位于core/components/files-regular/add.js文件中，在 《精通IPFS：系统启动之概览》 那篇文章中，我们说过，系统会把core/components/files-regular目录下的所有文件扩展到 IPFS 对象上面，这其中自然包括这里的add.js文件。下面，我们直接看这个函数的执行流程。

这个函数返回了一个内部定义的函数，在这个内部定义的函数中对参数做了一些处理，然后就调用内部的add函数，后者才是主体，它的逻辑如下：

if(typeofoptions==='function'){ callback=options options={} }

constisBufferOrStream=obj=>Buffer.isBuffer(obj)||isStream.readable(obj)||isSource(obj) constisContentObject=obj=>{ if(typeofobj!=='object')returnfalse if(obj.content)returnisBufferOrStream(obj.content) returnBoolean(obj.path)&&typeofobj.path==='string' }

constisInput=obj=>isBufferOrStream(obj)||isContentObject(obj) constok=isInput(data)||(Array.isArray(data)&&data.every(isInput))

if(!ok){ returncallback(newError('invalidinput:expectedbuffer,readablestream,pullstream,objectorarrayofobjects')) }

第二个参数是 IPFS 对象的addPullStream方法，这个方法也是在启动时候使用同样的方法扩展到 IPFS 对象，它的主体是当前目录的add-pull-stream.js文件中的函数。接下来，我们会详细看这个函数，现在我们只需要知道这个函数返回了一个部分化的管道。

第三个参数是pull-sort中定义的函数，这是一个依赖于pull-stream的库，根据一定规则来排序，这个函数我们不用管。

最后一个参数是pull.collect函数执行的结果，这个collect函数就是一个 sink 流。它把最终的结果放入一个数组中，然后调用回调函数。我们在前面代码中看到的filesAdded之所以是一个数组就是拜这个函数所赐。

上面逻辑的代码如下：

pull( pull.values([data]), self.addPullStream(options), sort((a,b)=>{ if(a.path<b.path)return1 if(a.path>b.path)return-1 return0 }), pull.collect(callback) )

现在，我们来看addPullStream方法，这个方法是保存内容的主体，add方法是只开胃小菜。addPullStream方法执行逻辑如下：

parseChunkerString=(chunker)=>{ if(!chunker){ return{ chunker:'fixed' } }elseif(chunker.startsWith('size-')){ constsizeStr=chunker.split('-')[1] constsize=parseInt(sizeStr) if(isNaN(size)){ thrownewError('Chunkerparametersizemustbeaninteger') } return{ chunker:'fixed', chunkerOptions:{ maxChunkSize:size } } }elseif(chunker.startsWith('rabin')){ return{ chunker:'rabin', chunkerOptions:parseRabinString(chunker) } }else{ thrownewError(Unrecognizedchunkeroption:${chunker}) } }

constopts=Object.assign({},{ shardSplitThreshold:self._options.EXPERIMENTAL.sharding ?1000 :Infinity },options,chunkerOptions)

if(opts.hashAlg&&opts.cidVersion!==1){ opts.cidVersion=1 }

constprog=opts.progress||noop constprogress=(bytes)=>{ total+=bytes prog(total) }

opts.progress=progress

if(Buffer.isBuffer(data)){ data={path:'',content:pull.values([data])} }

if(isStream.readable(data)){ data={path:'',content:toPull.source(data)} }

if(isSource(data)){ data={path:'',content:data} }

if(data&&data.content&&typeofdata.content!=='function'){ if(Buffer.isBuffer(data.content)){ data.content=pull.values([data.content]) }

if(isStream.readable(data.content)){ data.content=toPull.source(data.content) } }

if(opts.wrapWithDirectory&&!data.path){ thrownewError('Mustprovideapathwhenwrappingwithadirectory') }

if(opts.wrapWithDirectory){ data.path=WRAPPER+data.path }

[ [1,2,3], [4,5,6], [7,8,9] ]

[1,2,3,4,5,6,7,8,9]

这个函数定义在同一个文件中，它的处理具体如下：

letcid=newCID(file.multihash)

if(opts.cidVersion===1){ cid=cid.toV1() }

上面代码如下，比较简单，可自己阅读。

waterfall([ (cb)=>opts.onlyHash ?cb(null,file) :self.object.get(file.multihash,Object.assign({},opts,{preload:false}),cb), (node,cb)=>{ constb58Hash=cid.toBaseEncodedString()

letsize=node.size

if(Buffer.isBuffer(node)){ size=node.length }

cb(null,{ path:opts.wrapWithDirectory ?file.path.substring(WRAPPER.length) :(file.path||b58Hash), hash:b58Hash, size }) }

],callback)

—-

编译者/作者：星鉴网

玩币族申明：玩币族作为开放的资讯翻译/分享平台，所提供的所有资讯仅代表作者个人观点，与玩币族平台立场无关，且不构成任何投资理财建议。文章版权归原作者所有。