Completable futures added everywhere

Author: marat-gainullin

README.md

@@ -66,7 +66,7 @@ Prepared statements use native PostgreSQL syntax `$index`. Supported parameter t
 
 ```java
 db.querySet("insert into message(id, body) values($1, $2)", 123, "hello")
-    .subscribe(result -> out.printf("Inserted %d rows", result.updatedRows() ));
+    .subscribe(result -> out.printf("Inserted %d rows", result.affectedRows() ));
 ```
 
 ### Transactions

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

@@ -14,16 +14,19 @@
 
 package com.github.pgasync;
 
+import com.github.pgasync.impl.Oid;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
 /**
- * A single physical connection to PostgreSQL backend.
+ * A single physical connection to Postgres backend.
  *
  * @author Antti Laisi
  */
 public interface Connection extends Db {
 
-    /**
-     * Closes the connection. Doesn't block current thread.
-     */
-    void close();
+    CompletableFuture<PreparedStatement> prepareStatement(String sql, Oid... parametersTypes);
 
-}
+    CompletableFuture<Listening> subscribe(String channel, Consumer<String> onNotification);
+}

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

@@ -20,23 +20,15 @@
  * Pool of backend {@link Connection}s. Pools implement {@link Db} so
  * queries can be issued directly to the pool if using the same connection
  * is not required.
- * 
+ *
  * @author Antti Laisi
  */
 public interface ConnectionPool extends Db {
 
     /**
-     * Executes a {@link java.util.function.Consumer} callback when a connection is
-     * available. Connection passed to callback must be freed with
-     * {@link com.github.pgasync.ConnectionPool#release(Connection)}
+     * Gets a connection when available.
+     * {@link Connection#close()} method will return the connection into this pool instead of closing it.
      */
     CompletableFuture<Connection> getConnection();
 
-    /**
-     * Releases a connection back to the pool.
-     * 
-     * @param connection Connection fetched with getConnection
-     */
-    void release(Connection connection);
-
 }

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

@@ -63,7 +63,7 @@ public ConnectionPoolBuilder database(String database) {
     }
 
     public ConnectionPoolBuilder poolSize(int poolSize) {
-        properties.poolSize = poolSize;
+        properties.maxConnections = poolSize;
         return this;
     }
 
@@ -82,11 +82,6 @@ public ConnectionPoolBuilder ssl(boolean ssl) {
         return this;
     }
 
