fix: cleanup code, headers, chunkSize options

Author: hugomrdias

package.json

@@ -29,7 +29,6 @@
     "async": "^2.6.1",
     "big.js": "^5.1.2",
     "bs58": "^4.0.1",
-    "buffer-to-stream": "^1.0.0",
     "cids": "~0.5.3",
     "concat-stream": "^1.6.2",
     "debug": "^3.1.0",

src/add2/add2.js

@@ -1,7 +1,6 @@
 'use strict'
 
 const { Readable, Transform } = require('stream')
-const toStream = require('buffer-to-stream')
 const pump = require('pump')
 const Multipart = require('./multipart2')
 const {prepareWithHeaders} = require('./../utils/prepare-file')
@@ -24,106 +23,101 @@ const prepareTransform = (options) => new Transform({
 })
 
 module.exports = (send) => (files, options) => {
-  const multipart = new Multipart()
-
-  // add pump
-  arrayToStream([].concat(files))
-    .pipe(prepareTransform(options))
-    .pipe(multipart)
+  const multipart = pump(
+    arrayToStream([].concat(files)),
+    prepareTransform(options),
+    new Multipart(options),
+    (err) => {
+      if (err) {
+        // probably better to create a rejected Promise to return
+        console.error(err)
+      }
+    }
+  )
 
   return sendChunked(multipart, send, options)
 }
 
 const sendChunked = (multipartStream, send, options) => {
   return new Promise((resolve, reject) => {
-    const boundary = multipartStream._boundary
-    let index = 0
-    let rangeStart = 0
-    let rangeEnd = 0
-    let size = 0
-    let ended = false
-    let running = false
-    const name = createName()
+    const state = {
+      boundary: multipartStream._boundary,
+      id: uuid(),
+      index: 0,
+      rangeStart: 0,
+      rangeEnd: 0,
+      rangeTotal: 0,
+      ended: false,
+      running: false
+    }
 
+    multipartStream.on('error', reject)
     multipartStream.on('end', () => {
-      ended = true
-      console.log('end', size)
+      state.ended = true
+      console.log('end', state.rangeTotal)
 
       // multipart ended and no request is running send last request
-      if (!running) {
-        // sendChunk('', -1, rangeEnd, rangeEnd, name, boundary, size)
-        sendChunkRequest(send, options, '', -1, rangeEnd, rangeEnd, name, boundary, size)
-          .then(rsp => {
-            resolve(rsp)
-          })
+      if (!state.running) {
+        sendChunkRequest(send, options, '', state)
+          .then(resolve)
+          .catch(reject)
       }
     })
 
     multipartStream.on('data', (chunk) => {
       console.log('Sending ', chunk.length)
+      // stop producing chunks
       multipartStream.pause()
-      index++
-      rangeEnd = rangeStart + chunk.length
-      size += chunk.length
-      running = true
+      state.index++
+      state.rangeEnd = state.rangeStart + chunk.length
+      state.rangeTotal += chunk.length
+      state.running = true
 
-      //   sendChunk(chunk, index, rangeStart, rangeEnd, name, boundary)
-      sendChunkRequest(send, options, chunk, index, rangeStart, rangeEnd, name, boundary)
+      sendChunkRequest(send, options, chunk, state)
         .then(rsp => {
           console.log('Response', rsp)
-          rangeStart = rangeEnd
+          state.running = false
+          state.rangeStart = state.rangeEnd
+          // resume producing chunks
           multipartStream.resume()
+
           // if multipart already ended send last request
-          if (ended) {
-            console.log('sending last')
-            // sendChunk('', -1, rangeEnd, rangeEnd, name, boundary, size)
-            sendChunkRequest(send, options, '', -1, rangeEnd, rangeEnd, name, boundary, size)
-              .then(rsp => {
-                resolve(rsp)
-              })
+          if (state.ended) {
+            return sendChunkRequest(send, options, '', state)
+              .then(resolve)
           }
-          running = false
         })
         .catch(reject)
     })
   })
 }
 
-const sendChunk = (chunk, id, start, end, name, boundary, size = '*') => {
-  const url = new URL('http://localhost')
-  const search = new URLSearchParams()
-  search.set('stream-channels', true)
-  url.port = 5002
-  url.pathname = 'api/v0/add-chunked'
-  url.search = search
-
-  return window.fetch(url.href, {
-    method: 'POST',
-    body: chunk,
-    headers: {
-      'Content-Type': 'application/octet-stream',
-      'Content-Range': `bytes ${start}-${end}/${size}`,
-      'Ipfs-Chunk-Name': name,
-      'Ipfs-Chunk-Id': id,
-      'Ipfs-Chunk-Boundary': boundary
-    }
-  })
-    .then(res => res.json())
-}
-
-function createName () {
-  const date = new Date(Date.now()).toISOString()
+/**
+ * Poor man's uuid
+ *
+ * @returns {String}
+ */
+function uuid () {
   function chr4 () {
     return Math.random().toString(16).slice(-4)
   }
-  return date + '--' + chr4() + chr4() +
+  return chr4() + chr4() +
       '-' + chr4() +
       '-' + chr4() +
       '-' + chr4() +
       '-' + chr4() + chr4() + chr4()
 }
 
-const sendChunkRequest = (send, options, chunk, id, start, end, name, boundary, size = '*') => {
+/**
+ * Send http request
+ *
+ * @param {function} send
+ * @param {Object} options - http request options
+ * @param {Uint8Array} chunk - chunk to send
+ * @param {Object} {id, start, end, name, boundary, size = '*'} - uploading session state
+ * @returns {Promise}
+ */
+const sendChunkRequest = (send, options, chunk, { boundary, id, index, rangeStart, rangeEnd, rangeTotal = '*' }) => {
   return new Promise((resolve, reject) => {
     const qs = {
       'cid-version': options['cid-version'],
@@ -141,21 +135,21 @@ const sendChunkRequest = (send, options, chunk, id, start, end, name, boundary,
       progress: options.progress,
       headers: {
         'Content-Type': 'application/octet-stream',
-        'Content-Range': `bytes ${start}-${end}/${size}`,
-        'Ipfs-Chunk-Name': name,
-        'Ipfs-Chunk-Id': id,
-        'Ipfs-Chunk-Boundary': boundary
+        'Content-Range': `bytes ${rangeStart}-${rangeEnd}/${rangeTotal}`,
+        'X-Ipfs-Chunk-Group-Uuid': id,
+        'X-Ipfs-Chunk-Index': index,
+        'X-Ipfs-Chunk-Boundary': boundary
       }
     }
 
     const req = send(args, (err, res) => {
       if (err) {
         return reject(err)
       }
-
       resolve(res)
     })
 
+    // write and send
     req.write(Buffer.from(chunk))
     req.end()
   })

src/add2/multipart2.js

@@ -33,12 +33,12 @@ const leading = (headers = {}, boundary) => {
 }
 
 class Multipart extends Duplex {
-  constructor (options) {
-    super(Object.assign({}, options, { writableObjectMode: true, writableHighWaterMark: 1 }))
+  constructor ({ chunkSize = 256000 } = {}) {
+    super({ writableObjectMode: true, writableHighWaterMark: 1 })
 
     this._boundary = generateBoundary()
     this.source = null
-    this.chunkSize = 256000
+    this.chunkSize = chunkSize
     this.buffer = Buffer.alloc(this.chunkSize)
     this.bufferOffset = 0
     this.running = true
@@ -116,8 +116,8 @@ class Multipart extends Duplex {
       content = toStream.source(content)
     }
     this.source = content
-    // From now on we assume content is a stream
 
+    // From now on we assume content is a stream
     content.on('data', (data) => {
       if (!this.pushChunk(data)) {
         content.pause()