Merge pull request #1610 from jmordica/master

Types, tests, and commands Oh my! 🌪️

Author: sidorares

index.d.ts

@@ -1,7 +1,7 @@
 import {
   Connection as PromiseConnection,
   Pool as PromisePool,
-  PoolConnection as PromisePoolConnection
+  PoolConnection as PromisePoolConnection,
 } from './promise';
 
 import * as mysql from './typings/mysql';
@@ -74,6 +74,13 @@ export interface Connection extends mysql.Connection {
   promise(promiseImpl?: PromiseConstructor): PromiseConnection;
   unprepare(sql: string): mysql.PrepareStatementInfo;
   prepare(sql: string, callback?: (err: mysql.QueryError | null, statement: mysql.PrepareStatementInfo) => any): mysql.Prepare;
+  serverHandshake(args: any): any;
+  writeOk(args?: mysql.OkPacketParams): void;
+  writeError(args?: mysql.ErrorPacketParams): void;
+  writeEof(warnings?: number, statusFlags?: number): void;
+  writeTextResult(rows?: Array<any>, columns?: Array<any>): void;
+  writePacket(packet: any): void;
+  sequenceId: number;
 }
 
 export interface PoolConnection extends mysql.PoolConnection, Connection {
@@ -153,6 +160,8 @@ export interface Pool extends mysql.Connection {
   promise(promiseImpl?: PromiseConstructor): PromisePool;
   unprepare(sql: string): mysql.PrepareStatementInfo;
   prepare(sql: string, callback?: (err: mysql.QueryError | null, statement: mysql.PrepareStatementInfo) => any): mysql.Prepare;
+
+  config: mysql.PoolOptions;
 }
 
 type authPlugins = (pluginMetadata: {

lib/commands/prepare.js

@@ -55,7 +55,8 @@ class Prepare extends Command {
     }
     const cmdPacket = new Packets.PrepareStatement(
       this.query,
-      connection.config.charsetNumber
+      connection.config.charsetNumber,
+      this.options.values
     );
     connection.writePacket(cmdPacket.toPacket(1));
     return Prepare.prototype.prepareHeader;

lib/commands/server_handshake.js

@@ -69,12 +69,41 @@ class ServerHandshake extends Command {
     return ServerHandshake.prototype.dispatchCommands;
   }
 
+  _isStatement(query, name) {
+    const firstWord = query.split(' ')[0].toUpperCase();
+    return firstWord === name;
+  }
+
   dispatchCommands(packet, connection) {
     // command from client to server
     let knownCommand = true;
     const encoding = connection.clientHelloReply.encoding;
     const commandCode = packet.readInt8();
     switch (commandCode) {
+      case CommandCode.STMT_PREPARE:
+        if (connection.listeners('stmt_prepare').length) {
+          const query = packet.readString(undefined, encoding);
+          connection.emit('stmt_prepare', query);
+        } else {
+          connection.writeError({
+            code: Errors.HA_ERR_INTERNAL_ERROR,
+            message:
+              'No query handler for prepared statements.'
+          });
+        }
+        break;
+      case CommandCode.STMT_EXECUTE:
+        if (connection.listeners('stmt_execute').length) {
+          const { stmtId, flags, iterationCount, values } = Packets.Execute.fromPacket(packet, encoding);
+          connection.emit('stmt_execute', stmtId, flags, iterationCount, values);
+        } else {
+          connection.writeError({
+            code: Errors.HA_ERR_INTERNAL_ERROR,
+            message:
+              'No query handler for execute statements.'
+          });
+        }
+        break;
       case CommandCode.QUIT:
         if (connection.listeners('quit').length) {
           connection.emit('quit');
@@ -93,7 +122,13 @@ class ServerHandshake extends Command {
       case CommandCode.QUERY:
         if (connection.listeners('query').length) {
           const query = packet.readString(undefined, encoding);
-          connection.emit('query', query);
+          if (this._isStatement(query, 'PREPARE') || this._isStatement(query, 'SET')) {
+            connection.emit('stmt_prepare', query);
+          }
+          else if (this._isStatement(query, 'EXECUTE')) {
+            connection.emit('stmt_execute', null, null, null, null, query);
+          }
+          else connection.emit('query', query);
         } else {
           connection.writeError({
             code: Errors.HA_ERR_INTERNAL_ERROR,

lib/packets/execute.js

@@ -81,6 +81,68 @@ class Execute {
     this.timezone = timezone;
   }
 
+  static fromPacket(packet, encoding) {
+    const stmtId = packet.readInt32();
+    const flags = packet.readInt8();
+    const iterationCount = packet.readInt32();
+
+    let i = packet.offset;
+    while (i < packet.end - 1) {
+      if((packet.buffer[i+1] === Types.VAR_STRING 
+        || packet.buffer[i+1] === Types.NULL 
+        || packet.buffer[i+1] === Types.DOUBLE
+        || packet.buffer[i+1] === Types.TINY
+        || packet.buffer[i+1] === Types.DATETIME
+        || packet.buffer[i+1] === Types.JSON) && packet.buffer[i] === 1 && packet.buffer[i+2] === 0) {
+        break;
+      }
+      else {
+        packet.readInt8()
+      }
+      i++;
+    }
+
+    const types = [];
+
+    for(let i = packet.offset + 1; i < packet.end - 1; i++) {
+      if((packet.buffer[i] === Types.VAR_STRING 
+        || packet.buffer[i] === Types.NULL 
+        || packet.buffer[i] === Types.DOUBLE
+        || packet.buffer[i] === Types.TINY
+        || packet.buffer[i] === Types.DATETIME
+        || packet.buffer[i] === Types.JSON) && packet.buffer[i + 1] === 0) {
+        types.push(packet.buffer[i]);
+        packet.skip(2);
+      }
+    }
+
+    packet.skip(1);
+
+    const values = [];
+    for(let i = 0; i < types.length; i++) {
+      if(types[i] === Types.VAR_STRING) {
+        values.push(packet.readLengthCodedString(encoding))
+      }
+      else if(types[i] === Types.DOUBLE) {
+        values.push(packet.readDouble())
+      }
+      else if(types[i] === Types.TINY) {
+        values.push(packet.readInt8())
+      }
+      else if(types[i] === Types.DATETIME) {
+        values.push(packet.readDateTime())
+      }
+      else if(types[i] === Types.JSON) {
+        values.push(JSON.parse(packet.readLengthCodedString(encoding)))
+      }
+      if(types[i] === Types.NULL) {
+        values.push(null)
+      }
+    }
+
+    return { stmtId, flags, iterationCount, values };
+  }
+
   toPacket() {
     // TODO: don't try to calculate packet length in advance, allocate some big buffer in advance (header + 256 bytes?)
     // and copy + reallocate if not enough

package.json

@@ -19,7 +19,7 @@
     "precommit": "lint-staged",
     "eslint-check": "eslint --print-config .eslintrc | eslint-config-prettier-check",
     "wait-port": "wait-on",
-    "type-test": "node ./node_modules/typescript/bin/tsc -p tests.json && mocha typings/test"
+    "type-test": "node ./node_modules/typescript/bin/tsc -p tests.json && mocha typings/test --timeout 10000"
   },
   "lint-staged": {
     "*.js": [

promise.d.ts

@@ -5,7 +5,8 @@ import {
   FieldPacket,
   QueryOptions,
   ConnectionOptions,
-  PoolOptions
+  PoolOptions,
+  Pool as CorePool
 } from './index';
 
 import { EventEmitter } from 'events';
@@ -131,6 +132,13 @@ export interface Pool extends EventEmitter {
   on(event: 'release', listener: (connection: PoolConnection) => any): this;
   on(event: 'enqueue', listener: () => any): this;
   end(): Promise<void>;
+
+  escape(value: any): string;
+  escapeId(value: string): string;
+  escapeId(values: string[]): string;
+  format(sql: string, values?: any | any[] | { [param: string]: any }): string;
+
+  pool: CorePool;
 }
 
 export function createConnection(connectionUri: string): Promise<Connection>;
@@ -143,3 +151,7 @@ export interface PreparedStatementInfo {
   close(): Promise<void>;
   execute(parameters: any[]): Promise<[RowDataPacket[][] | RowDataPacket[] | OkPacket | OkPacket[] | ResultSetHeader, FieldPacket[]]>;
 }
+
+export interface PromisePoolConnection extends Connection {
+  destroy(): any;
+} 

tests.json

@@ -6,7 +6,9 @@
 		"listEmittedFiles": false
 	},
 	"files": [
-		"typings/test/server.ts"
+		"typings/test/server.ts",
+		"typings/test/connection.ts",
+		"typings/test/pool.ts"
 	],
 	"typeRoots": [
 		"./typings",

typings/mysql/index.d.ts

@@ -14,7 +14,7 @@ import Server = require('./lib/Server');
 
 export function createConnection(connectionUri: string): Connection;
 export function createConnection(config: BaseConnection.ConnectionOptions): Connection;
-export function createPool(config: BasePool.PoolOptions): Pool;
+export function createPool(config: BasePool.PoolOptions): BasePool;
 export function createPoolCluster(config?: BasePoolCluster.PoolClusterOptions): PoolCluster;
 export function escape(value: any): string;
 export function escapeId(value: any): string;
@@ -24,6 +24,7 @@ export function format(sql: string, values: any, stringifyObjects?: boolean, tim
 export function raw(sql: string): {
     toSqlString: () => string
 };
+export function createServer(handler: (conn: BaseConnection) => any): Server;
 
 export {
     ConnectionOptions,
@@ -42,6 +43,4 @@ export interface PoolConnection extends BasePoolConnection {}
 export interface Pool extends BasePool {}
 export interface PoolCluster extends BasePoolCluster {}
 export interface Query extends BaseQuery {}
-export interface Prepare extends BasePrepare {}
-
-export function createServer(handler: (conn: BaseConnection) => any): Server;
+export interface Prepare extends BasePrepare {}

typings/mysql/lib/Connection.d.ts

@@ -271,6 +271,12 @@ declare class Connection extends EventEmitter {
     on(event: string, listener: Function): this;
 
     rollback(callback: (err: Query.QueryError | null) => void): void;
+
+    execute(sql: string, values: Array<any>, cb: (err: any, rows: Array<any>, fields: Array<any>) => any): any;
+
+    unprepare(sql: string): any;
+
+    serverHandshake(args: any): any;
 }
 
 export = Connection;

typings/mysql/lib/Pool.d.ts

@@ -60,6 +60,8 @@ declare class Pool extends EventEmitter {
 
     on(event: string, listener: Function): this;
     on(event: 'connection', listener: (connection: PoolConnection) => any): this;
+
+    promise(promiseImpl?: any): any;
 }
 
 export = Pool;