AsyncSQLConnectionImpl shouldn't require the application to serialize calls.

[P4Awd]

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.asyncsql.MySQLClient;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.RunTestOnContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.net.HttpURLConnection;
import java.util.concurrent.atomic.AtomicInteger;

@RunWith(VertxUnitRunner.class)
public class AsyncSQLTest {

    @Rule
    public RunTestOnContext runTestOnContext = new RunTestOnContext();

    /**
     * AsyncSQLConnectionImpl doesn't handle non serialized access
     *
     * @param testContext
     */
    @Test
    public void test(TestContext testContext) {
        Vertx vertx = runTestOnContext.vertx();

        Async async = testContext.async();

        vertx.deployVerticle(new AbstractVerticle() {

            SQLClient sqlClient;

            @Override
            public void start(Future<Void> startFuture) throws Exception {
                JsonObject jsonObject = new JsonObject();
                jsonObject.put("host", "localhost");
                jsonObject.put("port", 3306);
                jsonObject.put("username", "root");
                jsonObject.put("password", "password");
                jsonObject.put("database", "mysql");

                sqlClient = MySQLClient.createShared(getVertx(), jsonObject);

                HttpServerOptions httpServerOptions = new HttpServerOptions();
                httpServerOptions.setHost("localhost");
                httpServerOptions.setPort(9091);

                getVertx().createHttpServer(httpServerOptions)
                        .requestHandler(httpServerRequest -> {
                            try {
                                sqlClient.getConnection(connectedResult -> {
                                    testContext.assertTrue(connectedResult.succeeded());

                                    SQLConnection connection = connectedResult.result();
                                    try {
                                        connection.setAutoCommit(false, disableAutoCommitResult -> {
                                            testContext.assertTrue(disableAutoCommitResult.succeeded());

                                            try {
                                                AtomicInteger counter = new AtomicInteger();
                                                getVertx()
                                                        .setPeriodic(500, tickResult -> {
                                                            if (counter.get() >= 1000) {
                                                                connection.commit(commitResult -> {
                                                                    testContext.assertTrue(commitResult.succeeded());
                                                                    connection.close(closeResult -> httpServerRequest.response()
                                                                            .setStatusCode(HttpURLConnection.HTTP_OK)
                                                                            .end());
                                                                });
                                                            }
                                                        });
                                                for (int i = 0; i < 1000; i++) {
                                                    // These could all be inserts running in a single transaction.
                                                    //
                                                    // In more complex rxified flows... aka real world applications
                                                    // it's unreasonable to expect the application to
                                                    // serialize queries to a connection which is maintaining
                                                    // a reference to the active transaction
                                                    //
                                                    // Consider a case like Observable.combineLatestDelayError
                                                    // where multiple queries are dispatched then collected and
                                                    // require repeatable-read transaction isolation
                                                    connection.query("select 1", queryResult -> {
                                                        if (!queryResult.succeeded()) {
                                                            queryResult.cause().printStackTrace();
                                                        }
                                                        testContext.assertTrue(queryResult.succeeded());
                                                        counter.incrementAndGet();
                                                    });
                                                }
                                            } catch (Throwable e) {
                                                handleError(httpServerRequest, e);
                                            }
                                        });
                                    } catch (Throwable e) {
                                        handleError(httpServerRequest, e);
                                    }
                                });
                            } catch (Throwable e) {
                                handleError(httpServerRequest, e);
                            }
                        })
                        .listen(9091, "localhost", event -> {
                            if (event.succeeded()) {
                                startFuture.complete();
                            } else {
                                startFuture.fail(event.cause());
                            }
                        });
            }

            @Override
            public void stop(Future<Void> stopFuture) throws Exception {
                sqlClient.close();
                stopFuture.complete();
            }
        }, event -> {
            if (!event.succeeded()) {
                event.cause().printStackTrace();
            }
            testContext.assertTrue(event.succeeded());
            HttpClient httpClient = vertx.createHttpClient();
            httpClient.postAbs("http://localhost:9091", httpClientResponse -> {
                int status = httpClientResponse.statusCode();
                testContext.assertEquals(HttpURLConnection.HTTP_OK, status);
                async.complete();
            }).end();
        });

    }

    private void handleError(HttpServerRequest httpServerRequest, Throwable event) {
        event.printStackTrace();
        httpServerRequest.response()
                .setStatusCode(HttpURLConnection.HTTP_INTERNAL_ERROR)
                .end();
    }
}

[P4Awd]

Perhaps com.github.mauricio.async.db.pool.ConnectionPool should be used by io.vertx.ext.asyncsql.impl.AsyncSQLConnectionImpl.