pxsStream.js

var
    fs = require('fs'),
    util = require('util'),
    stream = require('stream');

util.inherits(PxyStream,stream.Readable);
function PxyStream(path,readStream,writeStream,start,end){
    stream.Readable.call(this);
    this.path = path;
    this.readStream = readStream;
    this.writeStream = writeStream;
    this._offset = start;
    this.endPos = end;
    this.writeStream.on('finish',function(){
        this._writeStreamFinished = true;
        this._writeStreamBytes = this.writeStream.bytesWritten;
        //this.nextStream();
       // if(this._callReadAgain) this._read(this._cran);
    }.bind(this))
}

PxyStream.prototype._read = function(n){
    this._callReadAgain = false;
    clearTimeout(this._tm);
    if(!this._stream){
        process.nextTick(function(){
            this.nextStream();
        }.bind(this));
    }
    var chunk;
    if(this._stream) chunk = this._stream.read(n);
    if(chunk){
        this.push(chunk);
        console.log('chunk['+this._offset.toString(16)+']='+chunk[0].toString(16));
        this._offset += chunk.length;
    }
    else {

        if(this._lastEnded) {
            console.log('end pxyStream');
            this.push(null);
            return;
        }
        this._callReadAgain = true;
        this._cran = n;
        this._tm = setTimeout(function(){
            console.log('timeout on read next call');
        },1000);
    }
}


PxyStream.prototype.nextStream = function(){
    if(!this._stream && !this._lastEnded){
        var options = {
            start: this._offset
        };
        var last = this._writeStreamFinished;
        console.log('new stream ',this._offset);
        this._stream = fs.createReadStream(this.path,options);
        this._stream.on('readable',function(){
            if(this._callReadAgain) this._read(this._cran);
        }.bind(this));
        this._stream.on('end',function(){
            this._stream = null;
            this._lastEnded = last || (this._writeStreamFinished && (this._writeStreamBytes==this._offset));
            console.log('ended',last,this._writeStreamFinished,this._lastEnded,this._writeStreamBytes,this._offset);
            if(this._callReadAgain) this._read(this._cran);
        }.bind(this));
    }
}