210 lines
		
	
	
		
			4.1 KiB
		
	
	
	
		
			JavaScript
		
	
	
	
	
	
			
		
		
	
	
			210 lines
		
	
	
		
			4.1 KiB
		
	
	
	
		
			JavaScript
		
	
	
	
	
	
| var test = require('tape');
 | |
| var exhaust = require('./index.js');
 | |
| var Stream = require('stream');
 | |
| var Readable = Stream.Readable;
 | |
| var Writable = Stream.Writable;
 | |
| var Duplex = Stream.Duplex;
 | |
| var through = require('through2');
 | |
| var S2Readable = require('readable-stream').Readable;
 | |
| 
 | |
| test('it should cause a Readable stream to complete if it\'s not piped anywhere', function(assert) {
 | |
|   var rs = new Readable({highWaterMark: 2});
 | |
|   var a = 0;
 | |
|   var ended = false;
 | |
|   rs._read = function() {
 | |
|     if (a++ < 100) {
 | |
|       rs.push(a + "");
 | |
|     } else {
 | |
|       ended = true;
 | |
|       rs.push(null);
 | |
|     }
 | |
|   };
 | |
| 
 | |
|   rs.on("end", function() {
 | |
|     assert.ok(a > 99, 'a should be > 99');
 | |
|     assert.ok(ended, 'it should end');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   exhaust(rs);
 | |
| });
 | |
| 
 | |
| test('should work with Readable streams in objectMode', function(assert) {
 | |
|   var rs = new Readable({highWaterMark: 2, objectMode: true});
 | |
|   var a = 0;
 | |
|   var ended = false;
 | |
|   rs._read = function() {
 | |
|     if (a++ < 100) {
 | |
|       rs.push(a);
 | |
|     } else {
 | |
|       ended = true;
 | |
|       rs.push(null);
 | |
|     }
 | |
|   };
 | |
| 
 | |
|   rs.on("end", function() {
 | |
|     assert.ok(a > 99, 'a > 99');
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   exhaust(rs);
 | |
| });
 | |
| 
 | |
| test('should not interfere with a Readable stream that is piped somewhere', function(assert) {
 | |
|   var rs = new Readable({highWaterMark: 2});
 | |
|   var a = 0;
 | |
|   var ended = false;
 | |
|   rs._read = function() {
 | |
|     if (a++ < 100) {
 | |
|       rs.push(".");
 | |
|     } else {
 | |
|       ended = true;
 | |
|       rs.push(null);
 | |
|     }
 | |
|   };
 | |
| 
 | |
|   var sizeRead = 0;
 | |
|   var ws = new Writable({highWaterMark: 2});
 | |
|   ws._write = function(chunk, enc, next) {
 | |
|     sizeRead += chunk.length;
 | |
|     next();
 | |
|   }
 | |
| 
 | |
|   ws.on("finish", function() {
 | |
|     assert.ok(a > 99, 'a > 99');
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.equal(sizeRead, 100, 'sizeRead === 100');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   rs.pipe(ws);
 | |
| 
 | |
|   exhaust(rs);
 | |
| });
 | |
| 
 | |
| test('should not interfere with a Writable stream', function(assert) {
 | |
|   var rs = new Readable({highWaterMark: 2});
 | |
|   var a = 0;
 | |
|   var ended = false;
 | |
|   rs._read = function() {
 | |
|     if (a++ < 100) {
 | |
|       rs.push(".");
 | |
|     } else {
 | |
|       ended = true;
 | |
|       rs.push(null);
 | |
|     }
 | |
|   };
 | |
| 
 | |
|   var sizeRead = 0;
 | |
|   var ws = new Writable({highWaterMark: 2});
 | |
|   ws._write = function(chunk, enc, next) {
 | |
|     sizeRead += chunk.length;
 | |
|     next();
 | |
|   }
 | |
| 
 | |
|   ws.on("finish", function() {
 | |
|     assert.ok(a > 99, 'a > 99');
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.equal(sizeRead, 100, 'sizeRead === 100');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   rs.pipe(ws);
 | |
| 
 | |
|   exhaust(ws);
 | |
| });
 | |
| 
 | |
| test('should handle a Transform stream', function(assert) {
 | |
|   var rs = new Readable({highWaterMark: 2});
 | |
|   var a = 0;
 | |
|   var ended = false;
 | |
|   rs._read = function() {
 | |
|     if (a++ < 100) {
 | |
|       rs.push(".");
 | |
|     } else {
 | |
|       ended = true;
 | |
|       rs.push(null);
 | |
|     }
 | |
|   };
 | |
| 
 | |
|   var sizeRead = 0;
 | |
|   var flushed = false;
 | |
|   var ts = through({highWaterMark: 2}, function(chunk, enc, cb) {
 | |
|     sizeRead += chunk.length;
 | |
|     this.push(chunk);
 | |
|     cb();
 | |
|   }, function(cb) {
 | |
|     flushed = true;
 | |
|     cb();
 | |
|   });
 | |
| 
 | |
|   ts.on("end", function() {
 | |
|     assert.ok(a > 99, 'a > 99');
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.equal(sizeRead, 100, 'sizeRead === 100');
 | |
|     assert.ok(flushed, 'flushed is true');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   rs.pipe(ts);
 | |
| 
 | |
|   exhaust(ts);
 | |
| });
 | |
| 
 | |
| test('should handle a classic stream', function(assert) {
 | |
|   var rs = new Stream();
 | |
|   var ended = false;
 | |
|   var i;
 | |
| 
 | |
|   rs.on("end", function() {
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   exhaust(rs);
 | |
| 
 | |
|   for (i = 0; i < 100; i++) {
 | |
|     rs.emit("data", i);
 | |
|   }
 | |
|   ended = true;
 | |
|   rs.emit("end");
 | |
| });
 | |
| 
 | |
| test('should not modify .pipe', function(assert) {
 | |
|   var stream = new S2Readable;
 | |
| 
 | |
|   var pipe = stream.pipe;
 | |
| 
 | |
|   stream._read = function() {
 | |
|     stream.push('ending');
 | |
|     stream.push(null);
 | |
|   };
 | |
| 
 | |
|   exhaust(stream);
 | |
| 
 | |
|   assert.equal(stream.pipe, pipe);
 | |
|   assert.end();
 | |
| });
 | |
| 
 | |
| test('does not error on no resume but readable set to true', function(assert) {
 | |
|   var rs = new Stream();
 | |
|   rs.readable = true;
 | |
| 
 | |
|   var ended = false;
 | |
|   var i;
 | |
| 
 | |
|   rs.on("end", function() {
 | |
|     assert.ok(ended, 'ended is true');
 | |
|     assert.end();
 | |
|   });
 | |
| 
 | |
|   exhaust(rs);
 | |
| 
 | |
|   for (i = 0; i < 100; i++) {
 | |
|     rs.emit("data", i);
 | |
|   }
 | |
|   ended = true;
 | |
|   rs.emit("end");
 | |
| });
 |