Skip to content

Commit 58a6a94

Browse files
porsagerporsager
committed
Build cjs + deno
1 parent b6080d4 commit 58a6a94

File tree

13 files changed

+465
-186
lines changed

13 files changed

+465
-186
lines changed

‎cjs/src/connection.js

+7-24
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ const tls = require('tls')
33
const crypto = require('crypto')
44
const Stream = require('stream')
55

6-
const { Identifier, Builder, handleValue, arrayParser, arraySerializer } = require('./types.js')
6+
const { stringify, handleValue, arrayParser, arraySerializer } = require('./types.js')
77
const { Errors } = require('./errors.js')
88
const Result = require('./result.js')
99
const Queue = require('./queue.js')
@@ -218,9 +218,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
218218
const parameters = []
219219
, types = []
220220

221-
const string = stringify(q, q.strings[0], q.args[0], parameters, types)
221+
const string = stringify(q, q.strings[0], q.args[0], parameters, types, options)
222222

223-
!q.tagged && q.args.forEach(x => handleValue(x, parameters, types))
223+
!q.tagged && q.args.forEach(x => handleValue(x, parameters, types, options))
224224

225225
q.prepare = options.prepare && ('prepare' in q.options ? q.options.prepare : true)
226226
q.string = string
@@ -236,25 +236,6 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
236236
typeof options.debug === 'function' && options.debug(id, string, parameters, types)
237237
}
238238

239-
function stringify(q, string, value, parameters, types) {
240-
for (let i = 1; i < q.strings.length; i++) {
241-
string += (
242-
value instanceof Query ? fragment(string, value, parameters, types) :
243-
value instanceof Identifier ? value.value :
244-
value instanceof Builder ? value.build(string, parameters, types, options.transform) :
245-
handleValue(value, parameters, types)
246-
) + q.strings[i]
247-
value = q.args[i]
248-
}
249-
250-
return string
251-
}
252-
253-
function fragment(string, q, parameters, types) {
254-
q.fragment = true
255-
return stringify(q, q.strings[0], q.args[0], parameters, types)
256-
}
257-
258239
function write(x, fn) {
259240
chunk = chunk ? Buffer.concat([chunk, x]) : Buffer.from(x)
260241
if (fn || chunk.length >= 1024)
@@ -497,7 +478,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
497478

498479
value = length === -1
499480
? null
500-
: query.isRaw
481+
: query.isRaw === true
501482
? x.slice(index, index += length)
502483
: column.parser === undefined
503484
? x.toString('utf8', index, index += length)
@@ -506,7 +487,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
506487
: column.parser(x.toString('utf8', index, index += length))
507488

508489
query.isRaw
509-
? (row[i] = value)
490+
? (row[i] = query.isRaw === true
491+
? value
492+
: transform.value.from ? transform.value.from(value) : value)
510493
: (row[column.name] = transform.value.from ? transform.value.from(value) : value)
511494
}
512495

‎cjs/src/index.js

+8-2
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ module.exports = Postgres
4242

4343
function Postgres(a, b) {
4444
const options = parseOptions(a, b)
45-
, subscribe = Subscribe(Postgres, { ...options })
45+
, subscribe = options.no_subscribe || Subscribe(Postgres, { ...options })
4646

4747
let ending = false
4848

@@ -71,6 +71,7 @@ function Postgres(a, b) {
7171
listen,
7272
notify,
7373
begin,
74+
close,
7475
end
7576
})
7677

@@ -325,6 +326,10 @@ function Postgres(a, b) {
325326
]).then(() => clearTimeout(timer))
326327
}
327328

