promise.js 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584
  1. 'use strict';
  2. const core = require('./index.js');
  3. const EventEmitter = require('events').EventEmitter;
  4. const parserCache = require('./lib/parsers/parser_cache.js');
  5. function makeDoneCb(resolve, reject, localErr) {
  6. return function (err, rows, fields) {
  7. if (err) {
  8. localErr.message = err.message;
  9. localErr.code = err.code;
  10. localErr.errno = err.errno;
  11. localErr.sql = err.sql;
  12. localErr.sqlState = err.sqlState;
  13. localErr.sqlMessage = err.sqlMessage;
  14. reject(localErr);
  15. } else {
  16. resolve([rows, fields]);
  17. }
  18. };
  19. }
  20. function inheritEvents(source, target, events) {
  21. const listeners = {};
  22. target
  23. .on('newListener', eventName => {
  24. if (events.indexOf(eventName) >= 0 && !target.listenerCount(eventName)) {
  25. source.on(
  26. eventName,
  27. (listeners[eventName] = function () {
  28. const args = [].slice.call(arguments);
  29. args.unshift(eventName);
  30. target.emit.apply(target, args);
  31. })
  32. );
  33. }
  34. })
  35. .on('removeListener', eventName => {
  36. if (events.indexOf(eventName) >= 0 && !target.listenerCount(eventName)) {
  37. source.removeListener(eventName, listeners[eventName]);
  38. delete listeners[eventName];
  39. }
  40. });
  41. }
  42. class PromisePreparedStatementInfo {
  43. constructor(statement, promiseImpl) {
  44. this.statement = statement;
  45. this.Promise = promiseImpl;
  46. }
  47. execute(parameters) {
  48. const s = this.statement;
  49. const localErr = new Error();
  50. return new this.Promise((resolve, reject) => {
  51. const done = makeDoneCb(resolve, reject, localErr);
  52. if (parameters) {
  53. s.execute(parameters, done);
  54. } else {
  55. s.execute(done);
  56. }
  57. });
  58. }
  59. close() {
  60. return new this.Promise(resolve => {
  61. this.statement.close();
  62. resolve();
  63. });
  64. }
  65. }
  66. class PromiseConnection extends EventEmitter {
  67. constructor(connection, promiseImpl) {
  68. super();
  69. this.connection = connection;
  70. this.Promise = promiseImpl || Promise;
  71. inheritEvents(connection, this, [
  72. 'error',
  73. 'drain',
  74. 'connect',
  75. 'end',
  76. 'enqueue'
  77. ]);
  78. }
  79. release() {
  80. this.connection.release();
  81. }
  82. query(query, params) {
  83. const c = this.connection;
  84. const localErr = new Error();
  85. if (typeof params === 'function') {
  86. throw new Error(
  87. 'Callback function is not available with promise clients.'
  88. );
  89. }
  90. return new this.Promise((resolve, reject) => {
  91. const done = makeDoneCb(resolve, reject, localErr);
  92. if (params !== undefined) {
  93. c.query(query, params, done);
  94. } else {
  95. c.query(query, done);
  96. }
  97. });
  98. }
  99. execute(query, params) {
  100. const c = this.connection;
  101. const localErr = new Error();
  102. if (typeof params === 'function') {
  103. throw new Error(
  104. 'Callback function is not available with promise clients.'
  105. );
  106. }
  107. return new this.Promise((resolve, reject) => {
  108. const done = makeDoneCb(resolve, reject, localErr);
  109. if (params !== undefined) {
  110. c.execute(query, params, done);
  111. } else {
  112. c.execute(query, done);
  113. }
  114. });
  115. }
  116. end() {
  117. return new this.Promise(resolve => {
  118. this.connection.end(resolve);
  119. });
  120. }
  121. beginTransaction() {
  122. const c = this.connection;
  123. const localErr = new Error();
  124. return new this.Promise((resolve, reject) => {
  125. const done = makeDoneCb(resolve, reject, localErr);
  126. c.beginTransaction(done);
  127. });
  128. }
  129. commit() {
  130. const c = this.connection;
  131. const localErr = new Error();
  132. return new this.Promise((resolve, reject) => {
  133. const done = makeDoneCb(resolve, reject, localErr);
  134. c.commit(done);
  135. });
  136. }
  137. rollback() {
  138. const c = this.connection;
  139. const localErr = new Error();
  140. return new this.Promise((resolve, reject) => {
  141. const done = makeDoneCb(resolve, reject, localErr);
  142. c.rollback(done);
  143. });
  144. }
  145. ping() {
  146. const c = this.connection;
  147. const localErr = new Error();
  148. return new this.Promise((resolve, reject) => {
  149. c.ping(err => {
  150. if (err) {
  151. localErr.message = err.message;
  152. localErr.code = err.code;
  153. localErr.errno = err.errno;
  154. localErr.sqlState = err.sqlState;
  155. localErr.sqlMessage = err.sqlMessage;
  156. reject(localErr);
  157. } else {
  158. resolve(true);
  159. }
  160. });
  161. });
  162. }
  163. connect() {
  164. const c = this.connection;
  165. const localErr = new Error();
  166. return new this.Promise((resolve, reject) => {
  167. c.connect((err, param) => {
  168. if (err) {
  169. localErr.message = err.message;
  170. localErr.code = err.code;
  171. localErr.errno = err.errno;
  172. localErr.sqlState = err.sqlState;
  173. localErr.sqlMessage = err.sqlMessage;
  174. reject(localErr);
  175. } else {
  176. resolve(param);
  177. }
  178. });
  179. });
  180. }
  181. prepare(options) {
  182. const c = this.connection;
  183. const promiseImpl = this.Promise;
  184. const localErr = new Error();
  185. return new this.Promise((resolve, reject) => {
  186. c.prepare(options, (err, statement) => {
  187. if (err) {
  188. localErr.message = err.message;
  189. localErr.code = err.code;
  190. localErr.errno = err.errno;
  191. localErr.sqlState = err.sqlState;
  192. localErr.sqlMessage = err.sqlMessage;
  193. reject(localErr);
  194. } else {
  195. const wrappedStatement = new PromisePreparedStatementInfo(
  196. statement,
  197. promiseImpl
  198. );
  199. resolve(wrappedStatement);
  200. }
  201. });
  202. });
  203. }
  204. changeUser(options) {
  205. const c = this.connection;
  206. const localErr = new Error();
  207. return new this.Promise((resolve, reject) => {
  208. c.changeUser(options, err => {
  209. if (err) {
  210. localErr.message = err.message;
  211. localErr.code = err.code;
  212. localErr.errno = err.errno;
  213. localErr.sqlState = err.sqlState;
  214. localErr.sqlMessage = err.sqlMessage;
  215. reject(localErr);
  216. } else {
  217. resolve();
  218. }
  219. });
  220. });
  221. }
  222. get config() {
  223. return this.connection.config;
  224. }
  225. get threadId() {
  226. return this.connection.threadId;
  227. }
  228. }
  229. function createConnection(opts) {
  230. const coreConnection = core.createConnection(opts);
  231. const createConnectionErr = new Error();
  232. const thePromise = opts.Promise || Promise;
  233. if (!thePromise) {
  234. throw new Error(
  235. 'no Promise implementation available.' +
  236. 'Use promise-enabled node version or pass userland Promise' +
  237. " implementation as parameter, for example: { Promise: require('bluebird') }"
  238. );
  239. }
  240. return new thePromise((resolve, reject) => {
  241. coreConnection.once('connect', () => {
  242. resolve(new PromiseConnection(coreConnection, thePromise));
  243. });
  244. coreConnection.once('error', err => {
  245. createConnectionErr.message = err.message;
  246. createConnectionErr.code = err.code;
  247. createConnectionErr.errno = err.errno;
  248. createConnectionErr.sqlState = err.sqlState;
  249. reject(createConnectionErr);
  250. });
  251. });
  252. }
  253. // note: the callback of "changeUser" is not called on success
  254. // hence there is no possibility to call "resolve"
  255. // patching PromiseConnection
  256. // create facade functions for prototype functions on "Connection" that are not yet
  257. // implemented with PromiseConnection
  258. // proxy synchronous functions only
  259. (function (functionsToWrap) {
  260. for (let i = 0; functionsToWrap && i < functionsToWrap.length; i++) {
  261. const func = functionsToWrap[i];
  262. if (
  263. typeof core.Connection.prototype[func] === 'function' &&
  264. PromiseConnection.prototype[func] === undefined
  265. ) {
  266. PromiseConnection.prototype[func] = (function factory(funcName) {
  267. return function () {
  268. return core.Connection.prototype[funcName].apply(
  269. this.connection,
  270. arguments
  271. );
  272. };
  273. })(func);
  274. }
  275. }
  276. })([
  277. // synchronous functions
  278. 'close',
  279. 'createBinlogStream',
  280. 'destroy',
  281. 'escape',
  282. 'escapeId',
  283. 'format',
  284. 'pause',
  285. 'pipe',
  286. 'resume',
  287. 'unprepare'
  288. ]);
  289. class PromisePoolConnection extends PromiseConnection {
  290. constructor(connection, promiseImpl) {
  291. super(connection, promiseImpl);
  292. }
  293. destroy() {
  294. return core.PoolConnection.prototype.destroy.apply(
  295. this.connection,
  296. arguments
  297. );
  298. }
  299. }
  300. class PromisePool extends EventEmitter {
  301. constructor(pool, thePromise) {
  302. super();
  303. this.pool = pool;
  304. this.Promise = thePromise || Promise;
  305. inheritEvents(pool, this, ['acquire', 'connection', 'enqueue', 'release']);
  306. }
  307. getConnection() {
  308. const corePool = this.pool;
  309. return new this.Promise((resolve, reject) => {
  310. corePool.getConnection((err, coreConnection) => {
  311. if (err) {
  312. reject(err);
  313. } else {
  314. resolve(new PromisePoolConnection(coreConnection, this.Promise));
  315. }
  316. });
  317. });
  318. }
  319. releaseConnection(connection) {
  320. if (connection instanceof PromisePoolConnection) connection.release();
  321. }
  322. query(sql, args) {
  323. const corePool = this.pool;
  324. const localErr = new Error();
  325. if (typeof args === 'function') {
  326. throw new Error(
  327. 'Callback function is not available with promise clients.'
  328. );
  329. }
  330. return new this.Promise((resolve, reject) => {
  331. const done = makeDoneCb(resolve, reject, localErr);
  332. if (args !== undefined) {
  333. corePool.query(sql, args, done);
  334. } else {
  335. corePool.query(sql, done);
  336. }
  337. });
  338. }
  339. execute(sql, args) {
  340. const corePool = this.pool;
  341. const localErr = new Error();
  342. if (typeof args === 'function') {
  343. throw new Error(
  344. 'Callback function is not available with promise clients.'
  345. );
  346. }
  347. return new this.Promise((resolve, reject) => {
  348. const done = makeDoneCb(resolve, reject, localErr);
  349. if (args) {
  350. corePool.execute(sql, args, done);
  351. } else {
  352. corePool.execute(sql, done);
  353. }
  354. });
  355. }
  356. end() {
  357. const corePool = this.pool;
  358. const localErr = new Error();
  359. return new this.Promise((resolve, reject) => {
  360. corePool.end(err => {
  361. if (err) {
  362. localErr.message = err.message;
  363. localErr.code = err.code;
  364. localErr.errno = err.errno;
  365. localErr.sqlState = err.sqlState;
  366. localErr.sqlMessage = err.sqlMessage;
  367. reject(localErr);
  368. } else {
  369. resolve();
  370. }
  371. });
  372. });
  373. }
  374. }
  375. function createPool(opts) {
  376. const corePool = core.createPool(opts);
  377. const thePromise = opts.Promise || Promise;
  378. if (!thePromise) {
  379. throw new Error(
  380. 'no Promise implementation available.' +
  381. 'Use promise-enabled node version or pass userland Promise' +
  382. " implementation as parameter, for example: { Promise: require('bluebird') }"
  383. );
  384. }
  385. return new PromisePool(corePool, thePromise);
  386. }
  387. (function (functionsToWrap) {
  388. for (let i = 0; functionsToWrap && i < functionsToWrap.length; i++) {
  389. const func = functionsToWrap[i];
  390. if (
  391. typeof core.Pool.prototype[func] === 'function' &&
  392. PromisePool.prototype[func] === undefined
  393. ) {
  394. PromisePool.prototype[func] = (function factory(funcName) {
  395. return function () {
  396. return core.Pool.prototype[funcName].apply(this.pool, arguments);
  397. };
  398. })(func);
  399. }
  400. }
  401. })([
  402. // synchronous functions
  403. 'escape',
  404. 'escapeId',
  405. 'format'
  406. ]);
  407. class PromisePoolCluster extends EventEmitter {
  408. constructor(poolCluster, thePromise) {
  409. super();
  410. this.poolCluster = poolCluster;
  411. this.Promise = thePromise || Promise;
  412. inheritEvents(poolCluster, this, ['warn', 'remove']);
  413. }
  414. getConnection(pattern, selector) {
  415. const corePoolCluster = this.poolCluster;
  416. return new this.Promise((resolve, reject) => {
  417. corePoolCluster.getConnection(pattern, selector, (err, coreConnection) => {
  418. if (err) {
  419. reject(err);
  420. } else {
  421. resolve(new PromisePoolConnection(coreConnection, this.Promise));
  422. }
  423. });
  424. });
  425. }
  426. query(sql, args) {
  427. const corePoolCluster = this.poolCluster;
  428. const localErr = new Error();
  429. if (typeof args === 'function') {
  430. throw new Error(
  431. 'Callback function is not available with promise clients.'
  432. );
  433. }
  434. return new this.Promise((resolve, reject) => {
  435. const done = makeDoneCb(resolve, reject, localErr);
  436. corePoolCluster.query(sql, args, done);
  437. });
  438. }
  439. execute(sql, args) {
  440. const corePoolCluster = this.poolCluster;
  441. const localErr = new Error();
  442. if (typeof args === 'function') {
  443. throw new Error(
  444. 'Callback function is not available with promise clients.'
  445. );
  446. }
  447. return new this.Promise((resolve, reject) => {
  448. const done = makeDoneCb(resolve, reject, localErr);
  449. corePoolCluster.execute(sql, args, done);
  450. });
  451. }
  452. of(pattern, selector) {
  453. return new PromisePoolCluster(
  454. this.poolCluster.of(pattern, selector),
  455. this.Promise
  456. );
  457. }
  458. end() {
  459. const corePoolCluster = this.poolCluster;
  460. const localErr = new Error();
  461. return new this.Promise((resolve, reject) => {
  462. corePoolCluster.end(err => {
  463. if (err) {
  464. localErr.message = err.message;
  465. localErr.code = err.code;
  466. localErr.errno = err.errno;
  467. localErr.sqlState = err.sqlState;
  468. localErr.sqlMessage = err.sqlMessage;
  469. reject(localErr);
  470. } else {
  471. resolve();
  472. }
  473. });
  474. });
  475. }
  476. }
  477. /**
  478. * proxy poolCluster synchronous functions
  479. */
  480. (function (functionsToWrap) {
  481. for (let i = 0; functionsToWrap && i < functionsToWrap.length; i++) {
  482. const func = functionsToWrap[i];
  483. if (
  484. typeof core.PoolCluster.prototype[func] === 'function' &&
  485. PromisePoolCluster.prototype[func] === undefined
  486. ) {
  487. PromisePoolCluster.prototype[func] = (function factory(funcName) {
  488. return function () {
  489. return core.PoolCluster.prototype[funcName].apply(this.poolCluster, arguments);
  490. };
  491. })(func);
  492. }
  493. }
  494. })([
  495. 'add'
  496. ]);
  497. function createPoolCluster(opts) {
  498. const corePoolCluster = core.createPoolCluster(opts);
  499. const thePromise = (opts && opts.Promise) || Promise;
  500. if (!thePromise) {
  501. throw new Error(
  502. 'no Promise implementation available.' +
  503. 'Use promise-enabled node version or pass userland Promise' +
  504. " implementation as parameter, for example: { Promise: require('bluebird') }"
  505. );
  506. }
  507. return new PromisePoolCluster(corePoolCluster, thePromise);
  508. }
  509. exports.createConnection = createConnection;
  510. exports.createPool = createPool;
  511. exports.createPoolCluster = createPoolCluster;
  512. exports.escape = core.escape;
  513. exports.escapeId = core.escapeId;
  514. exports.format = core.format;
  515. exports.raw = core.raw;
  516. exports.PromisePool = PromisePool;
  517. exports.PromiseConnection = PromiseConnection;
  518. exports.PromisePoolConnection = PromisePoolConnection;
  519. exports.__defineGetter__('Types', () => require('./lib/constants/types.js'));
  520. exports.__defineGetter__('Charsets', () =>
  521. require('./lib/constants/charsets.js')
  522. );
  523. exports.__defineGetter__('CharsetToEncoding', () =>
  524. require('./lib/constants/charset_encodings.js')
  525. );
  526. exports.setMaxParserCache = function(max) {
  527. parserCache.setMaxCache(max);
  528. };
  529. exports.clearParserCache = function() {
  530. parserCache.clearCache();
  531. };