-    public ConnectionPoolBuilder pipeline(boolean pipeline) {
-        properties.usePipelining = pipeline;
-        return this;
-    }
-
     public ConnectionPoolBuilder validationQuery(String validationQuery) {
         properties.validationQuery = validationQuery;
         return this;
@@ -102,11 +97,11 @@ public static class PoolProperties {
         String username;
         String password;
         String database;
-        int poolSize = 20;
+        int maxConnections = 20;
+        int maxStatements = 20;
         DataConverter dataConverter = null;
         List<Converter<?>> converters = new ArrayList<>();
         boolean useSsl;
-        boolean usePipelining;
         String validationQuery;
 
         public String getHostname() {
@@ -129,16 +124,16 @@ public String getDatabase() {
             return database;
         }
 
-        public int getPoolSize() {
-            return poolSize;
+        public int getMaxConnections() {
+            return maxConnections;
         }
 
-        public boolean getUseSsl() {
-            return useSsl;
+        public int getMaxStatements() {
+            return maxStatements;
         }
 
-        public boolean getUsePipelining() {
-            return usePipelining;
+        public boolean getUseSsl() {
+            return useSsl;
         }
 
         public DataConverter getDataConverter() {

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

@@ -1,16 +1,13 @@
 package com.github.pgasync;
 
+import java.util.concurrent.CompletableFuture;
+
 /**
- * Main interface to PostgreSQL backend.
+ * Main interface to Postgres backend.
  *
  * @author Antti Laisi
  */
-public interface Db extends QueryExecutor, AutoCloseable {
-
-    /**
-     * Closes the pool, blocks the calling thread until connections are closed.
-     */
-    @Override
-    void close() throws Exception;
+public interface Db extends QueryExecutor {
 
+    CompletableFuture<Void> close();
 }

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

@@ -0,0 +1,9 @@
+package com.github.pgasync;
+
+import java.util.concurrent.CompletableFuture;
+
+@FunctionalInterface
+public interface Listening {
+
+    CompletableFuture<Void> unlisten();
+}

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

@@ -0,0 +1,37 @@
+package com.github.pgasync;
+
+import com.github.pgasync.impl.PgColumn;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/**
+ * Prepared statement in terms of Postgres.
+ * It lives during database session. It should be reused multiple times and it should be closed after using.
+ * Doesn't support function call feature because of its deprecation See <a href="https://www.postgresql.org/docs/11/protocol-flow.html#id-1.10.5.7.6"/>.
+ */
+public interface PreparedStatement {
+
+    /**
+     * Fetches the whole row set and returns a {@link CompletableFuture} with an instance of {@link ResultSet}.
+     * This future may be completed with an error. Use this method if you are sure, that all data, returned by the query can be placed into memory.
+     *
+     * @param params Array of query parameters values.
+     * @return An instance of {@link ResultSet} with data.
+     */
+    CompletableFuture<ResultSet> query(Object... params);
+
+    /**
+     * Fetches data rows from Postgres one by one. Use this method when you are unsure, that all data, returned by the query can be placed into memory.
+     *
+     * @param onColumns {@link Consumer} of parameters by name map. Gets called when bind/describe chain succeeded.
+     * @param processor {@link Consumer} of single data row. Performs some processing of data.
+     * @param params Array of query parameters values.
+     * @return CompletableFuture that completes when the whole process ends or when an error occurs. Future's value will indicate the number of affected rows by the query.
+     */
+    CompletableFuture<Integer> fetch(Consumer<Map<String, PgColumn>> onColumns, Consumer<Row> processor, Object... params);
+
+    CompletableFuture<Void> close();
+
+}

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

@@ -1,7 +1,14 @@
 package com.github.pgasync;
 
+import com.github.pgasync.impl.PgColumn;
+import com.github.pgasync.impl.PgResultSet;
+
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 /**
@@ -17,69 +24,79 @@ public interface QueryExecutor {
     CompletableFuture<Transaction> begin();
 
     /**
-     * Executes an anonymous prepared statement. Uses native PostgreSQL syntax with $arg instead of ?
-     * to mark parameters. Supported parameter types are String, Character, Number, Time, Date, Timestamp
-     * and byte[].
+     * Sends parameter less query script. The script may be multi query. Queries are separated with semicolons.
+     * Accumulates fetched columns, rows and affected rows counts into memory and transforms them into a ResultSet when each {@link ResultSet} is fetched.
+     * Completes returned {@link CompletableFuture} when the whole process of multiple {@link ResultSet}s fetching ends.
      *
-     * @param sql    SQL to execute
-     * @param params Parameter values
-     * @return Cold observable that emits 0-n rows.
+     * @param sql Sql Script text.
+     * @return CompletableFuture that is completed with a collection of fetched {@link ResultSet}s.
      */
-    CompletableFuture<Row> queryRows(String sql, Object... params);
+    default CompletableFuture<Collection<ResultSet>> completeScript(String sql) {
+        List<ResultSet> results = new ArrayList<>();
+        AtomicReference<Map<String, PgColumn>> columnsRef = new AtomicReference<>();
+        AtomicReference<List<Row>> rowsRef = new AtomicReference<>();
+        return script(
+                columns -> {
+                    columnsRef.set(columns);
+                    rowsRef.set(new ArrayList<>());
+                },
+                rowsRef.get()::add,
+                affected -> results.add(new PgResultSet(columnsRef.get(), rowsRef.get(), affected)),
+                sql
+        )
+                .thenApply(v -> results);
 
-    /**
-     * Executes an anonymous prepared statement. Uses native PostgreSQL syntax with $arg instead of ?
-     * to mark parameters. Supported parameter types are String, Character, Number, Time, Date, Timestamp
-     * and byte[].
-     *
-     * @param sql    SQL to execute
-     * @param params Parameter values
-     * @return Cold observable that emits a single result set.
-     */
-    CompletableFuture<ResultSet> querySet(String sql, Object... params);
+    }
 
     /**
-     * Begins a transaction.
+     * Sends parameter less query script. The script may be multi query. Queries are separated with semicolons.
+     * Unlike {@link #completeScript(String)} doesn't accumulate fetched columns, rows and affected rows counts into memory.
+     * Instead it calls passed in consumers, when columns, or particular row or an affected rows count is fetched from Postgres.
+     * Completes returned {@link CompletableFuture} when the whole process of multiple {@link ResultSet}s fetching ends.
      *
-     * @param onTransaction Called when transaction is successfully started.
-     * @param onError       Called on exception thrown
+     * @param onColumns  Columns fetched callback consumer.
+     * @param onRow      A row fetched callback consumer.
+     * @param onAffected An affected rows callback consumer. It is called when a particular {@link ResultSet} is completely fetched with its affected rows count. This callback should be used to create a {@link ResultSet} instance from already fetched columns, rows and affected rows count.
+     * @param sql        Sql Script text.
+     * @return CompletableFuture that is completed when the whole process of multiple {@link ResultSet}s fetching ends.
      */
-    default void begin(Consumer<Transaction> onTransaction, Consumer<Throwable> onError) {
-        begin()
-                .thenAccept(onTransaction)
-                .exceptionally(th -> {
-                    onError.accept(th);
-                    return null;
-                });
-    }
+    CompletableFuture<Void> script(Consumer<Map<String, PgColumn>> onColumns, Consumer<Row> onRow, Consumer<Integer> onAffected, String sql);
 
     /**
-     * Executes a simple query.
+     * Sends single query with parameters. Uses extended query protocol of Postgres.
+     * Accumulates fetched columns, rows and affected rows count into memory and transforms them into ResultSet when it is fetched.
+     * Completes returned {@link CompletableFuture} when the whole process of {@link ResultSet} fetching ends.
      *
-     * @param sql      SQL to execute.
-     * @param onResult Called when query is completed
-     * @param onError  Called on exception thrown
+     * @param sql    Sql query text with parameters substituted with ?.
+     * @param params Parameters of the query.
+     * @return CompletableFuture of {@link ResultSet}.
      */
-    default void query(String sql, Consumer<ResultSet> onResult, Consumer<Throwable> onError) {
-        query(sql, null, onResult, onError);
+    default CompletableFuture<ResultSet> completeQuery(String sql, Object... params) {
+        AtomicReference<Map<String, PgColumn>> columnsRef = new AtomicReference<>();
+        List<Row> rows = new ArrayList<>();
+        return query(
+                columnsRef::set,
+                rows::add,
+                sql,
+                params
+        )
+                .thenApply(affected -> new PgResultSet(columnsRef.get(), rows, affected));
+
     }
 
     /**
-     * Executes an anonymous prepared statement. Uses native PostgreSQL syntax with $arg instead of ?
-     * to mark parameters. Supported parameter types are String, Character, Number, Time, Date, Timestamp
-     * and byte[].
+     * Sends single query with parameters. Uses extended query protocol of Postgres.
+     * Unlike {@link #completeQuery(String, Object...)} doesn't accumulate columns, rows and affected rows counts into memory.
+     * Instead it calls passed in consumers, when columns, or particular row is fetched from Postgres.
+     * Completes returned {@link CompletableFuture} when the process of single {@link ResultSet}s fetching ends.
      *
-     * @param sql      SQL to execute
-     * @param params   List of parameters
-     * @param onResult Called when query is completed
-     * @param onError  Called on exception thrown
+     * @param onColumns Columns fetched callback consumer.
+     * @param onRow     A row fetched callback consumer.
+     * @param sql       Sql query text with parameters substituted with ?.
+     * @param params    Parameters of the query
+     * @return CompletableFuture of affected rows count.
+     * This future is used by implementation to create a {@link ResultSet} instance from already fetched columns, rows and affected rows count.
+     * Affected rows count is this future's completion value.
      */
-    default void query(String sql, List/*<Object>*/ params, Consumer<ResultSet> onResult, Consumer<Throwable> onError) {
-        querySet(sql, params != null ? params.toArray() : null)
-                .thenAccept(onResult)
-                .exceptionally(th -> {
-                    onError.accept(th);
-                    return null;
-                });
-    }
+    CompletableFuture<Integer> query(Consumer<Map<String, PgColumn>> onColumns, Consumer<Row> onRow, String sql, Object... params);
 }

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

@@ -18,9 +18,9 @@
 import java.util.Iterator;
 
 /**
- * SQL result set. Consists of 0-n result rows and amount of updated
+ * SQL result set. Consists of 0-n result rows and amount of affected
  * (INSERT/UPDATE/DELETE) rows.
- * 
+ *
  * @author Antti Laisi
  */
 public interface ResultSet extends Iterable<Row> {
@@ -30,16 +30,11 @@ public interface ResultSet extends Iterable<Row> {
      */
     Collection<String> getColumns();
 
-    /**
-     * @return Row iterator
-     */
-    Iterator<Row> iterator();
-
     /**
      * @param index Row index starting from 0
      * @return Row, never null
      */
-    Row row(int index);
+    Row at(int index);
 
     /**
      * @return Amount of result rows.
@@ -49,6 +44,6 @@ public interface ResultSet extends Iterable<Row> {
     /**
      * @return Amount of modified rows.
      */
-    int updatedRows();
+    int affectedRows();
 
 }

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

@@ -21,7 +21,7 @@
 import java.sql.Timestamp;
 
 /**
- * Row in a queryRows result set. A row consist of 0-n columns of a single type.
+ * Row in a query result set. A row consist of 0-n columns of a single type.
  * Column values can be accessed with a 0-based index or column label.
  * 
  * @author Antti Laisi