1berry/nodejs
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fs: streams2

Browse Source
This commit is contained in:
isaacs 2012-10-04 17:44:48 -07:00
parent d58f2654bc
commit 44b308b1f7
4 changed files with 197 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

425
lib/fs.js
View File

@ -34,6 +34,9 @@ var fs = exports;
var Stream = require('stream').Stream; var Stream = require('stream').Stream;
var EventEmitter = require('events').EventEmitter; var EventEmitter = require('events').EventEmitter;
var Readable = Stream.Readable;
var Writable = Stream.Writable;
var kMinPoolSpace = 128; var kMinPoolSpace = 128;
var kPoolSize = 40 * 1024; var kPoolSize = 40 * 1024;
@ -1386,34 +1389,30 @@ fs.createReadStream = function(path, options) {
return new ReadStream(path, options); return new ReadStream(path, options);
}; };
var ReadStream = fs.ReadStream = function(path, options) { util.inherits(ReadStream, Readable);
if (!(this instanceof ReadStream)) return new ReadStream(path, options); fs.ReadStream = ReadStream;
Stream.call(this); function ReadStream(path, options) {
if (!(this instanceof ReadStream))
return new ReadStream(path, options);
var self = this; // a little bit bigger buffer and water marks by default
options = util._extend({
bufferSize: 64 * 1024,
lowWaterMark: 16 * 1024,
highWaterMark: 64 * 1024
}, options || {});
Readable.call(this, options);
this.path = path; this.path = path;
this.fd = null; this.fd = options.hasOwnProperty('fd') ? options.fd : null;
this.readable = true; this.flags = options.hasOwnProperty('flags') ? options.flags : 'r';
this.paused = false; this.mode = options.hasOwnProperty('mode') ? options.mode : 438; /*=0666*/
this.flags = 'r'; this.start = options.hasOwnProperty('start') ? options.start : undefined;
this.mode = 438; /*=0666*/ this.end = options.hasOwnProperty('start') ? options.end : undefined;
this.bufferSize = 64 * 1024; this.pos = undefined;
options = options || {};
// Mixin options into this
var keys = Object.keys(options);
for (var index = 0, length = keys.length; index < length; index++) {
var key = keys[index];
this[key] = options[key];
}
assertEncoding(this.encoding);
if (this.encoding) this.setEncoding(this.encoding);
if (this.start !== undefined) { if (this.start !== undefined) {
if ('number' !== typeof this.start) { if ('number' !== typeof this.start) {
@ -1432,41 +1431,40 @@ var ReadStream = fs.ReadStream = function(path, options) {
this.pos = this.start; this.pos = this.start;
} }
if (this.fd !== null) { if (typeof this.fd !== 'number')
process.nextTick(function() { this.open();
self._read();
this.on('end', function() {
this.destroy();
}); });
return;
} }
fs.open(this.path, this.flags, this.mode, function(err, fd) { fs.FileReadStream = fs.ReadStream; // support the legacy name
if (err) {
self.emit('error', err); ReadStream.prototype.open = function() {
self.readable = false; var self = this;
fs.open(this.path, this.flags, this.mode, function(er, fd) {
if (er) {
self.destroy();
self.emit('error', er);
return; return;
} }
self.fd = fd; self.fd = fd;
self.emit('open', fd); self.emit('open', fd);
self._read(); // start the flow of data.
self.read();
}); });
}; };
util.inherits(ReadStream, Stream);
fs.FileReadStream = fs.ReadStream; // support the legacy name ReadStream.prototype._read = function(n, cb) {
if (typeof this.fd !== 'number')
return this.once('open', function() {
this._read(n, cb);
});
ReadStream.prototype.setEncoding = function(encoding) { if (this.destroyed)
assertEncoding(encoding); return;
var StringDecoder = require('string_decoder').StringDecoder; // lazy load
this._decoder = new StringDecoder(encoding);
};
ReadStream.prototype._read = function() {
var self = this;
if (!this.readable || this.paused || this.reading) return;
this.reading = true;
if (!pool || pool.length - pool.used < kMinPoolSpace) { if (!pool || pool.length - pool.used < kMinPoolSpace) {
// discard the old pool. Can't add to the free list because // discard the old pool. Can't add to the free list because
@ -1475,150 +1473,111 @@ ReadStream.prototype._read = function() {
allocNewPool(); allocNewPool();
} }
// Grab another reference to the pool in the case that while we're in the // Grab another reference to the pool in the case that while we're
// thread pool another read() finishes up the pool, and allocates a new // in the thread pool another read() finishes up the pool, and
// one. // allocates a new one.
var thisPool = pool; var thisPool = pool;
var toRead = Math.min(pool.length - pool.used, ~~this.bufferSize); var toRead = Math.min(pool.length - pool.used, n);
var start = pool.used; var start = pool.used;
if (this.pos !== undefined) { if (this.pos !== undefined)
toRead = Math.min(this.end - this.pos + 1, toRead); toRead = Math.min(this.end - this.pos + 1, toRead);
}
function afterRead(err, bytesRead) { // already read everything we were supposed to read!
self.reading = false; // treat as EOF.
if (err) { if (toRead <= 0)
fs.close(self.fd, function() { return cb();
self.fd = null;
self.emit('error', err);
self.readable = false;
});
return;
}
if (bytesRead === 0) { // the actual read.
if (this._decoder) { var self = this;
var ret = this._decoder.end(); fs.read(this.fd, pool, pool.used, toRead, this.pos, onread);
if (ret)
this.emit('data', ret);
}
self.emit('end');
self.destroy();
return;
}
var b = thisPool.slice(start, start + bytesRead); // move the pool positions, and internal position for reading.
if (this.pos !== undefined)
// Possible optimizition here?
// Reclaim some bytes if bytesRead < toRead?
// Would need to ensure that pool === thisPool.
// do not emit events if the stream is paused
if (self.paused) {
self.buffer = b;
return;
}
// do not emit events anymore after we declared the stream unreadable
if (!self.readable) return;
self._emitData(b);
self._read();
}
fs.read(this.fd, pool, pool.used, toRead, this.pos, afterRead);
if (this.pos !== undefined) {
this.pos += toRead; this.pos += toRead;
}
pool.used += toRead; pool.used += toRead;
};
function onread(er, bytesRead) {
if (er) {
self.destroy();
return cb(er);
}
ReadStream.prototype._emitData = function(d) { var b = null;
if (this._decoder) { if (bytesRead > 0)
var string = this._decoder.write(d); b = thisPool.slice(start, start + bytesRead);
if (string.length) this.emit('data', string);
} else { cb(null, b);
this.emit('data', d);
} }
}; };
ReadStream.prototype.destroy = function() { ReadStream.prototype.destroy = function() {
var self = this; if (this.destroyed)
return;
this.destroyed = true;
if (!this.readable) return; if ('number' === typeof this.fd)
this.readable = false; this.close();
};
ReadStream.prototype.close = function(cb) {
if (cb)
this.once('close', cb);
if (this.closed || 'number' !== typeof this.fd) {
if ('number' !== typeof this.fd)
this.once('open', close);
return process.nextTick(this.emit.bind(this, 'close'));
}
this.closed = true;
var self = this;
close();
function close() { function close() {
fs.close(self.fd, function(err) { fs.close(self.fd, function(er) {
if (err) { if (er)
self.emit('error', err); self.emit('error', er);
} else { else
self.emit('close'); self.emit('close');
}
}); });
} self.fd = null;
if (this.fd === null) {
this.addListener('open', close);
} else {
close();
} }
}; };
ReadStream.prototype.pause = function() {
this.paused = true;
};
ReadStream.prototype.resume = function() {
this.paused = false;
if (this.buffer) {
var buffer = this.buffer;
this.buffer = null;
this._emitData(buffer);
}
// hasn't opened yet.
if (null == this.fd) return;
this._read();
};
fs.createWriteStream = function(path, options) { fs.createWriteStream = function(path, options) {
return new WriteStream(path, options); return new WriteStream(path, options);
}; };
var WriteStream = fs.WriteStream = function(path, options) { util.inherits(WriteStream, Writable);
if (!(this instanceof WriteStream)) return new WriteStream(path, options); fs.WriteStream = WriteStream;
function WriteStream(path, options) {
if (!(this instanceof WriteStream))
return new WriteStream(path, options);
Stream.call(this); // a little bit bigger buffer and water marks by default
options = util._extend({
bufferSize: 64 * 1024,
lowWaterMark: 16 * 1024,
highWaterMark: 64 * 1024
}, options || {});
Writable.call(this, options);
this.path = path; this.path = path;
this.fd = null; this.fd = null;
this.writable = true;
this.flags = 'w'; this.fd = options.hasOwnProperty('fd') ? options.fd : null;
this.encoding = 'binary'; this.flags = options.hasOwnProperty('flags') ? options.flags : 'w';
this.mode = 438; /*=0666*/ this.mode = options.hasOwnProperty('mode') ? options.mode : 438; /*=0666*/
this.start = options.hasOwnProperty('start') ? options.start : undefined;
this.pos = undefined;
this.bytesWritten = 0; this.bytesWritten = 0;
options = options || {};
// Mixin options into this
var keys = Object.keys(options);
for (var index = 0, length = keys.length; index < length; index++) {
var key = keys[index];
this[key] = options[key];
}
if (this.start !== undefined) { if (this.start !== undefined) {
if ('number' !== typeof this.start) { if ('number' !== typeof this.start) {
throw TypeError('start must be a Number'); throw TypeError('start must be a Number');
@ -1630,154 +1589,54 @@ var WriteStream = fs.WriteStream = function(path, options) {
this.pos = this.start; this.pos = this.start;
} }
this.busy = false; if ('number' !== typeof this.fd)
this._queue = []; this.open();
if (this.fd === null) { // dispose on finish.
this._open = fs.open; this.once('finish', this.close);
this._queue.push([this._open, this.path, this.flags, this.mode, undefined]);
this.flush();
} }
};
util.inherits(WriteStream, Stream);
fs.FileWriteStream = fs.WriteStream; // support the legacy name fs.FileWriteStream = fs.WriteStream; // support the legacy name
WriteStream.prototype.flush = function() {
if (this.busy) return;
var self = this;
var args = this._queue.shift(); WriteStream.prototype.open = function() {
if (!args) { fs.open(this.path, this.flags, this.mode, function(er, fd) {
if (this.drainable) { this.emit('drain'); } if (er) {
this.destroy();
this.emit('error', er);
return; return;
} }
this.busy = true; this.fd = fd;
this.emit('open', fd);
var method = args.shift(), }.bind(this));
cb = args.pop();
args.push(function(err) {
self.busy = false;
if (err) {
self.writable = false;
function emit() {
self.fd = null;
if (cb) cb(err);
self.emit('error', err);
}
if (self.fd === null) {
emit();
} else {
fs.close(self.fd, emit);
}
return;
}
if (method == fs.write) {
self.bytesWritten += arguments[1];
if (cb) {
// write callback
cb(null, arguments[1]);
}
} else if (method === self._open) {
// save reference for file pointer
self.fd = arguments[1];
self.emit('open', self.fd);
} else if (method === fs.close) {
// stop flushing after close
if (cb) {
cb(null);
}
self.emit('close');
return;
}
self.flush();
});
// Inject the file pointer
if (method !== self._open) {
args.unshift(this.fd);
}
method.apply(this, args);
}; };
WriteStream.prototype.write = function(data) {
if (!this.writable) { WriteStream.prototype._write = function(data, cb) {
this.emit('error', new Error('stream not writable')); if (!Buffer.isBuffer(data))
return false; return this.emit('error', new Error('Invalid data'));
if (typeof this.fd !== 'number')
return this.once('open', this._write.bind(this, data, cb));
var self = this;
fs.write(this.fd, data, 0, data.length, this.pos, function(er, bytes) {
if (er) {
self.destroy();
return cb(er);
} }
self.bytesWritten += bytes;
cb();
});
this.drainable = true; if (this.pos !== undefined)
var cb;
if (typeof(arguments[arguments.length - 1]) == 'function') {
cb = arguments[arguments.length - 1];
}
if (!Buffer.isBuffer(data)) {
var encoding = 'utf8';
if (typeof(arguments[1]) == 'string') encoding = arguments[1];
assertEncoding(encoding);
data = new Buffer('' + data, encoding);
}
this._queue.push([fs.write, data, 0, data.length, this.pos, cb]);
if (this.pos !== undefined) {
this.pos += data.length; this.pos += data.length;
}
this.flush();
return false;
}; };
WriteStream.prototype.end = function(data, encoding, cb) {
if (typeof(data) === 'function') {
cb = data;
} else if (typeof(encoding) === 'function') {
cb = encoding;
this.write(data);
} else if (arguments.length > 0) {
this.write(data, encoding);
}
this.writable = false;
this._queue.push([fs.close, cb]);
this.flush();
};
WriteStream.prototype.destroy = function() { WriteStream.prototype.destroy = ReadStream.prototype.destroy;
var self = this; WriteStream.prototype.close = ReadStream.prototype.close;
if (!this.writable) return;
this.writable = false;
function close() {
fs.close(self.fd, function(err) {
if (err) {
self.emit('error', err);
} else {
self.emit('close');
}
});
}
if (this.fd === null) {
this.addListener('open', close);
} else {
close();
}
};
// There is no shutdown() for files. // There is no shutdown() for files.
WriteStream.prototype.destroySoon = WriteStream.prototype.end; WriteStream.prototype.destroySoon = WriteStream.prototype.end;

35
test/simple/test-file-write-stream.js
View File

@ -22,46 +22,50 @@
var common = require('../common'); var common = require('../common');
var assert = require('assert'); var assert = require('assert');
var path = require('path'), var path = require('path');
fs = require('fs'), var fs = require('fs');
fn = path.join(common.tmpDir, 'write.txt'), var fn = path.join(common.tmpDir, 'write.txt');
file = fs.createWriteStream(fn), var file = fs.createWriteStream(fn, {
lowWaterMark: 3,
highWaterMark: 10
});
EXPECTED = '012345678910', var EXPECTED = '012345678910';
callbacks = { var callbacks = {
open: -1, open: -1,
drain: -2, drain: -2,
close: -1, close: -1
endCb: -1
}; };
file file
.on('open', function(fd) { .on('open', function(fd) {
console.error('open!');
callbacks.open++; callbacks.open++;
assert.equal('number', typeof fd); assert.equal('number', typeof fd);
}) })
.on('error', function(err) { .on('error', function(err) {
throw err; throw err;
console.error('error!', err.stack);
}) })
.on('drain', function() { .on('drain', function() {
console.error('drain!', callbacks.drain);
callbacks.drain++; callbacks.drain++;
if (callbacks.drain == -1) { if (callbacks.drain == -1) {
assert.equal(EXPECTED, fs.readFileSync(fn)); assert.equal(EXPECTED, fs.readFileSync(fn, 'utf8'));
file.write(EXPECTED); file.write(EXPECTED);
} else if (callbacks.drain == 0) { } else if (callbacks.drain == 0) {
assert.equal(EXPECTED + EXPECTED, fs.readFileSync(fn)); assert.equal(EXPECTED + EXPECTED, fs.readFileSync(fn, 'utf8'));
file.end(function(err) { file.end();
assert.ok(!err);
callbacks.endCb++;
});
} }
}) })
.on('close', function() { .on('close', function() {
console.error('close!');
assert.strictEqual(file.bytesWritten, EXPECTED.length * 2); assert.strictEqual(file.bytesWritten, EXPECTED.length * 2);
callbacks.close++; callbacks.close++;
assert.throws(function() { assert.throws(function() {
console.error('write after end should not be allowed');
file.write('should not work anymore'); file.write('should not work anymore');
}); });
@ -70,7 +74,7 @@ file
for (var i = 0; i < 11; i++) { for (var i = 0; i < 11; i++) {
(function(i) { (function(i) {
assert.strictEqual(false, file.write(i)); file.write('' + i);
})(i); })(i);
} }
@ -78,4 +82,5 @@ process.on('exit', function() {
for (var k in callbacks) { for (var k in callbacks) {
assert.equal(0, callbacks[k], k + ' count off by ' + callbacks[k]); assert.equal(0, callbacks[k], k + ' count off by ' + callbacks[k]);
} }
console.log('ok');
}); });

40
test/simple/test-file-write-stream2.js
View File

@ -22,18 +22,18 @@
var common = require('../common'); var common = require('../common');
var assert = require('assert'); var assert = require('assert');
var path = require('path'), var path = require('path');
fs = require('fs'), var fs = require('fs');
util = require('util'); var util = require('util');
var filepath = path.join(common.tmpDir, 'write.txt'), var filepath = path.join(common.tmpDir, 'write.txt');
file; var file;
var EXPECTED = '012345678910'; var EXPECTED = '012345678910';
var cb_expected = 'write open drain write drain close error ', var cb_expected = 'write open drain write drain close error ';
cb_occurred = ''; var cb_occurred = '';
var countDrains = 0; var countDrains = 0;
@ -47,6 +47,8 @@ process.on('exit', function() {
assert.strictEqual(cb_occurred, cb_expected, assert.strictEqual(cb_occurred, cb_expected,
'events missing or out of order: "' + 'events missing or out of order: "' +
cb_occurred + '" !== "' + cb_expected + '"'); cb_occurred + '" !== "' + cb_expected + '"');
} else {
console.log('ok');
} }
}); });
@ -59,22 +61,30 @@ function removeTestFile() {
removeTestFile(); removeTestFile();
file = fs.createWriteStream(filepath); // drain at 0, return false at 10.
file = fs.createWriteStream(filepath, {
lowWaterMark: 0,
highWaterMark: 11
});
file.on('open', function(fd) { file.on('open', function(fd) {
console.error('open');
cb_occurred += 'open '; cb_occurred += 'open ';
assert.equal(typeof fd, 'number'); assert.equal(typeof fd, 'number');
}); });
file.on('drain', function() { file.on('drain', function() {
console.error('drain');
cb_occurred += 'drain '; cb_occurred += 'drain ';
++countDrains; ++countDrains;
if (countDrains === 1) { if (countDrains === 1) {
assert.equal(fs.readFileSync(filepath), EXPECTED); console.error('drain=1, write again');
file.write(EXPECTED); assert.equal(fs.readFileSync(filepath, 'utf8'), EXPECTED);
console.error('ondrain write ret=%j', file.write(EXPECTED));
cb_occurred += 'write '; cb_occurred += 'write ';
} else if (countDrains == 2) { } else if (countDrains == 2) {
assert.equal(fs.readFileSync(filepath), EXPECTED + EXPECTED); console.error('second drain, end');
assert.equal(fs.readFileSync(filepath, 'utf8'), EXPECTED + EXPECTED);
file.end(); file.end();
} }
}); });
@ -88,11 +98,15 @@ file.on('close', function() {
file.on('error', function(err) { file.on('error', function(err) {
cb_occurred += 'error '; cb_occurred += 'error ';
assert.ok(err.message.indexOf('not writable') >= 0); assert.ok(err.message.indexOf('write after end') >= 0);
}); });
for (var i = 0; i < 11; i++) { for (var i = 0; i < 11; i++) {
assert.strictEqual(file.write(i), false); var ret = file.write(i + '');
console.error('%d %j', i, ret);
// return false when i hits 10
assert(ret === (i != 10));
} }
cb_occurred += 'write '; cb_occurred += 'write ';

4
test/simple/test-fs-read-stream.js
View File

@ -60,12 +60,10 @@ file.on('data', function(data) {
paused = true; paused = true;
file.pause(); file.pause();
assert.ok(file.paused);
setTimeout(function() { setTimeout(function() {
paused = false; paused = false;
file.resume(); file.resume();
assert.ok(!file.paused);
}, 10); }, 10);
}); });
@ -77,7 +75,6 @@ file.on('end', function(chunk) {
file.on('close', function() { file.on('close', function() {
callbacks.close++; callbacks.close++;
assert.ok(!file.readable);
//assert.equal(fs.readFileSync(fn), fileContent); //assert.equal(fs.readFileSync(fn), fileContent);
}); });
@ -104,6 +101,7 @@ process.on('exit', function() {
assert.equal(2, callbacks.close); assert.equal(2, callbacks.close);
assert.equal(30000, file.length); assert.equal(30000, file.length);
assert.equal(10000, file3.length); assert.equal(10000, file3.length);
console.error('ok');
}); });
var file4 = fs.createReadStream(rangeFile, {bufferSize: 1, start: 1, end: 2}); var file4 = fs.createReadStream(rangeFile, {bufferSize: 1, start: 1, end: 2});