68 lines
2.8 KiB
Java
68 lines
2.8 KiB
Java
package com.cygnus.cloud.database;
|
|
|
|
import io.vertx.sqlclient.Pool;
|
|
import io.vertx.sqlclient.Row;
|
|
import io.vertx.sqlclient.RowSet;
|
|
import io.vertx.sqlclient.SqlConnection;
|
|
import io.vertx.sqlclient.Tuple;
|
|
import java.util.function.Function;
|
|
import org.springframework.stereotype.Service;
|
|
|
|
import reactor.core.publisher.Mono;
|
|
|
|
/**
|
|
* Reactor boundary around the Vert.x PostgreSQL pool. Feature repositories
|
|
* should use parameterized SQL and pass values through {@link Tuple}.
|
|
*/
|
|
@Service
|
|
public class ReactiveDatabaseClient {
|
|
|
|
private final Pool pool;
|
|
|
|
public ReactiveDatabaseClient(Pool pool) {
|
|
this.pool = pool;
|
|
}
|
|
|
|
public Mono<RowSet<Row>> query(String sql) {
|
|
return Mono.fromCompletionStage(() -> pool.query(sql).execute().toCompletionStage());
|
|
}
|
|
|
|
public Mono<RowSet<Row>> preparedQuery(String sql, Tuple parameters) {
|
|
return Mono.fromCompletionStage(
|
|
() -> pool.preparedQuery(sql).execute(parameters).toCompletionStage());
|
|
}
|
|
|
|
public Mono<Integer> preparedUpdate(String sql, Tuple parameters) {
|
|
return preparedQuery(sql, parameters).map(RowSet::rowCount);
|
|
}
|
|
|
|
public <T> Mono<T> inTransaction(Function<SqlConnection, Mono<T>> work) {
|
|
return Mono.usingWhen(
|
|
Mono.fromCompletionStage(() -> pool.getConnection().toCompletionStage()),
|
|
connection -> Mono.fromCompletionStage(
|
|
() -> connection.begin().toCompletionStage())
|
|
.flatMap(transaction -> work.apply(connection)
|
|
.flatMap(result -> Mono.fromCompletionStage(
|
|
() -> transaction.commit().toCompletionStage())
|
|
.thenReturn(result))
|
|
.onErrorResume(error -> Mono.fromCompletionStage(
|
|
() -> transaction.rollback().toCompletionStage())
|
|
.onErrorResume(ignored -> Mono.empty())
|
|
.then(Mono.error(error)))),
|
|
connection -> Mono.fromCompletionStage(
|
|
() -> connection.close().toCompletionStage()),
|
|
(connection, error) -> Mono.fromCompletionStage(
|
|
() -> connection.close().toCompletionStage()),
|
|
connection -> Mono.fromCompletionStage(
|
|
() -> connection.close().toCompletionStage()));
|
|
}
|
|
|
|
public Mono<RowSet<Row>> preparedQuery(
|
|
SqlConnection connection, String sql, Tuple parameters) {
|
|
return Mono.fromCompletionStage(
|
|
() -> connection.preparedQuery(sql)
|
|
.execute(parameters)
|
|
.toCompletionStage());
|
|
}
|
|
}
|