Set the query result as not suspended to avoid infinite streams - fixes

#207

src/main/java/io/vertx/jdbcclient/impl/ConnectionImpl.java

@@ -123,7 +123,7 @@ private <R> void handle(JDBCQueryAction<?, R> action, QueryResultHandler<R> hand
         ar.result()
           .handle(handler);
 
-        promise.complete(true);
+        promise.complete(false);
       } else {
         promise.fail(ar.cause());
       }

src/test/java/io/vertx/jdbcclient/ClientTest.java

@@ -1,10 +1,13 @@
 package io.vertx.jdbcclient;
 
+import io.vertx.core.Future;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.sqlclient.*;
 import org.junit.Test;
 
 import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 public class ClientTest extends ClientTestBase {
@@ -97,4 +100,26 @@ private void testTransaction(TestContext testCtx, boolean commit) throws Excepti
         }));
     }));
   }
+
+  @Test
+  public void testStream(TestContext should) {
+    client.<List<Row>>withTransaction(tx ->
+      tx.prepare("SELECT CURRENT_DATE AS today, CURRENT_TIME AS now FROM (VALUES(0))")
+        .map(pS -> pS.createStream(200))
+        .flatMap(stream -> Future
+          .future(promise -> {
+            List<Row> rows = new ArrayList<>();
+            stream.exceptionHandler(promise::fail);
+            stream.endHandler(v -> promise.complete(rows));
+            stream.handler(row -> {
+              should.assertEquals(0, rows.size());
+              rows.add(row);
+            });
+          })
+        )
+    )
+      .onComplete(should.asyncAssertSuccess(rows -> {
+        should.assertEquals(1, rows.size());
+      }));
+  }
 }