WIP - 0

Hacking around the API to make Future<Buffer> coming out

Author: psolomin

vertx-pg-client/src/main/java/examples/PgClientExamples.java

@@ -748,7 +748,7 @@ public void importDataToDb(Vertx vertx, PgConnection client) {
 
   public void exportDataFromDb(Vertx vertx, PgConnection client) {
     String path = "path/to/file";
-    client.copyToBytes("COPY my_table TO STDOUT (FORMAT csv, HEADER)").execute()
+    client.copyToBytes("COPY my_table TO STDOUT (FORMAT csv, HEADER)")
       .flatMap(buffer -> vertx.fileSystem().writeFile("path/to/file.csv", buffer))
       .onSuccess(res -> System.out.println("Data exported to " + path));
   }

vertx-pg-client/src/main/java/io/vertx/pgclient/PgConnection.java

@@ -131,7 +131,7 @@ static Future<PgConnection> connect(Vertx vertx, String connectionUri) {
    * @param sql COPY command (example {@code COPY my_table TO STDOUT (FORMAT csv)})
    * @return async result of bytes container data will be written to
    */
-  Query<Buffer> copyToBytes(String sql);
+  Future<Buffer> copyToBytes(String sql);
 
   /**
    * Send a request cancellation message to tell the server to cancel processing request in this connection.

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java

@@ -19,6 +19,7 @@
 import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.impl.future.PromiseInternal;
 import io.vertx.pgclient.PgConnectOptions;
 import io.vertx.pgclient.PgConnection;
 import io.vertx.pgclient.PgNotice;
@@ -29,10 +30,16 @@
 import io.vertx.sqlclient.Query;
 import io.vertx.sqlclient.Row;
 import io.vertx.sqlclient.RowSet;
+import io.vertx.sqlclient.SqlResult;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.Notification;
+import io.vertx.sqlclient.impl.QueryResultBuilder;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
 import io.vertx.sqlclient.impl.SqlConnectionBase;
+import io.vertx.sqlclient.impl.SqlResultImpl;
+import io.vertx.sqlclient.impl.command.CommandBase;
+
+import java.util.function.Function;
 
 public class PgConnectionImpl extends SqlConnectionBase<PgConnectionImpl> implements PgConnection  {
 
@@ -120,8 +127,14 @@ public Query<RowSet<Row>> copyToRows(String sql) {
   }
 
   @Override
-  public Query<Buffer> copyToBytes(String sql) {
-    return null;
+  public Future<Buffer> copyToBytes(String sql) {
+    Function<Buffer, SqlResultImpl<Buffer>> factory = null;
+    PromiseInternal<SqlResult<Buffer>> promise = null;
+
+    QueryResultBuilder<Buffer, SqlResultImpl<Buffer>, SqlResult<Buffer>> resultHandler =
+      new QueryResultBuilder<>(factory, promise);
+    CopyOutCommand cmd = new CopyOutCommand(sql, resultHandler);
+    return this.schedule(context, cmd);
   }
 
   @Override
@@ -150,4 +163,14 @@ public PgConnection cancelRequest(Handler<AsyncResult<Void>> handler) {
     }
     return this;
   }
+
+  private class CopyOutCommand extends CommandBase<Buffer> {
+    private final String sql;
+    private final QueryResultBuilder<Buffer, SqlResultImpl<Buffer>, SqlResult<Buffer>> resultHandler;
+
+    CopyOutCommand(String sql, QueryResultBuilder<Buffer, SqlResultImpl<Buffer>, SqlResult<Buffer>> resultHandler) {
+      this.sql = sql;
+      this.resultHandler = resultHandler;
+    }
+  }
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgProtocolConstants.java

@@ -72,7 +72,13 @@ public class PgProtocolConstants {
   public static final byte MESSAGE_TYPE_SSL_YES = 'S';
   public static final byte MESSAGE_TYPE_SSL_NO = 'N';
 
-  // COPY-related
+  /**
+   * COPY-out messages.
+   *
+   * <p>Other messages which might appear in between CopyData:
+   * <li>NoticeResponse
+   * <li>ParameterStatus
+   */
   public static final byte MESSAGE_TYPE_COPY_OUT_RESPONSE = 'H';
   public static final byte MESSAGE_TYPE_COPY_DATA = 'd';
   public static final byte MESSAGE_TYPE_COPY_COMPLETION = 'c';

vertx-pg-client/src/test/java/io/vertx/pgclient/PgConnectionCopyTest.java

@@ -2,6 +2,7 @@
 
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
+import org.junit.Ignore;
 import org.junit.Test;
 
 public class PgConnectionCopyTest extends PgConnectionTestBase {
@@ -32,6 +33,24 @@ public void testCopyToRows(TestContext ctx) {
     }));
   }
 
+  @Test
+  @Ignore("For now it just hangs forever")
+  public void testCopyToCsvBytes(TestContext ctx) {
+    Async async = ctx.async();
+    connector.accept(ctx.asyncAssertSuccess(conn -> {
+      deleteFromTestTable(ctx, conn, () -> {
+        insertIntoTestTable(ctx, conn, 10, () -> {
+          PgConnection pgConn = (PgConnection) conn;
+          pgConn.copyToBytes("COPY my_table TO STDOUT (FORMAT csv HEADER)")
+            .onComplete(ctx.asyncAssertSuccess(buffer -> {
+              buffer.getBytes();
+              async.complete();
+            }));
+        });
+      });
+    }));
+  }
+
   /**
    * Just a thingy to eavesdrop protocol interactions.
    *
@@ -54,9 +73,24 @@ public void testSimpleQuery(TestContext ctx) {
     Async async = ctx.async();
     connector.accept(ctx.asyncAssertSuccess(conn -> {
       conn
-        .query("COPY world TO STDOUT (FORMAT csv)")
+        .query("select 1")
+        .execute()
+        .onComplete(ctx.asyncAssertSuccess(result1 -> {
+          ctx.assertEquals(1, result1.size());
+          async.complete();
+        }));
+    }));
+  }
+
+  @Test
+  public void testMakeSureCopyOutProtocolIsDefined(TestContext ctx) {
+    Async async = ctx.async();
+    connector.accept(ctx.asyncAssertSuccess(conn -> {
+      conn
+        .query("copy fortune to stdout (format csv)")
         .execute()
         .onComplete(ctx.asyncAssertSuccess(result1 -> {
+          // nothing comes out
           ctx.assertEquals(0, result1.size());
           async.complete();
         }));

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryResultBuilder.java

@@ -38,7 +38,8 @@ public class QueryResultBuilder<T, R extends SqlResultBase<T>, L extends SqlResu
   private Throwable failure;
   private boolean suspended;
 
-  QueryResultBuilder(Function<T, R> factory, PromiseInternal<L> handler) {
+  // TODO: public-ing this shouldn't be needed
+  public QueryResultBuilder(Function<T, R> factory, PromiseInternal<L> handler) {
     this.factory = factory;
     this.handler = handler;
   }