Reactive SQL Clients

quarkus.datasource.db-kind=postgresql
quarkus.datasource.username=quarkus_test
quarkus.datasource.password=quarkus_test
quarkus.datasource.reactive.url=postgresql://localhost:5432/quarkus_test

package org.acme.reactive.crud;

import java.net.URI;

import jakarta.ws.rs.DELETE;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.PUT;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.Response.ResponseBuilder;
import jakarta.ws.rs.core.Response.Status;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.vertx.mutiny.sqlclient.Pool;

@Path("fruits")
public class FruitResource {

    private final Pool client;

    public FruitResource(Pool client) {
        this.client = client;
    }
}

package org.acme.reactive.crud;

import io.quarkus.runtime.StartupEvent;
import io.vertx.mutiny.sqlclient.Pool;
import org.eclipse.microprofile.config.inject.ConfigProperty;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Observes;

@ApplicationScoped
public class DBInit {

    private final Pool client;
    private final boolean schemaCreate;

    public DBInit(Pool client, @ConfigProperty(name = "myapp.schema.create", defaultValue = "true") boolean schemaCreate) {
        this.client = client;
        this.schemaCreate = schemaCreate;
    }

    void onStart(@Observes StartupEvent ev) {
        if (schemaCreate) {
            initdb();
        }
    }

    private void initdb() {
        // TODO
    }
}

client.query("DROP TABLE IF EXISTS fruits").execute()
    .flatMap(r -> client.query("CREATE TABLE fruits (id SERIAL PRIMARY KEY, name TEXT NOT NULL)").execute())
    .flatMap(r -> client.query("INSERT INTO fruits (name) VALUES ('Kiwi')").execute())
    .flatMap(r -> client.query("INSERT INTO fruits (name) VALUES ('Durian')").execute())
    .flatMap(r -> client.query("INSERT INTO fruits (name) VALUES ('Pomelo')").execute())
    .flatMap(r -> client.query("INSERT INTO fruits (name) VALUES ('Lychee')").execute())
    .await().indefinitely();

public static Uni<Void> insertTwoFruits(Pool client, Fruit fruit1, Fruit fruit2) {
    return client.withTransaction(conn -> {
        Uni<RowSet<Row>> insertOne = conn.preparedQuery("INSERT INTO fruits (name) VALUES ($1) RETURNING id")
                .execute(Tuple.of(fruit1.name));
        Uni<RowSet<Row>> insertTwo = conn.preparedQuery("INSERT INTO fruits (name) VALUES ($1) RETURNING id")
                .execute(Tuple.of(fruit2.name));

        return Uni.combine().all().unis(insertOne, insertTwo)
                // Ignore the results (the two ids)
                .discardItems();
    });
}

return client.withTransaction(conn -> conn

        .preparedQuery("INSERT INTO person (firstname,lastname) VALUES ($1,$2) RETURNING id")
        .execute(Tuple.of(person.getFirstName(), person.getLastName()))

        .onItem().transformToUni(id -> conn.preparedQuery("INSERT INTO addr (person_id,addrline1) VALUES ($1,$2)")
                .execute(Tuple.of(id.iterator().next().getLong("id"), person.getLastName())))

        .onItem().ignore().andContinueWithNull());

PreparedQuery<RowSet<Row>> preparedQuery = client.preparedQuery("UPDATE fruits SET name = $1 WHERE id = $2");

Uni<RowSet<Row>> rowSet = preparedQuery.executeBatch(Arrays.asList(
        Tuple.of("Orange", 1),
        Tuple.of("Pear", 2),
        Tuple.of("Apple", 3)));

Uni<Integer> totalAffected = rowSet.onItem().transform(res -> {
    int total = 0;
    do {
        total += res.rowCount(); (1)
    } while ((res = res.next()) != null); (2)
    return total;
});

PreparedQuery<RowSet<Row>> preparedQuery = client.preparedQuery("INSERT INTO fruits (name) VALUES ($1) RETURNING *");

