Refactoring of EventGroup using

Author: marat-gainullin

pom.xml

@@ -66,7 +66,7 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-handler</artifactId>
-            <version>4.1.11.Final</version>
+            <version>4.1.33.Final</version>
         </dependency>
         <!-- https://mvnrepository.com/artifact/javax.xml.bind/jaxb-api -->
         <dependency>

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

@@ -2,12 +2,11 @@
 
 import com.github.pgasync.conversion.DataConverter;
 import com.pgasync.Connection;
-import com.pgasync.ConnectibleBuilder;
+import com.pgasync.NettyConnectibleBuilder;
 import com.pgasync.Connectible;
 import com.pgasync.Row;
 import com.pgasync.Transaction;
 
-import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -19,23 +18,23 @@
 public abstract class PgConnectible implements Connectible {
 
     final String validationQuery;
-    final InetSocketAddress address;
     final String username;
     final DataConverter dataConverter;
+    final Executor futuresExecutor;
+    final Function<Executor, ProtocolStream> toStream;
+
     protected final String password;
     protected final String database;
     protected final Charset encoding;
 
-    protected final Executor futuresExecutor;
-
-    PgConnectible(ConnectibleBuilder.ConnectibleProperties properties, Executor futuresExecutor) {
-        this.address = InetSocketAddress.createUnresolved(properties.getHostname(), properties.getPort());
+    PgConnectible(NettyConnectibleBuilder.ConnectibleProperties properties, Function<Executor, ProtocolStream> toStream, Executor futuresExecutor) {
         this.username = properties.getUsername();
         this.password = properties.getPassword();
         this.database = properties.getDatabase();
         this.dataConverter = properties.getDataConverter();
         this.validationQuery = properties.getValidationQuery();
         this.encoding = Charset.forName(properties.getEncoding());
+        this.toStream = toStream;
         this.futuresExecutor = futuresExecutor;
     }
 
@@ -83,11 +82,4 @@ public CompletableFuture<Integer> query(BiConsumer<Map<String, PgColumn>, PgColu
                 .thenCompose(Function.identity());
     }
 
-    /**
-     * Creates a new socket stream to the backend.
-     *
-     * @param address Server address
-     * @return Stream with no pending messages
-     */
-    protected abstract PgProtocolStream openStream(InetSocketAddress address);
 }

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

@@ -129,13 +129,13 @@ private String next() {
 
     private static final NameSequence preparedStatementNames = new NameSequence("s-");
 
-    private final PgProtocolStream stream;
+    private final ProtocolStream stream;
     private final DataConverter dataConverter;
     private final Charset encoding;
 
     private Columns currentColumns;
 
-    PgConnection(PgProtocolStream stream, DataConverter dataConverter, Charset encoding) {
+    PgConnection(ProtocolStream stream, DataConverter dataConverter, Charset encoding) {
         this.stream = stream;
         this.dataConverter = dataConverter;
         this.encoding = encoding;

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

@@ -19,7 +19,7 @@
 import com.pgasync.PreparedStatement;
 import com.pgasync.Row;
 import com.pgasync.SqlException;
-import com.pgasync.ConnectibleBuilder;
+import com.pgasync.NettyConnectibleBuilder;
 import com.pgasync.ResultSet;
 import com.pgasync.Transaction;
 
@@ -45,7 +45,7 @@
  *
  * @author Antti Laisi
  */
-public abstract class PgConnectionPool extends PgConnectible {
+public class PgConnectionPool extends PgConnectible {
 
     private class PooledPgConnection implements Connection {
 
@@ -254,8 +254,8 @@ public CompletableFuture<Integer> fetch(BiConsumer<Map<String, PgColumn>, PgColu
     @GuardedBy("lock")
     private final Queue<PooledPgConnection> connections = new LinkedList<>();
 
-    public PgConnectionPool(ConnectibleBuilder.ConnectibleProperties properties, Executor futuresExecutor) {
-        super(properties, futuresExecutor);
+    public PgConnectionPool(NettyConnectibleBuilder.ConnectibleProperties properties, Function<Executor, ProtocolStream> addressToStream, Executor futuresExecutor) {
+        super(properties, addressToStream, futuresExecutor);
         this.maxConnections = properties.getMaxConnections();
         this.maxStatements = properties.getMaxStatements();
     }
@@ -295,7 +295,7 @@ public CompletableFuture<Connection> getConnection() {
                     uponAvailable.completeAsync(() -> connection, futuresExecutor);
                 } else {
                     if (tryIncreaseSize()) {
-                        new PooledPgConnection(new PgConnection(openStream(address), dataConverter, encoding))
+                        new PooledPgConnection(new PgConnection(toStream.apply(futuresExecutor), dataConverter, encoding))
                                 .connect(username, password, database)
                                 .thenApply(pooledConnection -> {
                                     if (validationQuery != null && !validationQuery.isBlank()) {

src/main/java/com/github/pgasync/PgDatabase.java

@@ -1,20 +1,21 @@
 package com.github.pgasync;
 
 import com.pgasync.Connection;
-import com.pgasync.ConnectibleBuilder;
+import com.pgasync.NettyConnectibleBuilder;
+
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
 import java.util.function.Function;
 
-public abstract class PgDatabase extends PgConnectible {
+public class PgDatabase extends PgConnectible {
 
-    public PgDatabase(ConnectibleBuilder.ConnectibleProperties properties, Executor futuresExecutor) {
-        super(properties, futuresExecutor);
+    public PgDatabase(NettyConnectibleBuilder.ConnectibleProperties properties, Function<Executor, ProtocolStream> toStream, Executor futuresExecutor) {
+        super(properties, toStream, futuresExecutor);
     }
 
     @Override
     public CompletableFuture<Connection> getConnection() {
-        return new PgConnection(openStream(address), dataConverter, encoding)
+        return new PgConnection(toStream.apply(futuresExecutor), dataConverter, encoding)
                 .connect(username, password, database)
                 .thenApply(connection -> {
                     if (validationQuery != null && !validationQuery.isBlank()) {