@databases/mysql - npm Package Compare versions

Comparing version 1.0.4 to 1.1.0

lib/index.d.ts

@@ -0,12 +1,27 @@
+/// <reference types="node" />
 import sql, { SQLQuery } from '@databases/sql';
 import { Pool, PoolConnection } from './raw';
+import { PassThrough } from 'stream';
 export { sql };
 export declare class Connection {
     private readonly conn;
-    constructor(conn: Pick<PoolConnection, 'query'>);
+    constructor(conn: Pick<PoolConnection, 'query' | 'connection'>);
     query(query: SQLQuery): Promise<any[]>;
+    queryStream(query: SQLQuery, options?: {
+        highWaterMark?: number;
+    }): AsyncGenerator<any, void, unknown>;
+    queryNodeStream(query: SQLQuery, options?: {
+        highWaterMark?: number;
+    }): NodeJS.ReadableStream;
 }
-export declare class ConnectionPool extends Connection {
+export declare class ConnectionPool {
     private readonly pool;
     constructor(pool: Pool);
+    query(query: SQLQuery): Promise<any[]>;
+    queryStream(query: SQLQuery, options?: {
+        highWaterMark?: number;
+    }): AsyncGenerator<any, void, unknown>;
+    queryNodeStream(query: SQLQuery, options?: {
+        highWaterMark?: number;
+    }): PassThrough;
     task<T>(fn: (connection: Connection) => Promise<T>): Promise<T>;
@@ -13,0 +28,0 @@ tx<T>(fn: (connection: Connection) => Promise<T>): Promise<T>;

lib/index.js

@@ -30,10 +30,93 @@ "use strict";
 };
+var __await = undefined && undefined.__await || function (v) {
+    return this instanceof __await ? (this.v = v, this) : new __await(v);
+};
+var __asyncValues = undefined && undefined.__asyncValues || function (o) {
+    if (!Symbol.asyncIterator) throw new TypeError("Symbol.asyncIterator is not defined.");
+    var m = o[Symbol.asyncIterator],
+        i;
+    return m ? m.call(o) : (o = typeof __values === "function" ? __values(o) : o[Symbol.iterator](), i = {}, verb("next"), verb("throw"), verb("return"), i[Symbol.asyncIterator] = function () {
+        return this;
+    }, i);
+    function verb(n) {
+        i[n] = o[n] && function (v) {
+            return new Promise(function (resolve, reject) {
+                v = o[n](v), settle(resolve, reject, v.done, v.value);
+            });
+        };
+    }
+    function settle(resolve, reject, d, v) {
+        Promise.resolve(v).then(function (v) {
+            resolve({ value: v, done: d });
+        }, reject);
+    }
+};
+var __asyncGenerator = undefined && undefined.__asyncGenerator || function (thisArg, _arguments, generator) {
+    if (!Symbol.asyncIterator) throw new TypeError("Symbol.asyncIterator is not defined.");
+    var g = generator.apply(thisArg, _arguments || []),
+        i,
+        q = [];
+    return i = {}, verb("next"), verb("throw"), verb("return"), i[Symbol.asyncIterator] = function () {
+        return this;
+    }, i;
+    function verb(n) {
+        if (g[n]) i[n] = function (v) {
+            return new Promise(function (a, b) {
+                q.push([n, v, a, b]) > 1 || resume(n, v);
+            });
+        };
+    }
+    function resume(n, v) {
+        try {
+            step(g[n](v));
+        } catch (e) {
+            settle(q[0][3], e);
+        }
+    }
+    function step(r) {
+        r.value instanceof __await ? Promise.resolve(r.value.v).then(fulfill, reject) : settle(q[0][2], r);
+    }
+    function fulfill(value) {
+        resume("next", value);
+    }
+    function reject(value) {
+        resume("throw", value);
+    }
+    function settle(f, v) {
+        if (f(v), q.shift(), q.length) resume(q[0][0], q[0][1]);
+    }
+};
 Object.defineProperty(exports, "__esModule", { value: true });
 const url_1 = require("url");
 const mysql_config_1 = require("@databases/mysql-config");
+const push_to_async_iterable_1 = require("@databases/push-to-async-iterable");
 const sql_1 = require("@databases/sql");
 exports.sql = sql_1.default;
 const raw_1 = require("./raw");
+const stream_1 = require("stream");
 const { codeFrameColumns } = require('@babel/code-frame');
 const { connectionStringEnvironmentVariable } = mysql_config_1.getMySqlConfigSync();
+function transformError(text, ex) {
+    // TODO: consider using https://github.com/Vincit/db-errors
+    if (ex.code === 'ER_PARSE_ERROR' && ex.sqlState === '42000' && typeof ex.sqlMessage === 'string') {
+        const match = / near \'((?:.|\n)+)\' at line (\d+)$/.exec(ex.sqlMessage);
+        if (match) {
+            const index = text.indexOf(match[1]);
+            if (index === text.lastIndexOf(match[1])) {
+                const linesUptoStart = text.substr(0, index).split('\n');
+                const line = linesUptoStart.length;
+                const start = {
+                    line,
+                    column: linesUptoStart[linesUptoStart.length - 1].length + 1
+                };
+                const linesUptoEnd = text.substr(0, index + match[1].length).split('\n');
+                const end = {
+                    line: linesUptoEnd.length,
+                    column: linesUptoEnd[linesUptoEnd.length - 1].length + 1
+                };
+                ex.message = ex.message.replace(/ near \'((?:.|\n)+)\' at line (\d+)$/, ` near:\n\n${codeFrameColumns(text, { start, end })}\n`);
+            }
+        }
+    }
+}
 class Connection {
@@ -52,23 +135,3 @@ constructor(conn) {
             } catch (ex) {
-                // TODO: consider using https://github.com/Vincit/db-errors
-                if (ex.code === 'ER_PARSE_ERROR' && ex.sqlState === '42000' && typeof ex.sqlMessage === 'string') {
-                    const match = / near \'((?:.|\n)+)\' at line (\d+)$/.exec(ex.sqlMessage);
-                    if (match) {
-                        const index = text.indexOf(match[1]);
-                        if (index === text.lastIndexOf(match[1])) {
-                            const linesUptoStart = text.substr(0, index).split('\n');
-                            const line = linesUptoStart.length;
-                            const start = {
-                                line,
-                                column: linesUptoStart[linesUptoStart.length - 1].length + 1
-                            };
-                            const linesUptoEnd = text.substr(0, index + match[1].length).split('\n');
-                            const end = {
-                                line: linesUptoEnd.length,
-                                column: linesUptoEnd[linesUptoEnd.length - 1].length + 1
-                            };
-                            ex.message = ex.message.replace(/ near \'((?:.|\n)+)\' at line (\d+)$/, ` near:\n\n${codeFrameColumns(text, { start, end })}\n`);
-                        }
-                    }
-                }
+                transformError(text, ex);
                 throw ex;
@@ -78,9 +141,119 @@ }
     }
+    queryStream(query, options) {
+        if (!(query instanceof sql_1.SQLQuery)) {
+            throw new Error('Invalid query, you must use @databases/sql to create your queries.');
+        }
+        const { text, values } = query.compileMySQL();
+        const highWaterMark = options && options.highWaterMark || 5;
+        const stream = this.conn.connection.query(text, values);
+        return push_to_async_iterable_1.default({
+            onData(fn) {
+                stream.on('result', fn);
+            },
+            onError(fn) {
+                stream.on('error', fn);
+            },
+            onEnd(fn) {
+                stream.on('end', fn);
+            },
+            pause: () => {
+                this.conn.connection.pause();
+            },
+            resume: () => {
+                this.conn.connection.resume();
+            },
+            highWaterMark
+        });
+    }
+    queryNodeStream(query, options) {
+        if (!(query instanceof sql_1.SQLQuery)) {
+            throw new Error('Invalid query, you must use @databases/sql to create your queries.');
+        }
+        const { text, values } = query.compileMySQL();
+        const result = this.conn.connection.query(text, values).stream(options);
+        // tslint:disable-next-line:no-unbound-method
+        const on = result.on;
+        const transformedExceptions = new Set();
+        return Object.assign(result, {
+            on(event, cb) {
+                if (event !== 'error') return on.call(this, event, cb);
+                return on.call(this, event, ex => {
+                    // TODO: consider using https://github.com/Vincit/db-errors
+                    if (!transformedExceptions.has(ex)) {
+                        transformedExceptions.add(ex);
+                        transformError(text, ex);
+                    }
+                    cb(ex);
+                });
+            }
+        });
+    }
 }
 exports.Connection = Connection;
-class ConnectionPool extends Connection {
+class ConnectionPool {
     constructor(pool) {
-        super(pool);
         this.pool = pool;
     }
+    query(query) {
+        return __awaiter(this, void 0, void 0, function* () {
+            if (!(query instanceof sql_1.SQLQuery)) {
+                throw new Error('Invalid query, you must use @databases/sql to create your queries.');
+            }
+            const { text, values } = query.compileMySQL();
+            try {
+                return (yield this.pool.query(text, values))[0];
+            } catch (ex) {
+                transformError(text, ex);
+                throw ex;
+            }
+        });
+    }
+    queryStream(query, options) {
+        return __asyncGenerator(this, arguments, function* queryStream_1() {
+            var e_1, _a;
+            const connection = yield __await(this.pool.getConnection());
+            const c = new Connection(connection);
+            try {
+                try {
+                    for (var _b = __asyncValues(c.queryStream(query, options)), _c; _c = yield __await(_b.next()), !_c.done;) {
+                        const record = _c.value;
+                        yield yield __await(record);
+                    }
+                } catch (e_1_1) {
+                    e_1 = { error: e_1_1 };
+                } finally {
+                    try {
+                        if (_c && !_c.done && (_a = _b.return)) yield __await(_a.call(_b));
+                    } finally {
+                        if (e_1) throw e_1.error;
+                    }
+                }
+            } finally {
+                connection.release();
+            }
+        });
+    }
+    queryNodeStream(query, options) {
+        const stream = new stream_1.PassThrough({ objectMode: true, highWaterMark: 2 });
+        this.pool.getConnection().then(connection => {
+            const c = new Connection(connection);
+            let released = false;
+            return c.queryNodeStream(query, options).on('fields', fields => {
+                stream.emit('fields', fields);
+            }).on('error', err => {
+                if (!released) {
+                    released = true;
+                    connection.release();
+                }
+                stream.emit('error', err);
+            }).on('end', () => {
+                if (!released) {
+                    released = true;
+                    connection.release();
+                }
+                stream.emit('end');
+            }).pipe(stream);
+        }).catch(ex => stream.emit('error', ex));
+        return stream;
+    }
     task(fn) {
@@ -87,0 +260,0 @@ return __awaiter(this, void 0, void 0, function* () {

lib/raw.d.ts

@@ -0,5 +1,29 @@
+/// <reference types="node" />
+import { ReadableOptions } from 'stream';
+export interface ColumnDefinition {
+    characterSet: number;
+    encoding: string;
+    name: string;
+    columnLength: number;
+    columnType: number;
+    flags: number;
+    decimals: number;
+}
+export interface QueryCmd {
+    on(event: 'result', fn: (row: any) => void): this;
+    on(event: 'error', fn: (err: any) => void): this;
+    on(event: 'end', fn: () => void): this;
+    on(event: 'fields', fn: (fields: undefined | ColumnDefinition[]) => void): this;
+    stream(options?: ReadableOptions): NodeJS.ReadableStream;
+}
+export interface CoreConnection {
+    query(sql: string, args: any[]): QueryCmd;
+    pause(): void;
+    resume(): void;
+}
 export interface PoolConnection {
+    readonly connection: CoreConnection;
     release(): void;
     destroy(): void;
-    query(sql: string, args: any[]): Promise<[unknown[], unknown[]]>;
+    query(sql: string, args: any[]): Promise<[unknown[], undefined | ColumnDefinition[]]>;
     execute(sql: string, args: any[]): Promise<unknown>;
@@ -13,3 +37,3 @@ beginTransaction(): Promise<void>;
     getConnection(): Promise<PoolConnection>;
-    query(sql: string, args: any[]): Promise<[unknown[], unknown[]]>;
+    query(sql: string, args: any[]): Promise<[unknown[], undefined | ColumnDefinition[]]>;
     execute(sql: string, args: any[]): Promise<unknown>;
@@ -16,0 +40,0 @@ end(): Promise<void>;

package.json

 {
   "name": "@databases/mysql",
-  "version": "1.0.4",
+  "version": "1.1.0",
   "description": "",
@@ -10,2 +10,3 @@ "main": "./lib/index.js",
     "@databases/mysql-config": "^1.0.1",
+    "@databases/push-to-async-iterable": "^1.0.0",
     "@databases/sql": "^1.0.3",
@@ -12,0 +13,0 @@ "@types/mysql": "^2.15.5",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

34053

increased by95.46%

Number of package files

15

increased by25%

Lines of code

581

increased by115.19%

Worsened metrics

Dependency count

6

increased by20%

Dependency changes

• + Added@databases/push-to-async-iterable@^1.0.0

• + Added@databases/push-to-async-iterable@1.0.0(transitive)

• + Addedasap@1.0.0(transitive)

• + Addedpromise@6.1.0(transitive)

• + Addedthen-queue@1.3.0(transitive)