Skip to content

Commit cf4777b

Browse files
committed
stream: add Readable.readv
A faster alternative to Readable.read for certain use cases. e.g. fs.writevSync(fd, readable.readv())
1 parent 6176222 commit cf4777b

File tree

1 file changed

+93
-3
lines changed

1 file changed

+93
-3
lines changed

lib/internal/streams/readable.js

Lines changed: 93 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -641,8 +641,16 @@ function howMuchToRead(n, state) {
641641
return (state[kState] & kEnded) !== 0 ? state.length : 0;
642642
}
643643

644+
Readable.prototype.readv = function readv () {
645+
return _read.call(this, true);
646+
};
647+
648+
Readable.prototype.read = function read () {
649+
return _read.call(this, false);
650+
}
651+
644652
// You can override either this method, or the async _read(n) below.
645-
Readable.prototype.read = function(n) {
653+
function _read (n, returnArr) {
646654
debug('read', n);
647655
// Same as parseInt(undefined, 10), however V8 7.3 performance regressed
648656
// in this scenario, so we are doing it manually.
@@ -748,7 +756,7 @@ Readable.prototype.read = function(n) {
748756

749757
let ret;
750758
if (n > 0)
751-
ret = fromList(n, state);
759+
ret = returnArr ? arrFromList(n, state) : fromList(n, state);
752760
else
753761
ret = null;
754762

@@ -777,7 +785,13 @@ Readable.prototype.read = function(n) {
777785

778786
if (ret !== null && (state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) {
779787
state[kState] |= kDataEmitted;
780-
this.emit('data', ret);
788+
if (returnArr) {
789+
for (let i = 0; i < ret.length; ++i) {
790+
this.emit('data', ret[i]);
791+
}
792+
} else {
793+
this.emit('data', ret);
794+
}
781795
}
782796

783797
return ret;
@@ -1682,6 +1696,82 @@ function fromList(n, state) {
16821696
return ret;
16831697
}
16841698

1699+
function arrFromList(n, state) {
1700+
// nothing buffered.
1701+
if (state.length === 0)
1702+
return null;
1703+
1704+
let idx = state.bufferIndex;
1705+
let ret;
1706+
1707+
const buf = state.buffer;
1708+
const len = buf.length;
1709+
1710+
if ((state[kState] & kObjectMode) !== 0 || !n || n >= state.length) {
1711+
ret = buf.slice(idx);
1712+
idx += ret.length;
1713+
} else if (n < buf[idx].length) {
1714+
// `slice` is the same for buffers and strings.
1715+
ret = [buf[idx].slice(0, n)];
1716+
buf[idx] = buf[idx].slice(n);
1717+
} else if (n === buf[idx].length) {
1718+
// First chunk is a perfect match.
1719+
ret = [buf[idx]];
1720+
buf[idx++] = null;
1721+
} else if ((state[kState] & kDecoder) !== 0) {
1722+
ret = [];
1723+
while (idx < len) {
1724+
const str = buf[idx];
1725+
if (n > str.length) {
1726+
ret.push(str);
1727+
n -= str.length;
1728+
buf[idx++] = null;
1729+
} else {
1730+
if (n === buf.length) {
1731+
ret.push(str);
1732+
buf[idx++] = null;
1733+
} else {
1734+
ret.push(str.slice(0, n));
1735+
buf[idx] = str.slice(n);
1736+
}
1737+
break;
1738+
}
1739+
}
1740+
} else {
1741+
ret = [];
1742+
const retLen = n;
1743+
while (idx < len) {
1744+
const data = buf[idx];
1745+
if (n > data.length) {
1746+
ret.push(data);
1747+
n -= data.length;
1748+
buf[idx++] = null;
1749+
} else {
1750+
if (n === data.length) {
1751+
ret.push(data);
1752+
buf[idx++] = null;
1753+
} else {
1754+
ret.push(new FastBuffer(data.buffer, data.byteOffset, n));
1755+
buf[idx] = new FastBuffer(data.buffer, data.byteOffset + n, data.length - n);
1756+
}
1757+
break;
1758+
}
1759+
}
1760+
}
1761+
1762+
if (idx === len) {
1763+
state.buffer.length = 0;
1764+
state.bufferIndex = 0;
1765+
} else if (idx > 1024) {
1766+
state.buffer.splice(0, idx);
1767+
state.bufferIndex = 0;
1768+
} else {
1769+
state.bufferIndex = idx;
1770+
}
1771+
1772+
return ret;
1773+
}
1774+
16851775
function endReadable(stream) {
16861776
const state = stream._readableState;
16871777

0 commit comments

Comments
 (0)