VladPolskiy/mariadb-connector-nodejs
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-connector-nodejs.git synced 2025-08-19 01:23:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

small code correction

Browse Source
This commit is contained in:
rusher
2018-04-13 17:04:11 +02:00
parent 859e0827eb
commit c9b45e91ae
11 changed files with 609 additions and 552 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.travis/wait-for-docker-up.js
View File

@ -6,7 +6,7 @@ const Conf = require('../test/conf');
const connOptions = new ConnOptions(Conf.baseConfig);
let decrement = 20;
var callback = () => console.log("docker db server up");
const callback = () => console.log("docker db server up");
const checkConnection = function() {
decrement-=1;

2
benchmarks/benchs/bench_select_one_user_random.js
View File

@ -5,7 +5,7 @@ module.exports.displaySql =
"select <all mysql.user fields>, <random field> from mysql.user u LIMIT 1";
module.exports.benchFct = function(conn, deferred) {
var rand = Math.floor(Math.random() * 50000000);
const rand = Math.floor(Math.random() * 50000000);
conn.query(
"select u.Host,\n" +
"u.User,\n" +

2
benchmarks/common_benchmarks.js
View File

@ -35,7 +35,7 @@ function Bench(callback) {
console.log(config);
this.CONN = {};
var bench = this;
const bench = this;
this.CONN["MYSQL"] = { drv: mysql.createConnection(config), desc: "mysql" };
this.CONN.MYSQL.drv.connect(() => ready("mysql"));
this.CONN.MYSQL.drv.on("error", err => console.log("driver mysql error :" + err));

10
documentation/readme.md
View File

@ -44,16 +44,16 @@ This permit to avoid TCP-IP layer.
If not on localhost, then hostname must be set, port is optional with default 3306, connector will then use TCP/IP socket.
```javascript
var mariadb = require('mariadb');
const mariadb = require('mariadb');
//localhost on windows
var conn1 = mariadb.createConnection({socketPath: '\\\\.\\pipe\\MySQL'});
const conn1 = mariadb.createConnection({socketPath: '\\\\.\\pipe\\MySQL'});
//localhost on unix
var conn2 = mariadb.createConnection({socketPath: '/tmp/mysql.sock'});
const conn2 = mariadb.createConnection({socketPath: '/tmp/mysql.sock'});
//not localhost
var conn3 = mariadb.createConnection({host: 'mydb.com', port:9999});
const conn3 = mariadb.createConnection({host: 'mydb.com', port:9999});
```
### Connection options
@ -67,7 +67,7 @@ var conn3 = mariadb.createConnection({host: 'mydb.com', port:9999});
* `database`: string. default database when establishing connection.
* `password`: string. user password
* `socketPath`: string. Permits connecting to the database via Unix domain socket or named pipe, if the server allows it.
* `compress`: boolean. exchanges with database must be gzip. (=> when database is not localhost). default: false
* `compress`: boolean. The exchanges with database will be gzipped. That permit better performance when database is distant (not in same location). default: false
* `connectTimeout`: integer. connection timeout in ms. default: 10 000.
Support for big integer:

1
src/cmd/query.js
View File

@ -498,6 +498,7 @@ class Query extends ResultSet {
* @param sql sql with placeholders
* @param info connection information
* @param initialValues placeholder object
* @param displaySql display sql function
* @returns {{parts: Array, values: Array}}
*/
static splitQueryPlaceholder(sql, info, initialValues, displaySql) {

4
src/connection.js
View File

@ -194,7 +194,7 @@ class Connection {
let sock = this._socket;
this._clear();
if (callback) setImmediate(callback);
sock.destroy;
sock.destroy();
}.bind(this)
);
this._sendQueue.push(cmd);
@ -231,7 +231,7 @@ class Connection {
}
}
process.nextTick(() => {
if (self._socket) self._socket.destroy;
if (self._socket) self._socket.destroy();
});
killCon.end();
});

292
src/io/compression-output-stream.js
View File

@ -12,158 +12,192 @@ const MAX_BUFFER_SIZE = 16777222; //16M + 7
/**
/**
* MySQL compression filter.
*
* @param socket current socket
* @constructor
* see https://mariadb.com/kb/en/library/0-packet/#compressed-packet
*/
function CompressionOutputStream(socket, info, opts) {
this.info = info;
this.opts = opts;
this.pos = 7;
this.header = Buffer.allocUnsafe(7);
this.smallBuffer = Buffer.allocUnsafe(SMALL_BUFFER_SIZE);
this.buf = this.smallBuffer;
this.writer = buffer => socket.write(buffer);
this.gzip = ZLib.createGzip();
}
class CompressionOutputStream {
/**
* Constructor
*
* @param socket current socket
* @param info current connection informations
* @param opts current connection options
* @constructor
*/
constructor(socket, info, opts) {
this.info = info;
this.opts = opts;
this.pos = 7;
this.header = Buffer.allocUnsafe(7);
this.smallBuffer = Buffer.allocUnsafe(SMALL_BUFFER_SIZE);
this.buf = this.smallBuffer;
this.writer = buffer => socket.write(buffer);
}
CompressionOutputStream.prototype.growBuffer = function(len) {
let newCapacity;
if (len + this.pos < MEDIUM_BUFFER_SIZE) {
newCapacity = MEDIUM_BUFFER_SIZE;
} else if (len + this.pos < LARGE_BUFFER_SIZE) {
newCapacity = LARGE_BUFFER_SIZE;
} else newCapacity = MAX_BUFFER_SIZE;
growBuffer(len) {
let newCapacity;
if (len + this.pos < MEDIUM_BUFFER_SIZE) {
newCapacity = MEDIUM_BUFFER_SIZE;
} else if (len + this.pos < LARGE_BUFFER_SIZE) {
newCapacity = LARGE_BUFFER_SIZE;
} else newCapacity = MAX_BUFFER_SIZE;
let newBuf = Buffer.allocUnsafe(newCapacity);
this.buf.copy(newBuf, 0, 0, this.pos);
this.buf = newBuf;
};
let newBuf = Buffer.allocUnsafe(newCapacity);
this.buf.copy(newBuf, 0, 0, this.pos);
this.buf = newBuf;
}
CompressionOutputStream.prototype.writeBuf = function(arr, cmd) {
let off = 0,
len = arr.length;
if (len > this.buf.length - this.pos) {
if (this.buf.length !== MAX_BUFFER_SIZE) {
this.growBuffer(len);
}
//max buffer size
writeBuf(arr, cmd) {
let off = 0,
len = arr.length;
if (len > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
// fill buffer and flush until all data are snd
let remainingLen = len;
if (this.buf.length !== MAX_BUFFER_SIZE) {
this.growBuffer(len);
}
while (true) {
//filling buffer
let lenToFillBuffer = Math.min(MAX_BUFFER_SIZE - this.pos, remainingLen);
arr.copy(this.buf, this.pos, off, off + lenToFillBuffer);
remainingLen -= lenToFillBuffer;
off += lenToFillBuffer;
this.pos += lenToFillBuffer;
//max buffer size
if (len > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
// fill buffer and flush until all data are snd
let remainingLen = len;
if (remainingLen === 0) return;
this.flush(false, cmd);
while (true) {
//filling buffer
let lenToFillBuffer = Math.min(MAX_BUFFER_SIZE - this.pos, remainingLen);
arr.copy(this.buf, this.pos, off, off + lenToFillBuffer);
remainingLen -= lenToFillBuffer;
off += lenToFillBuffer;
this.pos += lenToFillBuffer;
if (remainingLen === 0) return;
this.flush(false, cmd);
}
}
}
arr.copy(this.buf, this.pos, off, off + len);
this.pos += len;
}
arr.copy(this.buf, this.pos, off, off + len);
this.pos += len;
};
/**
* Flush the internal buffer.
*/
CompressionOutputStream.prototype.flush = function(cmdEnd, cmd) {
if (this.pos < 1536) {
//*******************************************************************************
// small packet, no compression
//*******************************************************************************
/**
* Flush the internal buffer.
*/
flush(cmdEnd, cmd) {
if (this.pos < 1536) {
//*******************************************************************************
// small packet, no compression
//*******************************************************************************
this.buf[0] = this.pos - 7;
this.buf[1] = (this.pos - 7) >>> 8;
this.buf[2] = (this.pos - 7) >>> 16;
this.buf[3] = cmd.compressSequenceNo;
this.buf[4] = 0;
this.buf[5] = 0;
this.buf[6] = 0;
cmd.incrementCompressSequenceNo(1);
this.buf[0] = this.pos - 7;
this.buf[1] = (this.pos - 7) >>> 8;
this.buf[2] = (this.pos - 7) >>> 16;
this.buf[3] = cmd.compressSequenceNo;
this.buf[4] = 0;
this.buf[5] = 0;
this.buf[6] = 0;
cmd.incrementCompressSequenceNo(1);
if (this.opts.debugCompress) {
console.log(
"==> conn:%d %s (compress)\n%s",
this.info.threadId ? this.info.threadId : -1,
cmd
? (cmd.onPacketReceive
? cmd.constructor.name + "." + cmd.onPacketReceive.name
: cmd.constructor.name) +
"(0," +
this.pos +
")"
: "unknown",
Utils.log(this.buf, 0, this.pos)
);
}
if (this.opts.debugCompress) {
console.log(
"==> conn:%d %s (compress)\n%s",
this.info.threadId ? this.info.threadId : -1,
cmd
? (cmd.onPacketReceive
? cmd.constructor.name + "." + cmd.onPacketReceive.name
: cmd.constructor.name) +
"(0," +
this.pos +
")"
: "unknown",
Utils.log(this.buf, 0, this.pos)
);
}
try {
this.writer(this.buf.slice(0, this.pos));
try {
this.writer(this.buf.slice(0, this.pos));
if (this.pos === MAX_BUFFER_SIZE) this.writeEmptyPacket();
//reset buffer
this.buf = this.smallBuffer;
this.pos = 7;
} catch (err) {
//eat exception : thrown by socket.on('error');
}
} else {
//*******************************************************************************
// compressing packet
//*******************************************************************************
//use synchronous inflating, to ensure FIFO packet order
const compressChunk = ZLib.deflateSync(this.buf.slice(7, this.pos));
const compressChunkLen = compressChunk.length;
this.header[0] = compressChunkLen;
this.header[1] = compressChunkLen >>> 8;
this.header[2] = compressChunkLen >>> 16;
this.header[3] = cmd.compressSequenceNo;
this.header[4] = this.pos - 7;
this.header[5] = (this.pos - 7) >>> 8;
this.header[6] = (this.pos - 7) >>> 16;
cmd.incrementCompressSequenceNo(1);
if (this.opts.debugCompress) {
console.log(
"==> conn:%d %s (compress)\n%s",
this.info.threadId ? this.info.threadId : -1,
cmd
? (cmd.onPacketReceive
? cmd.constructor.name + "." + cmd.onPacketReceive.name
: cmd.constructor.name) +
"(0," +
this.pos +
")"
: "unknown",
Utils.log(this.header, 0, 7) + Utils.log(compressChunk, 0, compressChunkLen)
);
}
try {
this.writer(this.header);
this.writer(compressChunk);
if (cmdEnd) {
if (this.pos === MAX_BUFFER_SIZE) this.writeEmptyPacket();
//reset buffer
this.buf = this.smallBuffer;
this.pos = 7;
} catch (err) {
//eat exception : thrown by socket.on('error');
}
this.pos = 7;
} else {
//*******************************************************************************
// compressing packet
//*******************************************************************************
//use synchronous inflating, to ensure FIFO packet order
const compressChunk = ZLib.deflateSync(this.buf.slice(7, this.pos));
const compressChunkLen = compressChunk.length;
this.header[0] = compressChunkLen;
this.header[1] = compressChunkLen >>> 8;
this.header[2] = compressChunkLen >>> 16;
this.header[3] = cmd.compressSequenceNo;
this.header[4] = this.pos - 7;
this.header[5] = (this.pos - 7) >>> 8;
this.header[6] = (this.pos - 7) >>> 16;
cmd.incrementCompressSequenceNo(1);
if (this.opts.debugCompress) {
console.log(
"==> conn:%d %s (compress)\n%s",
this.info.threadId ? this.info.threadId : -1,
cmd
? (cmd.onPacketReceive
? cmd.constructor.name + "." + cmd.onPacketReceive.name
: cmd.constructor.name) +
"(0," +
this.pos +
")"
: "unknown",
Utils.log(this.header, 0, 7) + Utils.log(compressChunk, 0, compressChunkLen)
);
}
try {
this.writer(this.header);
this.writer(compressChunk);
if (cmdEnd) {
if (this.pos === MAX_BUFFER_SIZE) this.writeEmptyPacket(cmd);
//reset buffer
this.buf = this.smallBuffer;
}
this.pos = 7;
} catch (err) {
//eat exception : thrown by socket.on('error');
}
}
}
writeEmptyPacket(cmd) {
const emptyBuf = new Buffer([0x00, 0x00, 0x00, cmd.compressSequenceNo, 0x00, 0x00, 0x00]);
cmd.incrementCompressSequenceNo(1);
if (this.opts.debugCompress) {
console.log(
"==> conn:%d %s (compress)\n%s",
this.info.threadId ? this.info.threadId : -1,
cmd
? (cmd.onPacketReceive
? cmd.constructor.name + "." + cmd.onPacketReceive.name
: cmd.constructor.name) +
"(0," +
this.pos +
")"
: "unknown",
Utils.log(emptyBuf, 0, 7)
);
}
try {
this.writer(emptyBuf);
} catch (err) {
//eat exception : thrown by socket.on('error');
}
}
};
}
module.exports = CompressionOutputStream;

839
src/io/packet-output-stream.js
View File

@ -13,6 +13,7 @@ const SMALL_BUFFER_SIZE = 2042;
const MEDIUM_BUFFER_SIZE = 131072; //128k
const LARGE_BUFFER_SIZE = 1048576; //1M
const MAX_BUFFER_SIZE = 16777219; //16M + 4
const CHARS_GLOBAL_REGEXP = /[\0\"\'\\]/g; // eslint-disable-line no-control-regex
/**
* MySQL packet builder.
@ -21,473 +22,495 @@ const MAX_BUFFER_SIZE = 16777219; //16M + 4
* @param info connection info
* @constructor
*/
function PacketOutputStream(opts, info) {
this.opts = opts;
this.info = info;
this.pos = 4;
this.smallBuffer = Buffer.allocUnsafe(SMALL_BUFFER_SIZE);
this.buf = this.smallBuffer;
this.writeDate = opts.timezone === "local" ? this.writeLocalDate : this.writeTimezoneDate;
this.encoding = this.opts.collation.encoding;
if (this.encoding === "utf8") {
this.writeString = this.writeDefaultBufferString;
this.writeStringEscape = this.writeUtf8StringEscape;
} else if (Buffer.isEncoding(this.encoding)) {
this.writeString = this.writeDefaultBufferString;
this.writeStringEscape = this.writeDefaultStringEscape;
} else {
this.writeString = this.writeDefaultIconvString;
this.writeStringEscape = this.writeDefaultStringEscape;
}
}
PacketOutputStream.prototype.setStreamer = function(stream) {
this.stream = stream;
};
PacketOutputStream.prototype.growBuffer = function(len) {
let newCapacity;
if (len + this.pos < MEDIUM_BUFFER_SIZE) {
newCapacity = MEDIUM_BUFFER_SIZE;
} else if (len + this.pos < LARGE_BUFFER_SIZE) {
newCapacity = LARGE_BUFFER_SIZE;
} else newCapacity = MAX_BUFFER_SIZE;
let newBuf = Buffer.allocUnsafe(newCapacity);
this.buf.copy(newBuf, 0, 0, this.pos);
this.buf = newBuf;
};
PacketOutputStream.prototype.startPacket = function(cmd) {
this.cmd = cmd;
this.pos = 4;
};
PacketOutputStream.prototype.writeInt8 = function(value) {
if (this.pos + 1 >= this.buf.length) {
if (this.pos >= MAX_BUFFER_SIZE) {
//buffer is more than a Packet, must flushBuffer()
this.flushBuffer(false);
} else this.growBuffer(1);
}
this.buf[this.pos++] = value;
};
PacketOutputStream.prototype.writeInt16 = function(value) {
if (this.pos + 2 >= this.buf.length) {
let b = Buffer.allocUnsafe(2);
b.writeUInt16LE(value, 0);
this.writeBuffer(b, 0, 2);
return;
}
this.buf.writeUInt16LE(value, this.pos);
this.pos += 2;
};
PacketOutputStream.prototype.writeInt32 = function(value) {
if (this.pos + 4 >= this.buf.length) {
//not enough space remaining
let arr = Buffer.allocUnsafe(4);
arr.writeInt32LE(value, 0);
this.writeBuffer(arr, 0, 4);
return;
class PacketOutputStream {
constructor(opts, info) {
this.opts = opts;
this.info = info;
this.pos = 4;
this.smallBuffer = Buffer.allocUnsafe(SMALL_BUFFER_SIZE);
this.buf = this.smallBuffer;
this.writeDate = opts.timezone === "local" ? this.writeLocalDate : this.writeTimezoneDate;
this.encoding = this.opts.collation.encoding;
if (this.encoding === "utf8") {
this.writeString = this.writeDefaultBufferString;
this.writeStringEscape = this.writeUtf8StringEscape;
} else if (Buffer.isEncoding(this.encoding)) {
this.writeString = this.writeDefaultBufferString;
this.writeStringEscape = this.writeDefaultStringEscape;
} else {
this.writeString = this.writeDefaultIconvString;
this.writeStringEscape = this.writeDefaultStringEscape;
}
}
this.buf.writeInt32LE(value, this.pos);
this.pos += 4;
};
PacketOutputStream.prototype.writeLengthCoded = function(len) {
if (len < 0xfb) {
this.writeInt8(len);
return;
setStreamer(stream) {
this.stream = stream;
}
if (len < 0xffff) {
this.writeInt8(0xfc);
this.writeInt16(len);
return;
growBuffer(len) {
let newCapacity;
if (len + this.pos < MEDIUM_BUFFER_SIZE) {
newCapacity = MEDIUM_BUFFER_SIZE;
} else if (len + this.pos < LARGE_BUFFER_SIZE) {
newCapacity = LARGE_BUFFER_SIZE;
} else newCapacity = MAX_BUFFER_SIZE;
let newBuf = Buffer.allocUnsafe(newCapacity);
this.buf.copy(newBuf, 0, 0, this.pos);
this.buf = newBuf;
}
if (len < 0xffffff) {
this.writeInt8(0xfd);
this.writeInt24(len);
return;
startPacket(cmd) {
this.cmd = cmd;
this.pos = 4;
}
if (len === null) {
this.writeInt8(0xfb);
return;
writeInt8(value) {
if (this.pos + 1 >= this.buf.length) {
if (this.pos >= MAX_BUFFER_SIZE) {
//buffer is more than a Packet, must flushBuffer()
this.flushBuffer(false);
} else this.growBuffer(1);
}
this.buf[this.pos++] = value;
}
this.writeInt8(0xfe);
this.buffer.writeUInt32LE(len, this.pos);
this.buffer.writeUInt32LE(len >> 32, this.pos + 4);
this.pos += 8;
};
writeInt16(value) {
if (this.pos + 2 >= this.buf.length) {
let b = Buffer.allocUnsafe(2);
b.writeUInt16LE(value, 0);
this.writeBuffer(b, 0, 2);
return;
}
this.buf.writeUInt16LE(value, this.pos);
this.pos += 2;
}
PacketOutputStream.prototype.writeLocalDate = function(date, opts) {
const year = date.getFullYear();
const mon = date.getMonth() + 1;
const day = date.getDate();
const hour = date.getHours();
const min = date.getMinutes();
const sec = date.getSeconds();
const ms = date.getMilliseconds();
this._writeDatePart(year, mon, day, hour, min, sec, ms);
};
writeInt24(value) {
if (this.pos + 3 >= this.buf.length) {
let b = Buffer.allocUnsafe(3);
PacketOutputStream.prototype._writeDatePart = function(year, mon, day, hour, min, sec, ms) {
//return 'YYYY-MM-DD HH:MM:SS' datetime format
//see https://mariadb.com/kb/en/library/datetime/
this.writeStringAscii(
(year > 999 ? year : year > 99 ? "0" + year : year > 9 ? "00" + year : "000" + year) +
"-" +
(mon < 10 ? "0" : "") +
mon +
"-" +
(day < 10 ? "0" : "") +
day +
" " +
(hour < 10 ? "0" : "") +
hour +
":" +
(min < 10 ? "0" : "") +
min +
":" +
(sec < 10 ? "0" : "") +
sec +
"." +
(ms > 99 ? ms : ms > 9 ? "0" + ms : "00" + ms)
);
};
PacketOutputStream.prototype.writeTimezoneDate = function(date, opts) {
if (opts.timezoneMillisOffset) date.setTime(date.getTime() + opts.timezoneMillisOffset);
const year = date.getUTCFullYear();
const mon = date.getUTCMonth() + 1;
const day = date.getUTCDate();
const hour = date.getUTCHours();
const min = date.getUTCMinutes();
const sec = date.getUTCSeconds();
const ms = date.getUTCMilliseconds();
this._writeDatePart(year, mon, day, hour, min, sec, ms);
};
PacketOutputStream.prototype.writeLengthCodedBuffer = function(arr) {
this.writeLengthCoded(arr.length);
this.writeBuffer(arr, 0, arr.length);
};
PacketOutputStream.prototype.writeBuffer = function(arr, off, len) {
if (len > this.buf.length - this.pos) {
if (this.buf.length !== MAX_BUFFER_SIZE) {
this.growBuffer(len);
b[0] = value;
b[1] = value >>> 8;
b[2] = value >>> 16;
this.writeBuffer(b, 0, 2);
return;
}
//max buffer size
b[this.pos] = value;
b[this.pos + 1] = value >>> 8;
b[this.pos + 2] = value >>> 16;
this.pos += 3;
}
writeInt32(value) {
if (this.pos + 4 >= this.buf.length) {
//not enough space remaining
let arr = Buffer.allocUnsafe(4);
arr.writeInt32LE(value, 0);
this.writeBuffer(arr, 0, 4);
return;
}
this.buf.writeInt32LE(value, this.pos);
this.pos += 4;
}
writeLengthCoded(len) {
if (len < 0xfb) {
this.writeInt8(len);
return;
}
if (len < 0xffff) {
this.writeInt8(0xfc);
this.writeInt16(len);
return;
}
if (len < 0xffffff) {
this.writeInt8(0xfd);
this.writeInt24(len);
return;
}
if (len === null) {
this.writeInt8(0xfb);
return;
}
this.writeInt8(0xfe);
this.buf.writeUInt32LE(len, this.pos);
this.buf.writeUInt32LE(len >> 32, this.pos + 4);
this.pos += 8;
}
writeLocalDate(date, opts) {
const year = date.getFullYear();
const mon = date.getMonth() + 1;
const day = date.getDate();
const hour = date.getHours();
const min = date.getMinutes();
const sec = date.getSeconds();
const ms = date.getMilliseconds();
this._writeDatePart(year, mon, day, hour, min, sec, ms);
}
_writeDatePart(year, mon, day, hour, min, sec, ms) {
//return 'YYYY-MM-DD HH:MM:SS' datetime format
//see https://mariadb.com/kb/en/library/datetime/
this.writeStringAscii(
(year > 999 ? year : year > 99 ? "0" + year : year > 9 ? "00" + year : "000" + year) +
"-" +
(mon < 10 ? "0" : "") +
mon +
"-" +
(day < 10 ? "0" : "") +
day +
" " +
(hour < 10 ? "0" : "") +
hour +
":" +
(min < 10 ? "0" : "") +
min +
":" +
(sec < 10 ? "0" : "") +
sec +
"." +
(ms > 99 ? ms : ms > 9 ? "0" + ms : "00" + ms)
);
}
writeTimezoneDate(date, opts) {
if (opts.timezoneMillisOffset) date.setTime(date.getTime() + opts.timezoneMillisOffset);
const year = date.getUTCFullYear();
const mon = date.getUTCMonth() + 1;
const day = date.getUTCDate();
const hour = date.getUTCHours();
const min = date.getUTCMinutes();
const sec = date.getUTCSeconds();
const ms = date.getUTCMilliseconds();
this._writeDatePart(year, mon, day, hour, min, sec, ms);
}
writeLengthCodedBuffer(arr) {
this.writeLengthCoded(arr.length);
this.writeBuffer(arr, 0, arr.length);
}
writeBuffer(arr, off, len) {
if (len > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
// fill buffer and flush until all data are snd
let remainingLen = len;
if (this.buf.length !== MAX_BUFFER_SIZE) {
this.growBuffer(len);
}
while (true) {
//filling buffer
let lenToFillBuffer = Math.min(MAX_BUFFER_SIZE - this.pos, remainingLen);
arr.copy(this.buf, this.pos, off, off + lenToFillBuffer);
remainingLen -= lenToFillBuffer;
off += lenToFillBuffer;
this.pos += lenToFillBuffer;
//max buffer size
if (len > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
// fill buffer and flush until all data are snd
let remainingLen = len;
if (remainingLen === 0) return;
this.flushBuffer(false);
while (true) {
//filling buffer
let lenToFillBuffer = Math.min(MAX_BUFFER_SIZE - this.pos, remainingLen);
arr.copy(this.buf, this.pos, off, off + lenToFillBuffer);
remainingLen -= lenToFillBuffer;
off += lenToFillBuffer;
this.pos += lenToFillBuffer;
if (remainingLen === 0) return;
this.flushBuffer(false);
}
}
}
}
arr.copy(this.buf, this.pos, off, off + len);
this.pos += len;
};
/**
* Write ascii string to socket (no escaping)
*
* @param str string
*/
PacketOutputStream.prototype.writeStringAscii = function writeStringAscii(str) {
let len = str.length;
//not enough space remaining
if (len >= this.buf.length - this.pos) {
let strBuf = Buffer.from(str, "ascii");
this.writeBuffer(strBuf, 0, strBuf.length);
return;
arr.copy(this.buf, this.pos, off, off + len);
this.pos += len;
}
for (let off = 0; off < len; ) {
this.buf[this.pos++] = str.charCodeAt(off++);
}
};
/**
* Write ascii string to socket (no escaping)
*
* @param str string
*/
writeStringAscii(str) {
let len = str.length;
PacketOutputStream.prototype.writeUtf8StringEscape = function(str) {
const charsLength = str.length;
//not enough space remaining
if (charsLength * 3 >= this.buf.length - this.pos) {
const arr = Buffer.from(str, "utf8");
this.writeBufferEscape(arr);
return;
}
//create UTF-8 byte array
//since java char are internally using UTF-16 using surrogate's pattern, 4 bytes unicode characters will
//represent 2 characters : example "\uD83C\uDFA4" = 🎤 unicode 8 "no microphones"
//so max size is 3 * charLength
//(escape characters are 1 byte encoded, so length might only be 2 when escape)
// + 2 for the quotes for text protocol
let charsOffset = 0;
let currChar;
//quick loop if only ASCII chars for faster escape
for (
;
charsOffset < charsLength && (currChar = str.charCodeAt(charsOffset)) < 0x80;
charsOffset++
) {
if (currChar == SLASH || currChar == QUOTE || currChar == ZERO_BYTE || currChar == DBL_QUOTE) {
this.buf[this.pos++] = SLASH;
//not enough space remaining
if (len >= this.buf.length - this.pos) {
let strBuf = Buffer.from(str, "ascii");
this.writeBuffer(strBuf, 0, strBuf.length);
return;
}
for (let off = 0; off < len; ) {
this.buf[this.pos++] = str.charCodeAt(off++);
}
this.buf[this.pos++] = currChar;
}
//if quick loop not finished
while (charsOffset < charsLength) {
currChar = str.charCodeAt(charsOffset++);
if (currChar < 0x80) {
writeUtf8StringEscape(str) {
const charsLength = str.length;
//not enough space remaining
if (charsLength * 3 >= this.buf.length - this.pos) {
const arr = Buffer.from(str, "utf8");
this.writeBufferEscape(arr);
return;
}
//create UTF-8 byte array
//since java char are internally using UTF-16 using surrogate's pattern, 4 bytes unicode characters will
//represent 2 characters : example "\uD83C\uDFA4" = 🎤 unicode 8 "no microphones"
//so max size is 3 * charLength
//(escape characters are 1 byte encoded, so length might only be 2 when escape)
// + 2 for the quotes for text protocol
let charsOffset = 0;
let currChar;
//quick loop if only ASCII chars for faster escape
for (
;
charsOffset < charsLength && (currChar = str.charCodeAt(charsOffset)) < 0x80;
charsOffset++
) {
if (
currChar == SLASH ||
currChar == QUOTE ||
currChar == ZERO_BYTE ||
currChar == DBL_QUOTE
currChar === SLASH ||
currChar === QUOTE ||
currChar === ZERO_BYTE ||
currChar === DBL_QUOTE
) {
this.buf[this.pos++] = SLASH;
}
this.buf[this.pos++] = currChar;
} else if (currChar < 0x800) {
this.buf[this.pos++] = 0xc0 | (currChar >> 6);
this.buf[this.pos++] = 0x80 | (currChar & 0x3f);
} else if (currChar >= 0xd800 && currChar < 0xe000) {
//reserved for surrogate - see https://en.wikipedia.org/wiki/UTF-16
if (currChar < 0xdc00) {
//is high surrogate
if (charsOffset + 1 > charsLength) {
this.buf[this.pos++] = 0x63;
} else {
const nextChar = str.charCodeAt(charsOffset);
if (nextChar >= 0xdc00 && nextChar < 0xe000) {
//is low surrogate
const surrogatePairs =
(currChar << 10) + nextChar + (0x010000 - (0xd800 << 10) - 0xdc00);
this.buf[this.pos++] = 0xf0 | (surrogatePairs >> 18);
this.buf[this.pos++] = 0x80 | ((surrogatePairs >> 12) & 0x3f);
this.buf[this.pos++] = 0x80 | ((surrogatePairs >> 6) & 0x3f);
this.buf[this.pos++] = 0x80 | (surrogatePairs & 0x3f);
charsOffset++;
} else {
//must have low surrogate
}
//if quick loop not finished
while (charsOffset < charsLength) {
currChar = str.charCodeAt(charsOffset++);
if (currChar < 0x80) {
if (
currChar === SLASH ||
currChar === QUOTE ||
currChar === ZERO_BYTE ||
currChar === DBL_QUOTE
) {
this.buf[this.pos++] = SLASH;
}
this.buf[this.pos++] = currChar;
} else if (currChar < 0x800) {
this.buf[this.pos++] = 0xc0 | (currChar >> 6);
this.buf[this.pos++] = 0x80 | (currChar & 0x3f);
} else if (currChar >= 0xd800 && currChar < 0xe000) {
//reserved for surrogate - see https://en.wikipedia.org/wiki/UTF-16
if (currChar < 0xdc00) {
//is high surrogate
if (charsOffset + 1 > charsLength) {
this.buf[this.pos++] = 0x63;
} else {
const nextChar = str.charCodeAt(charsOffset);
if (nextChar >= 0xdc00 && nextChar < 0xe000) {
//is low surrogate
const surrogatePairs =
(currChar << 10) + nextChar + (0x010000 - (0xd800 << 10) - 0xdc00);
this.buf[this.pos++] = 0xf0 | (surrogatePairs >> 18);
this.buf[this.pos++] = 0x80 | ((surrogatePairs >> 12) & 0x3f);
this.buf[this.pos++] = 0x80 | ((surrogatePairs >> 6) & 0x3f);
this.buf[this.pos++] = 0x80 | (surrogatePairs & 0x3f);
charsOffset++;
} else {
//must have low surrogate
this.buf[this.pos++] = 0x63;
}
}
} else {
//low surrogate without high surrogate before
this.buf[this.pos++] = 0x63;
}
} else {
//low surrogate without high surrogate before
this.buf[this.pos++] = 0x63;
this.buf[this.pos++] = 0xe0 | (currChar >> 12);
this.buf[this.pos++] = 0x80 | ((currChar >> 6) & 0x3f);
this.buf[this.pos++] = 0x80 | (currChar & 0x3f);
}
} else {
this.buf[this.pos++] = 0xe0 | (currChar >> 12);
this.buf[this.pos++] = 0x80 | ((currChar >> 6) & 0x3f);
this.buf[this.pos++] = 0x80 | (currChar & 0x3f);
}
}
};
PacketOutputStream.prototype.writeDefaultBufferString = function(str) {
//javascript use UCS-2 or UTF-16 string internal representation
//that means that string to byte will be a maximum of * 3
// (4 bytes utf-8 are represented on 2 UTF-16 characters)
if (str.length * 3 < this.buf.length - this.pos) {
this.pos += this.buf.write(str, this.pos, this.encoding);
return;
}
//checking real length
let byteLength = Buffer.byteLength(str, this.encoding);
if (byteLength > this.buf.length - this.pos) {
if (this.buf.length < MAX_BUFFER_SIZE) {
this.growBuffer(byteLength);
writeDefaultBufferString(str) {
//javascript use UCS-2 or UTF-16 string internal representation
//that means that string to byte will be a maximum of * 3
// (4 bytes utf-8 are represented on 2 UTF-16 characters)
if (str.length * 3 < this.buf.length - this.pos) {
this.pos += this.buf.write(str, this.pos, this.encoding);
return;
}
//checking real length
let byteLength = Buffer.byteLength(str, this.encoding);
if (byteLength > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
let strBuf = Buffer.from(str, this.encoding);
this.writeBuffer(strBuf, 0, strBuf.length);
if (this.buf.length < MAX_BUFFER_SIZE) {
this.growBuffer(byteLength);
}
if (byteLength > this.buf.length - this.pos) {
//not enough space in buffer, will stream :
let strBuf = Buffer.from(str, this.encoding);
this.writeBuffer(strBuf, 0, strBuf.length);
return;
}
}
this.pos += this.buf.write(str, this.pos, this.encoding);
}
writeDefaultIconvString(str) {
let buf = Iconv.encode(str, this.encoding);
this.writeBuffer(buf, 0, buf.length);
}
/**
* Parameters need to be properly escaped :
* following characters are to be escaped by "\" :
* - \0
* - \\
* - \'
* - \"
* regex split part of string writing part, and escaping special char.
* Those chars are <= 7f meaning that this will work even with multi-byte encoding
*
* @param str string to escape.
*/
writeDefaultStringEscape(str) {
let match;
let lastIndex = 0;
while ((match = CHARS_GLOBAL_REGEXP.exec(str)) !== null) {
this.writeString(str.slice(lastIndex, match.index));
this.writeInt8(SLASH);
this.writeInt8(match[0].charCodeAt(0));
lastIndex = CHARS_GLOBAL_REGEXP.lastIndex;
}
if (lastIndex === 0) {
// Nothing was escaped
this.writeString(str);
return;
}
}
this.pos += this.buf.write(str, this.pos, this.encoding);
};
PacketOutputStream.prototype.writeDefaultIconvString = function(str) {
let buf = Iconv.encode(str, this.encoding);
this.writeBuffer(buf, 0, buf.length);
};
const CHARS_GLOBAL_REGEXP = /[\0\"\'\\]/g; // eslint-disable-line no-control-regex
/**
* Parameters need to be properly escaped :
* following characters are to be escaped by "\" :
* - \0
* - \\
* - \'
* - \"
* regex split part of string writing part, and escaping special char.
* Those chars are <= 7f meaning that this will work even with multi-byte encoding
*
* @param str string to escape.
*/
PacketOutputStream.prototype.writeDefaultStringEscape = function(str) {
let match;
let lastIndex = 0;
while ((match = CHARS_GLOBAL_REGEXP.exec(str)) !== null) {
this.writeString(str.slice(lastIndex, match.index));
this.writeInt8(SLASH);
this.writeInt8(match[0].charCodeAt(0));
lastIndex = CHARS_GLOBAL_REGEXP.lastIndex;
if (lastIndex < str.length) {
this.writeString(str.slice(lastIndex));
}
}
if (lastIndex === 0) {
// Nothing was escaped
this.writeString(str);
return;
}
if (lastIndex < str.length) {
this.writeString(str.slice(lastIndex));
}
};
PacketOutputStream.prototype.writeBufferEscape = function(val) {
let valLen = val.length;
if (valLen * 2 > this.buf.length - this.pos) {
//makes buffer bigger (up to 16M)
if (this.buf.length !== MAX_BUFFER_SIZE) this.growBuffer(valLen * 2);
//data may still be bigger than buffer.
//must flush buffer when full (and reset position to 4)
writeBufferEscape(val) {
let valLen = val.length;
if (valLen * 2 > this.buf.length - this.pos) {
//not enough space in buffer, will fill buffer
for (let i = 0; i < valLen; i++) {
switch (val[i]) {
case QUOTE:
case SLASH:
case DBL_QUOTE:
case ZERO_BYTE:
if (this.pos >= this.buf.length) this.flushBuffer(false);
this.buf[this.pos++] = SLASH; //add escape slash
//makes buffer bigger (up to 16M)
if (this.buf.length !== MAX_BUFFER_SIZE) this.growBuffer(valLen * 2);
//data may still be bigger than buffer.
//must flush buffer when full (and reset position to 4)
if (valLen * 2 > this.buf.length - this.pos) {
//not enough space in buffer, will fill buffer
for (let i = 0; i < valLen; i++) {
switch (val[i]) {
case QUOTE:
case SLASH:
case DBL_QUOTE:
case ZERO_BYTE:
if (this.pos >= this.buf.length) this.flushBuffer(false);
this.buf[this.pos++] = SLASH; //add escape slash
}
if (this.pos >= this.buf.length) this.flushBuffer(false);
this.buf[this.pos++] = val[i];
}
if (this.pos >= this.buf.length) this.flushBuffer(false);
this.buf[this.pos++] = val[i];
return;
}
return;
}
//sure to have enough place to use buffer directly
for (let i = 0; i < valLen; i++) {
switch (val[i]) {
case QUOTE:
case SLASH:
case DBL_QUOTE:
case ZERO_BYTE:
this.buf[this.pos++] = SLASH; //add escape slash
}
this.buf[this.pos++] = val[i];
}
}
//sure to have enough place to use buffer directly
for (let i = 0; i < valLen; i++) {
switch (val[i]) {
case QUOTE:
case SLASH:
case DBL_QUOTE:
case ZERO_BYTE:
this.buf[this.pos++] = SLASH; //add escape slash
/**
* Indicate if buffer contain any data.
* @returns {boolean}
*/
isEmpty() {
return this.pos <= 4;
}
/**
* Flush the internal buffer.
*/
flushBuffer(commandEnd) {
this.buf[0] = this.pos - 4;
this.buf[1] = (this.pos - 4) >>> 8;
this.buf[2] = (this.pos - 4) >>> 16;
this.buf[3] = this.cmd.sequenceNo;
this.cmd.incrementSequenceNo(1);
if (this.opts.debug && !this.opts.debugCompress) {
console.log(
"==> conn:%d %s\n%s",
this.info.threadId ? this.info.threadId : -1,
(this.cmd.onPacketReceive
? this.cmd.constructor.name + "." + this.cmd.onPacketReceive.name
: this.cmd.constructor.name) +
"(0," +
this.pos +
")",
Utils.log(this.buf, 0, this.pos)
);
}
this.buf[this.pos++] = val[i];
}
};
/**
* Indicate if buffer contain any data.
* @returns {boolean}
*/
PacketOutputStream.prototype.isEmpty = function() {
return this.pos <= 4;
};
try {
this.stream.writeBuf(this.buf.slice(0, this.pos), this.cmd);
/**
* Flush the internal buffer.
*/
PacketOutputStream.prototype.flushBuffer = function(commandEnd) {
this.buf[0] = this.pos - 4;
this.buf[1] = (this.pos - 4) >>> 8;
this.buf[2] = (this.pos - 4) >>> 16;
this.buf[3] = this.cmd.sequenceNo;
this.cmd.incrementSequenceNo(1);
if (commandEnd) {
//if last com fill the max size, must send an empty com to indicate command end.
if (this.pos === MAX_BUFFER_SIZE) {
this.writeEmptyPacket();
} else {
this.stream.flush(true, this.cmd);
}
if (this.opts.debug && !this.opts.debugCompress) {
console.log(
"==> conn:%d %s\n%s",
this.info.threadId ? this.info.threadId : -1,
(this.cmd.onPacketReceive
? this.cmd.constructor.name + "." + this.cmd.onPacketReceive.name
: this.cmd.constructor.name) +
"(0," +
this.pos +
")",
Utils.log(this.buf, 0, this.pos)
);
}
try {
this.stream.writeBuf(this.buf.slice(0, this.pos), this.cmd);
if (commandEnd) {
//if last com fill the max size, must send an empty com to indicate command end.
if (this.pos === MAX_BUFFER_SIZE) {
this.writeEmptyPacket();
} else {
this.stream.flush(true, this.cmd);
//reset buffer
this.buf = this.smallBuffer;
}
//reset buffer
this.buf = this.smallBuffer;
this.pos = 4;
} catch (err) {
//eat exception : thrown by socket.on('error');
}
}
writeEmptyPacket() {
const emptyBuf = new Buffer([0x00, 0x00, 0x00, this.cmd.sequenceNo]);
this.cmd.incrementSequenceNo(1);
if (this.opts.debug && !this.opts.debugCompress) {
console.log(
"==> conn:%d %s\n%s",
this.info.threadId ? this.info.threadId : -1,
(this.cmd.onPacketReceive
? this.cmd.constructor.name + "." + this.cmd.onPacketReceive.name
: this.cmd.constructor.name) + "(0,4)",
Utils.log(emptyBuf, 0, 4)
);
}
this.pos = 4;
} catch (err) {
//eat exception : thrown by socket.on('error');
try {
this.stream.writeBuf(emptyBuf, this.cmd);
this.stream.flush(true, this.cmd);
} catch (err) {
//eat exception : thrown by socket.on('error');
}
}
};
PacketOutputStream.prototype.writeEmptyPacket = function() {
const emptyBuf = new Buffer([0x00, 0x00, 0x00, this.cmd.sequenceNo]);
this.cmd.incrementSequenceNo(1);
if (this.opts.debug && !this.opts.debugCompress) {
console.log(
"==> conn:%d %s\n%s",
this.info.threadId ? this.info.threadId : -1,
(this.cmd.onPacketReceive
? this.cmd.constructor.name + "." + this.cmd.onPacketReceive.name
: this.cmd.constructor.name) + "(0,4)",
Utils.log(emptyBuf, 0, 4)
);
}
try {
this.stream.writeBuf(emptyBuf, this.cmd);
this.stream.flush(true, this.cmd);
} catch (err) {
//eat exception : thrown by socket.on('error');
}
};
}
module.exports = PacketOutputStream;

3
test/conf.js
View File

@ -4,8 +4,7 @@ let baseConfig = {
user: "root",
database: "testn",
host: "localhost",
port: 3306,
compress: true
port: 3306
};
if (process.env.TEST_HOST) baseConfig["host"] = process.env.TEST_HOST;

2
test/integration/test-error.js
View File

@ -64,7 +64,7 @@ describe("Error", () => {
it("server close connection during query", function(done) {
this.timeout(10000);
var conn = base.createConnection();
const conn = base.createConnection();
conn.connect(function(err) {
conn.query("set @@wait_timeout = 1");
setTimeout(function() {

4
test/integration/test-ok-packet.js
View File

@ -185,7 +185,7 @@ describe("ok packet", () => {
});
it("update result text changedRows", function(done) {
var conn = base.createConnection({ foundRows: false });
const conn = base.createConnection({ foundRows: false });
conn.query("CREATE TEMPORARY TABLE updateResultSet1(id int(11))");
conn.query("INSERT INTO updateResultSet1 values (1), (1), (2), (3)");
conn.query("UPDATE updateResultSet1 set id = 1", function(err, res) {
@ -207,7 +207,7 @@ describe("ok packet", () => {
});
it("update result binary changedRows", function(done) {
var conn = base.createConnection({ foundRows: false });
const conn = base.createConnection({ foundRows: false });
conn.query("CREATE TEMPORARY TABLE updateResultSet2(id int(11))");
conn.query("INSERT INTO updateResultSet2 values (1), (1), (2), (3)");
conn.execute("UPDATE updateResultSet2 set id = 1", function(err, res) {