Proficient in IPFS: IPFS saves content

Pull-stream is a new type of stream library in which data is pulled from the source to the destination. It has two basic types of streams: the Source source and the Sink receiver. In addition, there are two composite types of streams: the Through channel stream (such as conversion) and the Duplex bidirectional stream.

The source stream, which returns an anonymous function called the read function, which is called by the subsequent sink stream function or through stream function to read the contents of the source stream.

• Market Analysis: Bitcoin continues to stand firm at 9000 points, and mainstream currencies are on the road to new highs.
• Babbitt column | Cai Weide: Blockchain application is not a wolf, but a tiger is coming.
• Babbitt Column | Facebook Digital Currency: Origin, Meaning and Consequences (4D Long Text)

The sink stream, which eventually returns to the sink function in the internal drain.js. This type of stream mainly reads data and processes each read data. If the stream has ended, it calls the user-specified end function for processing.

Through stream, the function of this type of stream returns a nested anonymous function. The first layer function receives the read function of a source stream or the first layer function returned by other through functions as parameters, and the second layer function receives the write provided by the final sink. The second layer function returned by a function or other through, the second layer function internally calls the read function to directly or indirectly retrieve data from the source, and directly or indirectly call the sink function after the data is acquired, thereby writing the data to the destination address.

In pull-streams, data must have a complete pipeline before it flows, which means one source, zero or more channels, and one receiver. But it's still very useful to be able to create a partial pipeline. In other words, you can create a complete pipeline, such as pull(source, sink) => undefined , or you can pull(through1, through2) => through the pipeline, such as pull(through, sink) => sink , or pull(through1, through2) => through We will encounter this part of the pipeline in large numbers below. Today, we look at the first most commonly used add command / action, we use IPFS just to save the file to IPFS, naturally indispensable to save operations, add command is to do this, a few gossip, let's look at a piece of code.

 const {createNode} = require('ipfs') 

Const node = createNode({
Libp2p:{
Config:{
Dht:{
Enabled: true
}
}
}
})

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

Const content = `I love black firefly`;

Const filesAdded = await node.add({
Content: Buffer.from(content)
},{
chunkerOptions:{
maxChunkSize: 1000,
avgChunkSize: 1000
}
})

Console.log('Added file:', filesAdded[0].path, filesAdded[0].hash)
})
This time we didn't use the default configuration completely, and started DHT. The readers who read my article know what DHT is, and I won't explain it in detail here. In the program, the content is uploaded by calling the add method of the IPFS node. The content can be a file or a direct content. There is a slight difference between the two. When we talk about the relevant code, we point out the difference. Here we For the sake of simply uploading content directly, let us explain.

add method is located in the core/components/files-regular/add.js file. In the article "Proficient in IPFS: System Startup", we said that the system will put all the core/components/files-regular directories. The file is extended to the IPFS object, which naturally includes the add.js file here. Below, we directly look at the execution flow of this function.

This function returns an internally defined function, does some processing on the parameters in this internally defined function, and then calls the internal add function, the latter is the body, its logic is as follows:

1. First, check if the option object is a function, and if so, regenerate the relevant variables.

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

2. Define a tool function that detects the content to detect what we want to upload.

    const isBufferOrStream = obj => Buffer.isBuffer(obj) || isStream.readable(obj) || isSource(obj) const isContentObject = obj => {  if (typeof obj !== 'object') return false  if (obj.content) return isBufferOrStream(obj.content)  return Boolean(obj.path) && typeof obj.path === 'string' } 

Const isInput = obj => isBufferOrStream(obj) || isContentObject(obj) const ok = isInput(data) || (Array.isArray(data) && data.every(isInput))

If (!ok) { return callback(new error('invalid input: expected buffer, readable stream, pull stream, object or array of objects')) }

The second parameter is the addPullStream method of the IPFS object. This method is also extended to the IPFS object using the same method at startup. Its body is the function in the add-pull-stream.js file of the current directory. Next, we'll look at this function in detail, and now we just need to know that this function returns a partial pipeline.

The third parameter is the function defined in pull-sort . This is a library that depends on the pull-stream . It is sorted according to certain rules. We don't care about this function.

The last parameter is the result of the pull.collect function, which is a sink stream. It puts the final result into an array and then calls the callback function. The reason why the filesAdded we saw in the previous code is an array is thanks to this function.

The above logic code is as follows:

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

In the above code, we make an array of the content to be saved, the specific reasons are explained below. Now, let's look at the addPullStream method, which is the main body for saving content. The add method is only an appetizer. addPullStream method execution logic is as follows:

1. Call the parseChunkerString function to handle the options related to content blocking. This function is located in the utils.js file in the same directory, which checks the user-specified blocking algorithm. If the user does not specify, the fixed blocking algorithm is used, the size is 262144 of the system default; if the size is specified, the fixed blocking algorithm is used, but the size is specified by the user; if the rabin class division method is specified, the variable length segmentation The method calls the internal function to generate the corresponding split option. The above logic code is as follows:

    parseChunkerString = (chunker) => {  if (!chunker) {    return {      chunker: 'fixed'    }  } else if (chunker.startsWith('size-')) {    const sizeStr = chunker.split('-')[1]    const size = parseInt(sizeStr)    if (isNaN(size)) {      throw new Error('Chunker parameter size must be an integer')    }    return {      chunker: 'fixed',      chunkerOptions: {        maxChunkSize: size      }    }  } else if (chunker.startsWith('rabin')) {    return {      chunker: 'rabin',      chunkerOptions: parseRabinString(chunker)    }  } else {    throw new Error( Unrecognized chunker option: ${chunker} )  } } 

   Note: We can also add our own segmentation algorithm by rewriting this function.

2. Merge the option variables.

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

3. Set the default CID version number. If the Hash algorithm is specified, but the CID version is not 1, then it is forced to 1. CID is a self-describing content addressing identifier for distributed systems. There are currently two versions 0 and 1. Version 0 is a backward compatible version that only supports the sha256 hash algorithm and cannot be specified.

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

4. Set the progress handler, the default is empty implementation.

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

Opts.progress = progress

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

 if (opts.wrapWithDirectory && !data.path) {  throw new Error('Must provide a path when wrapping with a directory') } 

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] 

This function is defined in the same file, and its processing is as follows:

• Generate a CID object using the multihash content of the generated file.

   let cid = new CID(file.multihash) 

  The CID constructor will check the passed parameters. If it is a CID object, it will take the version number, encoding mode, multi-hash and other attributes directly from the object; if it is a string, it will be divided into multibase encoding, if it is Need to decode first, and then separate out various attributes, if not multibase encoding, then it is definitely a base58 string, then set the version to 0, the encoding method is dag-pb , and then get the multi-hash value from the base58 string; Buffer object, then get the first byte, and convert it to an integer in hexadecimal. If the first byte is 0 or 1, the corresponding attribute is generated. Otherwise, if it is multi-hash, the version is set to 0. For dag-pb .

• If the user specifies a CID version of 1, then a CID object is generated to version 1.

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

• Next, call the waterfall method to process the function it specifies. The first function, check whether the configuration option specifies onlyHash , that is, does not actually upload the file to the IFS network, just calculate the HASH of this file, then directly call the second function, otherwise, call the object.get method of the IPFS object. Gets the node information of the specified file saved in the repository. This method will be explained in detail later. I will not talk about it here. The second function, which generates the object that is ultimately returned to the user, includes: path, size, hash, and so on.

The above code is as follows, it is relatively simple, you can read it yourself.

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

   let size = 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) 

We will continue to update Blocking; if you have any questions or suggestions, please contact us!