Spring/spring-data-r2dbc
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

#20 - Add Dialect initial support for H2, PostgreSQL, and Microsoft SQL Server.

Browse Source 
We now provide dialect support for H2, PostgreSQL, and Microsoft SQL Server databases, configurable through AbstractR2dbcConfiguration. By default, we obtain the Dialect by inspecting ConnectionFactoryMetadata to identify the database and the most likely dialect to use.

BindableOperation encapsulates statements/queries that can accept parameters. Use BindableOperation for statements through DatabaseClient.
Extract SQL creation.

Split integration test into abstract base class that can be implemented with a database-specific test class.

Original pull request: #24.
This commit is contained in:
Mark Paluch
2018-11-21 13:48:07 +01:00
parent dc3a0d5fc4
commit 68f1e0b0b3
51 changed files with 2594 additions and 445 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/java/org/springframework/data/r2dbc/repository/config/AbstractR2dbcConfiguration.java → src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java
View File

@@ -13,13 +13,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.r2dbc.repository.config;
package org.springframework.data.r2dbc.config;
import io.r2dbc.spi.ConnectionFactory;
import java.util.Optional;
import io.r2dbc.spi.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.dialect.Database;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.DefaultReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
@@ -37,7 +40,7 @@ import org.springframework.util.Assert;
* @author Mark Paluch
* @see ConnectionFactory
* @see DatabaseClient
* @see EnableR2dbcRepositories
* @see org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories
*/
@Configuration
public abstract class AbstractR2dbcConfiguration {
@@ -50,6 +53,24 @@ public abstract class AbstractR2dbcConfiguration {
*/
public abstract ConnectionFactory connectionFactory();
/**
* Return a {@link Dialect} for the given {@link ConnectionFactory}. This method attempts to resolve a {@link Dialect}
* from {@link io.r2dbc.spi.ConnectionFactoryMetadata}. Override this method to specify a dialect instead of
* attempting to resolve one.
*
* @param connectionFactory the configured {@link ConnectionFactory}.
* @return the resolved {@link Dialect}.
* @throws UnsupportedOperationException if the {@link Dialect} cannot be determined.
*/
public Dialect getDialect(ConnectionFactory connectionFactory) {
return Database.findDatabase(connectionFactory)
.orElseThrow(() -> new UnsupportedOperationException(
String.format("Cannot determine a dialect for %s using %s. Please provide a Dialect.",
connectionFactory.getMetadata().getName(), connectionFactory)))
.latestDialect();
}
/**
* Register a {@link DatabaseClient} using {@link #connectionFactory()} and {@link RelationalMappingContext}.
*
@@ -86,18 +107,21 @@ public abstract class AbstractR2dbcConfiguration {
}
/**
* Creates a {@link ReactiveDataAccessStrategy} using the configured {@link #r2dbcMappingContext(Optional) RelationalMappingContext}.
* Creates a {@link ReactiveDataAccessStrategy} using the configured {@link #r2dbcMappingContext(Optional)
* RelationalMappingContext}.
*
* @param mappingContext the configured {@link RelationalMappingContext}.
* @return must not be {@literal null}.
* @see #r2dbcMappingContext(Optional)
* @see #getDialect(ConnectionFactory)
* @throws IllegalArgumentException if any of the {@literal mappingContext} is {@literal null}.
*/
@Bean
public ReactiveDataAccessStrategy reactiveDataAccessStrategy(RelationalMappingContext mappingContext) {
Assert.notNull(mappingContext, "MappingContext must not be null!");
return new DefaultReactiveDataAccessStrategy(new BasicRelationalConverter(mappingContext));
return new DefaultReactiveDataAccessStrategy(getDialect(connectionFactory()),
new BasicRelationalConverter(mappingContext));
}
/**

6
src/main/java/org/springframework/data/r2dbc/config/package-info.java Normal file
View File

@@ -0,0 +1,6 @@
/**
* Configuration classes for Spring Data R2DBC.
*/
@org.springframework.lang.NonNullApi
@org.springframework.lang.NonNullFields
package org.springframework.data.r2dbc.config;

3
src/main/java/org/springframework/data/r2dbc/dialect/BindMarker.java
View File

@@ -28,7 +28,7 @@ public interface BindMarker {
* {@literal null} values.
* @see Statement#bind
*/
void bindValue(Statement<?> statement, Object value);
void bind(Statement<?> statement, Object value);
/**
* Bind a {@literal null} value to the {@link Statement} using the underlying binding strategy.
@@ -37,6 +37,5 @@ public interface BindMarker {
* @param valueType value type, must not be {@literal null}.
* @see Statement#bindNull
*/
void bindNull(Statement<?> statement, Class<?> valueType);
}

92
src/main/java/org/springframework/data/r2dbc/dialect/Database.java Normal file
View File

@@ -0,0 +1,92 @@
package org.springframework.data.r2dbc.dialect;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.ConnectionFactoryMetadata;
import java.util.Arrays;
import java.util.Locale;
import java.util.Optional;
import org.springframework.util.Assert;
/**
* Enumeration of known Databases for offline {@link Dialect} resolution. R2DBC {@link io.r2dbc.spi.ConnectionFactory}
* provides {@link io.r2dbc.spi.ConnectionFactoryMetadata metadata} that allows resolving an appropriate {@link Dialect}
* if none was configured explicitly.
*
* @author Mark Paluch
*/
public enum Database {
POSTGRES {
@Override
public String driverName() {
return "PostgreSQL";
}
@Override
public Dialect latestDialect() {
return PostgresDialect.INSTANCE;
}
},
SQL_SERVER {
@Override
public String driverName() {
return "Microsoft SQL Server";
}
@Override
public Dialect latestDialect() {
return SqlServerDialect.INSTANCE;
}
},
H2 {
@Override
public String driverName() {
return "H2";
}
@Override
public Dialect latestDialect() {
return H2Dialect.INSTANCE;
}
};
/**
* Find a {@link Database} type using {@link ConnectionFactory} and its metadata.
*
* @param connectionFactory must not be {@literal null}.
* @return the resolved {@link Database} or {@link Optional#empty()} if the database type cannot be determined from
* {@link ConnectionFactory}.
*/
public static Optional<Database> findDatabase(ConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "ConnectionFactor must not be null!");
ConnectionFactoryMetadata metadata = connectionFactory.getMetadata();
return Arrays.stream(values()).filter(it -> matches(metadata, it.driverName())).findFirst();
}
private static boolean matches(ConnectionFactoryMetadata metadata, String databaseType) {
return metadata.getName().toLowerCase(Locale.ENGLISH).contains(databaseType.toLowerCase(Locale.ENGLISH));
}
/**
* Returns the driver name.
*
* @return the driver name.
* @see ConnectionFactoryMetadata#getName()
*/
public abstract String driverName();
/**
* Returns the latest {@link Dialect} for the underlying database.
*
* @return the latest {@link Dialect} for the underlying database.
*/
public abstract Dialect latestDialect();
}

31
src/main/java/org/springframework/data/r2dbc/dialect/Dialect.java Normal file
View File

@@ -0,0 +1,31 @@
package org.springframework.data.r2dbc.dialect;
/**
* Represents a dialect that is implemented by a particular database.
*
* @author Mark Paluch
*/
public interface Dialect {
/**
* Returns the {@link BindMarkersFactory} used by this dialect.
*
* @return the {@link BindMarkersFactory} used by this dialect.
*/
BindMarkersFactory getBindMarkersFactory();
/**
* Returns the statement to include for returning generated keys. The returned query is directly appended to
* {@code INSERT} statements.
*
* @return the statement to include for returning generated keys.
*/
String returnGeneratedKeys();
/**
* Return the {@link LimitClause} used by this dialect.
*
* @return the {@link LimitClause} used by this dialect.
*/
LimitClause limit();
}

23
src/main/java/org/springframework/data/r2dbc/dialect/H2Dialect.java Normal file
View File

@@ -0,0 +1,23 @@
package org.springframework.data.r2dbc.dialect;
/**
* An SQL dialect for H2 in Postgres Compatibility mode.
*
* @author Mark Paluch
*/
public class H2Dialect extends PostgresDialect {
/**
* Singleton instance.
*/
public static final H2Dialect INSTANCE = new H2Dialect();
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#returnGeneratedKeys()
*/
@Override
public String returnGeneratedKeys() {
return "";
}
}

14
src/main/java/org/springframework/data/r2dbc/dialect/IndexedBindMarkers.java
View File

@@ -18,6 +18,7 @@ class IndexedBindMarkers implements BindMarkers {
// access via COUNTER_INCREMENTER
@SuppressWarnings("unused") private volatile int counter;
private final int offset;
private final String prefix;
/**
@@ -29,9 +30,10 @@ class IndexedBindMarkers implements BindMarkers {
IndexedBindMarkers(String prefix, int beginWith) {
this.counter = beginWith;
this.prefix = prefix;
this.offset = 0 - beginWith;
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.BindMarkers#next()
*/
@@ -40,7 +42,7 @@ class IndexedBindMarkers implements BindMarkers {
int index = COUNTER_INCREMENTER.getAndIncrement(this);
return new IndexedBindMarker(prefix + "" + index, index);
return new IndexedBindMarker(prefix + "" + index, index + offset);
}
/**
@@ -57,7 +59,7 @@ class IndexedBindMarkers implements BindMarkers {
this.index = index;
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.BindMarker#getPlaceholder()
*/
@@ -66,16 +68,16 @@ class IndexedBindMarkers implements BindMarkers {
return placeholder;
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.BindMarker#bindValue(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindValue(Statement<?> statement, Object value) {
public void bind(Statement<?> statement, Object value) {
statement.bind(this.index, value);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.BindMarker#bindNull(io.r2dbc.spi.Statement, java.lang.Class)
*/

42
src/main/java/org/springframework/data/r2dbc/dialect/LimitClause.java Normal file
View File

@@ -0,0 +1,42 @@
package org.springframework.data.r2dbc.dialect;
/**
* A clause representing Dialect-specific {@code LIMIT}.
*
* @author Mark Paluch
*/
public interface LimitClause {
/**
* Returns the {@code LIMIT} clause
*
* @param limit the actual limit to use.
* @return rendered limit clause.
*/
String getClause(long limit);
/**
* Returns the {@code LIMIT} clause
*
* @param limit the actual limit to use.
* @param offset the offset to start from.
* @return rendered limit clause.
*/
String getClause(long limit, long offset);
/**
* Returns the {@link Position} where to apply the {@link #getClause(long) clause}.
*/
Position getClausePosition();
/**
* Enumeration of where to render the clause within the SQL statement.
*/
enum Position {
/**
* Append the clause at the end of the statement.
*/
END
}
}

2
src/main/java/org/springframework/data/r2dbc/dialect/NamedBindMarkers.java
View File

@@ -101,7 +101,7 @@ class NamedBindMarkers implements BindMarkers {
* @see org.springframework.data.r2dbc.dialect.BindMarker#bindValue(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindValue(Statement<?> statement, Object value) {
public void bind(Statement<?> statement, Object value) {
statement.bind(this.identifier, value);
}

73
src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java Normal file
View File

@@ -0,0 +1,73 @@
package org.springframework.data.r2dbc.dialect;
/**
* An SQL dialect for Postgres.
*
* @author Mark Paluch
*/
public class PostgresDialect implements Dialect {
/**
* Singleton instance.
*/
public static final PostgresDialect INSTANCE = new PostgresDialect();
private static final BindMarkersFactory INDEXED = BindMarkersFactory.indexed("$", 1);
private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
*/
@Override
public String getClause(long limit, long offset) {
return String.format("LIMIT %d OFFSET %d", limit, offset);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
*/
@Override
public String getClause(long limit) {
return "LIMIT " + limit;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
*/
@Override
public Position getClausePosition() {
return Position.END;
}
};
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
*/
@Override
public BindMarkersFactory getBindMarkersFactory() {
return INDEXED;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#returnGeneratedKeys()
*/
@Override
public String returnGeneratedKeys() {
return "RETURNING *";
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#limit()
*/
@Override
public LimitClause limit() {
return LIMIT_CLAUSE;
}
}

95
src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java Normal file
View File

@@ -0,0 +1,95 @@
package org.springframework.data.r2dbc.dialect;
/**
* An SQL dialect for Microsoft SQL Server.
*
* @author Mark Paluch
*/
public class SqlServerDialect implements Dialect {
/**
* Singleton instance.
*/
public static final SqlServerDialect INSTANCE = new SqlServerDialect();
private static final BindMarkersFactory NAMED = BindMarkersFactory.named("@", "P", 32,
SqlServerDialect::filterBindMarker);
private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
*/
@Override
public String getClause(long limit) {
return "OFFSET 0 ROWS FETCH NEXT " + limit + " ROWS ONLY";
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
*/
@Override
public String getClause(long limit, long offset) {
return String.format("OFFSET %d ROWS FETCH NEXT %d ROWS ONLY", offset, limit);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
*/
@Override
public Position getClausePosition() {
return Position.END;
}
};
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
*/
@Override
public BindMarkersFactory getBindMarkersFactory() {
return NAMED;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#returnGeneratedKeys()
*/
@Override
public String returnGeneratedKeys() {
return "select SCOPE_IDENTITY() AS GENERATED_KEYS";
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.dialect.Dialect#limit()
*/
@Override
public LimitClause limit() {
return LIMIT_CLAUSE;
}
private static String filterBindMarker(CharSequence input) {
StringBuilder builder = new StringBuilder();
for (int i = 0; i < input.length(); i++) {
char ch = input.charAt(i);
// ascii letter or digit
if (Character.isLetterOrDigit(ch) && ch < 127) {
builder.append(ch);
}
}
if (builder.length() == 0) {
return "";
}
return "_" + builder.toString();
}
}

32
src/main/java/org/springframework/data/r2dbc/function/BindIdOperation.java Normal file
View File

@@ -0,0 +1,32 @@
package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Statement;
/**
* Extension to {@link BindableOperation} for operations that allow parameter substitution for a single {@code id}
* column that accepts either a single value or multiple values, depending on the underlying operation.
*
* @author Mark Paluch
* @see Statement#bind
* @see Statement#bindNull
*/
public interface BindIdOperation extends BindableOperation {
/**
* Bind the given {@code value} to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param value the actual value. Must not be {@literal null}.
* @see Statement#bind
*/
void bindId(Statement<?> statement, Object value);
/**
* Bind the given {@code values} to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param values the actual values.
* @see Statement#bind
*/
void bindIds(Statement<?> statement, Iterable<? extends Object> values);
}

36
src/main/java/org/springframework/data/r2dbc/function/BindableOperation.java Normal file
View File

@@ -0,0 +1,36 @@
package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Statement;
/**
* Extension to {@link QueryOperation} for operations that allow parameter substitution by binding parameter values.
* {@link BindableOperation} is typically created with a {@link Set} of column names or parameter names that accept bind
* parameters by calling {@link #bind(Statement, String, Object)}.
*
* @author Mark Paluch
* @see Statement#bind
* @see Statement#bindNull
*/
public interface BindableOperation extends QueryOperation {
/**
* Bind the given {@code value} to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param identifier named identifier that is considered by the underlying binding strategy.
* @param value the actual value. Must not be {@literal null}. Use {@link #bindNull(Statement, Class)} for
* {@literal null} values.
* @see Statement#bind
*/
void bind(Statement<?> statement, String identifier, Object value);
/**
* Bind a {@literal null} value to the {@link Statement} using the underlying binding strategy.
*
* @param statement the statement to bind the value to.
* @param identifier named identifier that is considered by the underlying binding strategy.
* @param valueType value type, must not be {@literal null}.
* @see Statement#bindNull
*/
void bindNull(Statement<?> statement, String identifier, Class<?> valueType);
}

160
src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
View File

@@ -34,14 +34,13 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -49,8 +48,6 @@ import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.NullHandling;
import org.springframework.data.domain.Sort.Order;
import org.springframework.data.r2dbc.UncategorizedR2dbcException;
import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
@@ -59,7 +56,6 @@ import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
import org.springframework.jdbc.core.SqlProvider;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Default implementation of {@link DatabaseClient}.
@@ -239,7 +235,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
return new DefaultGenericExecuteSpec(sqlSupplier);
}
private static void doBind(Statement statement, Map<String, SettableValue> byName,
private static void doBind(Statement<?> statement, Map<String, SettableValue> byName,
Map<Integer, SettableValue> byIndex) {
byIndex.forEach((i, o) -> {
@@ -308,13 +304,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
<T> SqlResult<T> exchange(String sql, BiFunction<Row, RowMetadata, T> mappingFunction) {
Function<Connection, Statement> executeFunction = it -> {
Function<Connection, Statement<?>> executeFunction = it -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing SQL statement [" + sql + "]");
}
Statement statement = it.createStatement(sql);
Statement<?> statement = it.createStatement(sql);
doBind(statement, byName, byIndex);
return statement;
@@ -571,37 +567,9 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
return createInstance(table, projectedFields, sort, page);
}
StringBuilder getLimitOffset(Pageable pageable) {
return new StringBuilder().append("LIMIT").append(' ').append(pageable.getPageSize()) //
.append(' ').append("OFFSET").append(' ').append(pageable.getOffset());
}
StringBuilder getSortClause(Sort sort) {
StringBuilder sortClause = new StringBuilder();
for (Order order : sort) {
if (sortClause.length() != 0) {
sortClause.append(',').append(' ');
}
sortClause.append(order.getProperty()).append(' ').append(order.getDirection().isAscending() ? "ASC" : "DESC");
if (order.getNullHandling() == NullHandling.NULLS_FIRST) {
sortClause.append(' ').append("NULLS FIRST");
}
if (order.getNullHandling() == NullHandling.NULLS_LAST) {
sortClause.append(' ').append("NULLS LAST");
}
}
return sortClause;
}
<R> SqlResult<R> execute(String sql, BiFunction<Row, RowMetadata, R> mappingFunction) {
Function<Connection, Statement> selectFunction = it -> {
Function<Connection, Statement<?>> selectFunction = it -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing SQL statement [" + sql + "]");
@@ -666,28 +634,17 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private <R> SqlResult<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
List<String> projectedFields;
Set<String> columns;
if (this.projectedFields.isEmpty()) {
projectedFields = Collections.singletonList("*");
columns = Collections.singleton("*");
} else {
projectedFields = this.projectedFields;
columns = new LinkedHashSet<>(this.projectedFields);
}
StringBuilder selectBuilder = new StringBuilder();
selectBuilder.append("SELECT").append(' ') //
.append(StringUtils.collectionToDelimitedString(projectedFields, ", ")).append(' ') //
.append("FROM").append(' ').append(table);
QueryOperation select = dataAccessStrategy.select(table, columns, sort, page);
if (sort.isSorted()) {
selectBuilder.append(' ').append("ORDER BY").append(' ').append(getSortClause(sort));
}
if (page.isPaged()) {
selectBuilder.append(' ').append(getLimitOffset(page));
}
return execute(selectBuilder.toString(), mappingFunction);
return execute(select.toQuery(), mappingFunction);
}
@Override
@@ -765,30 +722,18 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private <R> SqlResult<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
List<String> projectedFields;
List<String> columns;
if (this.projectedFields.isEmpty()) {
projectedFields = dataAccessStrategy.getAllFields(typeToRead);
columns = dataAccessStrategy.getAllColumns(typeToRead);
} else {
projectedFields = this.projectedFields;
columns = this.projectedFields;
}
Sort sortToUse = sort.isSorted() ? dataAccessStrategy.getMappedSort(typeToRead, sort) : Sort.unsorted();
StringBuilder selectBuilder = new StringBuilder();
selectBuilder.append("SELECT").append(' ') //
.append(StringUtils.collectionToDelimitedString(projectedFields, ", ")).append(' ') //
.append("FROM").append(' ').append(table);
QueryOperation select = dataAccessStrategy.select(table, new LinkedHashSet<>(columns), sortToUse, page);
if (sort.isSorted()) {
Sort mappedSort = dataAccessStrategy.getMappedSort(typeToRead, sort);
selectBuilder.append(' ').append("ORDER BY").append(' ').append(getSortClause(mappedSort));
}
if (page.isPaged()) {
selectBuilder.append(' ').append(getLimitOffset(page));
}
return execute(selectBuilder.toString(), mappingFunction);
return execute(select.get(), mappingFunction);
}
@Override
@@ -861,27 +806,29 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
throw new IllegalStateException("Insert fields is empty!");
}
StringBuilder builder = new StringBuilder();
String fieldNames = byName.keySet().stream().collect(Collectors.joining(","));
String placeholders = IntStream.range(0, byName.size()).mapToObj(i -> "$" + (i + 1))
.collect(Collectors.joining(","));
BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet());
builder.append("INSERT INTO ").append(table).append(" (").append(fieldNames).append(") ").append(" VALUES(")
.append(placeholders).append(") RETURNING *");
String sql = builder.toString();
String sql = bindableInsert.toQuery();
Function<Connection, Statement> insertFunction = it -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing SQL statement [" + sql + "]");
}
Statement statement = it.createStatement(sql);
doBind(statement);
Statement<?> statement = it.createStatement(sql);
byName.forEach((k, v) -> {
if (v.getValue() == null) {
bindableInsert.bindNull(statement, k, v.getType());
} else {
bindableInsert.bind(statement, k, v.getValue());
}
});
return statement;
};
Function<Connection, Flux<Result>> resultFunction = it -> Flux
.from(insertFunction.apply(it).execute());
Function<Connection, Flux<Result>> resultFunction = it -> Flux.from(insertFunction.apply(it).execute());
return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
sql, //
@@ -889,25 +836,6 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), //
mappingFunction);
}
/**
* PostgreSQL-specific bind.
*
* @param statement
*/
private void doBind(Statement statement) {
AtomicInteger index = new AtomicInteger();
for (SettableValue value : byName.values()) {
if (value.getValue() != null) {
statement.bind(index.getAndIncrement(), value.getValue());
} else {
statement.bindNull("$" + (index.getAndIncrement() + 1), value.getType());
}
}
}
}
/**
@@ -963,18 +891,16 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private <R> SqlResult<R> exchange(Object toInsert, BiFunction<Row, RowMetadata, R> mappingFunction) {
StringBuilder builder = new StringBuilder();
List<SettableValue> insertValues = dataAccessStrategy.getValuesToInsert(toInsert);
Set<String> columns = new LinkedHashSet<>();
List<SettableValue> insertValues = dataAccessStrategy.getInsert(toInsert);
String fieldNames = insertValues.stream().map(SettableValue::getIdentifier).map(Object::toString)
.collect(Collectors.joining(","));
String placeholders = IntStream.range(0, insertValues.size()).mapToObj(i -> "$" + (i + 1))
.collect(Collectors.joining(","));
for (SettableValue insertValue : insertValues) {
columns.add(insertValue.getIdentifier().toString());
}
builder.append("INSERT INTO ").append(table).append(" (").append(fieldNames).append(") ").append(" VALUES(")
.append(placeholders).append(") RETURNING *");
BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, columns);
String sql = builder.toString();
String sql = bindableInsert.toQuery();
Function<Connection, Statement> insertFunction = it -> {
@@ -982,16 +908,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
logger.debug("Executing SQL statement [" + sql + "]");
}
Statement statement = it.createStatement(sql);
AtomicInteger index = new AtomicInteger();
Statement<?> statement = it.createStatement(sql);
for (SettableValue settable : insertValues) {
if (settable.getValue() != null) {
statement.bind(index.getAndIncrement(), settable.getValue());
if (settable.getValue() == null) {
bindableInsert.bindNull(statement, settable.getIdentifier().toString(), settable.getType());
} else {
statement.bindNull("$" + (index.getAndIncrement() + 1), settable.getType());
bindableInsert.bind(statement, settable.getIdentifier().toString(), settable.getValue());
}
}

25
src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientBuilder.java
View File

@@ -20,6 +20,8 @@ import io.r2dbc.spi.ConnectionFactory;
import java.util.function.Consumer;
import org.springframework.data.r2dbc.dialect.Database;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.function.DatabaseClient.Builder;
import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
import org.springframework.data.r2dbc.support.SqlErrorCodeR2dbcExceptionTranslator;
@@ -33,9 +35,9 @@ import org.springframework.util.Assert;
*/
class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
private @Nullable ConnectionFactory connector;
private @Nullable ConnectionFactory connectionFactory;
private @Nullable R2dbcExceptionTranslator exceptionTranslator;
private ReactiveDataAccessStrategy accessStrategy = new DefaultReactiveDataAccessStrategy();
private ReactiveDataAccessStrategy accessStrategy;
DefaultDatabaseClientBuilder() {}
@@ -43,7 +45,7 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
Assert.notNull(other, "DefaultDatabaseClientBuilder must not be null!");
this.connector = other.connector;
this.connectionFactory = other.connectionFactory;
this.exceptionTranslator = other.exceptionTranslator;
this.accessStrategy = other.accessStrategy;
}
@@ -53,7 +55,7 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
Assert.notNull(factory, "ConnectionFactory must not be null!");
this.connector = factory;
this.connectionFactory = factory;
return this;
}
@@ -81,10 +83,21 @@ class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
R2dbcExceptionTranslator exceptionTranslator = this.exceptionTranslator;
if (exceptionTranslator == null) {
exceptionTranslator = new SqlErrorCodeR2dbcExceptionTranslator(connector);
exceptionTranslator = new SqlErrorCodeR2dbcExceptionTranslator(connectionFactory);
}
return doBuild(this.connector, exceptionTranslator, this.accessStrategy, new DefaultDatabaseClientBuilder(this));
ReactiveDataAccessStrategy accessStrategy = this.accessStrategy;
if (accessStrategy == null) {
Dialect dialect = Database.findDatabase(this.connectionFactory)
.orElseThrow(() -> new UnsupportedOperationException(
"Cannot determine a Dialect. Configure the dialect by providing DefaultReactiveDataAccessStrategy(Dialect)"))
.latestDialect();
accessStrategy = new DefaultReactiveDataAccessStrategy(dialect);
}
return doBuild(this.connectionFactory, exceptionTranslator, accessStrategy, new DefaultDatabaseClientBuilder(this));
}
protected DatabaseClient doBuild(ConnectionFactory connector, R2dbcExceptionTranslator exceptionTranslator,

484
src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
View File

@@ -17,16 +17,27 @@ package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.function.Function;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Order;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.r2dbc.dialect.BindMarker;
import org.springframework.data.r2dbc.dialect.BindMarkers;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.dialect.LimitClause;
import org.springframework.data.r2dbc.dialect.LimitClause.Position;
import org.springframework.data.r2dbc.function.convert.EntityRowMapper;
import org.springframework.data.r2dbc.function.convert.SettableValue;
import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
@@ -34,27 +45,51 @@ import org.springframework.data.relational.core.conversion.RelationalConverter;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.util.StreamUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
* Default {@link ReactiveDataAccessStrategy} implementation.
*
* @author Mark Paluch
*/
public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
private final RelationalConverter relationalConverter;
private final Dialect dialect;
public DefaultReactiveDataAccessStrategy() {
this(new BasicRelationalConverter(new RelationalMappingContext()));
/**
* Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect}.
*
* @param dialect the {@link Dialect} to use.
*/
public DefaultReactiveDataAccessStrategy(Dialect dialect) {
this(dialect, new BasicRelationalConverter(new RelationalMappingContext()));
}
public DefaultReactiveDataAccessStrategy(RelationalConverter converter) {
/**
* Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect} and {@link RelationalConverter}.
*
* @param dialect the {@link Dialect} to use.
* @param converter must not be {@literal null}.
*/
public DefaultReactiveDataAccessStrategy(Dialect dialect, RelationalConverter converter) {
Assert.notNull(dialect, "Dialect must not be null");
Assert.notNull(converter, "RelationalConverter must not be null");
this.relationalConverter = converter;
this.dialect = dialect;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getAllFields(java.lang.Class)
*/
@Override
public List<String> getAllFields(Class<?> typeToRead) {
public List<String> getAllColumns(Class<?> typeToRead) {
RelationalPersistentEntity<?> persistentEntity = getPersistentEntity(typeToRead);
@@ -62,13 +97,20 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return Collections.singletonList("*");
}
return StreamUtils.createStreamFromIterator(persistentEntity.iterator()) //
.map(RelationalPersistentProperty::getColumnName) //
.collect(Collectors.toList());
List<String> columnNames = new ArrayList<>();
for (RelationalPersistentProperty property : persistentEntity) {
columnNames.add(property.getColumnName());
}
return columnNames;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getValuesToInsert(java.lang.Object)
*/
@Override
public List<SettableValue> getInsert(Object object) {
public List<SettableValue> getValuesToInsert(Object object) {
Class<?> userClass = ClassUtils.getUserClass(object);
@@ -91,6 +133,10 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return values;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getMappedSort(java.lang.Class, org.springframework.data.domain.Sort)
*/
@Override
public Sort getMappedSort(Class<?> typeToRead, Sort sort) {
@@ -115,12 +161,20 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return Sort.by(mappedOrder);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getRowMapper(java.lang.Class)
*/
@Override
public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
return new EntityRowMapper<T>((RelationalPersistentEntity) getRequiredPersistentEntity(typeToRead),
relationalConverter);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getTableName(java.lang.Class)
*/
@Override
public String getTableName(Class<?> type) {
return getRequiredPersistentEntity(type).getTableName();
@@ -134,4 +188,414 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
private RelationalPersistentEntity<?> getPersistentEntity(Class<?> typeToRead) {
return relationalConverter.getMappingContext().getPersistentEntity(typeToRead);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#insertAndReturnGeneratedKeys(java.lang.String, java.util.Set)
*/
@Override
public BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns) {
return new DefaultBindableInsert(dialect.getBindMarkersFactory().create(), table, columns,
dialect.returnGeneratedKeys());
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#select(java.lang.String, java.util.Set, org.springframework.data.domain.Sort, org.springframework.data.domain.Pageable)
*/
@Override
public QueryOperation select(String table, Set<String> columns, Sort sort, Pageable page) {
StringBuilder selectBuilder = new StringBuilder();
selectBuilder.append("SELECT").append(' ') //
.append(StringUtils.collectionToDelimitedString(columns, ", ")).append(' ') //
.append("FROM").append(' ').append(table);
if (sort.isSorted()) {
selectBuilder.append(' ').append("ORDER BY").append(' ').append(getSortClause(sort));
}
if (page.isPaged()) {
LimitClause limitClause = dialect.limit();
if (limitClause.getClausePosition() == Position.END) {
selectBuilder.append(' ').append(limitClause.getClause(page.getPageSize(), page.getOffset()));
}
}
return selectBuilder::toString;
}
private StringBuilder getSortClause(Sort sort) {
StringBuilder sortClause = new StringBuilder();
for (Order order : sort) {
if (sortClause.length() != 0) {
sortClause.append(',').append(' ');
}
sortClause.append(order.getProperty()).append(' ').append(order.getDirection().isAscending() ? "ASC" : "DESC");
}
return sortClause;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#selectById(java.lang.String, java.util.Set, java.lang.String)
*/
@Override
public BindIdOperation selectById(String table, Set<String> columns, String idColumn) {
return new DefaultBindIdOperation(dialect.getBindMarkersFactory().create(), marker -> {
String columnClause = StringUtils.collectionToDelimitedString(columns, ", ");
return String.format("SELECT %s FROM %s WHERE %s = %s", columnClause, table, idColumn, marker.getPlaceholder());
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#selectById(java.lang.String, java.util.Set, java.lang.String, int)
*/
@Override
public BindIdOperation selectById(String table, Set<String> columns, String idColumn, int limit) {
LimitClause limitClause = dialect.limit();
return new DefaultBindIdOperation(dialect.getBindMarkersFactory().create(), marker -> {
String columnClause = StringUtils.collectionToDelimitedString(columns, ", ");
if (limitClause.getClausePosition() == Position.END) {
return String.format("SELECT %s FROM %s WHERE %s = %s ORDER BY %s %s", columnClause, table, idColumn,
marker.getPlaceholder(), idColumn, limitClause.getClause(limit));
}
throw new UnsupportedOperationException(
String.format("Limit clause position %s not supported!", limitClause.getClausePosition()));
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#selectByIdIn(java.lang.String, java.util.Set, java.lang.String)
*/
@Override
public BindIdOperation selectByIdIn(String table, Set<String> columns, String idColumn) {
String query = String.format("SELECT %s FROM %s", StringUtils.collectionToDelimitedString(columns, ", "), table);
return new DefaultBindIdIn(dialect.getBindMarkersFactory().create(), query, idColumn);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#updateById(java.lang.String, java.util.Set, java.lang.String)
*/
@Override
public BindIdOperation updateById(String table, Set<String> columns, String idColumn) {
return new DefaultBindableUpdate(dialect.getBindMarkersFactory().create(), table, columns, idColumn);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteById(java.lang.String, java.lang.String)
*/
@Override
public BindIdOperation deleteById(String table, String idColumn) {
return new DefaultBindIdOperation(dialect.getBindMarkersFactory().create(),
marker -> String.format("DELETE FROM %s WHERE %s = %s", table, idColumn, marker.getPlaceholder()));
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#deleteByIdIn(java.lang.String, java.lang.String)
*/
@Override
public BindIdOperation deleteByIdIn(String table, String idColumn) {
String query = String.format("DELETE FROM %s", table);
return new DefaultBindIdIn(dialect.getBindMarkersFactory().create(), query, idColumn);
}
/**
* Default {@link BindableOperation} implementation for a {@code INSERT} operation.
*/
static class DefaultBindableInsert implements BindableOperation {
private final Map<String, BindMarker> markers = new LinkedHashMap<>();
private final String query;
DefaultBindableInsert(BindMarkers bindMarkers, String table, Collection<String> columns,
String returningStatement) {
StringBuilder builder = new StringBuilder();
List<String> placeholders = new ArrayList<>(columns.size());
for (String column : columns) {
BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
placeholders.add(marker.getPlaceholder());
}
String columnsString = StringUtils.collectionToDelimitedString(columns, ", ");
String placeholdersString = StringUtils.collectionToDelimitedString(placeholders, ", ");
builder.append("INSERT INTO ").append(table).append(" (").append(columnsString).append(")").append(" VALUES(")
.append(placeholdersString).append(")");
if (StringUtils.hasText(returningStatement)) {
builder.append(' ').append(returningStatement);
}
this.query = builder.toString();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement<?> statement, String identifier, Object value) {
markers.get(identifier).bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
*/
@Override
public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
markers.get(identifier).bindNull(statement, valueType);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code UPDATE} operation using a single key.
*/
static class DefaultBindableUpdate implements BindIdOperation {
private final Map<String, BindMarker> markers = new LinkedHashMap<>();
private final BindMarker idMarker;
private final String query;
DefaultBindableUpdate(BindMarkers bindMarkers, String tableName, Set<String> columns, String idColumnName) {
this.idMarker = bindMarkers.next();
StringBuilder setClause = new StringBuilder();
for (String column : columns) {
BindMarker marker = markers.computeIfAbsent(column, bindMarkers::next);
if (setClause.length() != 0) {
setClause.append(", ");
}
setClause.append(column).append(" = ").append(marker.getPlaceholder());
}
this.query = String.format("UPDATE %s SET %s WHERE %s = %s", tableName, setClause, idColumnName,
idMarker.getPlaceholder());
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement<?> statement, String identifier, Object value) {
markers.get(identifier).bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
*/
@Override
public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
markers.get(identifier).bindNull(statement, valueType);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement<?> statement, Object value) {
idMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement<?> statement, Iterable<? extends Object> values) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code SELECT} or {@code DELETE} operation using a single key
* in the {@code WHERE} predicate.
*/
static class DefaultBindIdOperation implements BindIdOperation {
private final BindMarker idMarker;
private final String query;
DefaultBindIdOperation(BindMarkers bindMarkers, Function<BindMarker, String> queryFunction) {
this.idMarker = bindMarkers.next();
this.query = queryFunction.apply(this.idMarker);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement<?> statement, String identifier, Object value) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
*/
@Override
public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement<?> statement, Object value) {
idMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement<?> statement, Iterable<? extends Object> values) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
return this.query;
}
}
/**
* Default {@link BindIdOperation} implementation for a {@code SELECT … WHERE id IN (…)} or
* {@code DELETE … WHERE id IN (…)}.
*/
static class DefaultBindIdIn implements BindIdOperation {
private final List<String> markers = new ArrayList<>();
private final BindMarkers bindMarkers;
private final String baseQuery;
private final String idColumnName;
DefaultBindIdIn(BindMarkers bindMarkers, String baseQuery, String idColumnName) {
this.bindMarkers = bindMarkers;
this.baseQuery = baseQuery;
this.idColumnName = idColumnName;
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bind(io.r2dbc.spi.Statement, java.lang.String, java.lang.Object)
*/
@Override
public void bind(Statement<?> statement, String identifier, Object value) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindableOperation#bindNull(io.r2dbc.spi.Statement, java.lang.String, java.lang.Class)
*/
@Override
public void bindNull(Statement<?> statement, String identifier, Class<?> valueType) {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindId(io.r2dbc.spi.Statement, java.lang.Object)
*/
@Override
public void bindId(Statement<?> statement, Object value) {
BindMarker bindMarker = bindMarkers.next();
markers.add(bindMarker.getPlaceholder());
bindMarker.bind(statement, value);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.BindIdOperation#bindIds(io.r2dbc.spi.Statement, java.lang.Iterable)
*/
@Override
public void bindIds(Statement<?> statement, Iterable<? extends Object> values) {
for (Object value : values) {
bindId(statement, value);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.function.QueryOperation#toQuery()
*/
@Override
public String toQuery() {
if (this.markers.isEmpty()) {
throw new UnsupportedOperationException();
}
String in = StringUtils.collectionToDelimitedString(this.markers, ", ");
return String.format("%s WHERE %s IN (%s)", this.baseQuery, this.idColumnName, in);
}
}
}

26
src/main/java/org/springframework/data/r2dbc/function/QueryOperation.java Normal file
View File

@@ -0,0 +1,26 @@
package org.springframework.data.r2dbc.function;
import java.util.function.Supplier;
/**
* Interface declaring a query operation that can be represented with a query string. This interface is typically
* implemented by classes representing a SQL operation such as {@code SELECT}, {@code INSERT}, and such.
*
* @author Mark Paluch
*/
@FunctionalInterface
public interface QueryOperation extends Supplier<String> {
/**
* Returns the string-representation of this operation to be used with {@link io.r2dbc.spi.Statement} creation.
*
* @return the operation as SQL string.
* @see io.r2dbc.spi.Connection#createStatement(String)
*/
String toQuery();
@Override
default String get() {
return toQuery();
}
}

104
src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
View File

@@ -17,15 +17,23 @@ package org.springframework.data.r2dbc.function;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import java.util.List;
import java.util.Set;
import java.util.function.BiFunction;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.function.convert.SettableValue;
/**
* Draft of a data access strategy that generalizes convenience operations using mapped entities. Typically used
* internally by {@link DatabaseClient} and repository support. SQL creation is limited to single-table operations and
* single-column primary keys.
*
* @author Mark Paluch
* @see BindableOperation
*/
public interface ReactiveDataAccessStrategy {
@@ -33,13 +41,13 @@ public interface ReactiveDataAccessStrategy {
* @param typeToRead
* @return all field names for a specific type.
*/
List<String> getAllFields(Class<?> typeToRead);
List<String> getAllColumns(Class<?> typeToRead);
/**
* @param object
* @return {@link SettableValue} that represent an {@code INSERT} of {@code object}.
*/
List<SettableValue> getInsert(Object object);
List<SettableValue> getValuesToInsert(Object object);
/**
* Map the {@link Sort} object to apply field name mapping using {@link Class the type to read}.
@@ -58,4 +66,96 @@ public interface ReactiveDataAccessStrategy {
* @return the table name for the {@link Class entity type}.
*/
String getTableName(Class<?> type);
// -------------------------------------------------------------------------
// Methods creating SQL operations.
// Subject to be moved into a SQL creation DSL.
// -------------------------------------------------------------------------
/**
* Create an {@code INSERT} operation for the given {@code table} to insert {@code columns}.
*
* @param table the table to insert data to.
* @param columns column names that will be bound.
* @return the {@link BindableOperation} representing the {@code INSERT} statement.
*/
BindableOperation insertAndReturnGeneratedKeys(String table, Set<String> columns);
/**
* Create a {@code SELECT … ORDER BY … LIMIT …} operation for the given {@code table} using {@code columns} to
* project.
*
* @param table the table to insert data to.
* @param columns columns to return.
* @param sort
* @param page
* @return
*/
QueryOperation select(String table, Set<String> columns, Sort sort, Pageable page);
/**
* Create a {@code SELECT … WHERE id = ?} operation for the given {@code table} using {@code columns} to project and
* {@code idColumn}.
*
* @param table the table to insert data to.
* @param columns columns to return.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation selectById(String table, Set<String> columns, String idColumn);
/**
* Create a {@code SELECT … WHERE id = ?} operation for the given {@code table} using {@code columns} to project and
* {@code idColumn} applying a limit (TOP, LIMIT, …).
*
* @param table the table to insert data to.
* @param columns columns to return.
* @param idColumn name of the primary key.
* @param limit number of rows to return.
* @return
*/
BindIdOperation selectById(String table, Set<String> columns, String idColumn, int limit);
/**
* Create a {@code SELECT … WHERE id IN (?)} operation for the given {@code table} using {@code columns} to project
* and {@code idColumn}. The actual {@link BindableOperation#toQuery() query} string depends on
* {@link BindIdOperation#bindIds(Statement, Iterable) bound parameters}.
*
* @param table the table to insert data to.
* @param columns columns to return.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation selectByIdIn(String table, Set<String> columns, String idColumn);
/**
* Create a {@code UPDATE … SET … WHERE id = ?} operation for the given {@code table} updating {@code columns} and
* {@code idColumn}.
*
* @param table the table to insert data to.
* @param columns columns to update.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation updateById(String table, Set<String> columns, String idColumn);
/**
* Create a {@code DELETE … WHERE id = ?} operation for the given {@code table} and {@code idColumn}.
*
* @param table the table to insert data to.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation deleteById(String table, String idColumn);
/**
* Create a {@code DELETE … WHERE id IN (?)} operation for the given {@code table} and {@code idColumn}. The actual
* {@link BindableOperation#toQuery() query} string depends on {@link BindIdOperation#bindIds(Statement, Iterable)
* bound parameters}.
*
* @param table the table to insert data to.
* @param idColumn name of the primary key.
* @return
*/
BindIdOperation deleteByIdIn(String table, String idColumn);
}

68
src/main/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverter.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.r2dbc.function.convert;
import io.r2dbc.spi.ColumnMetadata;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
@@ -26,6 +27,7 @@ import java.util.function.BiFunction;
import org.springframework.core.convert.ConversionService;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
import org.springframework.data.relational.core.conversion.RelationalConverter;
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -41,7 +43,25 @@ public class MappingR2dbcConverter {
private final RelationalConverter relationalConverter;
/**
* Creates a new {@link MappingR2dbcConverter} given {@link MappingContext}.
*
* @param context must not be {@literal null}.
*/
public MappingR2dbcConverter(
MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context) {
this(new BasicRelationalConverter(context));
}
/**
* Creates a new {@link MappingR2dbcConverter} given {@link RelationalConverter}.
*
* @param converter must not be {@literal null}.
*/
public MappingR2dbcConverter(RelationalConverter converter) {
Assert.notNull(converter, "RelationalConverter must not be null!");
this.relationalConverter = converter;
}
@@ -52,7 +72,7 @@ public class MappingR2dbcConverter {
* @param object must not be {@literal null}.
* @return
*/
public Map<String, SettableValue> getFieldsToUpdate(Object object) {
public Map<String, SettableValue> getColumnsToUpdate(Object object) {
Assert.notNull(object, "Entity object must not be null!");
@@ -93,18 +113,52 @@ public class MappingR2dbcConverter {
if (propertyAccessor.getProperty(idProperty) == null) {
ConversionService conversionService = relationalConverter.getConversionService();
Object value = row.get(idProperty.getColumnName());
propertyAccessor.setProperty(idProperty, conversionService.convert(value, idProperty.getType()));
return (T) propertyAccessor.getBean();
if (potentiallySetId(row, metadata, propertyAccessor, idProperty)) {
return (T) propertyAccessor.getBean();
}
}
return object;
};
}
private boolean potentiallySetId(Row row, RowMetadata metadata, PersistentPropertyAccessor<?> propertyAccessor,
RelationalPersistentProperty idProperty) {
Map<String, ColumnMetadata> columns = createMetadataMap(metadata);
Object generatedIdValue = null;
if (columns.containsKey(idProperty.getColumnName())) {
generatedIdValue = row.get(idProperty.getColumnName());
}
if (columns.size() == 1) {
String key = columns.keySet().iterator().next();
generatedIdValue = row.get(key);
}
if (generatedIdValue != null) {
ConversionService conversionService = relationalConverter.getConversionService();
propertyAccessor.setProperty(idProperty, conversionService.convert(generatedIdValue, idProperty.getType()));
return true;
}
return false;
}
private static Map<String, ColumnMetadata> createMetadataMap(RowMetadata metadata) {
Map<String, ColumnMetadata> columns = new LinkedHashMap<>();
for (ColumnMetadata column : metadata.getColumnMetadatas()) {
columns.put(column.getName(), column);
}
return columns;
}
public MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
return relationalConverter.getMappingContext();
}

1
src/main/java/org/springframework/data/r2dbc/repository/config/R2dbcRepositoryConfigurationExtension.java
View File

@@ -98,6 +98,7 @@ public class R2dbcRepositoryConfigurationExtension extends RepositoryConfigurati
AnnotationAttributes attributes = config.getAttributes();
builder.addPropertyReference("databaseClient", attributes.getString("databaseClientRef"));
builder.addPropertyReference("dataAccessStrategy", "reactiveDataAccessStrategy");
}
/*

10
src/main/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQuery.java
View File

@@ -88,21 +88,19 @@ public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
T bindSpecToUse = bindSpec;
// TODO: Encapsulate PostgreSQL-specific bindings
Parameters<?, ?> bindableParameters = accessor.getBindableParameters();
int index = 1;
int index = 0;
for (Object value : accessor.getValues()) {
Parameter bindableParameter = bindableParameters.getBindableParameter(index - 1);
Parameter bindableParameter = bindableParameters.getBindableParameter(index);
if (value == null) {
if (accessor.hasBindableNullValue()) {
bindSpecToUse = bindSpecToUse.bindNull("$" + (index++), bindableParameter.getType());
bindSpecToUse = bindSpecToUse.bindNull(index++, bindableParameter.getType());
}
} else {
bindSpecToUse = bindSpecToUse.bind("$" + (index++), value);
bindSpecToUse = bindSpecToUse.bind(index++, value);
}
}

103
src/main/java/org/springframework/data/r2dbc/repository/support/BindSpecWrapper.java Normal file
View File

@@ -0,0 +1,103 @@
package org.springframework.data.r2dbc.repository.support;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Statement;
import org.reactivestreams.Publisher;
import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
/**
* Wrapper for {@link BindSpec} to be used with {@link org.springframework.data.r2dbc.dialect.BindMarker} binding.
* Binding parameters updates the {@link BindSpec}
*
* @param <S> type of the bind specification.
* @author Mark Paluch
*/
class BindSpecWrapper<S extends BindSpec<S>> implements Statement<BindSpecWrapper<S>> {
private S bindSpec;
private BindSpecWrapper(S bindSpec) {
this.bindSpec = bindSpec;
}
/**
* Create a new {@link BindSpecWrapper} for the given {@link BindSpec}.
*
* @param bindSpec the bind specification.
* @param <S> type of the bind spec to retain the type through {@link #getBoundOperation()}.
* @return {@link BindSpecWrapper} for the {@link BindSpec}.
*/
public static <S extends BindSpec<S>> BindSpecWrapper<S> create(S bindSpec) {
return new BindSpecWrapper<>(bindSpec);
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#add()
*/
@Override
public BindSpecWrapper<S> add() {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#execute()
*/
@Override
public Publisher<? extends Result> execute() {
throw new UnsupportedOperationException();
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#bind(java.lang.Object, java.lang.Object)
*/
@Override
public BindSpecWrapper<S> bind(Object identifier, Object value) {
this.bindSpec = bindSpec.bind((String) identifier, value);
return this;
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#bind(int, java.lang.Object)
*/
@Override
public BindSpecWrapper<S> bind(int index, Object value) {
this.bindSpec = bindSpec.bind(index, value);
return this;
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#bindNull(java.lang.Object, java.lang.Class)
*/
@Override
public BindSpecWrapper<S> bindNull(Object identifier, Class<?> type) {
this.bindSpec = bindSpec.bindNull((String) identifier, type);
return this;
}
/*
* (non-Javadoc)
* @see io.r2dbc.spi.Statement#bindNull(int, java.lang.Class)
*/
@Override
public BindSpecWrapper<S> bindNull(int index, Class<?> type) {
this.bindSpec = bindSpec.bindNull(index, type);
return this;
}
/**
* @return the bound (final) bind specification.
*/
public S getBoundOperation() {
return bindSpec;
}
}

10
src/main/java/org/springframework/data/r2dbc/repository/support/R2dbcRepositoryFactory.java
View File

@@ -24,6 +24,7 @@ import java.util.Optional;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.repository.R2dbcRepository;
import org.springframework.data.r2dbc.repository.query.R2dbcQueryMethod;
@@ -57,6 +58,7 @@ public class R2dbcRepositoryFactory extends ReactiveRepositoryFactorySupport {
private final DatabaseClient databaseClient;
private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
private final MappingR2dbcConverter converter;
private final ReactiveDataAccessStrategy dataAccessStrategy;
/**
* Creates a new {@link R2dbcRepositoryFactory} given {@link DatabaseClient} and {@link MappingContext}.
@@ -65,13 +67,16 @@ public class R2dbcRepositoryFactory extends ReactiveRepositoryFactorySupport {
* @param mappingContext must not be {@literal null}.
*/
public R2dbcRepositoryFactory(DatabaseClient databaseClient,
MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext,
ReactiveDataAccessStrategy dataAccessStrategy) {
Assert.notNull(databaseClient, "DatabaseClient must not be null!");
Assert.notNull(mappingContext, "MappingContext must not be null!");
Assert.notNull(dataAccessStrategy, "ReactiveDataAccessStrategy must not be null!");
this.databaseClient = databaseClient;
this.mappingContext = mappingContext;
this.dataAccessStrategy = dataAccessStrategy;
this.converter = new MappingR2dbcConverter(new BasicRelationalConverter(mappingContext));
}
@@ -94,7 +99,8 @@ public class R2dbcRepositoryFactory extends ReactiveRepositoryFactorySupport {
RelationalEntityInformation<?, ?> entityInformation = getEntityInformation(information.getDomainType(),
information);
return getTargetRepositoryViaReflection(information, entityInformation, databaseClient, converter);
return getTargetRepositoryViaReflection(information, entityInformation, databaseClient, converter,
dataAccessStrategy);
}
/*

12
src/main/java/org/springframework/data/r2dbc/repository/support/R2dbcRepositoryFactoryBean.java
View File

@@ -19,6 +19,7 @@ import java.io.Serializable;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -40,8 +41,8 @@ public class R2dbcRepositoryFactoryBean<T extends Repository<S, ID>, S, ID exten
extends RepositoryFactoryBeanSupport<T, S, ID> {
private @Nullable DatabaseClient client;
private @Nullable
MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
private @Nullable MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
private @Nullable ReactiveDataAccessStrategy dataAccessStrategy;
private boolean mappingContextConfigured = false;
@@ -80,6 +81,10 @@ public class R2dbcRepositoryFactoryBean<T extends Repository<S, ID>, S, ID exten
}
}
public void setDataAccessStrategy(@Nullable ReactiveDataAccessStrategy dataAccessStrategy) {
this.dataAccessStrategy = dataAccessStrategy;
}
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport#createRepositoryFactory()
@@ -98,7 +103,7 @@ public class R2dbcRepositoryFactoryBean<T extends Repository<S, ID>, S, ID exten
*/
protected RepositoryFactorySupport getFactoryInstance(DatabaseClient client,
MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
return new R2dbcRepositoryFactory(client, mappingContext);
return new R2dbcRepositoryFactory(client, mappingContext, dataAccessStrategy);
}
/*
@@ -109,6 +114,7 @@ public class R2dbcRepositoryFactoryBean<T extends Repository<S, ID>, S, ID exten
public void afterPropertiesSet() {
Assert.state(client != null, "DatabaseClient must not be null!");
Assert.state(dataAccessStrategy != null, "ReactiveDataAccessStrategy must not be null!");
if (!mappingContextConfigured) {
setMappingContext(new RelationalMappingContext());

159
src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
View File

@@ -15,25 +15,30 @@
*/
package org.springframework.data.r2dbc.repository.support;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import io.r2dbc.spi.Statement;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.function.BiConsumer;
import org.reactivestreams.Publisher;
import org.springframework.data.r2dbc.function.BindIdOperation;
import org.springframework.data.r2dbc.function.BindableOperation;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
import org.springframework.data.r2dbc.function.FetchSpec;
import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.function.convert.SettableValue;
import org.springframework.data.relational.repository.query.RelationalEntityInformation;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Simple {@link ReactiveCrudRepository} implementation using R2DBC through {@link DatabaseClient}.
@@ -46,6 +51,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
private final @NonNull RelationalEntityInformation<T, ID> entity;
private final @NonNull DatabaseClient databaseClient;
private final @NonNull MappingR2dbcConverter converter;
private final @NonNull ReactiveDataAccessStrategy accessStrategy;
/* (non-Javadoc)
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#save(S)
@@ -64,51 +70,24 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
.flatMap(it -> it.extract(converter.populateIdIfNecessary(objectToSave)).one());
}
// TODO: Extract in some kind of SQL generator
Object id = entity.getRequiredId(objectToSave);
Map<String, SettableValue> columns = converter.getColumnsToUpdate(objectToSave);
columns.remove(getIdColumnName()); // do not update the Id column.
String idColumnName = getIdColumnName();
BindIdOperation update = accessStrategy.updateById(entity.getTableName(), columns.keySet(), idColumnName);
Map<String, SettableValue> fields = converter.getFieldsToUpdate(objectToSave);
GenericExecuteSpec exec = databaseClient.execute().sql(update);
String setClause = getSetClause(fields);
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(exec);
columns.forEach(bind(update, wrapper));
update.bindId(wrapper, id);
GenericExecuteSpec exec = databaseClient.execute()
.sql(String.format("UPDATE %s SET %s WHERE %s = $1", entity.getTableName(), setClause, getIdColumnName())) //
.bind(0, id);
int index = 1;
for (SettableValue setValue : fields.values()) {
Object value = setValue.getValue();
if (value != null) {
exec = exec.bind(index++, value);
} else {
exec = exec.bindNull(index++, setValue.getType());
}
}
return exec.as(entity.getJavaType()) //
return wrapper.getBoundOperation().as(entity.getJavaType()) //
.exchange() //
.flatMap(FetchSpec::rowsUpdated) //
.thenReturn(objectToSave);
}
private static String getSetClause(Map<String, ?> fields) {
StringBuilder setClause = new StringBuilder();
int index = 2;
for (String field : fields.keySet()) {
if (setClause.length() != 0) {
setClause.append(", ");
}
setClause.append(field).append('=').append('$').append(index++);
}
return setClause.toString();
}
/* (non-Javadoc)
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#saveAll(java.lang.Iterable)
*/
@@ -139,14 +118,17 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Assert.notNull(id, "Id must not be null!");
// TODO: Generate proper SQL (select, where clause, parameter binding).
return databaseClient.execute()
.sql(String.format("SELECT * FROM %s WHERE %s = $1", entity.getTableName(), getIdColumnName())) //
.bind("$1", id) //
.as(entity.getJavaType()) //
Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
String idColumnName = getIdColumnName();
BindIdOperation select = accessStrategy.selectById(entity.getTableName(), columns, idColumnName);
GenericExecuteSpec sql = databaseClient.execute().sql(select);
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(sql);
select.bindId(wrapper, id);
return wrapper.getBoundOperation().as(entity.getJavaType()) //
.fetch() //
.one();
}
/* (non-Javadoc)
@@ -165,11 +147,15 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Assert.notNull(id, "Id must not be null!");
// TODO: Generate proper SQL (select, where clause, parameter binding).
return databaseClient.execute()
.sql(String.format("SELECT %s FROM %s WHERE %s = $1 LIMIT 1", getIdColumnName(), entity.getTableName(),
getIdColumnName())) //
.bind("$1", id) //
String idColumnName = getIdColumnName();
BindIdOperation select = accessStrategy.selectById(entity.getTableName(), Collections.singleton(idColumnName),
idColumnName, 10);
GenericExecuteSpec sql = databaseClient.execute().sql(select);
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(sql);
select.bindId(wrapper, id);
return wrapper.getBoundOperation().as(entity.getJavaType()) //
.exchange() //
.flatMap(it -> it.extract((r, md) -> r).first()).hasElement();
}
@@ -211,12 +197,18 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
return Flux.from(idPublisher).buffer().filter(ids -> !ids.isEmpty()).concatMap(ids -> {
String bindings = getInBinding(ids);
if (ids.isEmpty()) {
return Flux.empty();
}
GenericExecuteSpec exec = databaseClient.execute()
.sql(String.format("SELECT * FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
Set<String> columns = new LinkedHashSet<>(accessStrategy.getAllColumns(entity.getJavaType()));
String idColumnName = getIdColumnName();
BindIdOperation select = accessStrategy.selectByIdIn(entity.getTableName(), columns, idColumnName);
return bind(ids, exec).as(entity.getJavaType()).fetch().all();
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(databaseClient.execute().sql(select));
select.bindIds(wrapper, ids);
return wrapper.getBoundOperation().as(entity.getJavaType()).fetch().all();
});
}
@@ -242,9 +234,12 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
Assert.notNull(id, "Id must not be null!");
return databaseClient.execute()
.sql(String.format("DELETE FROM %s WHERE %s = $1", entity.getTableName(), getIdColumnName())) //
.bind("$1", id) //
BindIdOperation delete = accessStrategy.deleteById(entity.getTableName(), getIdColumnName());
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(databaseClient.execute().sql(delete));
delete.bindId(wrapper, id);
return wrapper.getBoundOperation() //
.fetch() //
.rowsUpdated() //
.then();
@@ -260,12 +255,17 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
return Flux.from(idPublisher).buffer().filter(ids -> !ids.isEmpty()).concatMap(ids -> {
String bindings = getInBinding(ids);
if (ids.isEmpty()) {
return Flux.empty();
}
GenericExecuteSpec exec = databaseClient.execute()
.sql(String.format("DELETE FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
String idColumnName = getIdColumnName();
BindIdOperation delete = accessStrategy.deleteByIdIn(entity.getTableName(), idColumnName);
return bind(ids, exec).as(entity.getJavaType()).fetch().rowsUpdated();
BindSpecWrapper<GenericExecuteSpec> wrapper = BindSpecWrapper.create(databaseClient.execute().sql(delete));
delete.bindIds(wrapper, ids);
return wrapper.getBoundOperation().as(entity.getJavaType()).fetch().rowsUpdated();
}).then();
}
@@ -273,7 +273,6 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#delete(java.lang.Object)
*/
@Override
@SuppressWarnings("unchecked")
public Mono<Void> delete(T objectToDelete) {
Assert.notNull(objectToDelete, "Object to delete must not be null!");
@@ -296,7 +295,6 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteAll(org.reactivestreams.Publisher)
*/
@Override
@SuppressWarnings("unchecked")
public Mono<Void> deleteAll(Publisher<? extends T> objectPublisher) {
Assert.notNull(objectPublisher, "The Object Publisher must not be null!");
@@ -318,22 +316,19 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
.then();
}
private String getInBinding(List<ID> ids) {
return IntStream.range(1, ids.size() + 1).mapToObj(i -> "$" + i).collect(Collectors.joining(", "));
}
@SuppressWarnings("unchecked")
private <S extends BindSpec<?>> S bind(List<ID> it, S bindSpec) {
for (int i = 0; i < it.size(); i++) {
bindSpec = (S) bindSpec.bind(i, it.get(i));
}
return bindSpec;
}
private String getIdColumnName() {
return converter.getMappingContext().getRequiredPersistentEntity(entity.getJavaType()).getRequiredIdProperty()
.getColumnName();
}
private BiConsumer<String, SettableValue> bind(BindableOperation operation, Statement<?> statement) {
return (k, v) -> {
if (v.getValue() == null) {
operation.bindNull(statement, k, v.getType());
} else {
operation.bind(statement, k, v.getValue());
}
};
}
}