Skip to content

Commit 4235468

Browse files
marat-gainullinmarat-gainullin
committed
Refactoring of interfaces
1 parent 0e06eb2 commit 4235468

26 files changed

+850
-282
lines changed

‎TODO.md‎

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,21 @@
11
## Refactoring
22
* Add arguments checks to most core methods.
3-
* Make Connection#shutdown to use futures properly
43

54
## Tests
6-
* Make tests for all message flow scenarios.
7-
+ Review other test cases.
8-
9-
### Test plan
10-
* Errors while authentication
11-
* Errors while simple query process
12-
* Errors while extended query process
13-
* Multiple result sets test
14-
* Termination process test
15-
* Prepared statement eviction test
5+
* Connection pool failed connections
6+
* Connection pool abandoned waiters
7+
+ Make tests for all message flow scenarios.
8+
+ Errors while validation query processing while getting connection in two cases (pool and plain connectible)
9+
+ Errors while authentication
10+
+ Errors while simple query process
11+
+ Errors while extended query process
12+
+ Multiple result sets test
13+
+ Connection pool statements eviction and duplicated statements
14+
+ Termination process test
15+
+ Prepared statement eviction test
1616
+ Auto rollback transaction review
1717
+ Ssl interaction test
1818
+ Notifications test
1919
+ Transactions general test
2020
+ Connection pool general test
2121
+ Queries pipelining vs errors test
22-
* Connection pool failed connections and abandoned waiters tests

‎src/main/java/com/github/pgasync/PgConnectible.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package com.github.pgasync;
2+
3+
import com.github.pgasync.conversion.DataConverter;
4+
import com.pgasync.Connection;
5+
import com.pgasync.ConnectibleBuilder;
6+
import com.pgasync.Connectible;
7+
import com.pgasync.Row;
8+
import com.pgasync.Transaction;
9+
10+
import java.net.InetSocketAddress;
11+
import java.nio.charset.Charset;
12+
import java.util.Map;
13+
import java.util.concurrent.CompletableFuture;
14+
import java.util.concurrent.Executor;
15+
import java.util.function.BiConsumer;
16+
import java.util.function.Consumer;
17+
import java.util.function.Function;
18+
19+
public abstract class PgConnectible implements Connectible {
20+
21+
final String validationQuery;
22+
final InetSocketAddress address;
23+
final String username;
24+
final DataConverter dataConverter;
25+
protected final String password;
26+
protected final String database;
27+
protected final Charset encoding;
28+
29+
protected final Executor futuresExecutor;
30+
31+
PgConnectible(ConnectibleBuilder.ConnectibleProperties properties, Executor futuresExecutor) {
32+
this.address = InetSocketAddress.createUnresolved(properties.getHostname(), properties.getPort());
33+
this.username = properties.getUsername();
34+
this.password = properties.getPassword();
35+
this.database = properties.getDatabase();
36+
this.dataConverter = properties.getDataConverter();
37+
this.validationQuery = properties.getValidationQuery();
38+
this.encoding = Charset.forName(properties.getEncoding());
39+
this.futuresExecutor = futuresExecutor;
40+
}
41+
42+
@Override
43+
public CompletableFuture<Transaction> begin() {
44+
return getConnection()
45+
.thenApply(Connection::begin)
46+
.thenCompose(Function.identity());
47+
}
48+
49+
@Override
50+
public CompletableFuture<Void> script(BiConsumer<Map<String, PgColumn>, PgColumn[]> onColumns, Consumer<Row> onRow, Consumer<Integer> onAffected, String sql) {
51+
return getConnection()
52+
.thenApply(connection ->
53+
connection.script(onColumns, onRow, onAffected, sql)
54+
.handle((message, th) ->
55+
connection.close()
56+
.thenApply(v -> {
57+
if (th == null) {
58+
return message;
59+
} else {
60+
throw new RuntimeException(th);
61+
}
62+
})
63+
).thenCompose(Function.identity())
64+
).thenCompose(Function.identity());
65+
}
66+
67+
@Override
68+
public CompletableFuture<Integer> query(BiConsumer<Map<String, PgColumn>, PgColumn[]> onColumns, Consumer<Row> onRow, String sql, Object... params) {
69+
return getConnection()
70+
.thenApply(connection ->
71+
connection.query(onColumns, onRow, sql, params)
72+
.handle((affected, th) ->
73+
connection.close()
74+
.thenApply(v -> {
75+
if (th == null) {
76+
return affected;
77+
} else {
78+
throw new RuntimeException(th);
79+
}
80+
})
81+
).thenCompose(Function.identity())
82+
)
83+
.thenCompose(Function.identity());
84+
}
85+
86+
/**
87+
* Creates a new socket stream to the backend.
88+
*
89+
* @param address Server address
90+
* @return Stream with no pending messages
91+
*/
92+
protected abstract PgProtocolStream openStream(InetSocketAddress address);
93+
}

