Repository navigation
add chunk into stream at a specific index #2311
Description
Activity
@Tresdin do you have a use case for this?
@vkurchatkin Yes. I got this issue when I was trying to upload multiple large files via WebSocket. And this is what I'm thinking about.
Client
inputElement.addEventListenner('change',function(){ var files = this.files; var chunkSize = 4000; for(var i=0;i<files.length;i++){ var fr = new FileReader(); fr.onload = function(){ var dataurl = this.result; var index = 0; for(var j=0, len = Math.ceil(dataurl.length/chunkSize));j<len;j++){ WebSocket.send(JSON.stringify([j,dataurl.substr(index,chunkSize),len])); index += chunkSize; } }; fr.readAsDataURL(files[i]); } });Server
var files = {}; var countChunks = {}; ws.on('message',function(data){ var chunkData = JSON.parse(data); var path = 'path/to/file'; if(!(path in files)){ files[path] = fs.createWriteStream(path); countChunks[path] = 0; } var stream = files[path]; //assume that `Stream.add(index,chunk)` is available stream.add(chunkData[0],chunkData[1]); countChunks[path]++; if(countChunks[path] === chunkData[2]){ stream.end(); } });The above approach, of course, is not possible at this time.
First of all,
pushis a method ofReadablestream, you are talking about writable.What you are trying to do has nothing to with streaming. If you expect that packets could arrive out of order (which, I believe, is not possible with websockets) you can just calculate offset and write data using
fs.write.Also, here is a simple code that might help you:
var Readable = require('stream').Readable; var inherits = require('util').inherits; function OrderStream() { Readable.call(this); this._buffer = {}; this._idx = 0; } inherits(OrderStream, Readable); OrderStream.prototype._read = function() {}; OrderStream.prototype.add = function(idx, chunk) { if (this._idx === idx) { push(this, idx, chunk); } else { this._buffer[idx] = chunk; } }; function push(stream, idx, chunk) { stream.push(chunk); stream._idx++; while (stream._buffer[stream._idx]) { stream.push(stream._buffer[stream._idx]); delete stream._buffer[stream._idx]; stream._idx++; } }
It buffers chunks that arrived early and pushes them later. It is a readable stream, so you need to pipe to writable fs stream in your case.
Reacted by Jimmy WärtingThanks for your informative responses @vkurchatkin .
Although your code can partially solve the issue, I still hope this feature will be supported.I've found that iojs docs recommends not to
fs.writedata into same source in parallel.Note that it is unsafe to use fs.write multiple times on the same file without waiting for the callback. For this scenario, fs.createWriteStream is strongly recommended.
@Tresdin right, you'll have to manage a queue for the calls yourself
I think this is more of a question, and it seems to be answered so I'm closing for now. @Tresdin let us know if this wasn't correct or if you have more questions. :)
@vkurchatkin I actually mean writing chunks into file in parallel is unreliable, and node should fix it. Queueing chunks on node will end up with higher memory usage. Queueing chunks on client will cause lower uploading speed.
I'm really surprised that you guys are not interested in out-of-order streaming approach. It can even be the alternative for traditional streaming paradigm in most cases because of its performance improvement. Handling files can be a good example of it.
I'm not sure what exactly you are proposing?
OK. I mainly asked for a solution to add chunks sent in parallel into stream. And we end up with these 3 approaches:
- Using
fs.writewrite chunks in parallel into file. Unfortunately, this approach is unreliable. - Queueing arrived chunks then push them later. This approach is inefficient in term of memory usage.
- Queueing chunks on client side then send them in order. This approach is inefficient in term of speed.
Then I proposed making
fs.writereliable.- Using
Node (iojs) just supports
Stream.pushwhich means chunks can only be added intoStreamat the last index. It would be nice to be able to add chunk intoStreamat a specific index.