test/parallel/test-stream3-pause-then-read.js

   1 var common = require('../common');
   2 var assert = require('assert');
   3
   4 var stream = require('stream');
   5 var Readable = stream.Readable;
   6 var Writable = stream.Writable;
   7
   8 var totalChunks = 100;
   9 var chunkSize = 99;
  10 var expectTotalData = totalChunks * chunkSize;
  11 var expectEndingData = expectTotalData;
  12
  13 var r = new Readable({ highWaterMark: 1000 });
  14 var chunks = totalChunks;
  15 r._read = function(n) {
  16   if (!(chunks % 2))
  17     setImmediate(push);
  18   else if (!(chunks % 3))
  19     process.nextTick(push);
  20   else
  21     push();
  22 };
  23
  24 var totalPushed = 0;
  25 function push() {
  26   var chunk = chunks-- > 0 ? new Buffer(chunkSize) : null;
  27   if (chunk) {
  28     totalPushed += chunk.length;
  29     chunk.fill('x');
  30   }
  31   r.push(chunk);
  32 }
  33
  34 read100();
  35
  36 // first we read 100 bytes
  37 function read100() {
  38   readn(100, onData);
  39 }
  40
  41 function readn(n, then) {
  42   console.error('read %d', n);
  43   expectEndingData -= n;
  44   ;(function read() {
  45     var c = r.read(n);
  46     if (!c)
  47       r.once('readable', read);
  48     else {
  49       assert.equal(c.length, n);
  50       assert(!r._readableState.flowing);
  51       then();
  52     }
  53   })();
  54 }
  55
  56 // then we listen to some data events
  57 function onData() {
  58   expectEndingData -= 100;
  59   console.error('onData');
  60   var seen = 0;
  61   r.on('data', function od(c) {
  62     seen += c.length;
  63     if (seen >= 100) {
  64       // seen enough
  65       r.removeListener('data', od);
  66       r.pause();
  67       if (seen > 100) {
  68         // oh no, seen too much!
  69         // put the extra back.
  70         var diff = seen - 100;
  71         r.unshift(c.slice(c.length - diff));
  72         console.error('seen too much', seen, diff);
  73       }
  74
  75       // Nothing should be lost in between
  76       setImmediate(pipeLittle);
  77     }
  78   });
  79 }
  80
  81 // Just pipe 200 bytes, then unshift the extra and unpipe
  82 function pipeLittle() {
  83   expectEndingData -= 200;
  84   console.error('pipe a little');
  85   var w = new Writable();
  86   var written = 0;
  87   w.on('finish', function() {
  88     assert.equal(written, 200);
  89     setImmediate(read1234);
  90   });
  91   w._write = function(chunk, encoding, cb) {
  92     written += chunk.length;
  93     if (written >= 200) {
  94       r.unpipe(w);
  95       w.end();
  96       cb();
  97       if (written > 200) {
  98         var diff = written - 200;
  99         written -= diff;
 100         r.unshift(chunk.slice(chunk.length - diff));
 101       }
 102     } else {
 103       setImmediate(cb);
 104     }
 105   };
 106   r.pipe(w);
 107 }
 108
 109 // now read 1234 more bytes
 110 function read1234() {
 111   readn(1234, resumePause);
 112 }
 113
 114 function resumePause() {
 115   console.error('resumePause');
 116   // don't read anything, just resume and re-pause a whole bunch
 117   r.resume();
 118   r.pause();
 119   r.resume();
 120   r.pause();
 121   r.resume();
 122   r.pause();
 123   r.resume();
 124   r.pause();
 125   r.resume();
 126   r.pause();
 127   setImmediate(pipe);
 128 }
 129
 130
 131 function pipe() {
 132   console.error('pipe the rest');
 133   var w = new Writable();
 134   var written = 0;
 135   w._write = function(chunk, encoding, cb) {
 136     written += chunk.length;
 137     cb();
 138   };
 139   w.on('finish', function() {
 140     console.error('written', written, totalPushed);
 141     assert.equal(written, expectEndingData);
 142     assert.equal(totalPushed, expectTotalData);
 143     console.log('ok');
 144   });
 145   r.pipe(w);
 146 }