www.aleph1.co.uk Git - yaffs-website/blob - node_modules/fd-slicer/index.js

   1 var fs = require('fs');
   2 var util = require('util');
   3 var stream = require('stream');
   4 var Readable = stream.Readable;
   5 var Writable = stream.Writable;
   6 var PassThrough = stream.PassThrough;
   7 var Pend = require('pend');
   8 var EventEmitter = require('events').EventEmitter;
   9
  10 exports.createFromBuffer = createFromBuffer;
  11 exports.createFromFd = createFromFd;
  12 exports.BufferSlicer = BufferSlicer;
  13 exports.FdSlicer = FdSlicer;
  14
  15 util.inherits(FdSlicer, EventEmitter);
  16 function FdSlicer(fd, options) {
  17   options = options || {};
  18   EventEmitter.call(this);
  19
  20   this.fd = fd;
  21   this.pend = new Pend();
  22   this.pend.max = 1;
  23   this.refCount = 0;
  24   this.autoClose = !!options.autoClose;
  25 }
  26
  27 FdSlicer.prototype.read = function(buffer, offset, length, position, callback) {
  28   var self = this;
  29   self.pend.go(function(cb) {
  30     fs.read(self.fd, buffer, offset, length, position, function(err, bytesRead, buffer) {
  31       cb();
  32       callback(err, bytesRead, buffer);
  33     });
  34   });
  35 };
  36
  37 FdSlicer.prototype.write = function(buffer, offset, length, position, callback) {
  38   var self = this;
  39   self.pend.go(function(cb) {
  40     fs.write(self.fd, buffer, offset, length, position, function(err, written, buffer) {
  41       cb();
  42       callback(err, written, buffer);
  43     });
  44   });
  45 };
  46
  47 FdSlicer.prototype.createReadStream = function(options) {
  48   return new ReadStream(this, options);
  49 };
  50
  51 FdSlicer.prototype.createWriteStream = function(options) {
  52   return new WriteStream(this, options);
  53 };
  54
  55 FdSlicer.prototype.ref = function() {
  56   this.refCount += 1;
  57 };
  58
  59 FdSlicer.prototype.unref = function() {
  60   var self = this;
  61   self.refCount -= 1;
  62
  63   if (self.refCount > 0) return;
  64   if (self.refCount < 0) throw new Error("invalid unref");
  65
  66   if (self.autoClose) {
  67     fs.close(self.fd, onCloseDone);
  68   }
  69
  70   function onCloseDone(err) {
  71     if (err) {
  72       self.emit('error', err);
  73     } else {
  74       self.emit('close');
  75     }
  76   }
  77 };
  78
  79 util.inherits(ReadStream, Readable);
  80 function ReadStream(context, options) {
  81   options = options || {};
  82   Readable.call(this, options);
  83
  84   this.context = context;
  85   this.context.ref();
  86
  87   this.start = options.start || 0;
  88   this.endOffset = options.end;
  89   this.pos = this.start;
  90   this.destroyed = false;
  91 }
  92
  93 ReadStream.prototype._read = function(n) {
  94   var self = this;
  95   if (self.destroyed) return;
  96
  97   var toRead = Math.min(self._readableState.highWaterMark, n);
  98   if (self.endOffset != null) {
  99     toRead = Math.min(toRead, self.endOffset - self.pos);
 100   }
 101   if (toRead <= 0) {
 102     self.destroyed = true;
 103     self.push(null);
 104     self.context.unref();
 105     return;
 106   }
 107   self.context.pend.go(function(cb) {
 108     if (self.destroyed) return cb();
 109     var buffer = new Buffer(toRead);
 110     fs.read(self.context.fd, buffer, 0, toRead, self.pos, function(err, bytesRead) {
 111       if (err) {
 112         self.destroy(err);
 113       } else if (bytesRead === 0) {
 114         self.destroyed = true;
 115         self.push(null);
 116         self.context.unref();
 117       } else {
 118         self.pos += bytesRead;
 119         self.push(buffer.slice(0, bytesRead));
 120       }
 121       cb();
 122     });
 123   });
 124 };
 125
 126 ReadStream.prototype.destroy = function(err) {
 127   if (this.destroyed) return;
 128   err = err || new Error("stream destroyed");
 129   this.destroyed = true;
 130   this.emit('error', err);
 131   this.context.unref();
 132 };
 133
 134 util.inherits(WriteStream, Writable);
 135 function WriteStream(context, options) {
 136   options = options || {};
 137   Writable.call(this, options);
 138
 139   this.context = context;
 140   this.context.ref();
 141
 142   this.start = options.start || 0;
 143   this.endOffset = (options.end == null) ? Infinity : +options.end;
 144   this.bytesWritten = 0;
 145   this.pos = this.start;
 146   this.destroyed = false;
 147
 148   this.on('finish', this.destroy.bind(this));
 149 }
 150
 151 WriteStream.prototype._write = function(buffer, encoding, callback) {
 152   var self = this;
 153   if (self.destroyed) return;
 154
 155   if (self.pos + buffer.length > self.endOffset) {
 156     var err = new Error("maximum file length exceeded");
 157     err.code = 'ETOOBIG';
 158     self.destroy();
 159     callback(err);
 160     return;
 161   }
 162   self.context.pend.go(function(cb) {
 163     if (self.destroyed) return cb();
 164     fs.write(self.context.fd, buffer, 0, buffer.length, self.pos, function(err, bytes) {
 165       if (err) {
 166         self.destroy();
 167         cb();
 168         callback(err);
 169       } else {
 170         self.bytesWritten += bytes;
 171         self.pos += bytes;
 172         self.emit('progress');
 173         cb();
 174         callback();
 175       }
 176     });
 177   });
 178 };
 179
 180 WriteStream.prototype.destroy = function() {
 181   if (this.destroyed) return;
 182   this.destroyed = true;
 183   this.context.unref();
 184 };
 185
 186 util.inherits(BufferSlicer, EventEmitter);
 187 function BufferSlicer(buffer) {
 188   EventEmitter.call(this);
 189
 190   this.refCount = 0;
 191   this.buffer = buffer;
 192 }
 193
 194 BufferSlicer.prototype.read = function(buffer, offset, length, position, callback) {
 195   var end = position + length;
 196   var delta = end - this.buffer.length;
 197   var written = (delta > 0) ? delta : length;
 198   this.buffer.copy(buffer, offset, position, end);
 199   setImmediate(function() {
 200     callback(null, written);
 201   });
 202 };
 203
 204 BufferSlicer.prototype.write = function(buffer, offset, length, position, callback) {
 205   buffer.copy(this.buffer, position, offset, offset + length);
 206   setImmediate(function() {
 207     callback(null, length, buffer);
 208   });
 209 };
 210
 211 BufferSlicer.prototype.createReadStream = function(options) {
 212   options = options || {};
 213   var readStream = new PassThrough(options);
 214   readStream.start = options.start || 0;
 215   readStream.endOffset = options.end;
 216   readStream.pos = readStream.endOffset || this.buffer.length; // yep, we're already done
 217   readStream.destroyed = false;
 218   readStream.write(this.buffer.slice(readStream.start, readStream.pos));
 219   readStream.end();
 220   readStream.destroy = function() {
 221     readStream.destroyed = true;
 222   };
 223   return readStream;
 224 };
 225
 226 BufferSlicer.prototype.createWriteStream = function(options) {
 227   var bufferSlicer = this;
 228   options = options || {};
 229   var writeStream = new Writable(options);
 230   writeStream.start = options.start || 0;
 231   writeStream.endOffset = (options.end == null) ? this.buffer.length : +options.end;
 232   writeStream.bytesWritten = 0;
 233   writeStream.pos = writeStream.start;
 234   writeStream.destroyed = false;
 235   writeStream._write = function(buffer, encoding, callback) {
 236     if (writeStream.destroyed) return;
 237
 238     var end = writeStream.pos + buffer.length;
 239     if (end > writeStream.endOffset) {
 240       var err = new Error("maximum file length exceeded");
 241       err.code = 'ETOOBIG';
 242       writeStream.destroyed = true;
 243       callback(err);
 244       return;
 245     }
 246     buffer.copy(bufferSlicer.buffer, writeStream.pos, 0, buffer.length);
 247
 248     writeStream.bytesWritten += buffer.length;
 249     writeStream.pos = end;
 250     writeStream.emit('progress');
 251     callback();
 252   };
 253   writeStream.destroy = function() {
 254     writeStream.destroyed = true;
 255   };
 256   return writeStream;
 257 };
 258
 259 BufferSlicer.prototype.ref = function() {
 260   this.refCount += 1;
 261 };
 262
 263 BufferSlicer.prototype.unref = function() {
 264   this.refCount -= 1;
 265
 266   if (this.refCount < 0) {
 267     throw new Error("invalid unref");
 268   }
 269 };
 270
 271 function createFromBuffer(buffer) {
 272   return new BufferSlicer(buffer);
 273 }
 274
 275 function createFromFd(fd, options) {
 276   return new FdSlicer(fd, options);
 277 }