Uni<RowSet<Row>> rowSet = preparedQuery.executeBatch(Arrays.asList(
        Tuple.of("Orange"),
        Tuple.of("Pear"),
        Tuple.of("Apple")));

// Generate a Multi of RowSet items
Multi<RowSet<Row>> rowSets = rowSet.onItem().transformToMulti(res -> {
    return Multi.createFrom().generator(() -> res, (rs, emitter) -> {
        RowSet<Row> next = null;
        if (rs != null) {
            emitter.emit(rs);
            next = rs.next();
        }
        if (next == null) {
            emitter.complete();
        }
        return next;
    });
});

// Transform each RowSet into Multi of Row items and Concatenate
Multi<Row> rows = rowSets.onItem().transformToMultiAndConcatenate(Multi.createFrom()::iterable);

quarkus.datasource.db-kind=postgresql (1)
quarkus.datasource.username=user-default
quarkus.datasource.password=password-default
quarkus.datasource.reactive.url=postgresql://localhost:5432/default

quarkus.datasource."additional1".db-kind=postgresql (2)
quarkus.datasource."additional1".username=user-additional1
quarkus.datasource."additional1".password=password-additional1
quarkus.datasource."additional1".reactive.url=postgresql://localhost:5432/additional1

quarkus.datasource."additional2".db-kind=mysql (3)
quarkus.datasource."additional2".username=user-additional2
quarkus.datasource."additional2".password=password-additional2
quarkus.datasource."additional2".reactive.url=mysql://localhost:3306/additional2

@Inject (1)
Pool defaultClient;

@Inject
@ReactiveDataSource("additional1") (2)
Pool additional1Client;

@Inject
@ReactiveDataSource("additional2")
MySQLPool additional2Client;

quarkus.datasource.reactive.url=postgresql://host1:5432/default,postgresql://host2:5432/default,postgresql://host3:5432/default

quarkus.datasource.reactive.url[0]=postgresql://host1:5432/default
quarkus.datasource.reactive.url[1]=postgresql://host2:5432/default
quarkus.datasource.reactive.url[2]=postgresql://host3:5432/default

quarkus.datasource.reactive.idle-timeout=PT60M

quarkus.datasource.reactive.max-lifetime=PT60M

import jakarta.inject.Singleton;

import io.quarkus.reactive.pg.client.PgPoolCreator;
import io.vertx.pgclient.PgConnectOptions;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.PoolOptions;

@Singleton
public class CustomPgPoolCreator implements PgPoolCreator {

    @Override
    public Pool create(Input input) {
        PgConnectOptions connectOptions = input.pgConnectOptions();
        PoolOptions poolOptions = input.poolOptions();
        // Customize connectOptions, poolOptions or both, as required
        return Pool.pool(input.vertx(), connectOptions, poolOptions);
    }
}

import jakarta.inject.Inject;

import io.smallrye.mutiny.Uni;
import io.vertx.mutiny.sqlclient.Pool;

public class PipeliningExample {

    @Inject
    Pool client;

    public Uni<String> favoriteFruitAndVegetable() {
        // Explicitly acquire a connection
        return client.withConnection(conn -> {
            Uni<String> favoriteFruit = conn.query("SELECT name FROM fruits WHERE preferred IS TRUE").execute()
                    .onItem().transform(rows -> rows.iterator().next().getString("name"));
            Uni<String> favoriteVegetable = conn.query("SELECT name FROM vegetables WHERE preferred IS TRUE").execute()
                    .onItem().transform(rows -> rows.iterator().next().getString("name"));
            // favoriteFruit and favoriteVegetable unis will be subscribed at the same time
            return Uni.combine().all().unis(favoriteFruit, favoriteVegetable)
                    .combinedWith(PipeliningExample::formatMessage);
        });
    }

    private static String formatMessage(String fruit, String vegetable) {
        return String.format("The favorite fruit is %s and the favorite vegetable is %s", fruit, vegetable);
    }
}

# For PostgreSQL
quarkus.datasource.reactive.postgresql.pipelining-limit=256
# For MariaDB/MySQL
quarkus.datasource.reactive.mysql.pipelining-limit=256