329+
async function close() {
330+
await Promise.all(connections.map(c => c.end()))
331+
}
332+
328333
async function destroy(resolve) {
329334
await Promise.all(connections.map(c => c.terminate()))
330335
while (queries.length)
@@ -397,7 +402,7 @@ function parseOptions(a, b) {
397402
onnotify : o.onnotify,
398403
onclose : o.onclose,
399404
onparameter : o.onparameter,
400-
transform : parseTransform(o.transform || {}),
405+
transform : parseTransform(o.transform || { undefined: undefined }),
401406
connection : Object.assign({ application_name: 'postgres.js' }, o.connection),
402407
target_session_attrs: tsa(o, url, env),
403408
debug : o.debug,
@@ -429,6 +434,7 @@ function max_lifetime() {
429434

430435
function parseTransform(x) {
431436
return {
437+
undefined: x.undefined,
432438
column: {
433439
from: typeof x.column === 'function' ? x.column : x.column && x.column.from,
434440
to: x.column && x.column.to

‎cjs/src/query.js

+5
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,11 @@ const Query = module.exports.Query = class Query extends Promise {
127127
return this
128128
}
129129

130+
values() {
131+
this.isRaw = 'values'
132+
return this
133+
}
134+
130135
async handle() {
131136
!this.executed && (this.executed = true) && await 1 && this.handler(this)
132137
}

‎cjs/src/subscribe.js

+67-38
Original file line numberDiff line numberDiff line change
@@ -1,59 +1,90 @@
1+
const noop = () => { /* noop */ }
2+
13
module.exports = Subscribe;function Subscribe(postgres, options) {
2-
const listeners = new Map()
4+
const subscribers = new Map()
5+
, slot = 'postgresjs_' + Math.random().toString(36).slice(2)
6+
, state = {}
37

48
let connection
5-
6-
return async function subscribe(event, fn) {
7-
event = parseEvent(event)
8-
9-
options.max = 1
10-
options.onclose = onclose
11-
options.fetch_types = false
12-
options.connection = {
9+
, stream
10+
, ended = false
11+
12+
const sql = subscribe.sql = postgres({
13+
...options,
14+
max: 1,
15+
fetch_types: false,
16+
idle_timeout: null,
17+
max_lifetime: null,
18+
connection: {
1319
...options.connection,
1420
replication: 'database'
15-
}
21+
},
22+
onclose: async function() {
23+
if (ended)
24+
return
25+
stream = null
26+
state.pid = state.secret = undefined
27+
!ended && connected(await init(sql, slot, options.publications))
28+
subscribers.forEach(event => event.forEach(({ onsubscribe }) => onsubscribe()))
29+
},
30+
no_subscribe: true
31+
})
1632

17-
let stream
18-
, ended = false
33+
const end = sql.end
34+
, close = sql.close
1935

20-
const sql = postgres(options)
21-
, slot = 'postgresjs_' + Math.random().toString(36).slice(2)
22-
, end = sql.end
36+
sql.end = async() => {
37+
ended = true
38+
stream && (await new Promise(r => (stream.once('end', r), stream.end())))
39+
return end()
40+
}
2341

24-
sql.end = async() => {
25-
ended = true
26-
stream && (await new Promise(r => (stream.once('end', r), stream.end())))
27-
return end()
28-
}
42+
sql.close = async() => {
43+
stream && (await new Promise(r => (stream.once('end', r), stream.end())))
44+
return close()
45+
}
46+
47+
return subscribe
2948

30-
!connection && (subscribe.sql = sql, connection = init(sql, slot, options.publications))
49+
async function subscribe(event, fn, onsubscribe = noop) {
50+
event = parseEvent(event)
3151

32-
const fns = listeners.has(event)
33-
? listeners.get(event).add(fn)
34-
: listeners.set(event, new Set([fn])).get(event)
52+
if (!connection)
53+
connection = init(sql, slot, options.publications)
54+
55+
const subscriber = { fn, onsubscribe }
56+
const fns = subscribers.has(event)
57+
? subscribers.get(event).add(subscriber)
58+
: subscribers.set(event, new Set([subscriber])).get(event)
3559

3660
const unsubscribe = () => {
37-
fns.delete(fn)
38-
fns.size === 0 && listeners.delete(event)
61+
fns.delete(subscriber)
62+
fns.size === 0 && subscribers.delete(event)
3963
}
4064

41-
return connection.then(x => (stream = x, { unsubscribe }))
65+
return connection.then(x => {
66+
connected(x)
67+
onsubscribe()
68+
return { unsubscribe, state, sql }
69+
})
70+
}
4271

43-
async function onclose() {
44-
stream = null
45-
!ended && (stream = await init(sql, slot, options.publications))
46-
}
72+
function connected(x) {
73+
stream = x.stream
74+
state.pid = x.state.pid
75+
state.secret = x.state.secret
4776
}
4877

4978
async function init(sql, slot, publications) {
5079
if (!publications)
5180
throw new Error('Missing publication names')
5281

53-
const [x] = await sql.unsafe(
82+
const xs = await sql.unsafe(
5483
`CREATE_REPLICATION_SLOT ${ slot } TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT`
5584
)
5685

86+
const [x] = xs
87+
5788
const stream = await sql.unsafe(
5889
`START_REPLICATION SLOT ${ slot } LOGICAL ${
5990
x.consistent_point
@@ -65,12 +96,10 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
6596
}
6697

6798
stream.on('data', data)
68-
stream.on('error', (error) => {
69-
console.error('Logical Replication Error - Reconnecting', error) // eslint-disable-line
70-
sql.end()
71-
})
99+
stream.on('error', sql.close)
100+
stream.on('close', sql.close)
72101

73-
return stream
102+
return { stream, state: xs.state }
74103

75104
function data(x) {
76105
if (x[0] === 0x77)
@@ -99,7 +128,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
99128
}
100129

101130
function call(x, a, b) {
102-
listeners.has(x) && listeners.get(x).forEach(fn => fn(a, b, x))
131+
subscribers.has(x) && subscribers.get(x).forEach(({ fn }) => fn(a, b, x))
103132
}
104133
}
105134

‎cjs/src/types.js

+47-22
Original file line numberDiff line numberDiff line change
@@ -64,19 +64,25 @@ const Builder = module.exports.Builder = class Builder extends NotTagged {
6464
this.rest = rest
6565
}
6666

67-
build(before, parameters, types, transform) {
67+
build(before, parameters, types, options) {
6868
const keyword = builders.map(([x, fn]) => ({ fn, i: before.search(x) })).sort((a, b) => a.i - b.i).pop()
6969
if (keyword.i === -1)
7070
throw new Error('Could not infer helper mode')
7171

72-
return keyword.fn(this.first, this.rest, parameters, types, transform)
72+
return keyword.fn(this.first, this.rest, parameters, types, options)
7373
}
7474
}
7575

76-
module.exports.handleValue = handleValue;function handleValue(x, parameters, types) {
77-
const value = x instanceof Parameter ? x.value : x
78-
if (value === undefined)
79-
throw Errors.generic('UNDEFINED_VALUE', 'Undefined values are not allowed')
76+
module.exports.handleValue = handleValue;function handleValue(x, parameters, types, options) {
77+
let value = x instanceof Parameter ? x.value : x
78+
if (value === undefined) {
79+
x instanceof Parameter
80+
? x.value = options.transform.undefined
81+
: value = x = options.transform.undefined
82+
83+
if (value === undefined)
84+
throw Errors.generic('UNDEFINED_VALUE', 'Undefined values are not allowed')
85+
}
8086

8187
return '$' + (types.push(
8288
x instanceof Parameter
@@ -90,40 +96,59 @@ module.exports.handleValue = handleValue;function handleValue(x, parameters, typ
9096

9197
const defaultHandlers = typeHandlers(types)
9298

93-
function valuesBuilder(first, parameters, types, transform, columns) {
99+
module.exports.stringify = stringify;function stringify(q, string, value, parameters, types, options) { // eslint-disable-line
100+
for (let i = 1; i < q.strings.length; i++) {
101+
string += (
102+
value instanceof Query ? fragment(value, parameters, types) :
103+
value instanceof Identifier ? value.value :
104+
value instanceof Builder ? value.build(string, parameters, types, options) :
105+
handleValue(value, parameters, types, options)
106+
) + q.strings[i]
107+
value = q.args[i]
108+
}
109+
110+
return string
111+
}
112+
113+
function fragment(q, parameters, types) {
114+
q.fragment = true
115+
return stringify(q, q.strings[0], q.args[0], parameters, types)
116+
}
117+
118+
function valuesBuilder(first, parameters, types, columns, options) {
94119
let value
95120
return first.map(row =>
96121
'(' + columns.map(column => {
97122
value = row[column]
98123
return (
99-
value instanceof Query ? value.strings[0] :
124+
value instanceof Query ? fragment(value, parameters, types) :
100125
value instanceof Identifier ? value.value :
101-
handleValue(value, parameters, types)
126+
handleValue(value, parameters, types, options)
102127
)
103128
}).join(',') + ')'
104129
).join(',')
105130
}
106131

107-
function values(first, rest, parameters, types, transform) {
132+
function values(first, rest, parameters, types, options) {
108133
const multi = Array.isArray(first[0])
109134
const columns = rest.length ? rest.flat() : Object.keys(multi ? first[0] : first)
110-
return valuesBuilder(multi ? first : [first], parameters, types, transform, columns)
135+
return valuesBuilder(multi ? first : [first], parameters, types, columns, options)
111136
}
112137

113-
function select(first, rest, parameters, types, transform) {
138+
function select(first, rest, parameters, types, options) {
114139
typeof first === 'string' && (first = [first].concat(rest))
115140
if (Array.isArray(first))
116-
return first.map(x => escapeIdentifier(transform.column.to ? transform.column.to(x) : x)).join(',')
141+
return first.map(x => escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)).join(',')
117142

118143
let value
119144
const columns = rest.length ? rest.flat() : Object.keys(first)
120145
return columns.map(x => {
121146
value = first[x]
122147
return (
123-
value instanceof Query ? value.strings[0] :
148+
value instanceof Query ? fragment(value, parameters, types) :
124149
value instanceof Identifier ? value.value :
125-
handleValue(value, parameters, types)
126-
) + ' as ' + escapeIdentifier(transform.column.to ? transform.column.to(x) : x)
150+
handleValue(value, parameters, types, options)
151+
) + ' as ' + escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)
127152
}).join(',')
128153
}
129154

@@ -133,19 +158,19 @@ const builders = Object.entries({
133158
select,
134159
returning: select,
135160

136-
update(first, rest, parameters, types, transform) {
161+
update(first, rest, parameters, types, options) {
137162
return (rest.length ? rest.flat() : Object.keys(first)).map(x =>
138-
escapeIdentifier(transform.column.to ? transform.column.to(x) : x) +
139-
'=' + handleValue(first[x], parameters, types)
163+
escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x) +
164+
'=' + handleValue(first[x], parameters, types, options)
140165
)
141166
},
142167

143-
insert(first, rest, parameters, types, transform) {
168+
insert(first, rest, parameters, types, options) {
144169
const columns = rest.length ? rest.flat() : Object.keys(Array.isArray(first) ? first[0] : first)
145170
return '(' + columns.map(x =>
146-
escapeIdentifier(transform.column.to ? transform.column.to(x) : x)
171+
escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)
147172
).join(',') + ')values' +
148-
valuesBuilder(Array.isArray(first) ? first : [first], parameters, types, transform, columns)
173+
valuesBuilder(Array.isArray(first) ? first : [first], parameters, types, columns, options)
149174
}
150175
}).map(([x, fn]) => ([new RegExp('(^|[\\s(])' + x + '($|[\\s(])', 'i'), fn]))
151176

0 commit comments

Comments
 (0)