add chunk into stream at a specific index

[lewispham]

Node (iojs) just supports Stream.push which means chunks can only be added into Stream at the last index. It would be nice to be able to add chunk into Stream at a specific index.

[vkurchatkin]

@Tresdin do you have a use case for this?

[lewispham]

@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.

[vkurchatkin]

First of all, push is a method of Readable stream, 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.

[vkurchatkin]

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.

[lewispham]

Thanks for your informative responses @vkurchatkin .
Although your code can partially solve the issue, I still hope this feature will be supported.

[lewispham]

I've found that iojs docs recommends not to fs.write data 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.

[vkurchatkin]

@Tresdin right, you'll have to manage a queue for the calls yourself

[Fishrock123]

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. :)

[lewispham]

@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.

[lewispham]

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.