lib/_stream_transform.js

   1 // Copyright Joyent, Inc. and other Node contributors.
   2 //
   3 // Permission is hereby granted, free of charge, to any person obtaining a
   4 // copy of this software and associated documentation files (the
   5 // "Software"), to deal in the Software without restriction, including
   6 // without limitation the rights to use, copy, modify, merge, publish,
   7 // distribute, sublicense, and/or sell copies of the Software, and to permit
   8 // persons to whom the Software is furnished to do so, subject to the
   9 // following conditions:
  10 //
  11 // The above copyright notice and this permission notice shall be included
  12 // in all copies or substantial portions of the Software.
  13 //
  14 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
  15 // OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
  17 // NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
  18 // DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
  19 // OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
  20 // USE OR OTHER DEALINGS IN THE SOFTWARE.
  21
  22
  23 // a transform stream is a readable/writable stream where you do
  24 // something with the data.  Sometimes it's called a "filter",
  25 // but that's not a great name for it, since that implies a thing where
  26 // some bits pass through, and others are simply ignored.  (That would
  27 // be a valid example of a transform, of course.)
  28 //
  29 // While the output is causally related to the input, it's not a
  30 // necessarily symmetric or synchronous transformation.  For example,
  31 // a zlib stream might take multiple plain-text writes(), and then
  32 // emit a single compressed chunk some time in the future.
  33 //
  34 // Here's how this works:
  35 //
  36 // The Transform stream has all the aspects of the readable and writable
  37 // stream classes.  When you write(chunk), that calls _write(chunk,cb)
  38 // internally, and returns false if there's a lot of pending writes
  39 // buffered up.  When you call read(), that calls _read(n) until
  40 // there's enough pending readable data buffered up.
  41 //
  42 // In a transform stream, the written data is placed in a buffer.  When
  43 // _read(n) is called, it transforms the queued up data, calling the
  44 // buffered _write cb's as it consumes chunks.  If consuming a single
  45 // written chunk would result in multiple output chunks, then the first
  46 // outputted bit calls the readcb, and subsequent chunks just go into
  47 // the read buffer, and will cause it to emit 'readable' if necessary.
  48 //
  49 // This way, back-pressure is actually determined by the reading side,
  50 // since _read has to be called to start processing a new chunk.  However,
  51 // a pathological inflate type of transform can cause excessive buffering
  52 // here.  For example, imagine a stream where every byte of input is
  53 // interpreted as an integer from 0-255, and then results in that many
  54 // bytes of output.  Writing the 4 bytes {ff,ff,ff,ff} would result in
  55 // 1kb of data being output.  In this case, you could write a very small
  56 // amount of input, and end up with a very large amount of output.  In
  57 // such a pathological inflating mechanism, there'd be no way to tell
  58 // the system to stop doing the transform.  A single 4MB write could
  59 // cause the system to run out of memory.
  60 //
  61 // However, even in such a pathological case, only a single written chunk
  62 // would be consumed, and then the rest would wait (un-transformed) until
  63 // the results of the previous transformed chunk were consumed.
  64
  65 module.exports = Transform;
  66
  67 var Duplex = require('_stream_duplex');
  68 var util = require('util');
  69 util.inherits(Transform, Duplex);
  70
  71
  72 function TransformState(options, stream) {
  73   this.afterTransform = function(er, data) {
  74     return afterTransform(stream, er, data);
  75   };
  76
  77   this.needTransform = false;
  78   this.transforming = false;
  79   this.writecb = null;
  80   this.writechunk = null;
  81 }
  82
  83 function afterTransform(stream, er, data) {
  84   var ts = stream._transformState;
  85   ts.transforming = false;
  86
  87   var cb = ts.writecb;
  88
  89   if (!cb)
  90     return stream.emit('error', new Error('no writecb in Transform class'));
  91
  92   ts.writechunk = null;
  93   ts.writecb = null;
  94
  95   if (!util.isNullOrUndefined(data))
  96     stream.push(data);
  97
  98   if (cb)
  99     cb(er);
 100
 101   var rs = stream._readableState;
 102   rs.reading = false;
 103   if (rs.needReadable || rs.length < rs.highWaterMark) {
 104     stream._read(rs.highWaterMark);
 105   }
 106 }
 107
 108
 109 function Transform(options) {
 110   if (!(this instanceof Transform))
 111     return new Transform(options);
 112
 113   Duplex.call(this, options);
 114
 115   this._transformState = new TransformState(options, this);
 116
 117   // when the writable side finishes, then flush out anything remaining.
 118   var stream = this;
 119
 120   // start out asking for a readable event once data is transformed.
 121   this._readableState.needReadable = true;
 122
 123   // we have implemented the _read method, and done the other things
 124   // that Readable wants before the first _read call, so unset the
 125   // sync guard flag.
 126   this._readableState.sync = false;
 127
 128   this.once('prefinish', function() {
 129     if (util.isFunction(this._flush))
 130       this._flush(function(er) {
 131         done(stream, er);
 132       });
 133     else
 134       done(stream);
 135   });
 136 }
 137
 138 Transform.prototype.push = function(chunk, encoding) {
 139   this._transformState.needTransform = false;
 140   return Duplex.prototype.push.call(this, chunk, encoding);
 141 };
 142
 143 // This is the part where you do stuff!
 144 // override this function in implementation classes.
 145 // 'chunk' is an input chunk.
 146 //
 147 // Call `push(newChunk)` to pass along transformed output
 148 // to the readable side.  You may call 'push' zero or more times.
 149 //
 150 // Call `cb(err)` when you are done with this chunk.  If you pass
 151 // an error, then that'll put the hurt on the whole operation.  If you
 152 // never call cb(), then you'll never get another chunk.
 153 Transform.prototype._transform = function(chunk, encoding, cb) {
 154   throw new Error('not implemented');
 155 };
 156
 157 Transform.prototype._write = function(chunk, encoding, cb) {
 158   var ts = this._transformState;
 159   ts.writecb = cb;
 160   ts.writechunk = chunk;
 161   ts.writeencoding = encoding;
 162   if (!ts.transforming) {
 163     var rs = this._readableState;
 164     if (ts.needTransform ||
 165         rs.needReadable ||
 166         rs.length < rs.highWaterMark)
 167       this._read(rs.highWaterMark);
 168   }
 169 };
 170
 171 // Doesn't matter what the args are here.
 172 // _transform does all the work.
 173 // That we got here means that the readable side wants more data.
 174 Transform.prototype._read = function(n) {
 175   var ts = this._transformState;
 176
 177   if (!util.isNull(ts.writechunk) && ts.writecb && !ts.transforming) {
 178     ts.transforming = true;
 179     this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
 180   } else {
 181     // mark that we need a transform, so that any data that comes in
 182     // will get processed, now that we've asked for it.
 183     ts.needTransform = true;
 184   }
 185 };
 186
 187
 188 function done(stream, er) {
 189   if (er)
 190     return stream.emit('error', er);
 191
 192   // if there's nothing in the write buffer, then that means
 193   // that nothing more will ever be provided
 194   var ws = stream._writableState;
 195   var ts = stream._transformState;
 196
 197   if (ws.length)
 198     throw new Error('calling transform done when ws.length != 0');
 199
 200   if (ts.transforming)
 201     throw new Error('calling transform done when still transforming');
 202
 203   return stream.push(null);
 204 }