This repository was archived by the owner on Feb 12, 2024. It is now read-only.
[WIP] feat: support chunked add requests #1540
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,8 +1,121 @@ | ||
| 'use strict' | ||
|
|
||
| const fs = require('fs') | ||
| const path = require('path') | ||
| const tempy = require('tempy') | ||
| const del = require('del') | ||
| const StreamConcat = require('stream-concat') | ||
| const boom = require('boom') | ||
| const pump = require('pump') | ||
| const glob = require('fast-glob') | ||
| const Joi = require('joi') | ||
| const content = require('content') | ||
| const { Parser } = require('ipfs-multipart') | ||
| const toPull = require('stream-to-pull-stream') | ||
| const toStream = require('pull-stream-to-stream') | ||
| const pull = require('pull-stream') | ||
| const pushable = require('pull-pushable') | ||
| const abortable = require('pull-abortable') | ||
| const { serialize } = require('pull-ndjson') | ||
| const mfs = require('ipfs-mfs/http') | ||
| const resources = require('./../resources') | ||
|
|
||
| const filesDir = tempy.directory() | ||
|
There was a problem hiding this comment. Just so that we don't forget: we should remove temporary files after successful upload, but also have a plan for what happens with temp files produced by failed/aborted ones. I don't think we can assume temp dir will be cleared by the OS. There was a problem hiding this comment. yes definitely should go like this
|
||
|
|
||
| const parseChunkedInput = (request) => { | ||
| const input = request.headers['x-chunked-input'] | ||
| const regex = /^uuid="([^"]+)";\s*index=(\d*)/i | ||
|
|
||
| if (!input) { | ||
| return null | ||
| } | ||
| const match = input.match(regex) | ||
|
|
||
| return [match[1], Number(match[2])] | ||
| } | ||
|
|
||
| const createMultipartReply = (readStream, request, reply, cb) => { | ||
| const fileAdder = pushable() | ||
| const boundary = content.type(request.headers['content-type']).boundary | ||
| const ipfs = request.server.app.ipfs | ||
| const query = request.query | ||
| const parser = new Parser({ boundary: boundary }) | ||
| readStream.pipe(parser) | ||
|
|
||
| parser.on('file', (fileName, fileStream) => { | ||
| fileAdder.push({ | ||
| path: decodeURIComponent(fileName), | ||
| content: toPull(fileStream) | ||
| }) | ||
| }) | ||
|
|
||
| parser.on('directory', (directory) => { | ||
| fileAdder.push({ | ||
| path: decodeURIComponent(directory), | ||
| content: '' | ||
| }) | ||
| }) | ||
|
|
||
| parser.on('end', () => { | ||
| fileAdder.end() | ||
| }) | ||
|
|
||
| parser.on('error', err => cb(err)) | ||
|
|
||
| const pushStream = pushable() | ||
| const abortStream = abortable() | ||
| const replyStream = toStream.source(pull( | ||
| pushStream, | ||
| abortStream, | ||
| serialize() | ||
| )) | ||
|
|
||
| // Fix Hapi Error: Stream must have a streams2 readable interface | ||
| if (!replyStream._read) { | ||
| replyStream._read = () => {} | ||
| replyStream._readableState = {} | ||
| replyStream.unpipe = () => {} | ||
| } | ||
|
|
||
| // setup reply | ||
| reply(replyStream) | ||
| .header('x-chunked-output', '1') | ||
| .header('content-type', 'application/json') | ||
|
|
||
| const progressHandler = (bytes) => { | ||
| pushStream.push({ Bytes: bytes }) | ||
| } | ||
| // ipfs add options | ||
| const options = { | ||
| cidVersion: query['cid-version'], | ||
| rawLeaves: query['raw-leaves'], | ||
| progress: query.progress ? progressHandler : null, | ||
| onlyHash: query['only-hash'], | ||
| hashAlg: query.hash, | ||
| wrapWithDirectory: query['wrap-with-directory'], | ||
| pin: query.pin, | ||
| chunker: query.chunker | ||
| } | ||
|
|
||
| pull( | ||
| fileAdder, | ||
| ipfs.files.addPullStream(options), | ||
| pull.collect((err, files) => { | ||
| if (err) { | ||
| pushStream.push({ | ||
| Message: err.toString(), | ||
| Code: 0, | ||
|
There was a problem hiding this comment. FYI I've tracked down the source of values used in Translating from iota notation, it is an enum: ErrNormal // ErrorType = 0
ErrClient // ErrorType = 1
ErrImplementation // ErrorType = 2
ErrNotFound // ErrorType = 3
ErrFatal // ErrorType = 3Not sure if that is a problem, probably hardcoding |
||
| Type: 'error' | ||
| }) | ||
| pushStream.end() | ||
| return | ||
| } | ||
| files.forEach((f) => pushStream.push(f)) | ||
| pushStream.end() | ||
| cb() | ||
| }) | ||
| ) | ||
| } | ||
| module.exports = (server) => { | ||
| const api = server.select('API') | ||
|
|
||
|
|
@@ -37,13 +150,109 @@ module.exports = (server) => { | |
| config: { | ||
| payload: { | ||
| parse: false, | ||
| // output: 'stream', | ||
| maxBytes: 10048576 | ||
| }, | ||
| handler: resources.files.add.handler, | ||
| validate: resources.files.add.validate | ||
| } | ||
| }) | ||
|
|
||
| api.route({ | ||
| method: 'POST', | ||
| path: '/api/v0/add-chunked', | ||
| config: { | ||
| payload: { | ||
| parse: false, | ||
| output: 'stream', | ||
| maxBytes: 1000 * 1024 * 1024 | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment.
we need this value to be high for the non chunked path
don't know. what you think ? from the hapi documentation when output=stream, maxBytes doesn't matter but this doesnt seem to be the case. Needs further investigation, maybe it's a matter of upgrading hapi dunno There was a problem hiding this comment. I can't think of anything right now. There was a problem hiding this comment. This should just be the maximum size we're prepared to accept for |
||
| // maxBytes: 10485760 | ||
| }, | ||
| validate: { | ||
| headers: { | ||
| 'content-range': Joi.string().regex(/(\d+)-(\d+)\/(\d+|\*)/), | ||
| 'x-chunked-input': Joi.string().regex(/^uuid="([^"]+)";\s*index=(\d*)/i) | ||
| }, | ||
| options: { | ||
| allowUnknown: true | ||
| } | ||
| }, | ||
| handler: (request, reply) => { | ||
| const chunkedInput = parseChunkedInput(request) | ||
|
|
||
| if (boom.isBoom(chunkedInput)) { | ||
| return reply(chunkedInput) | ||
| } | ||
|
|
||
| // non chunked | ||
| if (!chunkedInput) { | ||
| createMultipartReply( | ||
| request.payload, | ||
| request, | ||
| reply, | ||
| (err) => { | ||
| if (err) { | ||
| return reply(err) | ||
| } | ||
| console.log('Finished adding') | ||
| } | ||
| ) | ||
|
|
||
| return | ||
| } | ||
|
|
||
| // chunked | ||
| const [uuid, index] = chunkedInput | ||
| const [, start, , total] = request.headers['content-range'].match(/(\d+)-(\d+)\/(\d+|\*)/) | ||
| const file = path.join(filesDir, uuid) + '-' + index | ||
|
|
||
| // TODO validate duplicates, missing chunks | ||
|
|
||
| if (start === total) { | ||
| /* all chunks have been received */ | ||
| const base = path.join(filesDir, uuid) + '-' | ||
| const pattern = base + '*' | ||
| const files = glob.sync(pattern) | ||
|
|
||
| files.sort((a, b) => { | ||
| return Number(a.replace(base, '')) - Number(b.replace(base, '')) | ||
| }) | ||
|
|
||
| let fileIndex = 0 | ||
| const nextStream = () => fileIndex === files.length ? null : fs.createReadStream(files[fileIndex++]) | ||
| createMultipartReply( | ||
| new StreamConcat(nextStream), | ||
| request, | ||
| reply, | ||
| (err) => { | ||
| if (err) { | ||
| return reply(err) | ||
| } | ||
|
|
||
| console.log('Finished adding') | ||
| del(pattern, { force: true }) | ||
| .then(paths => { | ||
| console.log('Deleted files and folders:\n', paths.join('\n')) | ||
| }) | ||
| .catch(console.error) | ||
| } | ||
| ) | ||
| } else { | ||
| pump( | ||
| request.payload, | ||
| fs.createWriteStream(file), | ||
| (err) => { | ||
| if (err) { | ||
| return reply(err) | ||
| } | ||
| reply({ Bytes: total }) | ||
| } | ||
| ) | ||
| } | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| api.route({ | ||
| // TODO fix method | ||
| method: '*', | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we use
globor replaceglobwithfast-glob?