device_home/node_modules/pngjs/lib/chunkstream.js

   1 'use strict';
   2
   3
   4 var util = require('util');
   5 var Stream = require('stream');
   6
   7
   8 var ChunkStream = module.exports = function() {
   9   Stream.call(this);
  10
  11   this._buffers = [];
  12   this._buffered = 0;
  13
  14   this._reads = [];
  15   this._paused = false;
  16
  17   this._encoding = 'utf8';
  18   this.writable = true;
  19 };
  20 util.inherits(ChunkStream, Stream);
  21
  22
  23 ChunkStream.prototype.read = function(length, callback) {
  24
  25   this._reads.push({
  26     length: Math.abs(length), // if length < 0 then at most this length
  27     allowLess: length < 0,
  28     func: callback
  29   });
  30
  31   process.nextTick(function() {
  32     this._process();
  33
  34     // its paused and there is not enought data then ask for more
  35     if (this._paused && this._reads.length > 0) {
  36       this._paused = false;
  37
  38       this.emit('drain');
  39     }
  40   }.bind(this));
  41 };
  42
  43 ChunkStream.prototype.write = function(data, encoding) {
  44
  45   if (!this.writable) {
  46     this.emit('error', new Error('Stream not writable'));
  47     return false;
  48   }
  49
  50   var dataBuffer;
  51   if (Buffer.isBuffer(data)) {
  52     dataBuffer = data;
  53   }
  54   else {
  55     dataBuffer = new Buffer(data, encoding || this._encoding);
  56   }
  57
  58   this._buffers.push(dataBuffer);
  59   this._buffered += dataBuffer.length;
  60
  61   this._process();
  62
  63   // ok if there are no more read requests
  64   if (this._reads && this._reads.length === 0) {
  65     this._paused = true;
  66   }
  67
  68   return this.writable && !this._paused;
  69 };
  70
  71 ChunkStream.prototype.end = function(data, encoding) {
  72
  73   if (data) {
  74     this.write(data, encoding);
  75   }
  76
  77   this.writable = false;
  78
  79   // already destroyed
  80   if (!this._buffers) {
  81     return;
  82   }
  83
  84   // enqueue or handle end
  85   if (this._buffers.length === 0) {
  86     this._end();
  87   }
  88   else {
  89     this._buffers.push(null);
  90     this._process();
  91   }
  92 };
  93
  94 ChunkStream.prototype.destroySoon = ChunkStream.prototype.end;
  95
  96 ChunkStream.prototype._end = function() {
  97
  98   if (this._reads.length > 0) {
  99     this.emit('error',
 100       new Error('Unexpected end of input')
 101     );
 102   }
 103
 104   this.destroy();
 105 };
 106
 107 ChunkStream.prototype.destroy = function() {
 108
 109   if (!this._buffers) {
 110     return;
 111   }
 112
 113   this.writable = false;
 114   this._reads = null;
 115   this._buffers = null;
 116
 117   this.emit('close');
 118 };
 119
 120 ChunkStream.prototype._processReadAllowingLess = function(read) {
 121   // ok there is any data so that we can satisfy this request
 122   this._reads.shift(); // == read
 123
 124   // first we need to peek into first buffer
 125   var smallerBuf = this._buffers[0];
 126
 127   // ok there is more data than we need
 128   if (smallerBuf.length > read.length) {
 129
 130     this._buffered -= read.length;
 131     this._buffers[0] = smallerBuf.slice(read.length);
 132
 133     read.func.call(this, smallerBuf.slice(0, read.length));
 134
 135   }
 136   else {
 137     // ok this is less than maximum length so use it all
 138     this._buffered -= smallerBuf.length;
 139     this._buffers.shift(); // == smallerBuf
 140
 141     read.func.call(this, smallerBuf);
 142   }
 143 };
 144
 145 ChunkStream.prototype._processRead = function(read) {
 146   this._reads.shift(); // == read
 147
 148   var pos = 0;
 149   var count = 0;
 150   var data = new Buffer(read.length);
 151
 152   // create buffer for all data
 153   while (pos < read.length) {
 154
 155     var buf = this._buffers[count++];
 156     var len = Math.min(buf.length, read.length - pos);
 157
 158     buf.copy(data, pos, 0, len);
 159     pos += len;
 160
 161     // last buffer wasn't used all so just slice it and leave
 162     if (len !== buf.length) {
 163       this._buffers[--count] = buf.slice(len);
 164     }
 165   }
 166
 167   // remove all used buffers
 168   if (count > 0) {
 169     this._buffers.splice(0, count);
 170   }
 171
 172   this._buffered -= read.length;
 173
 174   read.func.call(this, data);
 175 };
 176
 177 ChunkStream.prototype._process = function() {
 178
 179   try {
 180     // as long as there is any data and read requests
 181     while (this._buffered > 0 && this._reads && this._reads.length > 0) {
 182
 183       var read = this._reads[0];
 184
 185       // read any data (but no more than length)
 186       if (read.allowLess) {
 187         this._processReadAllowingLess(read);
 188
 189       }
 190       else if (this._buffered >= read.length) {
 191         // ok we can meet some expectations
 192
 193         this._processRead(read);
 194       }
 195       else {
 196         // not enought data to satisfy first request in queue
 197         // so we need to wait for more
 198         break;
 199       }
 200     }
 201
 202     if (this._buffers && !this.writable) {
 203       this._end();
 204     }
 205   }
 206   catch (ex) {
 207     this.emit('error', ex);
 208   }
 209 };