123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
import { Client } from "./client.ts";import { log } from "./logger.ts";import { buildAuth } from "./packets/builders/auth.ts";import { buildQuery } from "./packets/builders/query.ts";import { ReceivePacket, SendPacket } from "./packets/packet.ts";import { parseError } from "./packets/parsers/err.ts";import { parseHandshake } from "./packets/parsers/handshake.ts";import { FieldInfo, parseField, parseRow } from "./packets/parsers/result.ts";
/** * Connection state */export enum ConnectionState {  CONNECTING,  CONNECTED,  COLSING,  CLOSED}
/** * Result for excute sql */export type ExecuteResult = {  affectedRows?: number;  lastInsertId?: number;  fields?: FieldInfo[];  rows?: any[];};
/** Connection for mysql */export class Connection {  state: ConnectionState = ConnectionState.CONNECTING;  capabilities: number = 0;
  private conn: Deno.Conn;
  constructor(readonly client: Client) {}
  private async _connect() {    const { hostname, port } = this.client.config;    log.info(`connecting ${hostname}:${port}`);    this.conn = await Deno.dial("tcp", `${hostname}:${port}`);
    let receive = await this.nextPacket();    const handshakePacket = parseHandshake(receive.body);    const data = buildAuth(handshakePacket, {      username: this.client.config.username,      password: this.client.config.password,      db: this.client.config.db    });    await new SendPacket(data, 0x1).send(this.conn);    this.state = ConnectionState.CONNECTING;    this.capabilities = handshakePacket.serverCapabilities;
    receive = await this.nextPacket();    const header = receive.body.readUint8();    if (header === 0xff) {      const error = parseError(receive.body, this);      log.error(`connect error(${error.code}): ${error.message}`);      this.close();      throw new Error(error.message);    } else {      log.info(`connected to ${this.client.config.hostname}`);      this.state = ConnectionState.CONNECTED;    }  }
  /** Connect to database */  async connect(): Promise<void> {    let { retry = 3, timeout = 10000 } = this.client.config;    let timer = 0;    while (retry--) {      try {        await Promise.race([          this._connect().then(() => clearTimeout(timer)),          new Promise(            (_, reject) =>              (timer = setTimeout(() => {                this.conn && this.conn.close();                reject(new Error("connect timeout"));              }, timeout))          )        ]);        if (this.state == ConnectionState.CONNECTED) {          break;        }        break;      } catch (err) {        log.error(err.message);        log.info(`retrying ${retry}`);      }    }    if (this.state !== ConnectionState.CONNECTED) {      throw new Error("connect fail");    }  }
  private async nextPacket(): Promise<ReceivePacket> {    while (this.conn) {      const packet = await new ReceivePacket().parse(this.conn);      if (packet) {        if (packet.type === "ERR") {          packet.body.skip(1);          const error = parseError(packet.body, this);          throw new Error(error.message);        }        return packet;      }    }  }
  /** Close database connection */  close(): void {    log.info("close connection");    this.state = ConnectionState.COLSING;    this.conn && this.conn.close();    this.state = ConnectionState.CLOSED;  }
  /**   * excute query sql   * @param sql query sql string   * @param params query params   */  async query(sql: string, params?: any[]): Promise<ExecuteResult | any[]> {    const result = await this.execute(sql, params);    if (result && result.rows) {      return result.rows;    } else {      return result;    }  }
  /**   * excute sql   * @param sql sql string   * @param params query params   */  async execute(sql: string, params?: any[]): Promise<ExecuteResult> {    const data = buildQuery(sql, params);    await new SendPacket(data, 0).send(this.conn);    let receive = await this.nextPacket();    if (receive.type === "OK") {      receive.body.skip(1);      return {        affectedRows: receive.body.readEncodedLen(),        lastInsertId: receive.body.readEncodedLen()      };    }    let fieldCount = receive.body.readEncodedLen();    const fields: FieldInfo[] = [];    while (fieldCount--) {      const packet = await this.nextPacket();      const field = parseField(packet.body);      fields.push(field);    }
    const rows = [];    receive = await this.nextPacket(); // EOF(less than 5.7)    if (receive.type == "RESULT") {      const row = parseRow(receive.body, fields);      rows.push(row);    }    while (true) {      receive = await this.nextPacket();      if (receive.type === "EOF") {        break;      } else {        const row = parseRow(receive.body, fields);        rows.push(row);      }    }    return { rows, fields };  }}