‎src/main/java/com/github/pgasync/PgConnection.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ public CompletableFuture<Integer> fetch(BiConsumer<Map<String, PgColumn>, PgColu
8080
Bind bind = new Bind(sname, dataConverter.fromParameters(params));
8181
Consumer<DataRow> rowProcessor = dataRow -> processor.accept(new PgRow(dataRow, columns.byName, columns.ordered, dataConverter));
8282
if (columns != null) {
83+
onColumns.accept(columns.byName, columns.ordered);
8384
return stream
8485
.send(bind, rowProcessor);
8586
} else {

‎src/main/java/com/github/pgasync/PgConnectionPool.java‎

Lines changed: 68 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -14,22 +14,19 @@
1414

1515
package com.github.pgasync;
1616

17-
import com.github.pgasync.conversion.DataConverter;
1817
import com.pgasync.Connection;
1918
import com.pgasync.Listening;
2019
import com.pgasync.PreparedStatement;
2120
import com.pgasync.Row;
2221
import com.pgasync.SqlException;
23-
import com.pgasync.ConnectionPool;
24-
import com.pgasync.ConnectionPoolBuilder;
22+
import com.pgasync.ConnectibleBuilder;
2523
import com.pgasync.ResultSet;
2624
import com.pgasync.Transaction;
2725

2826
import javax.annotation.concurrent.GuardedBy;
29-
import java.net.InetSocketAddress;
30-
import java.nio.charset.Charset;
3127
import java.util.ArrayList;
3228
import java.util.Collection;
29+
import java.util.Iterator;
3330
import java.util.LinkedHashMap;
3431
import java.util.LinkedList;
3532
import java.util.Map;
@@ -42,14 +39,13 @@
4239
import java.util.function.Function;
4340
import java.util.logging.Level;
4441
import java.util.logging.Logger;
45-
import java.util.stream.Collectors;
4642

4743
/**
48-
* Pool for backend connections.
44+
* Resource pool for backend connections.
4945
*
5046
* @author Antti Laisi
5147
*/
52-
public abstract class PgConnectionPool implements ConnectionPool {
48+
public abstract class PgConnectionPool extends PgConnectible {
5349

5450
private class PooledPgConnection implements Connection {
5551

@@ -118,15 +114,29 @@ boolean isConnected() {
118114
return delegate.isConnected();
119115
}
120116

117+
private void closeNextStatement(Iterator<PooledPgPreparedStatement> statementsSource, CompletableFuture<Void> onComplete) {
118+
if (statementsSource.hasNext()) {
119+
statementsSource.next().delegate.close()
120+
.thenAccept(v -> {
121+
statementsSource.remove();
122+
closeNextStatement(statementsSource, onComplete);
123+
})
124+
.exceptionally(th -> {
125+
futuresExecutor.execute(() -> onComplete.completeExceptionally(th));
126+
return null;
127+
});
128+
} else {
129+
onComplete.completeAsync(() -> null, futuresExecutor);
130+
}
131+
}
132+
121133
CompletableFuture<Void> shutdown() {
122-
CompletableFuture<?>[] closeTasks = statements.values().stream()
123-
.map(stmt -> stmt.delegate.close())
124-
.collect(Collectors.toList()).toArray(new CompletableFuture<?>[]{});
125-
statements.clear();
126-
return CompletableFuture.allOf(closeTasks)
134+
CompletableFuture<Void> onComplete = new CompletableFuture<>();
135+
closeNextStatement(statements.values().iterator(), onComplete);
136+
return onComplete
127137
.thenApply(v -> {
128138
if (!statements.isEmpty()) {
129-
Logger.getLogger(PooledPgConnection.class.getName()).log(Level.WARNING, "Stale prepared statements detected {0}", statements.size());
139+
throw new IllegalStateException("Stale prepared statements detected (" + statements.size() + ")");
130140
}
131141
return delegate.close();
132142
})
@@ -155,7 +165,20 @@ public CompletableFuture<Void> script(BiConsumer<Map<String, PgColumn>, PgColumn
155165

156166
@Override
157167
public CompletableFuture<Integer> query(BiConsumer<Map<String, PgColumn>, PgColumn[]> onColumns, Consumer<Row> onRow, String sql, Object... params) {
158-
return delegate.query(onColumns, onRow, sql, params);
168+
return prepareStatement(sql, dataConverter.assumeTypes(params))
169+
.thenApply(stmt ->
170+
stmt.fetch(onColumns, onRow, params)
171+
.handle((affected, th) ->
172+
stmt.close()
173+
.thenApply(v -> {
174+
if (th == null) {
175+
return affected;
176+
} else {
177+
throw new RuntimeException(th);
178+
}
179+
})
180+
).thenCompose(Function.identity())
181+
).thenCompose(Function.identity());
159182
}
160183

161184
@Override
@@ -171,6 +194,7 @@ public CompletableFuture<PreparedStatement> prepareStatement(String sql, Oid...
171194

172195
private class PooledPgPreparedStatement implements PreparedStatement {
173196

197+
private static final String DUPLICATED_PREPARED_STATEMENT_DETECTED = "Duplicated prepared statement detected. Closing extra instance. \n{0}";
174198
private final String sql;
175199
private final PgConnection.PgPreparedStatement delegate;
176200

@@ -181,15 +205,27 @@ private PooledPgPreparedStatement(String sql, PgConnection.PgPreparedStatement d
181205

182206
@Override
183207
public CompletableFuture<Void> close() {
184-
statements.put(sql, this);
208+
PooledPgPreparedStatement already = statements.put(sql, this);
185209
if (evicted != null) {
186210
try {
187-
return evicted.delegate.close();
211+
if (already != null && already != evicted) {
212+
Logger.getLogger(PgConnectionPool.class.getName()).log(Level.WARNING, DUPLICATED_PREPARED_STATEMENT_DETECTED, already.sql);
213+
return evicted.delegate.close()
214+
.thenApply(v -> already.delegate.close())
215+
.thenCompose(Function.identity());
216+
} else {
217+
return evicted.delegate.close();
218+
}
188219
} finally {
189220
evicted = null;
190221
}
191222
} else {
192-
return CompletableFuture.completedFuture(null);
223+
if (already != null) {
224+
Logger.getLogger(PgConnectionPool.class.getName()).log(Level.WARNING, DUPLICATED_PREPARED_STATEMENT_DETECTED, already.sql);
225+
return already.delegate.close();
226+
} else {
227+
return CompletableFuture.completedFuture(null);
228+
}
193229
}
194230
}
195231

@@ -207,9 +243,7 @@ public CompletableFuture<Integer> fetch(BiConsumer<Map<String, PgColumn>, PgColu
207243

208244
private final int maxConnections;
209245
private final int maxStatements;
210-
private final String validationQuery;
211246
private final ReentrantLock lock = new ReentrantLock();
212-
protected final Charset encoding;
213247

214248
@GuardedBy("lock")
215249
private int size;
@@ -220,25 +254,10 @@ public CompletableFuture<Integer> fetch(BiConsumer<Map<String, PgColumn>, PgColu
220254
@GuardedBy("lock")
221255
private final Queue<PooledPgConnection> connections = new LinkedList<>();
222256

223-
private final InetSocketAddress address;
224-
private final String username;
225-
private final String password;
226-
private final String database;
227-
private final DataConverter dataConverter;
228-
229-
protected final Executor futuresExecutor;
230-
231-
public PgConnectionPool(ConnectionPoolBuilder.PoolProperties properties, Executor futuresExecutor) {
232-
this.address = InetSocketAddress.createUnresolved(properties.getHostname(), properties.getPort());
233-
this.username = properties.getUsername();
234-
this.password = properties.getPassword();
235-
this.database = properties.getDatabase();
257+
public PgConnectionPool(ConnectibleBuilder.ConnectibleProperties properties, Executor futuresExecutor) {
258+
super(properties, futuresExecutor);
236259
this.maxConnections = properties.getMaxConnections();
237260
this.maxStatements = properties.getMaxStatements();
238-
this.dataConverter = properties.getDataConverter();
239-
this.validationQuery = properties.getValidationQuery();
240-
this.encoding = Charset.forName(properties.getEncoding());
241-
this.futuresExecutor = futuresExecutor;
242261
}
243262

244263
@Override
@@ -259,7 +278,7 @@ public CompletableFuture<Void> close() {
259278
} finally {
260279
lock.unlock();
261280
}
262-
return CompletableFuture.allOf(shutdownTasks.toArray(size -> new CompletableFuture<?>[size]));
281+
return CompletableFuture.allOf(shutdownTasks.toArray(CompletableFuture<?>[]::new));
263282
}
264283

265284
@Override
@@ -280,7 +299,17 @@ public CompletableFuture<Connection> getConnection() {
280299
.connect(username, password, database)
281300
.thenApply(pooledConnection -> {
282301
if (validationQuery != null && !validationQuery.isBlank()) {
283-
return pooledConnection.completeQuery(validationQuery).thenApply(rs -> pooledConnection);
302+
return pooledConnection.completeScript(validationQuery)
303+
.handle((rss, th) -> {
304+
if (th != null) {
305+
return ((PooledPgConnection) pooledConnection).delegate.close()
306+
.thenApply(v -> CompletableFuture.<Connection>failedFuture(th))
307+
.thenCompose(Function.identity());
308+
} else {
309+
return CompletableFuture.completedFuture(pooledConnection);
310+
}
311+
})
312+
.thenCompose(Function.identity());
284313
} else {
285314
return CompletableFuture.completedFuture(pooledConnection);
286315
}
@@ -347,56 +376,4 @@ private CompletableFuture<Void> release(PooledPgConnection connection) {
347376
return shutdownTask;
348377
}
349378

350-
@Override
351-
public CompletableFuture<Transaction> begin() {
352-
return getConnection()
353-
.thenApply(Connection::begin)
354-
.thenCompose(Function.identity());
355-
}
356-
357-
@Override
358-
public CompletableFuture<Void> script(BiConsumer<Map<String, PgColumn>, PgColumn[]> onColumns, Consumer<Row> onRow, Consumer<Integer> onAffected, String sql) {
359-
return getConnection()
360-
.thenApply(connection ->
361-
connection.script(onColumns, onRow, onAffected, sql)
362-
.handle((message, th) ->
363-
connection.close()
364-
.thenApply(v -> {
365-
if (th == null) {
366-
return message;
367-
} else {
368-
throw new RuntimeException(th);
369-
}
370-
})
371-
).thenCompose(Function.identity())
372-
)
373-
.thenCompose(Function.identity());
374-
}
375-
376-
@Override
377-
public CompletableFuture<Integer> query(BiConsumer<Map<String, PgColumn>, PgColumn[]> onColumns, Consumer<Row> onRow, String sql, Object... params) {
378-
return getConnection()
379-
.thenApply(connection ->
380-
connection.query(onColumns, onRow, sql, params)
381-
.handle((affected, th) ->
382-
connection.close()
383-
.thenApply(v -> {
384-
if (th == null) {
385-
return affected;
386-
} else {
387-
throw new RuntimeException(th);
388-
}
389-
})
390-
).thenCompose(Function.identity())
391-
)
392-
.thenCompose(Function.identity());
393-
}
394-
395-
/**
396-
* Creates a new socket stream to the backend.
397-
*
398-
* @param address Server address
399-
* @return Stream with no pending messages
400-
*/
401-
protected abstract PgProtocolStream openStream(InetSocketAddress address);
402379
}

0 commit comments

Comments
 (0)