Skip to content

Commit 02bcc9d

Browse files
briancbrianc
committed
Start working on promsie tests
1 parent bc2f550 commit 02bcc9d

File tree

2 files changed

+182
-56
lines changed

2 files changed

+182
-56
lines changed

‎lib/client.js

+77-56
Original file line numberDiff line numberDiff line change
@@ -108,56 +108,21 @@ Client.prototype.connect = function(callback) {
108108
self.secretKey = msg.secretKey;
109109
});
110110

111+
112+
con.on('readyForQuery', function() {
113+
var activeQuery = self.activeQuery;
114+
self.activeQuery = null;
115+
self.readyForQuery = true;
116+
self._pulseQueryQueue();
117+
if(activeQuery) {
118+
activeQuery.handleReadyForQuery(con);
119+
}
120+
});
121+
111122
//hook up query handling events to connection
112123
//after the connection initially becomes ready for queries
113124
con.once('readyForQuery', function() {
114-
self._connecting = false;
115-
116-
//delegate rowDescription to active query
117-
con.on('rowDescription', function(msg) {
118-
self.activeQuery.handleRowDescription(msg);
119-
});
120-
121-
//delegate dataRow to active query
122-
con.on('dataRow', function(msg) {
123-
self.activeQuery.handleDataRow(msg);
124-
});
125-
126-
//delegate portalSuspended to active query
127-
con.on('portalSuspended', function(msg) {
128-
self.activeQuery.handlePortalSuspended(con);
129-
});
130-
131-
//deletagate emptyQuery to active query
132-
con.on('emptyQuery', function(msg) {
133-
self.activeQuery.handleEmptyQuery(con);
134-
});
135-
136-
//delegate commandComplete to active query
137-
con.on('commandComplete', function(msg) {
138-
self.activeQuery.handleCommandComplete(msg, con);
139-
});
140-
141-
//if a prepared statement has a name and properly parses
142-
//we track that its already been executed so we don't parse
143-
//it again on the same client
144-
con.on('parseComplete', function(msg) {
145-
if(self.activeQuery.name) {
146-
con.parsedStatements[self.activeQuery.name] = true;
147-
}
148-
});
149-
150-
con.on('copyInResponse', function(msg) {
151-
self.activeQuery.handleCopyInResponse(self.connection);
152-
});
153-
154-
con.on('copyData', function (msg) {
155-
self.activeQuery.handleCopyData(msg, self.connection);
156-
});
157-
158-
con.on('notification', function(msg) {
159-
self.emit('notification', msg);
160-
});
125+
self._attachEventListeners(con)
161126

162127
//process possible callback argument to Client#connect
163128
if (callback) {
@@ -169,15 +134,15 @@ Client.prototype.connect = function(callback) {
169134
self.emit('connect');
170135
});
171136

172-
con.on('readyForQuery', function() {
173-
var activeQuery = self.activeQuery;
174-
self.activeQuery = null;
175-
self.readyForQuery = true;
176-
self._pulseQueryQueue();
177-
if(activeQuery) {
178-
activeQuery.handleReadyForQuery(con);
179-
}
180-
});
137+
if (!callback) {
138+
return new global.Promise(function (resolve, reject) {
139+
con.once('connect', () => {
140+
con.removeListener('error', reject)
141+
resolve()
142+
})
143+
con.once('error', reject)
144+
})
145+
}
181146

182147
con.on('error', function(error) {
183148
if(this.activeQuery) {
@@ -234,6 +199,58 @@ Client.prototype.connect = function(callback) {
234199

235200
};
236201

202+
// once a connection is established connect listeners
203+
Client.prototype._attachEventListeners = function(con) {
204+
var self = this;
205+
self._connecting = false;
206+
207+
//delegate rowDescription to active query
208+
con.on('rowDescription', function(msg) {
209+
self.activeQuery.handleRowDescription(msg);
210+
});
211+
212+
//delegate dataRow to active query
213+
con.on('dataRow', function(msg) {
214+
self.activeQuery.handleDataRow(msg);
215+
});
216+
217+
//delegate portalSuspended to active query
218+
con.on('portalSuspended', function(msg) {
219+
self.activeQuery.handlePortalSuspended(con);
220+
});
221+
222+
//deletagate emptyQuery to active query
223+
con.on('emptyQuery', function(msg) {
224+
self.activeQuery.handleEmptyQuery(con);
225+
});
226+
227+
//delegate commandComplete to active query
228+
con.on('commandComplete', function(msg) {
229+
self.activeQuery.handleCommandComplete(msg, con);
230+
});
231+
232+
//if a prepared statement has a name and properly parses
233+
//we track that its already been executed so we don't parse
234+
//it again on the same client
235+
con.on('parseComplete', function(msg) {
236+
if(self.activeQuery.name) {
237+
con.parsedStatements[self.activeQuery.name] = true;
238+
}
239+
});
240+
241+
con.on('copyInResponse', function(msg) {
242+
self.activeQuery.handleCopyInResponse(self.connection);
243+
});
244+
245+
con.on('copyData', function (msg) {
246+
self.activeQuery.handleCopyData(msg, self.connection);
247+
});
248+
249+
con.on('notification', function(msg) {
250+
self.emit('notification', msg);
251+
});
252+
}
253+
237254
Client.prototype.getStartupConf = function() {
238255
var params = this.connectionParameters;
239256

@@ -391,6 +408,10 @@ Client.prototype.end = function(cb) {
391408
this.connection.end();
392409
if (cb) {
393410
this.connection.once('end', cb);
411+
} else {
412+
return new global.Promise((resolve) => {
413+
this.connection.once('end', resolve);
414+
});
394415
}
395416
};
396417

‎test/integration/client/promise-api-tests.js

+105
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
const async = require('async')
2+
const helper = require('./test-helper')
3+
const pg = helper.pg;
4+
5+
class Test {
6+
constructor(name, cb) {
7+
this.name = name
8+
this.action = cb
9+
this.timeout = 5000
10+
}
11+
12+
run(cb) {
13+
try {
14+
this._run(cb)
15+
} catch (e) {
16+
cb(e)
17+
}
18+
}
19+
20+
_run(cb) {
21+
if (!this.action) {
22+
console.log(`${this.name} skipped`)
23+
return cb()
24+
}
25+
if (!this.action.length) {
26+
const result = this.action.call(this)
27+
if ((result || 0).then) {
28+
result
29+
.then(() => cb())
30+
.catch(err => cb(err || new Error('Unhandled promise rejection')))
31+
}
32+
} else {
33+
this.action.call(this, cb)
34+
}
35+
}
36+
}
37+
38+
class Suite {
39+
constructor() {
40+
console.log('')
41+
this._queue = async.queue(this.run.bind(this), 1)
42+
this._queue.drain = () => { }
43+
}
44+
45+
run(test, cb) {
46+
const tid = setTimeout(() => {
47+
const err = Error(`test: ${test.name} did not complete withint ${test.timeout}ms`)
48+
cb(err)
49+
}, test.timeout)
50+
test.run((err) => {
51+
clearTimeout(tid)
52+
if (err) {
53+
console.log(test.name + ' FAILED!', err.stack)
54+
} else {
55+
console.log(test.name)
56+
}
57+
cb(err)
58+
})
59+
}
60+
61+
test(name, cb) {
62+
this._queue.push(new Test(name, cb))
63+
}
64+
}
65+
66+
const suite = new Suite()
67+
68+
suite.test('valid connection completes promise', () => {
69+
const client = new pg.Client()
70+
return client.connect()
71+
.then(() => {
72+
return client.end()
73+
.then(() => { })
74+
})
75+
})
76+
77+
suite.test('valid connection completes promise', () => {
78+
const client = new pg.Client()
79+
return client.connect()
80+
.then(() => {
81+
return client.end()
82+
.then(() => { })
83+
})
84+
})
85+
86+
87+
suite.test('invalid connection rejects promise', (done) => {
88+
const client = new pg.Client({ host: 'alksdjflaskdfj' })
89+
return client.connect()
90+
.catch(e => {
91+
assert(e instanceof Error)
92+
done()
93+
})
94+
})
95+
96+
suite.test('connected client does not reject promise after', (done) => {
97+
const client = new pg.Client()
98+
return client.connect()
99+
.then(() => {
100+
setTimeout(() => {
101+
// manually kill the connection
102+
client.connection.stream.end()
103+
}, 50)
104+
})
105+
})

0 commit comments

Comments
 (0)