#220 - Introduce R2dbcEntityTemplate.
We now provide a Template API that exposes entity-centric methods. It complements DatabaseClient's simple object mapper methods. Original pull request: #287.
This commit is contained in:
Mark Paluch
@@ -204,6 +204,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
|
||||
return ConnectionFactoryUtils.getConnection(obtainConnectionFactory());
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtain the {@link ReactiveDataAccessStrategy}.
|
||||
*
|
||||
* @return a the ReactiveDataAccessStrategy.
|
||||
*/
|
||||
protected ReactiveDataAccessStrategy getDataAccessStrategy() {
|
||||
return dataAccessStrategy;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release the {@link Connection}.
|
||||
*
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.r2dbc.core;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.dao.TransientDataAccessResourceException;
|
||||
import org.springframework.data.r2dbc.query.Query;
|
||||
import org.springframework.data.r2dbc.query.Update;
|
||||
|
||||
/**
|
||||
* Interface specifying a basic set of reactive R2DBC operations using entities. Implemented by
|
||||
* {@link R2dbcEntityTemplate}. Not often used directly, but a useful option to enhance testability, as it can easily be
|
||||
* mocked or stubbed.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @since 1.1
|
||||
* @see DatabaseClient
|
||||
*/
|
||||
public interface R2dbcEntityOperations extends FluentR2dbcOperations {
|
||||
|
||||
/**
|
||||
* Expose the underlying {@link DatabaseClient} to allow SQL operations.
|
||||
*
|
||||
* @return the underlying {@link DatabaseClient}.
|
||||
* @see DatabaseClient
|
||||
*/
|
||||
DatabaseClient getDatabaseClient();
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Methods dealing with org.springframework.data.r2dbc.query.Query
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users
|
||||
* to further refine the selection Query using a {@link Query} predicate to determine how many entities of the given
|
||||
* {@link Class type} match the Query.
|
||||
*
|
||||
* @param query user-defined count {@link Query} to execute; must not be {@literal null}.
|
||||
* @param entityClass {@link Class type} of the entity; must not be {@literal null}.
|
||||
* @return the number of existing entities.
|
||||
* @throws DataAccessException if any problem occurs while executing the query.
|
||||
*/
|
||||
Mono<Long> count(Query query, Class<?> entityClass) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Determine whether the result for {@code entityClass} {@link Query} yields at least one row.
|
||||
*
|
||||
* @param query user-defined exists {@link Query} to execute; must not be {@literal null}.
|
||||
* @param entityClass {@link Class type} of the entity; must not be {@literal null}.
|
||||
* @return {@literal true} if the object exists.
|
||||
* @throws DataAccessException if any problem occurs while executing the query.
|
||||
* @since 2.1
|
||||
*/
|
||||
Mono<Boolean> exists(Query query, Class<?> entityClass) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Execute a {@code SELECT} query and convert the resulting items to a stream of entities.
|
||||
*
|
||||
* @param query must not be {@literal null}.
|
||||
* @param entityClass The entity type must not be {@literal null}.
|
||||
* @return the result objects returned by the action.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
*/
|
||||
<T> Flux<T> select(Query query, Class<T> entityClass) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Execute a {@code SELECT} query and convert the resulting item to an entity.
|
||||
*
|
||||
* @param query must not be {@literal null}.
|
||||
* @param entityClass The entity type must not be {@literal null}.
|
||||
* @return the result object returned by the action or {@link Mono#empty()}.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
*/
|
||||
<T> Mono<T> selectOne(Query query, Class<T> entityClass) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Update the queried entities and return {@literal true} if the update was applied.
|
||||
*
|
||||
* @param query must not be {@literal null}.
|
||||
* @param update must not be {@literal null}.
|
||||
* @param entityClass The entity type must not be {@literal null}.
|
||||
* @return the number of affected rows.
|
||||
* @throws DataAccessException if there is any problem executing the query.
|
||||
*/
|
||||
Mono<Integer> update(Query query, Update update, Class<?> entityClass) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Remove entities (rows)/columns from the table by {@link Query}.
|
||||
*
|
||||
* @param query must not be {@literal null}.
|
||||
* @param entityClass The entity type must not be {@literal null}.
|
||||
* @return the number of affected rows.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
*/
|
||||
Mono<Integer> delete(Query query, Class<?> entityClass) throws DataAccessException;
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Methods dealing with entities
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Insert the given entity and emit the entity if the insert was applied.
|
||||
*
|
||||
* @param entity The entity to insert, must not be {@literal null}.
|
||||
* @return the inserted entity.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
*/
|
||||
<T> Mono<T> insert(T entity) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Update the given entity and emit the entity if the update was applied.
|
||||
*
|
||||
* @param entity The entity to update, must not be {@literal null}.
|
||||
* @return the updated entity.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
* @throws TransientDataAccessResourceException if the update did not affect any rows.
|
||||
*/
|
||||
<T> Mono<T> update(T entity) throws DataAccessException;
|
||||
|
||||
/**
|
||||
* Delete the given entity and emit the entity if the delete was applied.
|
||||
*
|
||||
* @param entity must not be {@literal null}.
|
||||
* @return the deleted entity.
|
||||
* @throws DataAccessException if there is any problem issuing the execution.
|
||||
*/
|
||||
<T> Mono<T> delete(T entity) throws DataAccessException;
|
||||
}
|
||||
@@ -0,0 +1,491 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.r2dbc.core;
|
||||
|
||||
import io.r2dbc.spi.Row;
|
||||
import io.r2dbc.spi.RowMetadata;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.beans.FeatureDescriptor;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.dao.TransientDataAccessResourceException;
|
||||
import org.springframework.data.mapping.IdentifierAccessor;
|
||||
import org.springframework.data.mapping.MappingException;
|
||||
import org.springframework.data.mapping.context.MappingContext;
|
||||
import org.springframework.data.projection.ProjectionInformation;
|
||||
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
|
||||
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
|
||||
import org.springframework.data.r2dbc.query.Criteria;
|
||||
import org.springframework.data.r2dbc.query.Query;
|
||||
import org.springframework.data.r2dbc.query.Update;
|
||||
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
|
||||
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
|
||||
import org.springframework.data.relational.core.sql.Functions;
|
||||
import org.springframework.data.util.ProxyUtils;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Implementation of {@link R2dbcEntityOperations}. It simplifies the use of Reactive R2DBC usage through entities and
|
||||
* helps to avoid common errors. This class uses {@link DatabaseClient} to execute SQL queries or updates, initiating
|
||||
* iteration over {@link io.r2dbc.spi.Result}.
|
||||
* <p>
|
||||
* Can be used within a service implementation via direct instantiation with a {@link DatabaseClient} reference, or get
|
||||
* prepared in an application context and given to services as bean reference.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @since 1.1
|
||||
*/
|
||||
public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware {
|
||||
|
||||
private final DatabaseClient databaseClient;
|
||||
|
||||
private final ReactiveDataAccessStrategy dataAccessStrategy;
|
||||
|
||||
private final MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
|
||||
|
||||
private final SpelAwareProxyProjectionFactory projectionFactory;
|
||||
|
||||
/**
|
||||
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient}.
|
||||
*
|
||||
* @param databaseClient
|
||||
*/
|
||||
public R2dbcEntityTemplate(DatabaseClient databaseClient) {
|
||||
|
||||
Assert.notNull(databaseClient, "DatabaseClient must not be null");
|
||||
|
||||
this.databaseClient = databaseClient;
|
||||
this.dataAccessStrategy = getDataAccessStrategy(databaseClient);
|
||||
this.mappingContext = getMappingContext(this.dataAccessStrategy);
|
||||
this.projectionFactory = new SpelAwareProxyProjectionFactory();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient} and {@link ReactiveDataAccessStrategy}.
|
||||
*
|
||||
* @param databaseClient
|
||||
*/
|
||||
public R2dbcEntityTemplate(DatabaseClient databaseClient, ReactiveDataAccessStrategy strategy) {
|
||||
|
||||
Assert.notNull(databaseClient, "DatabaseClient must not be null");
|
||||
Assert.notNull(strategy, "ReactiveDataAccessStrategy must not be null");
|
||||
|
||||
this.databaseClient = databaseClient;
|
||||
this.dataAccessStrategy = strategy;
|
||||
this.mappingContext = strategy.getConverter().getMappingContext();
|
||||
this.projectionFactory = new SpelAwareProxyProjectionFactory();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#getDatabaseClient()
|
||||
*/
|
||||
@Override
|
||||
public DatabaseClient getDatabaseClient() {
|
||||
return this.databaseClient;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory)
|
||||
*/
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
|
||||
this.projectionFactory.setBeanFactory(beanFactory);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Methods dealing with org.springframework.data.r2dbc.core.FluentR2dbcOperations
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.ReactiveSelectOperation#select(java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public <T> ReactiveSelect<T> select(Class<T> domainType) {
|
||||
return new ReactiveSelectOperationSupport(this).select(domainType);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.ReactiveInsertOperation#insert(java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public <T> ReactiveInsert<T> insert(Class<T> domainType) {
|
||||
return new ReactiveInsertOperationSupport(this).insert(domainType);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.ReactiveUpdateOperation#update(java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public ReactiveUpdate update(Class<?> domainType) {
|
||||
return new ReactiveUpdateOperationSupport(this).update(domainType);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.ReactiveDeleteOperation#delete(java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public ReactiveDelete delete(Class<?> domainType) {
|
||||
return new ReactiveDeleteOperationSupport(this).delete(domainType);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Methods dealing with org.springframework.data.r2dbc.query.Query
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#count(org.springframework.data.r2dbc.query.Query, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> count(Query query, Class<?> entityClass) throws DataAccessException {
|
||||
|
||||
Assert.notNull(query, "Query must not be null");
|
||||
Assert.notNull(entityClass, "entity class must not be null");
|
||||
|
||||
return doCount(query, entityClass, getTableName(entityClass));
|
||||
}
|
||||
|
||||
Mono<Long> doCount(Query query, Class<?> entityClass, String tableName) {
|
||||
|
||||
RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
|
||||
StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
|
||||
|
||||
StatementMapper.SelectSpec selectSpec = statementMapper //
|
||||
.createSelect(tableName) //
|
||||
.doWithTable((table, spec) -> {
|
||||
return spec.withProjection(Functions.count(table.column(entity.getRequiredIdProperty().getColumnName())));
|
||||
});
|
||||
|
||||
Optional<Criteria> criteria = query.getCriteria();
|
||||
if (criteria.isPresent()) {
|
||||
selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec);
|
||||
}
|
||||
|
||||
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
|
||||
|
||||
return this.databaseClient.execute(operation) //
|
||||
.map((r, md) -> r.get(0, Long.class)) //
|
||||
.first() //
|
||||
.defaultIfEmpty(0L);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#exists(org.springframework.data.r2dbc.query.Query, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Boolean> exists(Query query, Class<?> entityClass) throws DataAccessException {
|
||||
|
||||
Assert.notNull(query, "Query must not be null");
|
||||
Assert.notNull(entityClass, "entity class must not be null");
|
||||
|
||||
return doExists(query, entityClass, getTableName(entityClass));
|
||||
}
|
||||
|
||||
Mono<Boolean> doExists(Query query, Class<?> entityClass, String tableName) {
|
||||
|
||||
RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
|
||||
StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
|
||||
|
||||
String columnName = entity.hasIdProperty() ? entity.getRequiredIdProperty().getColumnName() : "*";
|
||||
|
||||
StatementMapper.SelectSpec selectSpec = statementMapper //
|
||||
.createSelect(tableName) //
|
||||
.withProjection(columnName);
|
||||
|
||||
Optional<Criteria> criteria = query.getCriteria();
|
||||
if (criteria.isPresent()) {
|
||||
selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec);
|
||||
}
|
||||
|
||||
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
|
||||
|
||||
return this.databaseClient.execute(operation) //
|
||||
.map((r, md) -> r) //
|
||||
.first() //
|
||||
.hasElement();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#select(org.springframework.data.r2dbc.query.Query, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public <T> Flux<T> select(Query query, Class<T> entityClass) throws DataAccessException {
|
||||
|
||||
Assert.notNull(query, "Query must not be null");
|
||||
Assert.notNull(entityClass, "entity class must not be null");
|
||||
|
||||
return doSelect(query, entityClass, getTableName(entityClass), entityClass).all();
|
||||
}
|
||||
|
||||
<T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityClass, String tableName, Class<T> returnType) {
|
||||
|
||||
RelationalPersistentEntity<?> entity = getRequiredEntity(entityClass);
|
||||
StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
|
||||
|
||||
StatementMapper.SelectSpec selectSpec = statementMapper //
|
||||
.createSelect(tableName) //
|
||||
.withProjection(getSelectProjection(query, returnType));
|
||||
|
||||
if (query.getLimit() > 0) {
|
||||
selectSpec = selectSpec.limit(query.getLimit());
|
||||
}
|
||||
|
||||
if (query.getOffset() > 0) {
|
||||
selectSpec = selectSpec.offset(query.getOffset());
|
||||
}
|
||||
|
||||
if (query.isSorted()) {
|
||||
selectSpec = selectSpec.withSort(query.getSort());
|
||||
}
|
||||
|
||||
Optional<Criteria> criteria = query.getCriteria();
|
||||
if (criteria.isPresent()) {
|
||||
selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec);
|
||||
}
|
||||
|
||||
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
|
||||
|
||||
BiFunction<Row, RowMetadata, T> rowMapper;
|
||||
if (returnType.isInterface()) {
|
||||
rowMapper = dataAccessStrategy.getRowMapper(entityClass)
|
||||
.andThen(o -> projectionFactory.createProjection(returnType, o));
|
||||
} else {
|
||||
rowMapper = dataAccessStrategy.getRowMapper(returnType);
|
||||
}
|
||||
|
||||
return this.databaseClient.execute(operation).map(rowMapper);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#selectOne(org.springframework.data.r2dbc.query.Query, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public <T> Mono<T> selectOne(Query query, Class<T> entityClass) throws DataAccessException {
|
||||
return doSelect(query.limit(2), entityClass, getTableName(entityClass), entityClass).one();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(org.springframework.data.r2dbc.query.Query, org.springframework.data.r2dbc.query.Update, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Integer> update(Query query, Update update, Class<?> entityClass) throws DataAccessException {
|
||||
|
||||
Assert.notNull(query, "Query must not be null");
|
||||
Assert.notNull(update, "Update must not be null");
|
||||
Assert.notNull(entityClass, "entity class must not be null");
|
||||
|
||||
return doUpdate(query, update, entityClass, getTableName(entityClass));
|
||||
}
|
||||
|
||||
Mono<Integer> doUpdate(Query query, Update update, Class<?> entityClass, String tableName) {
|
||||
|
||||
StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
|
||||
|
||||
StatementMapper.UpdateSpec selectSpec = statementMapper //
|
||||
.createUpdate(tableName, update);
|
||||
|
||||
Optional<Criteria> criteria = query.getCriteria();
|
||||
if (criteria.isPresent()) {
|
||||
selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec);
|
||||
}
|
||||
|
||||
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
|
||||
return this.databaseClient.execute(operation).fetch().rowsUpdated();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(org.springframework.data.r2dbc.query.Query, java.lang.Class)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Integer> delete(Query query, Class<?> entityClass) throws DataAccessException {
|
||||
|
||||
Assert.notNull(query, "Query must not be null");
|
||||
Assert.notNull(entityClass, "entity class must not be null");
|
||||
|
||||
return doDelete(query, entityClass, getTableName(entityClass));
|
||||
}
|
||||
|
||||
Mono<Integer> doDelete(Query query, Class<?> entityClass, String tableName) {
|
||||
|
||||
StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
|
||||
|
||||
StatementMapper.DeleteSpec selectSpec = statementMapper //
|
||||
.createDelete(tableName);
|
||||
|
||||
Optional<Criteria> criteria = query.getCriteria();
|
||||
if (criteria.isPresent()) {
|
||||
selectSpec = criteria.map(selectSpec::withCriteria).orElse(selectSpec);
|
||||
}
|
||||
|
||||
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
|
||||
return this.databaseClient.execute(operation).fetch().rowsUpdated().defaultIfEmpty(0);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Methods dealing with entities
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#insert(java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public <T> Mono<T> insert(T entity) throws DataAccessException {
|
||||
|
||||
Assert.notNull(entity, "Entity must not be null");
|
||||
|
||||
return doInsert(entity, getRequiredEntity(entity).getTableName());
|
||||
}
|
||||
|
||||
<T> Mono<T> doInsert(T entity, String tableName) {
|
||||
|
||||
RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
|
||||
|
||||
return this.databaseClient.insert() //
|
||||
.into(persistentEntity.getType()) //
|
||||
.table(tableName).using(entity) //
|
||||
.map(this.dataAccessStrategy.getConverter().populateIdIfNecessary(entity)) //
|
||||
.first() //
|
||||
.defaultIfEmpty(entity);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#update(java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public <T> Mono<T> update(T entity) throws DataAccessException {
|
||||
|
||||
Assert.notNull(entity, "Entity must not be null");
|
||||
|
||||
RelationalPersistentEntity<T> persistentEntity = getRequiredEntity(entity);
|
||||
|
||||
return this.databaseClient.update() //
|
||||
.table(persistentEntity.getType()) //
|
||||
.table(persistentEntity.getTableName()).using(entity) //
|
||||
.fetch().rowsUpdated().handle((rowsUpdated, sink) -> {
|
||||
|
||||
if (rowsUpdated == 0) {
|
||||
sink.error(new TransientDataAccessResourceException(
|
||||
String.format("Failed to update table [%s]. Row with Id [%s] does not exist.",
|
||||
persistentEntity.getTableName(), persistentEntity.getIdentifierAccessor(entity).getIdentifier())));
|
||||
} else {
|
||||
sink.next(entity);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#delete(java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public <T> Mono<T> delete(T entity) throws DataAccessException {
|
||||
|
||||
Assert.notNull(entity, "Entity must not be null");
|
||||
|
||||
RelationalPersistentEntity<?> persistentEntity = getRequiredEntity(entity);
|
||||
|
||||
return delete(getByIdQuery(entity, persistentEntity), persistentEntity.getType()).thenReturn(entity);
|
||||
}
|
||||
|
||||
private <T> Query getByIdQuery(T entity, RelationalPersistentEntity<?> persistentEntity) {
|
||||
if (!persistentEntity.hasIdProperty()) {
|
||||
throw new MappingException("No id property found for object of type " + persistentEntity.getType() + "!");
|
||||
}
|
||||
|
||||
IdentifierAccessor identifierAccessor = persistentEntity.getIdentifierAccessor(entity);
|
||||
Object id = identifierAccessor.getRequiredIdentifier();
|
||||
|
||||
return Query.query(Criteria.where(persistentEntity.getRequiredIdProperty().getName()).is(id));
|
||||
}
|
||||
|
||||
String getTableName(Class<?> entityClass) {
|
||||
return getRequiredEntity(entityClass).getTableName();
|
||||
}
|
||||
|
||||
private RelationalPersistentEntity<?> getRequiredEntity(Class<?> entityClass) {
|
||||
return this.mappingContext.getRequiredPersistentEntity(entityClass);
|
||||
}
|
||||
|
||||
private <T> RelationalPersistentEntity<T> getRequiredEntity(T entity) {
|
||||
Class<?> entityType = ProxyUtils.getUserClass(entity);
|
||||
return (RelationalPersistentEntity) getRequiredEntity(entityType);
|
||||
}
|
||||
|
||||
private <T> List<String> getSelectProjection(Query query, Class<T> returnType) {
|
||||
|
||||
if (query.getColumns().isEmpty()) {
|
||||
|
||||
if (returnType.isInterface()) {
|
||||
|
||||
ProjectionInformation projectionInformation = projectionFactory.getProjectionInformation(returnType);
|
||||
|
||||
if (projectionInformation.isClosed()) {
|
||||
return projectionInformation.getInputProperties().stream().map(FeatureDescriptor::getName)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
}
|
||||
|
||||
return Collections.singletonList("*");
|
||||
}
|
||||
|
||||
return query.getColumns();
|
||||
}
|
||||
|
||||
private static ReactiveDataAccessStrategy getDataAccessStrategy(DatabaseClient databaseClient) {
|
||||
|
||||
if (databaseClient instanceof DefaultDatabaseClient) {
|
||||
|
||||
DefaultDatabaseClient client = (DefaultDatabaseClient) databaseClient;
|
||||
return client.getDataAccessStrategy();
|
||||
}
|
||||
|
||||
throw new IllegalStateException("Cannot obtain ReactiveDataAccessStrategy");
|
||||
}
|
||||
|
||||
private static MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext(
|
||||
ReactiveDataAccessStrategy strategy) {
|
||||
|
||||
if (strategy instanceof DefaultReactiveDataAccessStrategy) {
|
||||
DefaultReactiveDataAccessStrategy strategy1 = (DefaultReactiveDataAccessStrategy) strategy;
|
||||
return strategy1.getMappingContext();
|
||||
}
|
||||
return new R2dbcMappingContext();
|
||||
}
|
||||
|
||||
}
|
||||
234
src/main/java/org/springframework/data/r2dbc/query/Query.java
Normal file
234
src/main/java/org/springframework/data/r2dbc/query/Query.java
Normal file
@@ -0,0 +1,234 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.r2dbc.query;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Query object representing {@link Criteria}, columns, {@link Sort}, and limit/offset for a SQL query. {@link Query} is
|
||||
* created with a fluent API creating immutable objects.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @since 1.1
|
||||
* @see Criteria
|
||||
* @see Sort
|
||||
* @see Pageable
|
||||
*/
|
||||
public class Query {
|
||||
|
||||
private final @Nullable Criteria criteria;
|
||||
|
||||
// TODO: select list should be List<Expression>
|
||||
private final List<String> columns;
|
||||
private final Sort sort;
|
||||
private final int limit;
|
||||
private final long offset;
|
||||
|
||||
/**
|
||||
* Static factory method to create a {@link Query} using the provided {@link Criteria}.
|
||||
*
|
||||
* @param criteria must not be {@literal null}.
|
||||
* @return a new {@link Query} for the given {@link Criteria}.
|
||||
*/
|
||||
public static Query query(Criteria criteria) {
|
||||
return new Query(criteria);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new {@link Query} using the given {@link Criteria}.
|
||||
*
|
||||
* @param criteria must not be {@literal null}.
|
||||
*/
|
||||
private Query(@Nullable Criteria criteria) {
|
||||
this.criteria = criteria;
|
||||
this.sort = Sort.unsorted();
|
||||
this.columns = Collections.emptyList();
|
||||
this.limit = -1;
|
||||
this.offset = -1;
|
||||
}
|
||||
|
||||
private Query(Criteria criteria, List<String> columns, Sort sort, int limit, long offset) {
|
||||
this.criteria = criteria;
|
||||
this.columns = columns;
|
||||
this.sort = sort;
|
||||
this.limit = limit;
|
||||
this.offset = offset;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new empty {@link Query}.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static Query empty() {
|
||||
return new Query(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add columns to the query.
|
||||
*
|
||||
* @param columns
|
||||
* @return a new {@link Query} object containing the former settings with {@code columns} applied.
|
||||
*/
|
||||
public Query columns(String... columns) {
|
||||
|
||||
Assert.notNull(columns, "Columns must not be null");
|
||||
|
||||
return columns(Arrays.asList(columns));
|
||||
}
|
||||
|
||||
/**
|
||||
* Add columns to the query.
|
||||
*
|
||||
* @param columns
|
||||
* @return a new {@link Query} object containing the former settings with {@code columns} applied.
|
||||
*/
|
||||
public Query columns(Collection<String> columns) {
|
||||
|
||||
Assert.notNull(columns, "Columns must not be null");
|
||||
|
||||
List<String> newColumns = new ArrayList<>(this.columns);
|
||||
newColumns.addAll(columns);
|
||||
return new Query(this.criteria, newColumns, this.sort, this.limit, offset);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set number of rows to skip before returning results.
|
||||
*
|
||||
* @param offset
|
||||
* @return a new {@link Query} object containing the former settings with {@code offset} applied.
|
||||
*/
|
||||
public Query offset(long offset) {
|
||||
return new Query(this.criteria, this.columns, this.sort, this.limit, offset);
|
||||
}
|
||||
|
||||
/**
|
||||
* Limit the number of returned documents to {@code limit}.
|
||||
*
|
||||
* @param limit
|
||||
* @return a new {@link Query} object containing the former settings with {@code limit} applied.
|
||||
*/
|
||||
public Query limit(int limit) {
|
||||
return new Query(this.criteria, this.columns, this.sort, limit, this.offset);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the given pagination information on the {@link Query} instance. Will transparently set {@code offset} and
|
||||
* {@code limit} as well as applying the {@link Sort} instance defined with the {@link Pageable}.
|
||||
*
|
||||
* @param pageable
|
||||
* @return a new {@link Query} object containing the former settings with {@link Pageable} applied.
|
||||
*/
|
||||
public Query with(Pageable pageable) {
|
||||
|
||||
if (pageable.isUnpaged()) {
|
||||
return this;
|
||||
}
|
||||
|
||||
assertNoCaseSort(pageable.getSort());
|
||||
|
||||
return new Query(this.criteria, this.columns, this.sort.and(sort), pageable.getPageSize(), pageable.getOffset());
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a {@link Sort} to the {@link Query} instance.
|
||||
*
|
||||
* @param sort
|
||||
* @return a new {@link Query} object containing the former settings with {@link Sort} applied.
|
||||
*/
|
||||
public Query sort(Sort sort) {
|
||||
|
||||
Assert.notNull(sort, "Sort must not be null!");
|
||||
|
||||
if (sort.isUnsorted()) {
|
||||
return this;
|
||||
}
|
||||
|
||||
assertNoCaseSort(sort);
|
||||
|
||||
return new Query(this.criteria, this.columns, this.sort.and(sort), this.limit, this.offset);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the {@link Criteria} to be applied.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public Optional<Criteria> getCriteria() {
|
||||
return Optional.ofNullable(this.criteria);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the columns that this query should project.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public List<String> getColumns() {
|
||||
return columns;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return {@literal true} if the {@link Query} has a sort parameter.
|
||||
*
|
||||
* @return {@literal true} if sorted.
|
||||
* @see Sort#isSorted()
|
||||
*/
|
||||
public boolean isSorted() {
|
||||
return sort.isSorted();
|
||||
}
|
||||
|
||||
public Sort getSort() {
|
||||
return sort;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the number of rows to skip.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public long getOffset() {
|
||||
return this.offset;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the maximum number of rows to be return.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public int getLimit() {
|
||||
return this.limit;
|
||||
}
|
||||
|
||||
private static void assertNoCaseSort(Sort sort) {
|
||||
|
||||
for (Sort.Order order : sort) {
|
||||
if (order.isIgnoreCase()) {
|
||||
throw new IllegalArgumentException(String.format("Given sort contained an Order for %s with ignore case;"
|
||||
+ " R2DBC does not support sorting ignoring case currently", order.getProperty()));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.UnaryOperator;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.data.mapping.PersistentPropertyPath;
|
||||
import org.springframework.data.mapping.PropertyPath;
|
||||
@@ -39,12 +40,14 @@ import org.springframework.data.r2dbc.query.Criteria.Combinator;
|
||||
import org.springframework.data.r2dbc.query.Criteria.Comparator;
|
||||
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
|
||||
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
|
||||
import org.springframework.data.relational.core.sql.Aliased;
|
||||
import org.springframework.data.relational.core.sql.Column;
|
||||
import org.springframework.data.relational.core.sql.Condition;
|
||||
import org.springframework.data.relational.core.sql.Expression;
|
||||
import org.springframework.data.relational.core.sql.IdentifierProcessing;
|
||||
import org.springframework.data.relational.core.sql.SQL;
|
||||
import org.springframework.data.relational.core.sql.SqlIdentifier;
|
||||
import org.springframework.data.relational.core.sql.SimpleFunction;
|
||||
import org.springframework.data.relational.core.sql.Table;
|
||||
import org.springframework.data.util.ClassTypeInformation;
|
||||
import org.springframework.data.util.TypeInformation;
|
||||
@@ -119,6 +122,51 @@ public class QueryMapper {
|
||||
return Sort.by(mappedOrder);
|
||||
}
|
||||
|
||||
/**
|
||||
* Map the {@link Expression} object to apply field name mapping using {@link Class the type to read}.
|
||||
*
|
||||
* @param expression must not be {@literal null}.
|
||||
* @param entity related {@link RelationalPersistentEntity}, can be {@literal null}.
|
||||
* @return the mapped {@link Expression}.
|
||||
* @since 1.1
|
||||
*/
|
||||
public Expression getMappedObject(Expression expression, @Nullable RelationalPersistentEntity<?> entity) {
|
||||
|
||||
if (entity == null) {
|
||||
return expression;
|
||||
}
|
||||
|
||||
if (expression instanceof Column) {
|
||||
|
||||
Column column = (Column) expression;
|
||||
Field field = createPropertyField(entity, column.getName());
|
||||
|
||||
return column instanceof Aliased
|
||||
? Column.aliased(field.getMappedColumnName(), column.getTable(), ((Aliased) column).getAlias())
|
||||
: Column.create(field.getMappedColumnName(), column.getTable());
|
||||
}
|
||||
|
||||
if (expression instanceof SimpleFunction) {
|
||||
|
||||
// Revisit after https://jira.spring.io/browse/DATAJDBC-478
|
||||
SimpleFunction function = (SimpleFunction) expression;
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(function);
|
||||
|
||||
List<Expression> arguments = (List<Expression>) accessor.getPropertyValue("expressions");
|
||||
List<Expression> mappedArguments = new ArrayList<>(arguments.size());
|
||||
|
||||
for (Expression argument : arguments) {
|
||||
mappedArguments.add(getMappedObject(argument, entity));
|
||||
}
|
||||
|
||||
SimpleFunction mappedFunction = SimpleFunction.create(function.getFunctionName(), mappedArguments);
|
||||
|
||||
return function instanceof Aliased ? mappedFunction.as(((Aliased) function).getAlias()) : mappedFunction;
|
||||
}
|
||||
|
||||
throw new IllegalArgumentException(String.format("Cannot map %s", expression));
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a {@link Criteria} object into {@link Condition} and consider value/{@code NULL} {@link Bindings}.
|
||||
*
|
||||
@@ -287,6 +335,10 @@ public class QueryMapper {
|
||||
}
|
||||
}
|
||||
|
||||
Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, String key) {
|
||||
return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext);
|
||||
}
|
||||
|
||||
Field createPropertyField(@Nullable RelationalPersistentEntity<?> entity, String key,
|
||||
MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
|
||||
return entity == null ? new Field(key) : new MetadataBackedField(key, entity, mappingContext);
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.r2dbc.core;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import io.r2dbc.spi.test.MockColumnMetadata;
|
||||
import io.r2dbc.spi.test.MockResult;
|
||||
import io.r2dbc.spi.test.MockRow;
|
||||
import io.r2dbc.spi.test.MockRowMetadata;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.data.annotation.Id;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.data.r2dbc.dialect.PostgresDialect;
|
||||
import org.springframework.data.r2dbc.mapping.SettableValue;
|
||||
import org.springframework.data.r2dbc.query.Criteria;
|
||||
import org.springframework.data.r2dbc.query.Query;
|
||||
import org.springframework.data.r2dbc.query.Update;
|
||||
import org.springframework.data.r2dbc.testing.StatementRecorder;
|
||||
import org.springframework.data.relational.core.mapping.Column;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link R2dbcEntityTemplate}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class R2dbcEntityTemplateUnitTests {
|
||||
|
||||
DatabaseClient client;
|
||||
R2dbcEntityTemplate entityTemplate;
|
||||
StatementRecorder recorder;
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
|
||||
recorder = StatementRecorder.newInstance();
|
||||
client = DatabaseClient.builder().connectionFactory(recorder)
|
||||
.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)).build();
|
||||
entityTemplate = new R2dbcEntityTemplate(client);
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldCountBy() {
|
||||
|
||||
MockRowMetadata metadata = MockRowMetadata.builder()
|
||||
.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
|
||||
MockResult result = MockResult.builder().rowMetadata(metadata)
|
||||
.row(MockRow.builder().identified(0, Long.class, 1L).build()).build();
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("SELECT"), result);
|
||||
|
||||
entityTemplate.count(Query.query(Criteria.where("name").is("Walter")), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1L) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
|
||||
|
||||
assertThat(statement.getSql()).isEqualTo("SELECT COUNT(person.id) FROM person WHERE person.THE_NAME = $1");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldExistsByCriteria() {
|
||||
|
||||
MockRowMetadata metadata = MockRowMetadata.builder()
|
||||
.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
|
||||
MockResult result = MockResult.builder().rowMetadata(metadata)
|
||||
.row(MockRow.builder().identified(0, Long.class, 1L).build()).build();
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("SELECT"), result);
|
||||
|
||||
entityTemplate.exists(Query.query(Criteria.where("name").is("Walter")), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(true) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
|
||||
|
||||
assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldSelectByCriteria() {
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("SELECT"), Collections.emptyList());
|
||||
|
||||
entityTemplate.select(Query.query(Criteria.where("name").is("Walter")).sort(Sort.by("name")), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
|
||||
|
||||
assertThat(statement.getSql())
|
||||
.isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 ORDER BY THE_NAME ASC");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldSelectOne() {
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("SELECT"), Collections.emptyList());
|
||||
|
||||
entityTemplate.selectOne(Query.query(Criteria.where("name").is("Walter")).sort(Sort.by("name")), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
|
||||
|
||||
assertThat(statement.getSql())
|
||||
.isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1 ORDER BY THE_NAME ASC LIMIT 2");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldUpdateByQuery() {
|
||||
|
||||
MockRowMetadata metadata = MockRowMetadata.builder()
|
||||
.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
|
||||
MockResult result = MockResult.builder().rowMetadata(metadata).rowsUpdated(1).build();
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("UPDATE"), result);
|
||||
|
||||
entityTemplate
|
||||
.update(Query.query(Criteria.where("name").is("Walter")), Update.update("name", "Heisenberg"), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("UPDATE"));
|
||||
|
||||
assertThat(statement.getSql()).isEqualTo("UPDATE person SET THE_NAME = $1 WHERE person.THE_NAME = $2");
|
||||
assertThat(statement.getBindings()).hasSize(2).containsEntry(0, SettableValue.from("Heisenberg")).containsEntry(1,
|
||||
SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldDeleteByQuery() {
|
||||
|
||||
MockRowMetadata metadata = MockRowMetadata.builder()
|
||||
.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
|
||||
MockResult result = MockResult.builder().rowMetadata(metadata).rowsUpdated(1).build();
|
||||
|
||||
recorder.addStubbing(s -> s.startsWith("DELETE"), result);
|
||||
|
||||
entityTemplate.delete(Query.query(Criteria.where("name").is("Walter")), Person.class) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1) //
|
||||
.verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("DELETE"));
|
||||
|
||||
assertThat(statement.getSql()).isEqualTo("DELETE FROM person WHERE person.THE_NAME = $1");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
@Test // gh-220
|
||||
public void shouldDeleteEntity() {
|
||||
|
||||
Person person = new Person();
|
||||
person.id = "Walter";
|
||||
recorder.addStubbing(s -> s.startsWith("DELETE"), Collections.emptyList());
|
||||
|
||||
entityTemplate.delete(person) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(person).verifyComplete();
|
||||
|
||||
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("DELETE"));
|
||||
|
||||
assertThat(statement.getSql()).isEqualTo("DELETE FROM person WHERE person.id = $1");
|
||||
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
|
||||
}
|
||||
|
||||
static class Person {
|
||||
|
||||
@Id String id;
|
||||
|
||||
@Column("THE_NAME") String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,312 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.r2dbc.testing;
|
||||
|
||||
import io.r2dbc.spi.Batch;
|
||||
import io.r2dbc.spi.Connection;
|
||||
import io.r2dbc.spi.ConnectionFactory;
|
||||
import io.r2dbc.spi.ConnectionFactoryMetadata;
|
||||
import io.r2dbc.spi.ConnectionMetadata;
|
||||
import io.r2dbc.spi.IsolationLevel;
|
||||
import io.r2dbc.spi.Result;
|
||||
import io.r2dbc.spi.Statement;
|
||||
import io.r2dbc.spi.ValidationDepth;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
import org.springframework.data.r2dbc.mapping.SettableValue;
|
||||
|
||||
/**
|
||||
* Recorder utility for R2DBC {@link Statement}s. Allows stubbing and introspection.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class StatementRecorder implements ConnectionFactory {
|
||||
|
||||
private final Map<Predicate<String>, Supplier<List<Result>>> stubbings = new LinkedHashMap<>();
|
||||
private final List<RecordedStatement> createdStatements = new ArrayList<>();
|
||||
private final List<RecordedStatement> executedStatements = new ArrayList<>();
|
||||
|
||||
private StatementRecorder() {}
|
||||
|
||||
/**
|
||||
* Create a new {@link StatementRecorder}.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static StatementRecorder newInstance() {
|
||||
return new StatementRecorder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link StatementRecorder} accepting a {@link Consumer configurer}.
|
||||
*
|
||||
* @param configurer
|
||||
* @return
|
||||
*/
|
||||
public static StatementRecorder newInstance(Consumer<StatementRecorder> configurer) {
|
||||
|
||||
StatementRecorder statementRecorder = new StatementRecorder();
|
||||
|
||||
configurer.accept(statementRecorder);
|
||||
|
||||
return statementRecorder;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a stubbing rule given the {@link Predicate SQL Predicate} and a {@link Result} that is emitted by the executed
|
||||
* statement. Typical usage:
|
||||
*
|
||||
* <pre class="code">
|
||||
* recorder.addStubbing(sql -> sql.startsWith("SELECT"), result);
|
||||
* </pre>
|
||||
*
|
||||
* @param sqlPredicate
|
||||
* @param result
|
||||
*/
|
||||
public void addStubbing(Predicate<String> sqlPredicate, Result result) {
|
||||
this.stubbings.put(sqlPredicate, () -> Collections.singletonList(result));
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a stubbing rule given the {@link Predicate SQL Predicate} and a list of {@link Result results} that are emitted
|
||||
* by the executed statement. Typical usage:
|
||||
*
|
||||
* <pre class="code">
|
||||
* recorder.addStubbing(sql -> sql.startsWith("SELECT"), results);
|
||||
* </pre>
|
||||
*
|
||||
* @param sqlPredicate
|
||||
* @param result
|
||||
*/
|
||||
public void addStubbing(Predicate<String> sqlPredicate, List<Result> results) {
|
||||
this.stubbings.put(sqlPredicate, () -> results);
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a statement by {@code sql}.
|
||||
*
|
||||
* @param sql
|
||||
* @return
|
||||
*/
|
||||
public RecordedStatement getCreatedStatement(String sql) {
|
||||
return getCreatedStatement(it -> compareSql(sql, it));
|
||||
}
|
||||
|
||||
private static boolean compareSql(String pattern, String actual) {
|
||||
return actual.equals(pattern) || Pattern.compile(pattern).matcher(actual).find();
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a statement by a {@link Predicate SQL predicate}.
|
||||
*
|
||||
* @param sql
|
||||
* @return
|
||||
*/
|
||||
public RecordedStatement getCreatedStatement(Predicate<String> predicate) {
|
||||
|
||||
return createdStatements.stream().filter(recordedStatement -> {
|
||||
return predicate.test(recordedStatement.getSql()) || predicate.test(recordedStatement.getSql().toLowerCase())
|
||||
|| predicate.test(recordedStatement.getSql().toUpperCase());
|
||||
}).findFirst().orElseThrow(() -> new NoSuchElementException("No statement found"));
|
||||
}
|
||||
|
||||
public List<RecordedStatement> getCreatedStatements() {
|
||||
return createdStatements;
|
||||
}
|
||||
|
||||
public List<RecordedStatement> getExecutedStatements() {
|
||||
return executedStatements;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<? extends Connection> create() {
|
||||
return Mono.just(new RecorderConnection());
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConnectionFactoryMetadata getMetadata() {
|
||||
return () -> "StatementRecorder";
|
||||
}
|
||||
|
||||
class RecorderConnection implements Connection {
|
||||
@Override
|
||||
public Publisher<Void> beginTransaction() {
|
||||
return createStatement("BEGIN").execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> close() {
|
||||
return createStatement("CLOSE").execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> commitTransaction() {
|
||||
return createStatement("COMMIT").execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Batch createBatch() {
|
||||
throw new UnsupportedOperationException("createBatch not yet supported");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> createSavepoint(String name) {
|
||||
return createStatement("CREATE SAVEPOINT " + name).execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public RecordedStatement createStatement(String sql) {
|
||||
|
||||
RecordedStatement statement = doCreateStatement(sql);
|
||||
|
||||
createdStatements.add(statement);
|
||||
|
||||
return statement;
|
||||
}
|
||||
|
||||
private RecordedStatement doCreateStatement(String sql) {
|
||||
for (Map.Entry<Predicate<String>, Supplier<List<Result>>> entry : stubbings.entrySet()) {
|
||||
|
||||
if (entry.getKey().test(sql) || entry.getKey().test(sql.toLowerCase())
|
||||
|| entry.getKey().test(sql.toUpperCase())) {
|
||||
return new RecordedStatement(sql, entry.getValue().get());
|
||||
}
|
||||
|
||||
}
|
||||
return new RecordedStatement(sql, Collections.emptyList());
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoCommit() {
|
||||
throw new UnsupportedOperationException("isAutoCommit not yet supported");
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConnectionMetadata getMetadata() {
|
||||
throw new UnsupportedOperationException("getMetadata not yet supported");
|
||||
}
|
||||
|
||||
@Override
|
||||
public IsolationLevel getTransactionIsolationLevel() {
|
||||
throw new UnsupportedOperationException("getTransactionIsolationLevel not yet supported");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> releaseSavepoint(String name) {
|
||||
return createStatement("RELEASE SAVEPOINT " + name).execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> rollbackTransaction() {
|
||||
return createStatement("ROLLBACK").execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> rollbackTransactionToSavepoint(String name) {
|
||||
return createStatement("ROLLBACK TO " + name).execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> setAutoCommit(boolean autoCommit) {
|
||||
return createStatement("SET AUTOCOMMIT " + autoCommit).execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> setTransactionIsolationLevel(IsolationLevel isolationLevel) {
|
||||
return createStatement("SET TRANSACTION ISOLATION LEVEL " + isolationLevel.asSql()).execute().then();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> validate(ValidationDepth depth) {
|
||||
return Mono.just(true);
|
||||
}
|
||||
}
|
||||
|
||||
public class RecordedStatement implements Statement {
|
||||
|
||||
private final String sql;
|
||||
|
||||
private final List<Result> results;
|
||||
|
||||
private final Map<Object, SettableValue> bindings = new LinkedHashMap<>();
|
||||
|
||||
public RecordedStatement(String sql, Result result) {
|
||||
this(sql, Collections.singletonList(result));
|
||||
}
|
||||
|
||||
public RecordedStatement(String sql, List<Result> results) {
|
||||
this.sql = sql;
|
||||
this.results = results;
|
||||
}
|
||||
|
||||
public Map<Object, SettableValue> getBindings() {
|
||||
return bindings;
|
||||
}
|
||||
|
||||
public String getSql() {
|
||||
return sql;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement add() {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement bind(int index, Object o) {
|
||||
this.bindings.put(index, SettableValue.from(o));
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement bind(String identifier, Object o) {
|
||||
this.bindings.put(identifier, SettableValue.from(o));
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement bindNull(int index, Class<?> type) {
|
||||
this.bindings.put(index, SettableValue.empty(type));
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Statement bindNull(String identifier, Class<?> type) {
|
||||
this.bindings.put(identifier, SettableValue.empty(type));
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<Result> execute() {
|
||||
return Flux.fromIterable(results).doOnSubscribe(subscription -> executedStatements.add(this));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user