317 lines
14 KiB
Java
317 lines
14 KiB
Java
package com.cygnus.db;
|
|
|
|
import java.sql.Connection;
|
|
import java.sql.PreparedStatement;
|
|
import java.sql.ResultSet;
|
|
import java.sql.SQLException;
|
|
import java.sql.Statement;
|
|
import java.util.ArrayList;
|
|
import java.util.List;
|
|
import java.util.Optional;
|
|
import javax.sql.DataSource;
|
|
|
|
public final class JdbcCygnusDbExecutor implements CygnusDbExecutor {
|
|
private final DataSource dataSource;
|
|
private final QueryDefinitionProvider queryProvider;
|
|
private final ExecutorOptions options;
|
|
private final ParameterBinder binder = new ParameterBinder();
|
|
private final ObjectRowMapperFactory mapperFactory = new ObjectRowMapperFactory();
|
|
|
|
public JdbcCygnusDbExecutor(DataSource dataSource, QueryDefinitionProvider queryProvider) {
|
|
this(dataSource, queryProvider, ExecutorOptions.DEFAULT);
|
|
}
|
|
|
|
public JdbcCygnusDbExecutor(
|
|
DataSource dataSource, QueryDefinitionProvider queryProvider, ExecutorOptions options) {
|
|
this.dataSource = java.util.Objects.requireNonNull(dataSource, "dataSource");
|
|
this.queryProvider = java.util.Objects.requireNonNull(queryProvider, "queryProvider");
|
|
this.options = java.util.Objects.requireNonNull(options, "options");
|
|
}
|
|
|
|
@Override
|
|
public <T> DbResult<T> execute(int queryId, Object[] parameters, Class<T> responseType) {
|
|
QueryDefinition definition = queryProvider.get(queryId);
|
|
if (definition.type() == QueryType.SELECT) {
|
|
return withReadConnection(queryId, connection ->
|
|
executeRowsOrCount(connection, definition, parameters, responseType));
|
|
}
|
|
if (definition.type() == QueryType.PROCEDURE) {
|
|
return executeRowsOrCount(definition, parameters, responseType);
|
|
}
|
|
return DbResult.affected(definition.type(), executeUpdate(definition, parameters));
|
|
}
|
|
|
|
@Override
|
|
public DbResult<RowView> execute(int queryId, Object[] parameters) {
|
|
return execute(queryId, parameters, RowView.class);
|
|
}
|
|
|
|
@Override
|
|
public <T> List<T> query(int queryId, Object[] parameters, Class<T> responseType) {
|
|
return withReadConnection(queryId, connection -> query(
|
|
connection, require(queryId, QueryType.SELECT), parameters, responseType));
|
|
}
|
|
|
|
@Override
|
|
public <T> List<T> query(int queryId, Object[] parameters, RowMapper<T> mapper) {
|
|
return withReadConnection(queryId, connection -> query(
|
|
connection, require(queryId, QueryType.SELECT), parameters, mapper));
|
|
}
|
|
|
|
@Override
|
|
public List<RowView> query(int queryId, Object[] parameters) {
|
|
return query(queryId, parameters, RowView.class);
|
|
}
|
|
|
|
@Override
|
|
public <T> Optional<T> queryOne(int queryId, Object[] parameters, Class<T> responseType) {
|
|
List<T> rows = query(queryId, parameters, responseType);
|
|
if (rows.size() > 1) throw new DatabaseExecutionException(
|
|
queryId, "Expected one row but received " + rows.size(), null);
|
|
return rows.stream().findFirst();
|
|
}
|
|
|
|
@Override
|
|
public int insert(int queryId, Object[] parameters) {
|
|
return executeUpdate(require(queryId, QueryType.INSERT), parameters);
|
|
}
|
|
|
|
@Override
|
|
public <K> K insert(int queryId, Object[] parameters, Class<K> generatedKeyType) {
|
|
QueryDefinition definition = require(queryId, QueryType.INSERT);
|
|
return withConnection(queryId, connection -> {
|
|
try (PreparedStatement statement = prepare(
|
|
connection, definition, Statement.RETURN_GENERATED_KEYS)) {
|
|
binder.bind(statement, parameters);
|
|
statement.executeUpdate();
|
|
try (ResultSet keys = statement.getGeneratedKeys()) {
|
|
if (!keys.next()) throw new DatabaseExecutionException(
|
|
queryId, "Insert did not return a generated key", null);
|
|
return ValueConverter.convert(keys.getObject(1), generatedKeyType);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
@Override
|
|
public int update(int queryId, Object[] parameters) {
|
|
return executeUpdate(require(queryId, QueryType.UPDATE), parameters);
|
|
}
|
|
|
|
@Override
|
|
public int delete(int queryId, Object[] parameters) {
|
|
return executeUpdate(require(queryId, QueryType.DELETE), parameters);
|
|
}
|
|
|
|
@Override
|
|
public <T> T procedure(int queryId, Object[] parameters, Class<T> responseType) {
|
|
List<T> rows = procedureRows(queryId, parameters, responseType);
|
|
return rows.isEmpty() ? null : rows.getFirst();
|
|
}
|
|
|
|
@Override
|
|
public <T> List<T> procedureRows(int queryId, Object[] parameters, Class<T> responseType) {
|
|
DbResult<T> result = executeRowsOrCount(
|
|
require(queryId, QueryType.PROCEDURE), parameters, responseType);
|
|
return result.rows();
|
|
}
|
|
|
|
@Override
|
|
public <T> void stream(
|
|
int queryId, Object[] parameters, RowMapper<T> mapper, RowConsumer<T> consumer) {
|
|
QueryDefinition definition = require(queryId, QueryType.SELECT);
|
|
withReadConnection(queryId, connection -> {
|
|
try (PreparedStatement statement = prepare(connection, definition);
|
|
ResultSet result = executeQuery(statement, parameters)) {
|
|
while (result.next()) consumer.accept(mapper.map(result));
|
|
return null;
|
|
}
|
|
});
|
|
}
|
|
|
|
@Override
|
|
public <T> T transaction(TransactionCallback<T> callback) {
|
|
try (Connection connection = dataSource.getConnection()) {
|
|
boolean autoCommit = connection.getAutoCommit();
|
|
connection.setAutoCommit(false);
|
|
try {
|
|
T result = callback.execute(new TransactionOperationsImpl(connection));
|
|
connection.commit();
|
|
return result;
|
|
} catch (Exception exception) {
|
|
try { connection.rollback(); } catch (SQLException rollback) { exception.addSuppressed(rollback); }
|
|
if (exception instanceof RuntimeException runtime) throw runtime;
|
|
throw new DatabaseExecutionException(0, "Transaction failed", exception);
|
|
} finally {
|
|
connection.setAutoCommit(autoCommit);
|
|
}
|
|
} catch (SQLException exception) {
|
|
throw new DatabaseExecutionException(0, "Unable to manage transaction", exception);
|
|
}
|
|
}
|
|
|
|
private <T> DbResult<T> executeRowsOrCount(
|
|
QueryDefinition definition, Object[] parameters, Class<T> responseType) {
|
|
return withConnection(definition.queryId(), connection ->
|
|
executeRowsOrCount(connection, definition, parameters, responseType));
|
|
}
|
|
|
|
private <T> DbResult<T> executeRowsOrCount(
|
|
Connection connection,
|
|
QueryDefinition definition,
|
|
Object[] parameters,
|
|
Class<T> responseType) throws SQLException {
|
|
try (PreparedStatement statement = prepare(connection, definition)) {
|
|
binder.bind(statement, parameters);
|
|
boolean rows = statement.execute();
|
|
if (!rows) return DbResult.affected(definition.type(), statement.getUpdateCount());
|
|
try (ResultSet result = statement.getResultSet()) {
|
|
return DbResult.rows(definition.type(), map(result, responseType));
|
|
}
|
|
}
|
|
}
|
|
|
|
private int executeUpdate(QueryDefinition definition, Object[] parameters) {
|
|
return withConnection(definition.queryId(), connection ->
|
|
executeUpdate(connection, definition, parameters));
|
|
}
|
|
|
|
private int executeUpdate(
|
|
Connection connection, QueryDefinition definition, Object[] parameters) throws SQLException {
|
|
try (PreparedStatement statement = prepare(connection, definition)) {
|
|
binder.bind(statement, parameters);
|
|
return statement.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private <T> List<T> query(
|
|
Connection connection,
|
|
QueryDefinition definition,
|
|
Object[] parameters,
|
|
Class<T> responseType) throws SQLException {
|
|
try (PreparedStatement statement = prepare(connection, definition);
|
|
ResultSet result = executeQuery(statement, parameters)) {
|
|
return map(result, responseType);
|
|
}
|
|
}
|
|
|
|
private <T> List<T> query(
|
|
Connection connection,
|
|
QueryDefinition definition,
|
|
Object[] parameters,
|
|
RowMapper<T> mapper) throws SQLException {
|
|
try (PreparedStatement statement = prepare(connection, definition);
|
|
ResultSet result = executeQuery(statement, parameters)) {
|
|
List<T> rows = new ArrayList<>();
|
|
while (result.next()) rows.add(mapper.map(result));
|
|
return List.copyOf(rows);
|
|
}
|
|
}
|
|
|
|
private <T> List<T> map(ResultSet result, Class<T> responseType) throws SQLException {
|
|
RowMapper<T> mapper = mapperFactory.create(responseType, result.getMetaData());
|
|
List<T> rows = new ArrayList<>();
|
|
while (result.next()) rows.add(mapper.map(result));
|
|
return List.copyOf(rows);
|
|
}
|
|
|
|
private ResultSet executeQuery(PreparedStatement statement, Object[] parameters) throws SQLException {
|
|
binder.bind(statement, parameters);
|
|
return statement.executeQuery();
|
|
}
|
|
|
|
private PreparedStatement prepare(Connection connection, QueryDefinition definition) throws SQLException {
|
|
PreparedStatement statement = connection.prepareStatement(
|
|
definition.sql(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
|
configure(statement);
|
|
return statement;
|
|
}
|
|
|
|
private PreparedStatement prepare(
|
|
Connection connection, QueryDefinition definition, int generatedKeys) throws SQLException {
|
|
PreparedStatement statement = connection.prepareStatement(definition.sql(), generatedKeys);
|
|
configure(statement);
|
|
return statement;
|
|
}
|
|
|
|
private void configure(PreparedStatement statement) throws SQLException {
|
|
statement.setFetchSize(options.fetchSize());
|
|
statement.setQueryTimeout(options.queryTimeoutSeconds());
|
|
}
|
|
|
|
private QueryDefinition require(int queryId, QueryType expected) {
|
|
QueryDefinition definition = queryProvider.get(queryId);
|
|
if (definition.type() != expected) throw new QueryDefinitionException(
|
|
"Query " + queryId + " is " + definition.type() + ", not " + expected);
|
|
return definition;
|
|
}
|
|
|
|
private <T> T withConnection(int queryId, SqlWork<T> work) {
|
|
try (Connection connection = dataSource.getConnection()) {
|
|
return work.execute(connection);
|
|
} catch (DatabaseExecutionException exception) {
|
|
throw exception;
|
|
} catch (Exception exception) {
|
|
throw new DatabaseExecutionException(queryId, "Database operation failed for query " + queryId, exception);
|
|
}
|
|
}
|
|
|
|
private <T> T withReadConnection(int queryId, SqlWork<T> work) {
|
|
try (Connection connection = dataSource.getConnection()) {
|
|
boolean autoCommit = connection.getAutoCommit();
|
|
connection.setAutoCommit(false); // PostgreSQL requires this for cursor-based fetchSize.
|
|
try {
|
|
T value = work.execute(connection);
|
|
connection.commit();
|
|
return value;
|
|
} catch (Exception exception) {
|
|
try { connection.rollback(); } catch (SQLException rollback) { exception.addSuppressed(rollback); }
|
|
if (exception instanceof RuntimeException runtime) throw runtime;
|
|
throw new DatabaseExecutionException(queryId,
|
|
"Database read failed for query " + queryId, exception);
|
|
} finally {
|
|
connection.setAutoCommit(autoCommit);
|
|
}
|
|
} catch (SQLException exception) {
|
|
throw new DatabaseExecutionException(queryId,
|
|
"Unable to manage database read for query " + queryId, exception);
|
|
}
|
|
}
|
|
|
|
@FunctionalInterface
|
|
private interface SqlWork<T> { T execute(Connection connection) throws Exception; }
|
|
|
|
private final class TransactionOperationsImpl implements TransactionOperations {
|
|
private final Connection connection;
|
|
private TransactionOperationsImpl(Connection connection) { this.connection = connection; }
|
|
|
|
@Override
|
|
public <T> List<T> query(int queryId, Object[] parameters, Class<T> type) {
|
|
try { return JdbcCygnusDbExecutor.this.query(
|
|
connection, require(queryId, QueryType.SELECT), parameters, type); }
|
|
catch (SQLException exception) { throw failure(queryId, exception); }
|
|
}
|
|
|
|
@Override
|
|
public <T> Optional<T> queryOne(int queryId, Object[] parameters, Class<T> type) {
|
|
List<T> rows = query(queryId, parameters, type);
|
|
if (rows.size() > 1) throw new DatabaseExecutionException(
|
|
queryId, "Expected one row but received " + rows.size(), null);
|
|
return rows.stream().findFirst();
|
|
}
|
|
|
|
@Override public int insert(int id, Object[] values) { return updateType(id, values, QueryType.INSERT); }
|
|
@Override public int update(int id, Object[] values) { return updateType(id, values, QueryType.UPDATE); }
|
|
@Override public int delete(int id, Object[] values) { return updateType(id, values, QueryType.DELETE); }
|
|
|
|
private int updateType(int id, Object[] values, QueryType type) {
|
|
try { return executeUpdate(connection, require(id, type), values); }
|
|
catch (SQLException exception) { throw failure(id, exception); }
|
|
}
|
|
|
|
private DatabaseExecutionException failure(int id, SQLException exception) {
|
|
return new DatabaseExecutionException(id, "Transaction operation failed for query " + id, exception);
|
|
}
|
|
}
|
|
}
|