index.js
4.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
// MariaSQL Client
// -------
import inherits from 'inherits';
import Client_MySQL from '../mysql';
import Promise from 'bluebird';
import * as helpers from '../../helpers';
import Transaction from './transaction';
import { assign, map } from 'lodash'
function Client_MariaSQL(config) {
Client_MySQL.call(this, config)
}
inherits(Client_MariaSQL, Client_MySQL)
assign(Client_MariaSQL.prototype, {
dialect: 'mariadb',
driverName: 'mariasql',
transaction() {
return new Transaction(this, ...arguments)
},
_driver() {
return require('mariasql')
},
// Get a raw connection, called by the `pool` whenever a new
// connection needs to be added to the pool.
acquireRawConnection() {
return new Promise((resolver, rejecter) => {
const connection = new this.driver();
connection.connect(assign({metadata: true}, this.connectionSettings))
connection
.on('ready', function() {
connection.removeAllListeners('error');
resolver(connection);
})
.on('error', rejecter);
})
},
validateConnection(connection) {
return connection.connected === true
},
// Used to explicitly close a connection, called internally by the pool
// when a connection times out or the pool is shutdown.
destroyRawConnection(connection) {
connection.removeAllListeners()
connection.end()
},
// Return the database for the MariaSQL client.
database() {
return this.connectionSettings.db;
},
// Grab a connection, run the query via the MariaSQL streaming interface,
// and pass that through to the stream we've sent back to the client.
_stream(connection, sql, stream) {
return new Promise(function(resolver, rejecter) {
connection.query(sql.sql, sql.bindings)
.on('result', function(res) {
res
.on('error', rejecter)
.on('end', function() {
resolver(res.info);
})
.on('data', function (data) {
stream.write(handleRow(data, res.info.metadata));
})
})
.on('error', rejecter);
});
},
// Runs the query on the specified connection, providing the bindings
// and any other necessary prep work.
_query(connection, obj) {
const tz = this.connectionSettings.timezone || 'local';
return new Promise((resolver, rejecter) => {
if (!obj.sql) return resolver()
const sql = this._formatQuery(obj.sql, obj.bindings, tz)
connection.query(sql, function (err, rows) {
if (err) {
return rejecter(err);
}
handleRows(rows, rows.info.metadata);
obj.response = [rows, rows.info];
resolver(obj);
})
});
},
// Process the response as returned from the query.
processResponse(obj, runner) {
const { response } = obj;
const { method } = obj;
const rows = response[0];
const data = response[1];
if (obj.output) return obj.output.call(runner, rows/*, fields*/);
switch (method) {
case 'select':
case 'pluck':
case 'first': {
const resp = helpers.skim(rows);
if (method === 'pluck') return map(resp, obj.pluck);
return method === 'first' ? resp[0] : resp;
}
case 'insert':
return [data.insertId];
case 'del':
case 'update':
case 'counter':
return parseInt(data.affectedRows, 10);
default:
return response;
}
}
})
function parseType(value, type) {
switch (type) {
case 'DATETIME':
case 'TIMESTAMP':
return new Date(value);
case 'INTEGER':
return parseInt(value, 10);
default:
return value;
}
}
function handleRow(row, metadata) {
const keys = Object.keys(metadata);
for (let i = 0; i < keys.length; i++) {
const key = keys[i];
const { type } = metadata[key];
row[key] = parseType(row[key], type);
}
return row;
}
function handleRows(rows, metadata) {
for (let i = 0; i < rows.length; i++) {
const row = rows[i];
handleRow(row, metadata);
}
return rows;
}
export default Client_MariaSQL