connection.js 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947
  1. // This file was modified by Oracle on June 1, 2021.
  2. // The changes involve new logic to handle an additional ERR Packet sent by
  3. // the MySQL server when the connection is closed unexpectedly.
  4. // Modifications copyright (c) 2021, Oracle and/or its affiliates.
  5. // This file was modified by Oracle on June 17, 2021.
  6. // The changes involve logic to ensure the socket connection is closed when
  7. // there is a fatal error.
  8. // Modifications copyright (c) 2021, Oracle and/or its affiliates.
  9. // This file was modified by Oracle on September 21, 2021.
  10. // The changes involve passing additional authentication factor passwords
  11. // to the ChangeUser Command instance.
  12. // Modifications copyright (c) 2021, Oracle and/or its affiliates.
  13. 'use strict';
  14. const Net = require('net');
  15. const Tls = require('tls');
  16. const Timers = require('timers');
  17. const EventEmitter = require('events').EventEmitter;
  18. const Readable = require('stream').Readable;
  19. const Queue = require('denque');
  20. const SqlString = require('sqlstring');
  21. const { createLRU } = require('lru.min');
  22. const PacketParser = require('../packet_parser.js');
  23. const Packets = require('../packets/index.js');
  24. const Commands = require('../commands/index.js');
  25. const ConnectionConfig = require('../connection_config.js');
  26. const CharsetToEncoding = require('../constants/charset_encodings.js');
  27. let _connectionId = 0;
  28. let convertNamedPlaceholders = null;
  29. class BaseConnection extends EventEmitter {
  30. constructor(opts) {
  31. super();
  32. this.config = opts.config;
  33. // TODO: fill defaults
  34. // if no params, connect to /var/lib/mysql/mysql.sock ( /tmp/mysql.sock on OSX )
  35. // if host is given, connect to host:3306
  36. // TODO: use `/usr/local/mysql/bin/mysql_config --socket` output? as default socketPath
  37. // if there is no host/port and no socketPath parameters?
  38. if (!opts.config.stream) {
  39. if (opts.config.socketPath) {
  40. this.stream = Net.connect(opts.config.socketPath);
  41. } else {
  42. this.stream = Net.connect(opts.config.port, opts.config.host);
  43. // Optionally enable keep-alive on the socket.
  44. if (this.config.enableKeepAlive) {
  45. this.stream.on('connect', () => {
  46. this.stream.setKeepAlive(true, this.config.keepAliveInitialDelay);
  47. });
  48. }
  49. // Enable TCP_NODELAY flag. This is needed so that the network packets
  50. // are sent immediately to the server
  51. this.stream.setNoDelay(true);
  52. }
  53. // if stream is a function, treat it as "stream agent / factory"
  54. } else if (typeof opts.config.stream === 'function') {
  55. this.stream = opts.config.stream(opts);
  56. } else {
  57. this.stream = opts.config.stream;
  58. }
  59. this._internalId = _connectionId++;
  60. this._commands = new Queue();
  61. this._command = null;
  62. this._paused = false;
  63. this._paused_packets = new Queue();
  64. this._statements = createLRU({
  65. max: this.config.maxPreparedStatements,
  66. onEviction: function (_, statement) {
  67. statement.close();
  68. },
  69. });
  70. this.serverCapabilityFlags = 0;
  71. this.authorized = false;
  72. this.sequenceId = 0;
  73. this.compressedSequenceId = 0;
  74. this.threadId = null;
  75. this._handshakePacket = null;
  76. this._fatalError = null;
  77. this._protocolError = null;
  78. this._outOfOrderPackets = [];
  79. this.clientEncoding = CharsetToEncoding[this.config.charsetNumber];
  80. this.stream.on('error', this._handleNetworkError.bind(this));
  81. // see https://gist.github.com/khoomeister/4985691#use-that-instead-of-bind
  82. this.packetParser = new PacketParser((p) => {
  83. this.handlePacket(p);
  84. });
  85. this.stream.on('data', (data) => {
  86. if (this.connectTimeout) {
  87. Timers.clearTimeout(this.connectTimeout);
  88. this.connectTimeout = null;
  89. }
  90. this.packetParser.execute(data);
  91. });
  92. this.stream.on('end', () => {
  93. // emit the end event so that the pooled connection can close the connection
  94. this.emit('end');
  95. });
  96. this.stream.on('close', () => {
  97. // we need to set this flag everywhere where we want connection to close
  98. if (this._closing) {
  99. return;
  100. }
  101. if (!this._protocolError) {
  102. // no particular error message before disconnect
  103. this._protocolError = new Error(
  104. 'Connection lost: The server closed the connection.'
  105. );
  106. this._protocolError.fatal = true;
  107. this._protocolError.code = 'PROTOCOL_CONNECTION_LOST';
  108. }
  109. this._notifyError(this._protocolError);
  110. });
  111. let handshakeCommand;
  112. if (!this.config.isServer) {
  113. handshakeCommand = new Commands.ClientHandshake(this.config.clientFlags);
  114. handshakeCommand.on('end', () => {
  115. // this happens when handshake finishes early either because there was
  116. // some fatal error or the server sent an error packet instead of
  117. // an hello packet (for example, 'Too many connections' error)
  118. if (
  119. !handshakeCommand.handshake ||
  120. this._fatalError ||
  121. this._protocolError
  122. ) {
  123. return;
  124. }
  125. this._handshakePacket = handshakeCommand.handshake;
  126. this.threadId = handshakeCommand.handshake.connectionId;
  127. this.emit('connect', handshakeCommand.handshake);
  128. });
  129. handshakeCommand.on('error', (err) => {
  130. this._closing = true;
  131. this._notifyError(err);
  132. });
  133. this.addCommand(handshakeCommand);
  134. }
  135. // in case there was no initial handshake but we need to read sting, assume it utf-8
  136. // most common example: "Too many connections" error ( packet is sent immediately on connection attempt, we don't know server encoding yet)
  137. // will be overwritten with actual encoding value as soon as server handshake packet is received
  138. this.serverEncoding = 'utf8';
  139. if (this.config.connectTimeout) {
  140. const timeoutHandler = this._handleTimeoutError.bind(this);
  141. this.connectTimeout = Timers.setTimeout(
  142. timeoutHandler,
  143. this.config.connectTimeout
  144. );
  145. }
  146. }
  147. _addCommandClosedState(cmd) {
  148. const err = new Error(
  149. "Can't add new command when connection is in closed state"
  150. );
  151. err.fatal = true;
  152. if (cmd.onResult) {
  153. cmd.onResult(err);
  154. } else {
  155. this.emit('error', err);
  156. }
  157. }
  158. _handleFatalError(err) {
  159. err.fatal = true;
  160. // stop receiving packets
  161. this.stream.removeAllListeners('data');
  162. this.addCommand = this._addCommandClosedState;
  163. this.write = () => {
  164. this.emit('error', new Error("Can't write in closed state"));
  165. };
  166. this._notifyError(err);
  167. this._fatalError = err;
  168. }
  169. _handleNetworkError(err) {
  170. if (this.connectTimeout) {
  171. Timers.clearTimeout(this.connectTimeout);
  172. this.connectTimeout = null;
  173. }
  174. // Do not throw an error when a connection ends with a RST,ACK packet
  175. if (err.code === 'ECONNRESET' && this._closing) {
  176. return;
  177. }
  178. this._handleFatalError(err);
  179. }
  180. _handleTimeoutError() {
  181. if (this.connectTimeout) {
  182. Timers.clearTimeout(this.connectTimeout);
  183. this.connectTimeout = null;
  184. }
  185. this.stream.destroy && this.stream.destroy();
  186. const err = new Error('connect ETIMEDOUT');
  187. err.errorno = 'ETIMEDOUT';
  188. err.code = 'ETIMEDOUT';
  189. err.syscall = 'connect';
  190. this._handleNetworkError(err);
  191. }
  192. // notify all commands in the queue and bubble error as connection "error"
  193. // called on stream error or unexpected termination
  194. _notifyError(err) {
  195. if (this.connectTimeout) {
  196. Timers.clearTimeout(this.connectTimeout);
  197. this.connectTimeout = null;
  198. }
  199. // prevent from emitting 'PROTOCOL_CONNECTION_LOST' after EPIPE or ECONNRESET
  200. if (this._fatalError) {
  201. return;
  202. }
  203. let command;
  204. // if there is no active command, notify connection
  205. // if there are commands and all of them have callbacks, pass error via callback
  206. let bubbleErrorToConnection = !this._command;
  207. if (this._command && this._command.onResult) {
  208. this._command.onResult(err);
  209. this._command = null;
  210. // connection handshake is special because we allow it to be implicit
  211. // if error happened during handshake, but there are others commands in queue
  212. // then bubble error to other commands and not to connection
  213. } else if (
  214. !(
  215. this._command &&
  216. this._command.constructor === Commands.ClientHandshake &&
  217. this._commands.length > 0
  218. )
  219. ) {
  220. bubbleErrorToConnection = true;
  221. }
  222. while ((command = this._commands.shift())) {
  223. if (command.onResult) {
  224. command.onResult(err);
  225. } else {
  226. bubbleErrorToConnection = true;
  227. }
  228. }
  229. // notify connection if some comands in the queue did not have callbacks
  230. // or if this is pool connection ( so it can be removed from pool )
  231. if (bubbleErrorToConnection || this._pool) {
  232. this.emit('error', err);
  233. }
  234. // close connection after emitting the event in case of a fatal error
  235. if (err.fatal) {
  236. this.close();
  237. }
  238. }
  239. write(buffer) {
  240. const result = this.stream.write(buffer, (err) => {
  241. if (err) {
  242. this._handleNetworkError(err);
  243. }
  244. });
  245. if (!result) {
  246. this.stream.emit('pause');
  247. }
  248. }
  249. // http://dev.mysql.com/doc/internals/en/sequence-id.html
  250. //
  251. // The sequence-id is incremented with each packet and may wrap around.
  252. // It starts at 0 and is reset to 0 when a new command
  253. // begins in the Command Phase.
  254. // http://dev.mysql.com/doc/internals/en/example-several-mysql-packets.html
  255. _resetSequenceId() {
  256. this.sequenceId = 0;
  257. this.compressedSequenceId = 0;
  258. }
  259. _bumpCompressedSequenceId(numPackets) {
  260. this.compressedSequenceId += numPackets;
  261. this.compressedSequenceId %= 256;
  262. }
  263. _bumpSequenceId(numPackets) {
  264. this.sequenceId += numPackets;
  265. this.sequenceId %= 256;
  266. }
  267. writePacket(packet) {
  268. const MAX_PACKET_LENGTH = 16777215;
  269. const length = packet.length();
  270. let chunk, offset, header;
  271. if (length < MAX_PACKET_LENGTH) {
  272. packet.writeHeader(this.sequenceId);
  273. if (this.config.debug) {
  274. console.log(
  275. `${this._internalId} ${this.connectionId} <== ${this._command._commandName}#${this._command.stateName()}(${[this.sequenceId, packet._name, packet.length()].join(',')})`
  276. );
  277. console.log(
  278. `${this._internalId} ${this.connectionId} <== ${packet.buffer.toString('hex')}`
  279. );
  280. }
  281. this._bumpSequenceId(1);
  282. this.write(packet.buffer);
  283. } else {
  284. if (this.config.debug) {
  285. console.log(
  286. `${this._internalId} ${this.connectionId} <== Writing large packet, raw content not written:`
  287. );
  288. console.log(
  289. `${this._internalId} ${this.connectionId} <== ${this._command._commandName}#${this._command.stateName()}(${[this.sequenceId, packet._name, packet.length()].join(',')})`
  290. );
  291. }
  292. for (offset = 4; offset < 4 + length; offset += MAX_PACKET_LENGTH) {
  293. chunk = packet.buffer.slice(offset, offset + MAX_PACKET_LENGTH);
  294. if (chunk.length === MAX_PACKET_LENGTH) {
  295. header = Buffer.from([0xff, 0xff, 0xff, this.sequenceId]);
  296. } else {
  297. header = Buffer.from([
  298. chunk.length & 0xff,
  299. (chunk.length >> 8) & 0xff,
  300. (chunk.length >> 16) & 0xff,
  301. this.sequenceId,
  302. ]);
  303. }
  304. this._bumpSequenceId(1);
  305. this.write(header);
  306. this.write(chunk);
  307. }
  308. }
  309. }
  310. // 0.11+ environment
  311. startTLS(onSecure) {
  312. if (this.config.debug) {
  313. console.log('Upgrading connection to TLS');
  314. }
  315. const secureContext = Tls.createSecureContext({
  316. ca: this.config.ssl.ca,
  317. cert: this.config.ssl.cert,
  318. ciphers: this.config.ssl.ciphers,
  319. key: this.config.ssl.key,
  320. passphrase: this.config.ssl.passphrase,
  321. minVersion: this.config.ssl.minVersion,
  322. maxVersion: this.config.ssl.maxVersion,
  323. });
  324. const rejectUnauthorized = this.config.ssl.rejectUnauthorized;
  325. const verifyIdentity = this.config.ssl.verifyIdentity;
  326. const servername = Net.isIP(this.config.host)
  327. ? undefined
  328. : this.config.host;
  329. let secureEstablished = false;
  330. this.stream.removeAllListeners('data');
  331. const secureSocket = Tls.connect(
  332. {
  333. rejectUnauthorized,
  334. requestCert: rejectUnauthorized,
  335. checkServerIdentity: verifyIdentity
  336. ? Tls.checkServerIdentity
  337. : function () {
  338. return undefined;
  339. },
  340. secureContext,
  341. isServer: false,
  342. socket: this.stream,
  343. servername,
  344. },
  345. () => {
  346. secureEstablished = true;
  347. if (rejectUnauthorized) {
  348. if (typeof servername === 'string' && verifyIdentity) {
  349. const cert = secureSocket.getPeerCertificate(true);
  350. const serverIdentityCheckError = Tls.checkServerIdentity(
  351. servername,
  352. cert
  353. );
  354. if (serverIdentityCheckError) {
  355. onSecure(serverIdentityCheckError);
  356. return;
  357. }
  358. }
  359. }
  360. onSecure();
  361. }
  362. );
  363. // error handler for secure socket
  364. secureSocket.on('error', (err) => {
  365. if (secureEstablished) {
  366. this._handleNetworkError(err);
  367. } else {
  368. onSecure(err);
  369. }
  370. });
  371. secureSocket.on('data', (data) => {
  372. this.packetParser.execute(data);
  373. });
  374. this.stream = secureSocket;
  375. }
  376. protocolError(message, code) {
  377. // Starting with MySQL 8.0.24, if the client closes the connection
  378. // unexpectedly, the server will send a last ERR Packet, which we can
  379. // safely ignore.
  380. // https://dev.mysql.com/worklog/task/?id=12999
  381. if (this._closing) {
  382. return;
  383. }
  384. const err = new Error(message);
  385. err.fatal = true;
  386. err.code = code || 'PROTOCOL_ERROR';
  387. this.emit('error', err);
  388. }
  389. get fatalError() {
  390. return this._fatalError;
  391. }
  392. handlePacket(packet) {
  393. if (this._paused) {
  394. this._paused_packets.push(packet);
  395. return;
  396. }
  397. if (this.config.debug) {
  398. if (packet) {
  399. console.log(
  400. ` raw: ${packet.buffer
  401. .slice(packet.offset, packet.offset + packet.length())
  402. .toString('hex')}`
  403. );
  404. console.trace();
  405. const commandName = this._command
  406. ? this._command._commandName
  407. : '(no command)';
  408. const stateName = this._command
  409. ? this._command.stateName()
  410. : '(no command)';
  411. console.log(
  412. `${this._internalId} ${this.connectionId} ==> ${commandName}#${stateName}(${[packet.sequenceId, packet.type(), packet.length()].join(',')})`
  413. );
  414. }
  415. }
  416. if (!this._command) {
  417. const marker = packet.peekByte();
  418. // If it's an Err Packet, we should use it.
  419. if (marker === 0xff) {
  420. const error = Packets.Error.fromPacket(packet);
  421. this.protocolError(error.message, error.code);
  422. } else {
  423. // Otherwise, it means it's some other unexpected packet.
  424. this.protocolError(
  425. 'Unexpected packet while no commands in the queue',
  426. 'PROTOCOL_UNEXPECTED_PACKET'
  427. );
  428. }
  429. this.close();
  430. return;
  431. }
  432. if (packet) {
  433. // Note: when server closes connection due to inactivity, Err packet ER_CLIENT_INTERACTION_TIMEOUT from MySQL 8.0.24, sequenceId will be 0
  434. if (this.sequenceId !== packet.sequenceId) {
  435. const err = new Error(
  436. `Warning: got packets out of order. Expected ${this.sequenceId} but received ${packet.sequenceId}`
  437. );
  438. err.expected = this.sequenceId;
  439. err.received = packet.sequenceId;
  440. this.emit('warn', err); // REVIEW
  441. console.error(err.message);
  442. }
  443. this._bumpSequenceId(packet.numPackets);
  444. }
  445. try {
  446. if (this._fatalError) {
  447. // skip remaining packets after client is in the error state
  448. return;
  449. }
  450. const done = this._command.execute(packet, this);
  451. if (done) {
  452. this._command = this._commands.shift();
  453. if (this._command) {
  454. this.sequenceId = 0;
  455. this.compressedSequenceId = 0;
  456. this.handlePacket();
  457. }
  458. }
  459. } catch (err) {
  460. this._handleFatalError(err);
  461. this.stream.destroy();
  462. }
  463. }
  464. addCommand(cmd) {
  465. // this.compressedSequenceId = 0;
  466. // this.sequenceId = 0;
  467. if (this.config.debug) {
  468. const commandName = cmd.constructor.name;
  469. console.log(`Add command: ${commandName}`);
  470. cmd._commandName = commandName;
  471. }
  472. if (!this._command) {
  473. this._command = cmd;
  474. this.handlePacket();
  475. } else {
  476. this._commands.push(cmd);
  477. }
  478. return cmd;
  479. }
  480. format(sql, values) {
  481. if (typeof this.config.queryFormat === 'function') {
  482. return this.config.queryFormat.call(
  483. this,
  484. sql,
  485. values,
  486. this.config.timezone
  487. );
  488. }
  489. const opts = {
  490. sql: sql,
  491. values: values,
  492. };
  493. this._resolveNamedPlaceholders(opts);
  494. return SqlString.format(
  495. opts.sql,
  496. opts.values,
  497. this.config.stringifyObjects,
  498. this.config.timezone
  499. );
  500. }
  501. escape(value) {
  502. return SqlString.escape(value, false, this.config.timezone);
  503. }
  504. escapeId(value) {
  505. return SqlString.escapeId(value, false);
  506. }
  507. raw(sql) {
  508. return SqlString.raw(sql);
  509. }
  510. _resolveNamedPlaceholders(options) {
  511. let unnamed;
  512. if (this.config.namedPlaceholders || options.namedPlaceholders) {
  513. if (Array.isArray(options.values)) {
  514. // if an array is provided as the values, assume the conversion is not necessary.
  515. // this allows the usage of unnamed placeholders even if the namedPlaceholders flag is enabled.
  516. return;
  517. }
  518. if (convertNamedPlaceholders === null) {
  519. convertNamedPlaceholders = require('named-placeholders')();
  520. }
  521. unnamed = convertNamedPlaceholders(options.sql, options.values);
  522. options.sql = unnamed[0];
  523. options.values = unnamed[1];
  524. }
  525. }
  526. query(sql, values, cb) {
  527. let cmdQuery;
  528. if (sql.constructor === Commands.Query) {
  529. cmdQuery = sql;
  530. } else {
  531. cmdQuery = BaseConnection.createQuery(sql, values, cb, this.config);
  532. }
  533. this._resolveNamedPlaceholders(cmdQuery);
  534. const rawSql = this.format(
  535. cmdQuery.sql,
  536. cmdQuery.values !== undefined ? cmdQuery.values : []
  537. );
  538. cmdQuery.sql = rawSql;
  539. return this.addCommand(cmdQuery);
  540. }
  541. pause() {
  542. this._paused = true;
  543. this.stream.pause();
  544. }
  545. resume() {
  546. let packet;
  547. this._paused = false;
  548. while ((packet = this._paused_packets.shift())) {
  549. this.handlePacket(packet);
  550. // don't resume if packet handler paused connection
  551. if (this._paused) {
  552. return;
  553. }
  554. }
  555. this.stream.resume();
  556. }
  557. // TODO: named placeholders support
  558. prepare(options, cb) {
  559. if (typeof options === 'string') {
  560. options = { sql: options };
  561. }
  562. return this.addCommand(new Commands.Prepare(options, cb));
  563. }
  564. unprepare(sql) {
  565. let options = {};
  566. if (typeof sql === 'object') {
  567. options = sql;
  568. } else {
  569. options.sql = sql;
  570. }
  571. const key = BaseConnection.statementKey(options);
  572. const stmt = this._statements.get(key);
  573. if (stmt) {
  574. this._statements.delete(key);
  575. stmt.close();
  576. }
  577. return stmt;
  578. }
  579. execute(sql, values, cb) {
  580. let options = {
  581. infileStreamFactory: this.config.infileStreamFactory,
  582. };
  583. if (typeof sql === 'object') {
  584. // execute(options, cb)
  585. options = {
  586. ...options,
  587. ...sql,
  588. sql: sql.sql,
  589. values: sql.values,
  590. };
  591. if (typeof values === 'function') {
  592. cb = values;
  593. } else {
  594. options.values = options.values || values;
  595. }
  596. } else if (typeof values === 'function') {
  597. // execute(sql, cb)
  598. cb = values;
  599. options.sql = sql;
  600. options.values = undefined;
  601. } else {
  602. // execute(sql, values, cb)
  603. options.sql = sql;
  604. options.values = values;
  605. }
  606. this._resolveNamedPlaceholders(options);
  607. // check for values containing undefined
  608. if (options.values) {
  609. //If namedPlaceholder is not enabled and object is passed as bind parameters
  610. if (!Array.isArray(options.values)) {
  611. throw new TypeError(
  612. 'Bind parameters must be array if namedPlaceholders parameter is not enabled'
  613. );
  614. }
  615. options.values.forEach((val) => {
  616. //If namedPlaceholder is not enabled and object is passed as bind parameters
  617. if (!Array.isArray(options.values)) {
  618. throw new TypeError(
  619. 'Bind parameters must be array if namedPlaceholders parameter is not enabled'
  620. );
  621. }
  622. if (val === undefined) {
  623. throw new TypeError(
  624. 'Bind parameters must not contain undefined. To pass SQL NULL specify JS null'
  625. );
  626. }
  627. if (typeof val === 'function') {
  628. throw new TypeError(
  629. 'Bind parameters must not contain function(s). To pass the body of a function as a string call .toString() first'
  630. );
  631. }
  632. });
  633. }
  634. const executeCommand = new Commands.Execute(options, cb);
  635. const prepareCommand = new Commands.Prepare(options, (err, stmt) => {
  636. if (err) {
  637. // skip execute command if prepare failed, we have main
  638. // combined callback here
  639. executeCommand.start = function () {
  640. return null;
  641. };
  642. if (cb) {
  643. cb(err);
  644. } else {
  645. executeCommand.emit('error', err);
  646. }
  647. executeCommand.emit('end');
  648. return;
  649. }
  650. executeCommand.statement = stmt;
  651. });
  652. this.addCommand(prepareCommand);
  653. this.addCommand(executeCommand);
  654. return executeCommand;
  655. }
  656. changeUser(options, callback) {
  657. if (!callback && typeof options === 'function') {
  658. callback = options;
  659. options = {};
  660. }
  661. const charsetNumber = options.charset
  662. ? ConnectionConfig.getCharsetNumber(options.charset)
  663. : this.config.charsetNumber;
  664. return this.addCommand(
  665. new Commands.ChangeUser(
  666. {
  667. user: options.user || this.config.user,
  668. // for the purpose of multi-factor authentication, or not, the main
  669. // password (used for the 1st authentication factor) can also be
  670. // provided via the "password1" option
  671. password:
  672. options.password ||
  673. options.password1 ||
  674. this.config.password ||
  675. this.config.password1,
  676. password2: options.password2 || this.config.password2,
  677. password3: options.password3 || this.config.password3,
  678. passwordSha1: options.passwordSha1 || this.config.passwordSha1,
  679. database: options.database || this.config.database,
  680. timeout: options.timeout,
  681. charsetNumber: charsetNumber,
  682. currentConfig: this.config,
  683. },
  684. (err) => {
  685. if (err) {
  686. err.fatal = true;
  687. }
  688. if (callback) {
  689. callback(err);
  690. }
  691. }
  692. )
  693. );
  694. }
  695. // transaction helpers
  696. beginTransaction(cb) {
  697. return this.query('START TRANSACTION', cb);
  698. }
  699. commit(cb) {
  700. return this.query('COMMIT', cb);
  701. }
  702. rollback(cb) {
  703. return this.query('ROLLBACK', cb);
  704. }
  705. ping(cb) {
  706. return this.addCommand(new Commands.Ping(cb));
  707. }
  708. _registerSlave(opts, cb) {
  709. return this.addCommand(new Commands.RegisterSlave(opts, cb));
  710. }
  711. _binlogDump(opts, cb) {
  712. return this.addCommand(new Commands.BinlogDump(opts, cb));
  713. }
  714. // currently just alias to close
  715. destroy() {
  716. this.close();
  717. }
  718. close() {
  719. if (this.connectTimeout) {
  720. Timers.clearTimeout(this.connectTimeout);
  721. this.connectTimeout = null;
  722. }
  723. this._closing = true;
  724. this.stream.end();
  725. this.addCommand = this._addCommandClosedState;
  726. }
  727. createBinlogStream(opts) {
  728. // TODO: create proper stream class
  729. // TODO: use through2
  730. let test = 1;
  731. const stream = new Readable({ objectMode: true });
  732. stream._read = function () {
  733. return {
  734. data: test++,
  735. };
  736. };
  737. this._registerSlave(opts, () => {
  738. const dumpCmd = this._binlogDump(opts);
  739. dumpCmd.on('event', (ev) => {
  740. stream.push(ev);
  741. });
  742. dumpCmd.on('eof', () => {
  743. stream.push(null);
  744. // if non-blocking, then close stream to prevent errors
  745. if (opts.flags && opts.flags & 0x01) {
  746. this.close();
  747. }
  748. });
  749. // TODO: pipe errors as well
  750. });
  751. return stream;
  752. }
  753. connect(cb) {
  754. if (!cb) {
  755. return;
  756. }
  757. if (this._fatalError || this._protocolError) {
  758. return cb(this._fatalError || this._protocolError);
  759. }
  760. if (this._handshakePacket) {
  761. return cb(null, this);
  762. }
  763. let connectCalled = 0;
  764. function callbackOnce(isErrorHandler) {
  765. return function (param) {
  766. if (!connectCalled) {
  767. if (isErrorHandler) {
  768. cb(param);
  769. } else {
  770. cb(null, param);
  771. }
  772. }
  773. connectCalled = 1;
  774. };
  775. }
  776. this.once('error', callbackOnce(true));
  777. this.once('connect', callbackOnce(false));
  778. }
  779. // ===================================
  780. // outgoing server connection methods
  781. // ===================================
  782. writeColumns(columns) {
  783. this.writePacket(Packets.ResultSetHeader.toPacket(columns.length));
  784. columns.forEach((column) => {
  785. this.writePacket(
  786. Packets.ColumnDefinition.toPacket(column, this.serverConfig.encoding)
  787. );
  788. });
  789. this.writeEof();
  790. }
  791. // row is array of columns, not hash
  792. writeTextRow(column) {
  793. this.writePacket(
  794. Packets.TextRow.toPacket(column, this.serverConfig.encoding)
  795. );
  796. }
  797. writeBinaryRow(column) {
  798. this.writePacket(
  799. Packets.BinaryRow.toPacket(column, this.serverConfig.encoding)
  800. );
  801. }
  802. writeTextResult(rows, columns, binary = false) {
  803. this.writeColumns(columns);
  804. rows.forEach((row) => {
  805. const arrayRow = new Array(columns.length);
  806. columns.forEach((column) => {
  807. arrayRow.push(row[column.name]);
  808. });
  809. if (binary) {
  810. this.writeBinaryRow(arrayRow);
  811. } else this.writeTextRow(arrayRow);
  812. });
  813. this.writeEof();
  814. }
  815. writeEof(warnings, statusFlags) {
  816. this.writePacket(Packets.EOF.toPacket(warnings, statusFlags));
  817. }
  818. writeOk(args) {
  819. if (!args) {
  820. args = { affectedRows: 0 };
  821. }
  822. this.writePacket(Packets.OK.toPacket(args, this.serverConfig.encoding));
  823. }
  824. writeError(args) {
  825. // if we want to send error before initial hello was sent, use default encoding
  826. const encoding = this.serverConfig ? this.serverConfig.encoding : 'cesu8';
  827. this.writePacket(Packets.Error.toPacket(args, encoding));
  828. }
  829. serverHandshake(args) {
  830. this.serverConfig = args;
  831. this.serverConfig.encoding =
  832. CharsetToEncoding[this.serverConfig.characterSet];
  833. return this.addCommand(new Commands.ServerHandshake(args));
  834. }
  835. // ===============================================================
  836. end(callback) {
  837. if (this.config.isServer) {
  838. this._closing = true;
  839. const quitCmd = new EventEmitter();
  840. setImmediate(() => {
  841. this.stream.end();
  842. quitCmd.emit('end');
  843. });
  844. return quitCmd;
  845. }
  846. // trigger error if more commands enqueued after end command
  847. const quitCmd = this.addCommand(new Commands.Quit(callback));
  848. this.addCommand = this._addCommandClosedState;
  849. return quitCmd;
  850. }
  851. static createQuery(sql, values, cb, config) {
  852. let options = {
  853. rowsAsArray: config.rowsAsArray,
  854. infileStreamFactory: config.infileStreamFactory,
  855. };
  856. if (typeof sql === 'object') {
  857. // query(options, cb)
  858. options = {
  859. ...options,
  860. ...sql,
  861. sql: sql.sql,
  862. values: sql.values,
  863. };
  864. if (typeof values === 'function') {
  865. cb = values;
  866. } else if (values !== undefined) {
  867. options.values = values;
  868. }
  869. } else if (typeof values === 'function') {
  870. // query(sql, cb)
  871. cb = values;
  872. options.sql = sql;
  873. options.values = undefined;
  874. } else {
  875. // query(sql, values, cb)
  876. options.sql = sql;
  877. options.values = values;
  878. }
  879. return new Commands.Query(options, cb);
  880. }
  881. static statementKey(options) {
  882. return `${typeof options.nestTables}/${options.nestTables}/${options.rowsAsArray}${options.sql}`;
  883. }
  884. }
  885. module.exports = BaseConnection;