feat: support chunked add requests

Author: hugomrdias

package.json

@@ -170,6 +170,7 @@
     "stream-to-pull-stream": "^1.7.2",
     "tar-stream": "^1.6.1",
     "temp": "~0.8.3",
+    "tempy": "^0.2.1",
     "through2": "^2.0.3",
     "update-notifier": "^2.5.0",
     "yargs": "^12.0.1",

src/http/api/resources/files.js

@@ -165,6 +165,7 @@ exports.add = {
   },
 
   handler: (request, reply) => {
+    console.log('yoo')
     if (!request.payload) {
       return reply({
         Message: 'Array, Buffer, or String is required.',
@@ -182,6 +183,11 @@ exports.add = {
 
     parser.on('file', (fileName, fileStream) => {
       fileName = decodeURIComponent(fileName)
+
+      console.log('file', fileName)
+      fileStream.on('data', (d) => {
+        console.log(d.byteLength)
+      })
       const filePair = {
         path: fileName,
         content: toPull(fileStream)
@@ -192,7 +198,7 @@ exports.add = {
 
     parser.on('directory', (directory) => {
       directory = decodeURIComponent(directory)
-
+      console.log('directory', directory)
       fileAdder.push({
         path: directory,
         content: ''
@@ -220,7 +226,7 @@ exports.add = {
       rawLeaves: request.query['raw-leaves'],
       progress: request.query.progress ? progressHandler : null,
       onlyHash: request.query['only-hash'],
-      hashAlg: request.query['hash'],
+      hashAlg: request.query.hash,
       wrapWithDirectory: request.query['wrap-with-directory'],
       pin: request.query.pin,
       chunker: request.query.chunker

src/http/api/routes/files.js

@@ -2,7 +2,66 @@
 
 const resources = require('./../resources')
 const mfs = require('ipfs-mfs/http')
+const fs = require('fs')
+const path = require('path')
+const tempy = require('tempy')
+const multipart = require('ipfs-multipart')
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
+const pushable = require('pull-pushable')
 
+const streams = []
+const filesDir = tempy.directory()
+
+const createMultipartStream = (readStream, boundary, ipfs, cb) => {
+  const parser = new multipart.Parser({ boundary: boundary })
+  readStream.pipe(parser)
+  const fileAdder = pushable()
+
+  parser.on('file', (fileName, fileStream) => {
+    fileName = decodeURIComponent(fileName)
+
+    const filePair = {
+      path: fileName,
+      content: toPull(fileStream)
+    }
+    console.log(filePair)
+    fileAdder.push(filePair)
+  })
+
+  parser.on('directory', (directory) => {
+    directory = decodeURIComponent(directory)
+    fileAdder.push({
+      path: directory,
+      content: ''
+    })
+  })
+
+  parser.on('end', () => {
+    fileAdder.end()
+  })
+
+  pull(
+    fileAdder,
+    ipfs.files.addPullStream(),
+    pull.map((file) => {
+      return {
+        Name: file.path, // addPullStream already turned this into a hash if it wanted to
+        Hash: file.hash,
+        Size: file.size
+      }
+    }),
+    pull.collect((err, files) => {
+      if (err) {
+        cb(err)
+        return
+      }
+      cb(null, files)
+    })
+  )
+
+  return parser
+}
 module.exports = (server) => {
   const api = server.select('API')
 
@@ -37,13 +96,95 @@ module.exports = (server) => {
     config: {
       payload: {
         parse: false,
-        output: 'stream'
+        output: 'stream',
+        maxBytes: 10048576
       },
       handler: resources.files.add.handler,
       validate: resources.files.add.validate
     }
   })
 
+  api.route({
+    // TODO fix method
+    method: 'POST',
+    path: '/api/v0/add-chunked',
+    config: {
+      payload: {
+        parse: false,
+        maxBytes: 10048576
+      },
+      handler: (request, reply) => {
+        console.log('received')
+        console.log(request.headers['content-range'])
+        console.log(request.headers['ipfs-chunk-id'])
+        console.log(request.headers['ipfs-chunk-name'])
+        const boundary = request.headers['ipfs-chunk-boundary']
+        const id = request.headers['ipfs-chunk-name'] // change name to id
+        const index = Number(request.headers['ipfs-chunk-id'])
+        const file = path.join(filesDir, id)
+        const match = request.headers['content-range'].match(/(\d+)-(\d+)\/(\d+|\*)/)
+        const ipfs = request.server.app.ipfs
+        // if (!match || !match[1] || !match[2] || !match[3]) {
+        /* malformed content-range header */
+        // res.send('Bad Request', 400)
+        //   return;
+        // }
+
+        const start = parseInt(match[1])
+        const end = parseInt(match[2])
+        const total = parseInt(match[3])
+        // console.log(start, end, total, index, boundary)
+
+        let stream = streams[id]
+        if (!stream) {
+          console.log('create new stream', file)
+          stream = fs.createWriteStream(file, {flags: 'a+'})
+          streams[id] = stream
+        }
+
+        console.log('stream', file)
+        let size = 0
+        if (fs.existsSync(file)) {
+          size = fs.statSync(file).size
+        }
+
+        if ((end + 1) === size) {
+          /* duplicate chunk */
+          // res.send('Created', 201)
+          // return;
+        }
+
+        if (start !== size) {
+          /* missing chunk */
+          // res.send('Bad Request', 400)
+          // return;
+        }
+
+        if (index === -1) {
+          // check if size + payload.length === total
+          /* all chunks have been received */
+          stream.on('finish', function () {
+            console.log('add to ipfs from the file')
+            var readStream = fs.createReadStream(file)
+            createMultipartStream(readStream, boundary, ipfs, (err, files) => {
+              if (err) {
+                console.error(err)
+              }
+              console.log('finished adding to ipfs', files)
+              reply({files})
+            })
+          })
+
+          stream.end()
+        } else {
+          stream.write(request.payload)
+          /* this chunk has been processed successfully */
+          reply({ Bytes: request.payload.length })
+        }
+      }
+    }
+  })
+
   api.route({
     // TODO fix method
     method: '*',

src/http/index.js

@@ -103,7 +103,11 @@ function HttpApi (repo, config, cliArgs) {
           this.server = new Hapi.Server({
             connections: {
               routes: {
-                cors: true
+                cors: {
+                  origin: ['*'],
+                  additionalHeaders: ['X-Stream-Output, X-Chunked-Output, X-Content-Length', 'Content-Type', 'Content-Range', 'IPFS-CHUNK-NAME', 'IPFS-CHUNK-ID', 'ipfs-chunk-boundary'],
+                  additionalExposedHeaders: ['X-Stream-Output, X-Chunked-Output, X-Content-Length']
+                }
               }
             },
             debug: process.env.DEBUG ? {
@@ -137,14 +141,6 @@ function HttpApi (repo, config, cliArgs) {
           // load gateway routes
           require('./gateway/routes')(this.server)
 
-          // Set default headers
-          setHeader(this.server,
-            'Access-Control-Allow-Headers',
-            'X-Stream-Output, X-Chunked-Output, X-Content-Length')
-          setHeader(this.server,
-            'Access-Control-Expose-Headers',
-            'X-Stream-Output, X-Chunked-Output, X-Content-Length')
-
           this.server.start(cb)
         })
       },