connection.js 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945
  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 = this.config.host;
  327. let secureEstablished = false;
  328. this.stream.removeAllListeners('data');
  329. const secureSocket = Tls.connect(
  330. {
  331. rejectUnauthorized,
  332. requestCert: rejectUnauthorized,
  333. checkServerIdentity: verifyIdentity
  334. ? Tls.checkServerIdentity
  335. : function () {
  336. return undefined;
  337. },
  338. secureContext,
  339. isServer: false,
  340. socket: this.stream,
  341. servername,
  342. },
  343. () => {
  344. secureEstablished = true;
  345. if (rejectUnauthorized) {
  346. if (typeof servername === 'string' && verifyIdentity) {
  347. const cert = secureSocket.getPeerCertificate(true);
  348. const serverIdentityCheckError = Tls.checkServerIdentity(
  349. servername,
  350. cert,
  351. );
  352. if (serverIdentityCheckError) {
  353. onSecure(serverIdentityCheckError);
  354. return;
  355. }
  356. }
  357. }
  358. onSecure();
  359. },
  360. );
  361. // error handler for secure socket
  362. secureSocket.on('error', (err) => {
  363. if (secureEstablished) {
  364. this._handleNetworkError(err);
  365. } else {
  366. onSecure(err);
  367. }
  368. });
  369. secureSocket.on('data', (data) => {
  370. this.packetParser.execute(data);
  371. });
  372. this.write = (buffer) => secureSocket.write(buffer);
  373. }
  374. protocolError(message, code) {
  375. // Starting with MySQL 8.0.24, if the client closes the connection
  376. // unexpectedly, the server will send a last ERR Packet, which we can
  377. // safely ignore.
  378. // https://dev.mysql.com/worklog/task/?id=12999
  379. if (this._closing) {
  380. return;
  381. }
  382. const err = new Error(message);
  383. err.fatal = true;
  384. err.code = code || 'PROTOCOL_ERROR';
  385. this.emit('error', err);
  386. }
  387. get fatalError() {
  388. return this._fatalError;
  389. }
  390. handlePacket(packet) {
  391. if (this._paused) {
  392. this._paused_packets.push(packet);
  393. return;
  394. }
  395. if (this.config.debug) {
  396. if (packet) {
  397. console.log(
  398. ` raw: ${packet.buffer
  399. .slice(packet.offset, packet.offset + packet.length())
  400. .toString('hex')}`,
  401. );
  402. console.trace();
  403. const commandName = this._command
  404. ? this._command._commandName
  405. : '(no command)';
  406. const stateName = this._command
  407. ? this._command.stateName()
  408. : '(no command)';
  409. console.log(
  410. `${this._internalId} ${this.connectionId} ==> ${commandName}#${stateName}(${[packet.sequenceId, packet.type(), packet.length()].join(',')})`,
  411. );
  412. }
  413. }
  414. if (!this._command) {
  415. const marker = packet.peekByte();
  416. // If it's an Err Packet, we should use it.
  417. if (marker === 0xff) {
  418. const error = Packets.Error.fromPacket(packet);
  419. this.protocolError(error.message, error.code);
  420. } else {
  421. // Otherwise, it means it's some other unexpected packet.
  422. this.protocolError(
  423. 'Unexpected packet while no commands in the queue',
  424. 'PROTOCOL_UNEXPECTED_PACKET',
  425. );
  426. }
  427. this.close();
  428. return;
  429. }
  430. if (packet) {
  431. // Note: when server closes connection due to inactivity, Err packet ER_CLIENT_INTERACTION_TIMEOUT from MySQL 8.0.24, sequenceId will be 0
  432. if (this.sequenceId !== packet.sequenceId) {
  433. const err = new Error(
  434. `Warning: got packets out of order. Expected ${this.sequenceId} but received ${packet.sequenceId}`,
  435. );
  436. err.expected = this.sequenceId;
  437. err.received = packet.sequenceId;
  438. this.emit('warn', err); // REVIEW
  439. console.error(err.message);
  440. }
  441. this._bumpSequenceId(packet.numPackets);
  442. }
  443. try {
  444. if (this._fatalError) {
  445. // skip remaining packets after client is in the error state
  446. return;
  447. }
  448. const done = this._command.execute(packet, this);
  449. if (done) {
  450. this._command = this._commands.shift();
  451. if (this._command) {
  452. this.sequenceId = 0;
  453. this.compressedSequenceId = 0;
  454. this.handlePacket();
  455. }
  456. }
  457. } catch (err) {
  458. this._handleFatalError(err);
  459. this.stream.destroy();
  460. }
  461. }
  462. addCommand(cmd) {
  463. // this.compressedSequenceId = 0;
  464. // this.sequenceId = 0;
  465. if (this.config.debug) {
  466. const commandName = cmd.constructor.name;
  467. console.log(`Add command: ${commandName}`);
  468. cmd._commandName = commandName;
  469. }
  470. if (!this._command) {
  471. this._command = cmd;
  472. this.handlePacket();
  473. } else {
  474. this._commands.push(cmd);
  475. }
  476. return cmd;
  477. }
  478. format(sql, values) {
  479. if (typeof this.config.queryFormat === 'function') {
  480. return this.config.queryFormat.call(
  481. this,
  482. sql,
  483. values,
  484. this.config.timezone,
  485. );
  486. }
  487. const opts = {
  488. sql: sql,
  489. values: values,
  490. };
  491. this._resolveNamedPlaceholders(opts);
  492. return SqlString.format(
  493. opts.sql,
  494. opts.values,
  495. this.config.stringifyObjects,
  496. this.config.timezone,
  497. );
  498. }
  499. escape(value) {
  500. return SqlString.escape(value, false, this.config.timezone);
  501. }
  502. escapeId(value) {
  503. return SqlString.escapeId(value, false);
  504. }
  505. raw(sql) {
  506. return SqlString.raw(sql);
  507. }
  508. _resolveNamedPlaceholders(options) {
  509. let unnamed;
  510. if (this.config.namedPlaceholders || options.namedPlaceholders) {
  511. if (Array.isArray(options.values)) {
  512. // if an array is provided as the values, assume the conversion is not necessary.
  513. // this allows the usage of unnamed placeholders even if the namedPlaceholders flag is enabled.
  514. return;
  515. }
  516. if (convertNamedPlaceholders === null) {
  517. convertNamedPlaceholders = require('named-placeholders')();
  518. }
  519. unnamed = convertNamedPlaceholders(options.sql, options.values);
  520. options.sql = unnamed[0];
  521. options.values = unnamed[1];
  522. }
  523. }
  524. query(sql, values, cb) {
  525. let cmdQuery;
  526. if (sql.constructor === Commands.Query) {
  527. cmdQuery = sql;
  528. } else {
  529. cmdQuery = BaseConnection.createQuery(sql, values, cb, this.config);
  530. }
  531. this._resolveNamedPlaceholders(cmdQuery);
  532. const rawSql = this.format(
  533. cmdQuery.sql,
  534. cmdQuery.values !== undefined ? cmdQuery.values : [],
  535. );
  536. cmdQuery.sql = rawSql;
  537. return this.addCommand(cmdQuery);
  538. }
  539. pause() {
  540. this._paused = true;
  541. this.stream.pause();
  542. }
  543. resume() {
  544. let packet;
  545. this._paused = false;
  546. while ((packet = this._paused_packets.shift())) {
  547. this.handlePacket(packet);
  548. // don't resume if packet handler paused connection
  549. if (this._paused) {
  550. return;
  551. }
  552. }
  553. this.stream.resume();
  554. }
  555. // TODO: named placeholders support
  556. prepare(options, cb) {
  557. if (typeof options === 'string') {
  558. options = { sql: options };
  559. }
  560. return this.addCommand(new Commands.Prepare(options, cb));
  561. }
  562. unprepare(sql) {
  563. let options = {};
  564. if (typeof sql === 'object') {
  565. options = sql;
  566. } else {
  567. options.sql = sql;
  568. }
  569. const key = BaseConnection.statementKey(options);
  570. const stmt = this._statements.get(key);
  571. if (stmt) {
  572. this._statements.delete(key);
  573. stmt.close();
  574. }
  575. return stmt;
  576. }
  577. execute(sql, values, cb) {
  578. let options = {
  579. infileStreamFactory: this.config.infileStreamFactory,
  580. };
  581. if (typeof sql === 'object') {
  582. // execute(options, cb)
  583. options = {
  584. ...options,
  585. ...sql,
  586. sql: sql.sql,
  587. values: sql.values,
  588. };
  589. if (typeof values === 'function') {
  590. cb = values;
  591. } else {
  592. options.values = options.values || values;
  593. }
  594. } else if (typeof values === 'function') {
  595. // execute(sql, cb)
  596. cb = values;
  597. options.sql = sql;
  598. options.values = undefined;
  599. } else {
  600. // execute(sql, values, cb)
  601. options.sql = sql;
  602. options.values = values;
  603. }
  604. this._resolveNamedPlaceholders(options);
  605. // check for values containing undefined
  606. if (options.values) {
  607. //If namedPlaceholder is not enabled and object is passed as bind parameters
  608. if (!Array.isArray(options.values)) {
  609. throw new TypeError(
  610. 'Bind parameters must be array if namedPlaceholders parameter is not enabled',
  611. );
  612. }
  613. options.values.forEach((val) => {
  614. //If namedPlaceholder is not enabled and object is passed as bind parameters
  615. if (!Array.isArray(options.values)) {
  616. throw new TypeError(
  617. 'Bind parameters must be array if namedPlaceholders parameter is not enabled',
  618. );
  619. }
  620. if (val === undefined) {
  621. throw new TypeError(
  622. 'Bind parameters must not contain undefined. To pass SQL NULL specify JS null',
  623. );
  624. }
  625. if (typeof val === 'function') {
  626. throw new TypeError(
  627. 'Bind parameters must not contain function(s). To pass the body of a function as a string call .toString() first',
  628. );
  629. }
  630. });
  631. }
  632. const executeCommand = new Commands.Execute(options, cb);
  633. const prepareCommand = new Commands.Prepare(options, (err, stmt) => {
  634. if (err) {
  635. // skip execute command if prepare failed, we have main
  636. // combined callback here
  637. executeCommand.start = function () {
  638. return null;
  639. };
  640. if (cb) {
  641. cb(err);
  642. } else {
  643. executeCommand.emit('error', err);
  644. }
  645. executeCommand.emit('end');
  646. return;
  647. }
  648. executeCommand.statement = stmt;
  649. });
  650. this.addCommand(prepareCommand);
  651. this.addCommand(executeCommand);
  652. return executeCommand;
  653. }
  654. changeUser(options, callback) {
  655. if (!callback && typeof options === 'function') {
  656. callback = options;
  657. options = {};
  658. }
  659. const charsetNumber = options.charset
  660. ? ConnectionConfig.getCharsetNumber(options.charset)
  661. : this.config.charsetNumber;
  662. return this.addCommand(
  663. new Commands.ChangeUser(
  664. {
  665. user: options.user || this.config.user,
  666. // for the purpose of multi-factor authentication, or not, the main
  667. // password (used for the 1st authentication factor) can also be
  668. // provided via the "password1" option
  669. password:
  670. options.password ||
  671. options.password1 ||
  672. this.config.password ||
  673. this.config.password1,
  674. password2: options.password2 || this.config.password2,
  675. password3: options.password3 || this.config.password3,
  676. passwordSha1: options.passwordSha1 || this.config.passwordSha1,
  677. database: options.database || this.config.database,
  678. timeout: options.timeout,
  679. charsetNumber: charsetNumber,
  680. currentConfig: this.config,
  681. },
  682. (err) => {
  683. if (err) {
  684. err.fatal = true;
  685. }
  686. if (callback) {
  687. callback(err);
  688. }
  689. },
  690. ),
  691. );
  692. }
  693. // transaction helpers
  694. beginTransaction(cb) {
  695. return this.query('START TRANSACTION', cb);
  696. }
  697. commit(cb) {
  698. return this.query('COMMIT', cb);
  699. }
  700. rollback(cb) {
  701. return this.query('ROLLBACK', cb);
  702. }
  703. ping(cb) {
  704. return this.addCommand(new Commands.Ping(cb));
  705. }
  706. _registerSlave(opts, cb) {
  707. return this.addCommand(new Commands.RegisterSlave(opts, cb));
  708. }
  709. _binlogDump(opts, cb) {
  710. return this.addCommand(new Commands.BinlogDump(opts, cb));
  711. }
  712. // currently just alias to close
  713. destroy() {
  714. this.close();
  715. }
  716. close() {
  717. if (this.connectTimeout) {
  718. Timers.clearTimeout(this.connectTimeout);
  719. this.connectTimeout = null;
  720. }
  721. this._closing = true;
  722. this.stream.end();
  723. this.addCommand = this._addCommandClosedState;
  724. }
  725. createBinlogStream(opts) {
  726. // TODO: create proper stream class
  727. // TODO: use through2
  728. let test = 1;
  729. const stream = new Readable({ objectMode: true });
  730. stream._read = function () {
  731. return {
  732. data: test++,
  733. };
  734. };
  735. this._registerSlave(opts, () => {
  736. const dumpCmd = this._binlogDump(opts);
  737. dumpCmd.on('event', (ev) => {
  738. stream.push(ev);
  739. });
  740. dumpCmd.on('eof', () => {
  741. stream.push(null);
  742. // if non-blocking, then close stream to prevent errors
  743. if (opts.flags && opts.flags & 0x01) {
  744. this.close();
  745. }
  746. });
  747. // TODO: pipe errors as well
  748. });
  749. return stream;
  750. }
  751. connect(cb) {
  752. if (!cb) {
  753. return;
  754. }
  755. if (this._fatalError || this._protocolError) {
  756. return cb(this._fatalError || this._protocolError);
  757. }
  758. if (this._handshakePacket) {
  759. return cb(null, this);
  760. }
  761. let connectCalled = 0;
  762. function callbackOnce(isErrorHandler) {
  763. return function (param) {
  764. if (!connectCalled) {
  765. if (isErrorHandler) {
  766. cb(param);
  767. } else {
  768. cb(null, param);
  769. }
  770. }
  771. connectCalled = 1;
  772. };
  773. }
  774. this.once('error', callbackOnce(true));
  775. this.once('connect', callbackOnce(false));
  776. }
  777. // ===================================
  778. // outgoing server connection methods
  779. // ===================================
  780. writeColumns(columns) {
  781. this.writePacket(Packets.ResultSetHeader.toPacket(columns.length));
  782. columns.forEach((column) => {
  783. this.writePacket(
  784. Packets.ColumnDefinition.toPacket(column, this.serverConfig.encoding),
  785. );
  786. });
  787. this.writeEof();
  788. }
  789. // row is array of columns, not hash
  790. writeTextRow(column) {
  791. this.writePacket(
  792. Packets.TextRow.toPacket(column, this.serverConfig.encoding),
  793. );
  794. }
  795. writeBinaryRow(column) {
  796. this.writePacket(
  797. Packets.BinaryRow.toPacket(column, this.serverConfig.encoding),
  798. );
  799. }
  800. writeTextResult(rows, columns, binary = false) {
  801. this.writeColumns(columns);
  802. rows.forEach((row) => {
  803. const arrayRow = new Array(columns.length);
  804. columns.forEach((column) => {
  805. arrayRow.push(row[column.name]);
  806. });
  807. if (binary) {
  808. this.writeBinaryRow(arrayRow);
  809. } else this.writeTextRow(arrayRow);
  810. });
  811. this.writeEof();
  812. }
  813. writeEof(warnings, statusFlags) {
  814. this.writePacket(Packets.EOF.toPacket(warnings, statusFlags));
  815. }
  816. writeOk(args) {
  817. if (!args) {
  818. args = { affectedRows: 0 };
  819. }
  820. this.writePacket(Packets.OK.toPacket(args, this.serverConfig.encoding));
  821. }
  822. writeError(args) {
  823. // if we want to send error before initial hello was sent, use default encoding
  824. const encoding = this.serverConfig ? this.serverConfig.encoding : 'cesu8';
  825. this.writePacket(Packets.Error.toPacket(args, encoding));
  826. }
  827. serverHandshake(args) {
  828. this.serverConfig = args;
  829. this.serverConfig.encoding =
  830. CharsetToEncoding[this.serverConfig.characterSet];
  831. return this.addCommand(new Commands.ServerHandshake(args));
  832. }
  833. // ===============================================================
  834. end(callback) {
  835. if (this.config.isServer) {
  836. this._closing = true;
  837. const quitCmd = new EventEmitter();
  838. setImmediate(() => {
  839. this.stream.end();
  840. quitCmd.emit('end');
  841. });
  842. return quitCmd;
  843. }
  844. // trigger error if more commands enqueued after end command
  845. const quitCmd = this.addCommand(new Commands.Quit(callback));
  846. this.addCommand = this._addCommandClosedState;
  847. return quitCmd;
  848. }
  849. static createQuery(sql, values, cb, config) {
  850. let options = {
  851. rowsAsArray: config.rowsAsArray,
  852. infileStreamFactory: config.infileStreamFactory,
  853. };
  854. if (typeof sql === 'object') {
  855. // query(options, cb)
  856. options = {
  857. ...options,
  858. ...sql,
  859. sql: sql.sql,
  860. values: sql.values,
  861. };
  862. if (typeof values === 'function') {
  863. cb = values;
  864. } else if (values !== undefined) {
  865. options.values = values;
  866. }
  867. } else if (typeof values === 'function') {
  868. // query(sql, cb)
  869. cb = values;
  870. options.sql = sql;
  871. options.values = undefined;
  872. } else {
  873. // query(sql, values, cb)
  874. options.sql = sql;
  875. options.values = values;
  876. }
  877. return new Commands.Query(options, cb);
  878. }
  879. static statementKey(options) {
  880. return `${typeof options.nestTables}/${options.nestTables}/${options.rowsAsArray}${options.sql}`;
  881. }
  882. }
  883. module.exports = BaseConnection;