WIP: Stuck with copyToBytes

Author: psolomin

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

@@ -749,7 +749,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)")
-      .flatMap(buffer -> vertx.fileSystem().writeFile("path/to/file.csv", buffer))
+      .flatMap(result -> vertx.fileSystem().writeFile("path/to/file.csv", result.value()))
       .onSuccess(res -> System.out.println("Data exported to " + path));
   }
 

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

@@ -29,6 +29,7 @@
 import io.vertx.sqlclient.Row;
 import io.vertx.sqlclient.RowSet;
 import io.vertx.sqlclient.SqlConnection;
+import io.vertx.sqlclient.SqlResult;
 
 import java.util.List;
 
@@ -130,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
    */
-  Future<Buffer> copyToBytes(String sql);
+  Future<SqlResult<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

@@ -20,12 +20,14 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.buffer.impl.BufferImpl;
 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;
 import io.vertx.pgclient.PgNotification;
+import io.vertx.pgclient.impl.codec.CopyOutCommand;
 import io.vertx.pgclient.impl.codec.NoticeResponse;
 import io.vertx.pgclient.impl.codec.TxFailedEvent;
 import io.vertx.pgclient.spi.PgDriver;
@@ -35,13 +37,17 @@
 import io.vertx.sqlclient.SqlResult;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.Notification;
+import io.vertx.sqlclient.impl.QueryExecutor;
 import io.vertx.sqlclient.impl.QueryResultBuilder;
+import io.vertx.sqlclient.impl.QueryResultHandler;
 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 io.vertx.sqlclient.impl.command.QueryCommandBase;
+import io.vertx.sqlclient.impl.command.SimpleQueryCommand;
 
 import java.util.function.Function;
+import java.util.stream.Collector;
 
 public class PgConnectionImpl extends SqlConnectionBase<PgConnectionImpl> implements PgConnection  {
 
@@ -129,14 +135,29 @@ public Query<RowSet<Row>> copyToRows(String sql) {
   }
 
   @Override
-  public Future<Buffer> copyToBytes(String sql) {
-    Function<Buffer, SqlResultImpl<Buffer>> factory = null;
-    PromiseInternal<SqlResult<Buffer>> promise = null;
+  public Future<SqlResult<Buffer>> copyToBytes(String sql) {
+    Function<Buffer, SqlResultImpl<Buffer>> factory = SqlResultImpl::new;
+    PromiseInternal<SqlResult<Buffer>> promise = context.promise();
 
     QueryResultBuilder<Buffer, SqlResultImpl<Buffer>, SqlResult<Buffer>> resultHandler =
       new QueryResultBuilder<>(factory, promise);
-    CopyOutCommand cmd = new CopyOutCommand(sql, resultHandler);
-    return this.schedule(context, cmd);
+
+    Collector<Row, Buffer, Buffer> collector = Collector.of(
+      BufferImpl::new,
+      (v, row) -> {
+        System.out.println(row);
+      },
+      (v1, v2) -> null,
+      Function.identity()
+    );
+
+    SimpleQueryCommand<Buffer> cmd = new SimpleQueryCommand<>(
+      sql, true, false, collector, resultHandler);
+    // this.schedule(promise.context(), cmd);
+
+    QueryExecutor executor = new QueryExecutor(factory, collector);
+    executor.executeSimpleQuery(this, sql, false, false, promise);
+    return promise.future();
   }
 
   @Override
@@ -159,13 +180,4 @@ public Future<Void> cancelRequest() {
     return promise.future();
   }
 
-  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/CopyOutCommand.java

@@ -0,0 +1,17 @@
+package io.vertx.pgclient.impl.codec;
+
+import io.vertx.core.buffer.Buffer;
+import io.vertx.sqlclient.SqlResult;
+import io.vertx.sqlclient.impl.QueryResultBuilder;
+import io.vertx.sqlclient.impl.SqlResultImpl;
+import io.vertx.sqlclient.impl.command.CommandBase;
+
+public class CopyOutCommand extends CommandBase<Buffer> {
+  private final String sql;
+  private final QueryResultBuilder<Buffer, SqlResultImpl<Buffer>, SqlResult<Buffer>> resultHandler;
+
+  public 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/CopyOutCommandCodec.java

@@ -0,0 +1,4 @@
+package io.vertx.pgclient.impl.codec;
+
+public class CopyOutCommandCodec {
+}

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

@@ -34,16 +34,15 @@ 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, () -> {
+        insertIntoTestTable(ctx, conn, 1, () -> {
           PgConnection pgConn = (PgConnection) conn;
-          pgConn.copyToBytes("COPY my_table TO STDOUT (FORMAT csv HEADER)")
-            .onComplete(ctx.asyncAssertSuccess(buffer -> {
-              buffer.getBytes();
+          pgConn.copyToBytes("COPY Test TO STDOUT (FORMAT csv)")
+            .onComplete(ctx.asyncAssertSuccess(result -> {
+              result.value().getBytes();
               async.complete();
             }));
         });

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

@@ -34,7 +34,7 @@
 /**
  * Executes query.
  */
-class QueryExecutor<T, R extends SqlResultBase<T>, L extends SqlResult<T>> {
+public class QueryExecutor<T, R extends SqlResultBase<T>, L extends SqlResult<T>> {
 
   private final Function<T, R> factory;
   private final Collector<Row, ?, T> collector;
@@ -49,7 +49,7 @@ private QueryResultBuilder<T, R, L> createHandler(PromiseInternal<L> promise) {
     return new QueryResultBuilder<>(factory, promise);
   }
 
-  void executeSimpleQuery(CommandScheduler scheduler,
+  public void executeSimpleQuery(CommandScheduler scheduler,
                           String sql,
                           boolean autoCommit,
                           boolean singleton,