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

#368 - Use Spring R2DBC.

Browse Source 
We now use Spring R2DBC DatabaseClient and utilities to implement Spring Data R2DBC functionality.

Original pull request: #412.
This commit is contained in:
Mark Paluch
2020-07-21 14:21:43 +02:00
parent f331572167
commit 086c5a1d2d
50 changed files with 535 additions and 1450 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/asciidoc/new-features.adoc
View File

@@ -4,6 +4,7 @@
[[new-features.1-2-0]]
== What's New in Spring Data R2DBC 1.2.0
* Deprecate Spring Data R2DBC `DatabaseClient` and move off deprecated API in favor of Spring R2DBC. Consult the <<upgrading.1.1-1.2,Migration Guide>> for further details.
* Support for <<entity-callbacks>>.
* <<r2dbc.auditing,Auditing>> through `@EnableR2dbcAuditing`.

10
src/main/asciidoc/reference/r2dbc-upgrading.adoc
View File

@@ -1,4 +1,5 @@
[appendix]
[[migration-guide]]
= Migration Guide
The following sections explain how to migrate to a newer version of Spring Data R2DBC.
@@ -12,6 +13,10 @@ Spring Framework 5.3 ships with a new module: Spring R2DBC.
`spring-r2dbc` ships core R2DBC functionality (a slim variant of `DatabaseClient`, Transaction Manager, Connection Factory initialization, Exception translation) that was initially provided by Spring Data R2DBC. The 1.2.0 release aligns with what's provided in Spring R2DBC by making several changes outlined in the following sections.
Spring R2DBC's `DatabaseClient` is a more lightweight implementation that encapsulates a pure SQL-oriented interface.
You will notice that the method to run SQL statements changed from `DatabaseClient.execute(…)` to `DatabaseClient.sql(…)`.
The fluent API for CRUD operations has moved into `R2dbcEntityTemplate`.
[[upgrading.1.1-1.2.deprecation]]
=== Deprecations
@@ -40,6 +45,11 @@ Specifically the following classes are affected:
We recommend that you review your imports if you work with these types directly.
=== Breaking Changes
* `OutboundRow` and statement mappers switched from using `SettableValue` to `Parameter`
* Repository factory support requires `o.s.r2dbc.core.DatabaseClient` instead of `o.s.data.r2dbc.core.DatabaseClient`.
[[upgrading.1.1-1.2.dependencies]]
=== Dependency Changes

39
src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java
View File

@@ -30,19 +30,18 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.convert.CustomConversions.StoreConversions;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DefaultReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.dialect.DialectResolver;
import org.springframework.data.r2dbc.dialect.R2dbcDialect;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.r2dbc.support.R2dbcExceptionSubclassTranslator;
import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
import org.springframework.data.relational.core.mapping.NamingStrategy;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.util.Assert;
/**
@@ -98,24 +97,34 @@ public abstract class AbstractR2dbcConfiguration implements ApplicationContextAw
* @throws IllegalArgumentException if any of the required args is {@literal null}.
*/
@Bean({ "r2dbcDatabaseClient", "databaseClient" })
public DatabaseClient databaseClient(ReactiveDataAccessStrategy dataAccessStrategy) {
public DatabaseClient databaseClient() {
Assert.notNull(dataAccessStrategy, "DataAccessStrategy must not be null!");
SpelAwareProxyProjectionFactory projectionFactory = new SpelAwareProxyProjectionFactory();
if (context != null) {
projectionFactory.setBeanFactory(context);
projectionFactory.setBeanClassLoader(context.getClassLoader());
}
ConnectionFactory connectionFactory = lookupConnectionFactory();
return DatabaseClient.builder() //
.connectionFactory(lookupConnectionFactory()) //
.dataAccessStrategy(dataAccessStrategy) //
.exceptionTranslator(new R2dbcExceptionSubclassTranslator()) //
.projectionFactory(projectionFactory) //
.connectionFactory(connectionFactory) //
.bindMarkers(getDialect(connectionFactory).getBindMarkersFactory()) //
.build();
}
/**
* Register {@link R2dbcEntityTemplate} using {@link #databaseClient()} and {@link #connectionFactory()}.
*
* @param databaseClient must not be {@literal null}.
* @param dataAccessStrategy must not be {@literal null}.
* @return
* @since 1.2
*/
@Bean
public R2dbcEntityTemplate r2dbcEntityTemplate(DatabaseClient databaseClient,
ReactiveDataAccessStrategy dataAccessStrategy) {
Assert.notNull(databaseClient, "DatabaseClient must not be null!");
Assert.notNull(dataAccessStrategy, "ReactiveDataAccessStrategy must not be null!");
return new R2dbcEntityTemplate(databaseClient, dataAccessStrategy);
}
/**
* Register a {@link R2dbcMappingContext} and apply an optional {@link NamingStrategy}.
*

218
src/main/java/org/springframework/data/r2dbc/connectionfactory/ConnectionFactoryUtils.java
View File

@@ -22,11 +22,8 @@ import reactor.core.publisher.Mono;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.Ordered;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.lang.Nullable;
import org.springframework.transaction.NoTransactionException;
import org.springframework.transaction.reactive.TransactionSynchronization;
import org.springframework.transaction.reactive.TransactionSynchronizationManager;
import org.springframework.util.Assert;
@@ -68,8 +65,7 @@ public abstract class ConnectionFactoryUtils {
* @see #releaseConnection
*/
public static Mono<Connection> getConnection(ConnectionFactory connectionFactory) {
return doGetConnection(connectionFactory)
.onErrorMap(e -> new DataAccessResourceFailureException("Failed to obtain R2DBC Connection", e));
return org.springframework.r2dbc.connection.ConnectionFactoryUtils.getConnection(connectionFactory);
}
/**
@@ -83,64 +79,7 @@ public abstract class ConnectionFactoryUtils {
* @return a R2DBC {@link io.r2dbc.spi.Connection} from the given {@link io.r2dbc.spi.ConnectionFactory}.
*/
public static Mono<Connection> doGetConnection(ConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(synchronizationManager -> {
ConnectionHolder conHolder = (ConnectionHolder) synchronizationManager.getResource(connectionFactory);
if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
conHolder.requested();
if (!conHolder.hasConnection()) {
if (logger.isDebugEnabled()) {
logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory");
}
return fetchConnection(connectionFactory).doOnNext(conHolder::setConnection);
}
return Mono.just(conHolder.getConnection());
}
// Else we either got no holder or an empty thread-bound holder here.
if (logger.isDebugEnabled()) {
logger.debug("Fetching R2DBC Connection from ConnectionFactory");
}
Mono<Connection> con = fetchConnection(connectionFactory);
if (synchronizationManager.isSynchronizationActive()) {
return con.flatMap(it -> {
return Mono.just(it).doOnNext(conn -> {
// Use same Connection for further R2DBC actions within the transaction.
// Thread-bound object will get removed by synchronization at transaction completion.
ConnectionHolder holderToUse = conHolder;
if (holderToUse == null) {
holderToUse = new ConnectionHolder(conn);
} else {
holderToUse.setConnection(conn);
}
holderToUse.requested();
synchronizationManager
.registerSynchronization(new ConnectionSynchronization(holderToUse, connectionFactory));
holderToUse.setSynchronizedWithTransaction(true);
if (holderToUse != conHolder) {
synchronizationManager.bindResource(connectionFactory, holderToUse);
}
}).onErrorResume(e -> {
// Unexpected exception from external delegation call -> close Connection and rethrow.
return releaseConnection(it, connectionFactory).then(Mono.error(e));
});
});
}
return con;
}) //
.onErrorResume(NoTransactionException.class, e -> {
return Mono.from(connectionFactory.create());
});
return org.springframework.r2dbc.connection.ConnectionFactoryUtils.doGetConnection(connectionFactory);
}
/**
@@ -183,17 +122,8 @@ public abstract class ConnectionFactoryUtils {
public static Mono<Void> doReleaseConnection(io.r2dbc.spi.Connection connection,
ConnectionFactory connectionFactory) {
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(it -> {
ConnectionHolder conHolder = (ConnectionHolder) it.getResource(connectionFactory);
if (conHolder != null && connectionEquals(conHolder, connection)) {
// It's the transactional Connection: Don't close it.
conHolder.released();
}
return Mono.from(connection.close());
}).onErrorResume(NoTransactionException.class, e -> {
return doCloseConnection(connection, connectionFactory);
});
return org.springframework.r2dbc.connection.ConnectionFactoryUtils.doReleaseConnection(connection,
connectionFactory);
}
/**
@@ -246,37 +176,7 @@ public abstract class ConnectionFactoryUtils {
*/
public static Mono<ConnectionFactory> currentConnectionFactory(ConnectionFactory connectionFactory) {
return TransactionSynchronizationManager.forCurrentTransaction()
.filter(TransactionSynchronizationManager::isSynchronizationActive).filter(it -> {
ConnectionHolder conHolder = (ConnectionHolder) it.getResource(connectionFactory);
if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
return true;
}
return false;
}).map(it -> connectionFactory);
}
/**
* Determine whether the given two {@link io.r2dbc.spi.Connection}s are equal, asking the target
* {@link io.r2dbc.spi.Connection} in case of a proxy. Used to detect equality even if the user passed in a raw target
* Connection while the held one is a proxy.
*
* @param conHolder the {@link .ConnectionHolder} for the held {@link io.r2dbc.spi.Connection} (potentially a proxy).
* @param passedInCon the {@link io.r2dbc.spi.Connection} passed-in by the user (potentially a target
* {@link io.r2dbc.spi.Connection} without proxy).
* @return whether the given Connections are equal
* @see #getTargetConnection
*/
private static boolean connectionEquals(ConnectionHolder conHolder, Connection passedInCon) {
if (!conHolder.hasConnection()) {
return false;
}
Connection heldCon = conHolder.getConnection();
// Explicitly check for identity too: for Connection handles that do not implement
// "equals" properly).
return (heldCon == passedInCon || heldCon.equals(passedInCon) || getTargetConnection(heldCon).equals(passedInCon));
return org.springframework.r2dbc.connection.ConnectionFactoryUtils.currentConnectionFactory(connectionFactory);
}
/**
@@ -317,112 +217,4 @@ public abstract class ConnectionFactoryUtils {
return order;
}
/**
* Callback for resource cleanup at the end of a non-native R2DBC transaction.
*/
private static class ConnectionSynchronization implements TransactionSynchronization, Ordered {
private final ConnectionHolder connectionHolder;
private final ConnectionFactory connectionFactory;
private int order;
private boolean holderActive = true;
ConnectionSynchronization(ConnectionHolder connectionHolder, ConnectionFactory connectionFactory) {
this.connectionHolder = connectionHolder;
this.connectionFactory = connectionFactory;
this.order = getConnectionSynchronizationOrder(connectionFactory);
}
@Override
public int getOrder() {
return this.order;
}
@Override
public Mono<Void> suspend() {
if (this.holderActive) {
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(it -> {
it.unbindResource(this.connectionFactory);
if (this.connectionHolder.hasConnection() && !this.connectionHolder.isOpen()) {
// Release Connection on suspend if the application doesn't keep
// a handle to it anymore. We will fetch a fresh Connection if the
// application accesses the ConnectionHolder again after resume,
// assuming that it will participate in the same transaction.
return releaseConnection(this.connectionHolder.getConnection(), this.connectionFactory)
.doOnTerminate(() -> this.connectionHolder.setConnection(null));
}
return Mono.empty();
});
}
return Mono.empty();
}
@Override
public Mono<Void> resume() {
if (this.holderActive) {
return TransactionSynchronizationManager.forCurrentTransaction().doOnNext(it -> {
it.bindResource(this.connectionFactory, this.connectionHolder);
}).then();
}
return Mono.empty();
}
@Override
public Mono<Void> beforeCompletion() {
// Release Connection early if the holder is not open anymore
// (that is, not used by another resource
// that has its own cleanup via transaction synchronization),
// to avoid issues with strict transaction implementations that expect
// the close call before transaction completion.
if (!this.connectionHolder.isOpen()) {
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(it -> {
it.unbindResource(this.connectionFactory);
this.holderActive = false;
if (this.connectionHolder.hasConnection()) {
return releaseConnection(this.connectionHolder.getConnection(), this.connectionFactory);
}
return Mono.empty();
});
}
return Mono.empty();
}
@Override
public Mono<Void> afterCompletion(int status) {
// If we haven't closed the Connection in beforeCompletion,
// close it now. The holder might have been used for other
// cleanup in the meantime, for example by a Hibernate Session.
if (this.holderActive) {
// The thread-bound ConnectionHolder might not be available anymore,
// since afterCompletion might get called from a different thread.
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(it -> {
it.unbindResourceIfPossible(this.connectionFactory);
this.holderActive = false;
if (this.connectionHolder.hasConnection()) {
return releaseConnection(this.connectionHolder.getConnection(), this.connectionFactory)
.doOnTerminate(() -> {
// Reset the ConnectionHolder: It might remain bound to the context.
this.connectionHolder.setConnection(null);
});
}
return Mono.empty();
});
}
this.connectionHolder.reset();
return Mono.empty();
}
}
}

532
src/main/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManager.java
View File

@@ -17,27 +17,9 @@ package org.springframework.data.r2dbc.connectionfactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.IsolationLevel;
import io.r2dbc.spi.R2dbcException;
import io.r2dbc.spi.Result;
import reactor.core.publisher.Mono;
import java.time.Duration;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.support.R2dbcExceptionSubclassTranslator;
import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
import org.springframework.lang.Nullable;
import org.springframework.transaction.CannotCreateTransactionException;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionSystemException;
import org.springframework.transaction.reactive.AbstractReactiveTransactionManager;
import org.springframework.transaction.reactive.GenericReactiveTransaction;
import org.springframework.transaction.reactive.TransactionSynchronizationManager;
import org.springframework.util.Assert;
/**
* {@link org.springframework.transaction.ReactiveTransactionManager} implementation for a single R2DBC
@@ -73,16 +55,12 @@ import org.springframework.util.Assert;
* @see ConnectionFactoryUtils#releaseConnection
* @see TransactionAwareConnectionFactoryProxy
* @see DatabaseClient
* @deprecated since 1.2 in favor of Spring R2DBC. Use {@link org.springframework.r2dbc.connection} instead.
* @deprecated since 1.2 in favor of Spring R2DBC. Use
* {@link org.springframework.r2dbc.connection.R2dbcTransactionManager} instead.
*/
@Deprecated
public class R2dbcTransactionManager extends AbstractReactiveTransactionManager implements InitializingBean {
private ConnectionFactory connectionFactory;
private boolean enforceReadOnly = false;
private R2dbcExceptionTranslator exceptionTranslator = new R2dbcExceptionSubclassTranslator();
public class R2dbcTransactionManager extends org.springframework.r2dbc.connection.R2dbcTransactionManager
implements InitializingBean {
/**
* Create a new @link ConnectionFactoryTransactionManager} instance. A ConnectionFactory has to be set to be able to
@@ -102,506 +80,4 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager
setConnectionFactory(connectionFactory);
afterPropertiesSet();
}
/**
* Set the R2DBC {@link ConnectionFactory} that this instance should manage transactions for.
* <p>
* This will typically be a locally defined {@link ConnectionFactory}, for example an connection pool.
* <p>
* The {@link ConnectionFactory} specified here should be the target {@link ConnectionFactory} to manage transactions
* for, not a TransactionAwareConnectionFactoryProxy. Only data access code may work with
* TransactionAwareConnectionFactoryProxy, while the transaction manager needs to work on the underlying target
* {@link ConnectionFactory}. If there's nevertheless a TransactionAwareConnectionFactoryProxy passed in, it will be
* unwrapped to extract its target {@link ConnectionFactory}.
* <p>
* <b>The {@link ConnectionFactory} passed in here needs to return independent {@link Connection}s.</b> The
* {@link Connection}s may come from a pool (the typical case), but the {@link ConnectionFactory} must not return
* scoped {@link Connection} or the like.
*
* @see TransactionAwareConnectionFactoryProxy
*/
public void setConnectionFactory(@Nullable ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
/**
* Return the R2DBC {@link ConnectionFactory} that this instance manages transactions for.
*/
@Nullable
public ConnectionFactory getConnectionFactory() {
return this.connectionFactory;
}
/**
* Set the exception translator for this instance.
* <p>
* If no custom translator is provided, a default {@link R2dbcExceptionSubclassTranslator} is used which translates
* {@link R2dbcException}'s subclasses into Springs {@link DataAccessException} hierarchy.
*
* @see R2dbcExceptionSubclassTranslator
* @since 1.1
*/
public void setExceptionTranslator(R2dbcExceptionTranslator exceptionTranslator) {
this.exceptionTranslator = exceptionTranslator;
}
/**
* Obtain the {@link ConnectionFactory} for actual use.
*
* @return the {@link ConnectionFactory} (never {@literal null})
* @throws IllegalStateException in case of no ConnectionFactory set
*/
protected ConnectionFactory obtainConnectionFactory() {
ConnectionFactory connectionFactory = getConnectionFactory();
Assert.state(connectionFactory != null, "No ConnectionFactory set");
return connectionFactory;
}
/**
* Specify whether to enforce the read-only nature of a transaction (as indicated by
* {@link TransactionDefinition#isReadOnly()} through an explicit statement on the transactional connection: "SET
* TRANSACTION READ ONLY" as understood by Oracle, MySQL and Postgres.
* <p>
* The exact treatment, including any SQL statement executed on the connection, can be customized through through
* {@link #prepareTransactionalConnection}.
*
* @see #prepareTransactionalConnection
*/
public void setEnforceReadOnly(boolean enforceReadOnly) {
this.enforceReadOnly = enforceReadOnly;
}
/**
* Return whether to enforce the read-only nature of a transaction through an explicit statement on the transactional
* connection.
*
* @see #setEnforceReadOnly
*/
public boolean isEnforceReadOnly() {
return this.enforceReadOnly;
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() {
if (getConnectionFactory() == null) {
throw new IllegalArgumentException("Property 'connectionFactory' is required");
}
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doGetTransaction(org.springframework.transaction.reactive.TransactionSynchronizationManager)
*/
@Override
protected Object doGetTransaction(TransactionSynchronizationManager synchronizationManager)
throws TransactionException {
ConnectionFactoryTransactionObject txObject = new ConnectionFactoryTransactionObject();
ConnectionHolder conHolder = (ConnectionHolder) synchronizationManager.getResource(obtainConnectionFactory());
txObject.setConnectionHolder(conHolder, false);
return txObject;
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#isExistingTransaction(java.lang.Object)
*/
@Override
protected boolean isExistingTransaction(Object transaction) {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doBegin(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object, org.springframework.transaction.TransactionDefinition)
*/
@Override
protected Mono<Void> doBegin(TransactionSynchronizationManager synchronizationManager, Object transaction,
TransactionDefinition definition) throws TransactionException {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
return Mono.defer(() -> {
Mono<Connection> connection = null;
if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
Mono<Connection> newCon = Mono.from(obtainConnectionFactory().create());
connection = newCon.doOnNext(it -> {
if (this.logger.isDebugEnabled()) {
this.logger.debug("Acquired Connection [" + newCon + "] for R2DBC transaction");
}
txObject.setConnectionHolder(new ConnectionHolder(it), true);
});
} else {
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
connection = Mono.just(txObject.getConnectionHolder().getConnection());
}
return connection.flatMap(con -> {
return prepareTransactionalConnection(con, definition, transaction).then(Mono.from(con.beginTransaction()))
.doOnSuccess(v -> {
txObject.getConnectionHolder().setTransactionActive(true);
Duration timeout = determineTimeout(definition);
if (!timeout.isNegative() && !timeout.isZero()) {
txObject.getConnectionHolder().setTimeoutInMillis(timeout.toMillis());
}
// Bind the connection holder to the thread.
if (txObject.isNewConnectionHolder()) {
synchronizationManager.bindResource(obtainConnectionFactory(), txObject.getConnectionHolder());
}
}).thenReturn(con).onErrorResume(e -> {
if (txObject.isNewConnectionHolder()) {
return ConnectionFactoryUtils.releaseConnection(con, obtainConnectionFactory()).doOnTerminate(() -> {
txObject.setConnectionHolder(null, false);
}).then(Mono.error(e));
}
return Mono.error(e);
});
}).onErrorResume(e -> {
CannotCreateTransactionException ex = new CannotCreateTransactionException(
"Could not open R2DBC Connection for transaction",
e instanceof R2dbcException ? potentiallyTranslateException("Open R2DBC Connection", (R2dbcException) e)
: e);
return Mono.error(ex);
});
}).then();
}
/**
* Determine the actual timeout to use for the given definition. Will fall back to this manager's default timeout if
* the transaction definition doesn't specify a non-default value.
*
* @param definition the transaction definition
* @return the actual timeout to use
* @see org.springframework.transaction.TransactionDefinition#getTimeout()
*/
protected Duration determineTimeout(TransactionDefinition definition) {
if (definition.getTimeout() != TransactionDefinition.TIMEOUT_DEFAULT) {
return Duration.ofSeconds(definition.getTimeout());
}
return Duration.ZERO;
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doSuspend(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object)
*/
@Override
protected Mono<Object> doSuspend(TransactionSynchronizationManager synchronizationManager, Object transaction)
throws TransactionException {
return Mono.defer(() -> {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
txObject.setConnectionHolder(null);
return Mono.justOrEmpty(synchronizationManager.unbindResource(obtainConnectionFactory()));
});
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doResume(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object, java.lang.Object)
*/
@Override
protected Mono<Void> doResume(TransactionSynchronizationManager synchronizationManager, Object transaction,
Object suspendedResources) throws TransactionException {
return Mono.defer(() -> {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
txObject.setConnectionHolder(null);
synchronizationManager.bindResource(obtainConnectionFactory(), suspendedResources);
return Mono.empty();
});
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doCommit(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
*/
@Override
protected Mono<Void> doCommit(TransactionSynchronizationManager TransactionSynchronizationManager,
GenericReactiveTransaction status) throws TransactionException {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) status.getTransaction();
Connection connection = txObject.getConnectionHolder().getConnection();
if (status.isDebug()) {
this.logger.debug("Committing R2DBC transaction on Connection [" + connection + "]");
}
return Mono.from(connection.commitTransaction())
.onErrorMap(R2dbcException.class, ex -> translateException("R2DBC commit", ex));
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doRollback(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
*/
@Override
protected Mono<Void> doRollback(TransactionSynchronizationManager TransactionSynchronizationManager,
GenericReactiveTransaction status) throws TransactionException {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) status.getTransaction();
Connection connection = txObject.getConnectionHolder().getConnection();
if (status.isDebug()) {
this.logger.debug("Rolling back R2DBC transaction on Connection [" + connection + "]");
}
return Mono.from(connection.rollbackTransaction())
.onErrorMap(R2dbcException.class, ex -> translateException("R2DBC rollback", ex));
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doSetRollbackOnly(org.springframework.transaction.reactive.TransactionSynchronizationManager, org.springframework.transaction.reactive.GenericReactiveTransaction)
*/
@Override
protected Mono<Void> doSetRollbackOnly(TransactionSynchronizationManager synchronizationManager,
GenericReactiveTransaction status) throws TransactionException {
return Mono.fromRunnable(() -> {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) status.getTransaction();
if (status.isDebug()) {
this.logger
.debug("Setting R2DBC transaction [" + txObject.getConnectionHolder().getConnection() + "] rollback-only");
}
txObject.setRollbackOnly();
});
}
/*
* (non-Javadoc)
* @see org.springframework.transaction.reactive.AbstractReactiveTransactionManager#doCleanupAfterCompletion(org.springframework.transaction.reactive.TransactionSynchronizationManager, java.lang.Object)
*/
@Override
protected Mono<Void> doCleanupAfterCompletion(TransactionSynchronizationManager synchronizationManager,
Object transaction) {
return Mono.defer(() -> {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
// Remove the connection holder from the context, if exposed.
if (txObject.isNewConnectionHolder()) {
synchronizationManager.unbindResource(obtainConnectionFactory());
}
// Reset connection.
Connection con = txObject.getConnectionHolder().getConnection();
Mono<Void> afterCleanup = Mono.empty();
if (txObject.isMustRestoreAutoCommit()) {
afterCleanup = afterCleanup.then(Mono.from(con.setAutoCommit(true)));
}
if (txObject.getPreviousIsolationLevel() != null) {
afterCleanup = afterCleanup
.then(Mono.from(con.setTransactionIsolationLevel(txObject.getPreviousIsolationLevel())));
}
return afterCleanup.then(Mono.defer(() -> {
try {
if (txObject.isNewConnectionHolder()) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("Releasing R2DBC Connection [" + con + "] after transaction");
}
return ConnectionFactoryUtils.releaseConnection(con, obtainConnectionFactory());
}
} finally {
txObject.getConnectionHolder().clear();
}
return Mono.empty();
}));
});
}
/**
* Prepare the transactional {@link Connection} right after transaction begin.
* <p>
* The default implementation executes a "SET TRANSACTION READ ONLY" statement if the {@link #setEnforceReadOnly
* "enforceReadOnly"} flag is set to {@literal true} and the transaction definition indicates a read-only transaction.
* <p>
* The "SET TRANSACTION READ ONLY" is understood by Oracle, MySQL and Postgres and may work with other databases as
* well. If you'd like to adapt this treatment, override this method accordingly.
*
* @param con the transactional R2DBC Connection
* @param definition the current transaction definition
* @param transaction the transaction object
* @see #setEnforceReadOnly
*/
protected Mono<Void> prepareTransactionalConnection(Connection con, TransactionDefinition definition,
Object transaction) {
ConnectionFactoryTransactionObject txObject = (ConnectionFactoryTransactionObject) transaction;
Mono<Void> prepare = Mono.empty();
if (isEnforceReadOnly() && definition.isReadOnly()) {
prepare = Mono.from(con.createStatement("SET TRANSACTION READ ONLY").execute()) //
.flatMapMany(Result::getRowsUpdated) //
.then();
}
// Apply specific isolation level, if any.
IsolationLevel isolationLevelToUse = resolveIsolationLevel(definition.getIsolationLevel());
if (isolationLevelToUse != null && definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
if (this.logger.isDebugEnabled()) {
this.logger
.debug("Changing isolation level of R2DBC Connection [" + con + "] to " + isolationLevelToUse.asSql());
}
IsolationLevel currentIsolation = con.getTransactionIsolationLevel();
if (!currentIsolation.asSql().equalsIgnoreCase(isolationLevelToUse.asSql())) {
txObject.setPreviousIsolationLevel(currentIsolation);
prepare = prepare.then(Mono.from(con.setTransactionIsolationLevel(isolationLevelToUse)));
}
}
// Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
// so we don't want to do it unnecessarily (for example if we've explicitly
// configured the connection pool to set it already).
if (con.isAutoCommit()) {
txObject.setMustRestoreAutoCommit(true);
if (this.logger.isDebugEnabled()) {
this.logger.debug("Switching R2DBC Connection [" + con + "] to manual commit");
}
prepare = prepare.then(Mono.from(con.setAutoCommit(false)));
}
return prepare;
}
/**
* Resolve the {@link TransactionDefinition#getIsolationLevel() isolation level constant} to a R2DBC
* {@link IsolationLevel}. If you'd like to extend isolation level translation for vendor-specific
* {@link IsolationLevel}s, override this method accordingly.
*
* @param isolationLevel the isolation level to translate.
* @return the resolved isolation level. Can be {@literal null} if not resolvable or the isolation level should remain
* {@link TransactionDefinition#ISOLATION_DEFAULT default}.
* @see TransactionDefinition#getIsolationLevel()
*/
@Nullable
protected IsolationLevel resolveIsolationLevel(int isolationLevel) {
switch (isolationLevel) {
case TransactionDefinition.ISOLATION_READ_COMMITTED:
return IsolationLevel.READ_COMMITTED;
case TransactionDefinition.ISOLATION_READ_UNCOMMITTED:
return IsolationLevel.READ_UNCOMMITTED;
case TransactionDefinition.ISOLATION_REPEATABLE_READ:
return IsolationLevel.REPEATABLE_READ;
case TransactionDefinition.ISOLATION_SERIALIZABLE:
return IsolationLevel.SERIALIZABLE;
}
return null;
}
/**
* Translate the given R2DBC commit/rollback exception to a common Spring exception to propagate from the
* {@link #commit}/{@link #rollback} call.
* <p>
* The default implementation throws a {@link TransactionSystemException}. Subclasses may specifically identify
* concurrency failures etc.
*
* @param task the task description (commit or rollback).
* @param ex the SQLException thrown from commit/rollback.
* @return the translated exception to throw, either a {@link org.springframework.dao.DataAccessException} or a
* {@link org.springframework.transaction.TransactionException}
* @since 1.1
*/
protected RuntimeException translateException(String task, R2dbcException ex) {
Exception translated = potentiallyTranslateException(task, ex);
return new TransactionSystemException(task + " failed", translated);
}
private Exception potentiallyTranslateException(String task, R2dbcException ex) {
DataAccessException translated = exceptionTranslator.translate(task, null, ex);
return translated != null ? translated : ex;
}
/**
* ConnectionFactory transaction object, representing a ConnectionHolder. Used as transaction object by
* ConnectionFactoryTransactionManager.
*/
private static class ConnectionFactoryTransactionObject {
private @Nullable ConnectionHolder connectionHolder;
private @Nullable IsolationLevel previousIsolationLevel;
private boolean newConnectionHolder;
private boolean mustRestoreAutoCommit;
void setConnectionHolder(@Nullable ConnectionHolder connectionHolder, boolean newConnectionHolder) {
setConnectionHolder(connectionHolder);
this.newConnectionHolder = newConnectionHolder;
}
boolean isNewConnectionHolder() {
return this.newConnectionHolder;
}
void setRollbackOnly() {
getConnectionHolder().setRollbackOnly();
}
public void setConnectionHolder(@Nullable ConnectionHolder connectionHolder) {
this.connectionHolder = connectionHolder;
}
public ConnectionHolder getConnectionHolder() {
Assert.state(this.connectionHolder != null, "No ConnectionHolder available");
return this.connectionHolder;
}
public boolean hasConnectionHolder() {
return (this.connectionHolder != null);
}
public void setPreviousIsolationLevel(@Nullable IsolationLevel previousIsolationLevel) {
this.previousIsolationLevel = previousIsolationLevel;
}
@Nullable
public IsolationLevel getPreviousIsolationLevel() {
return this.previousIsolationLevel;
}
public void setMustRestoreAutoCommit(boolean mustRestoreAutoCommit) {
this.mustRestoreAutoCommit = mustRestoreAutoCommit;
}
public boolean isMustRestoreAutoCommit() {
return this.mustRestoreAutoCommit;
}
}
}

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

@@ -32,7 +32,7 @@ import org.springframework.data.convert.CustomConversions;
import org.springframework.data.mapping.MappingException;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.mapping.PreferredConstructor.Parameter;
import org.springframework.data.mapping.PreferredConstructor;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
import org.springframework.data.mapping.model.ParameterValueProvider;
@@ -48,6 +48,7 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing;
import org.springframework.data.util.ClassTypeInformation;
import org.springframework.data.util.TypeInformation;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
@@ -357,7 +358,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
}
private void writeSimpleInternal(OutboundRow sink, Object value, RelationalPersistentProperty property) {
sink.put(property.getColumnName(), SettableValue.from(getPotentiallyConvertedSimpleWrite(value)));
sink.put(property.getColumnName(), Parameter.from(getPotentiallyConvertedSimpleWrite(value)));
}
private void writePropertyInternal(OutboundRow sink, Object value, RelationalPersistentProperty property) {
@@ -374,7 +375,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
}
List<Object> collectionInternal = createCollection(asCollection(value), property);
sink.put(property.getColumnName(), SettableValue.from(collectionInternal));
sink.put(property.getColumnName(), Parameter.from(collectionInternal));
return;
}
@@ -431,7 +432,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
private void writeNullInternal(OutboundRow sink, RelationalPersistentProperty property) {
sink.put(property.getColumnName(), SettableValue.empty(getPotentiallyConvertedSimpleNullType(property.getType())));
sink.put(property.getColumnName(), Parameter.empty(getPotentiallyConvertedSimpleNullType(property.getType())));
}
private Class<?> getPotentiallyConvertedSimpleNullType(Class<?> type) {
@@ -640,7 +641,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
*/
@Override
@Nullable
public <T> T getParameterValue(Parameter<T, RelationalPersistentProperty> parameter) {
public <T> T getParameterValue(PreferredConstructor.Parameter<T, RelationalPersistentProperty> parameter) {
RelationalPersistentProperty property = this.entity.getRequiredPersistentProperty(parameter.getName());

8
src/main/java/org/springframework/data/r2dbc/core/DatabaseClient.java
View File

@@ -54,6 +54,14 @@ import org.springframework.util.Assert;
@Deprecated
public interface DatabaseClient {
/**
* Return the {@link ConnectionFactory} that this client uses.
*
* @return the connection factory.
* @since 1.2
*/
ConnectionFactory getConnectionFactory();
/**
* Specify a static {@code sql} string to execute. Contract for specifying a SQL call along with options leading to
* the exchange. The SQL string can contain either native parameter bind markers or named parameters (e.g.

12
src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java
View File

@@ -63,6 +63,7 @@ import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.query.CriteriaDefinition;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.r2dbc.core.PreparedOperation;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -73,7 +74,9 @@ import org.springframework.util.StringUtils;
* @author Mark Paluch
* @author Mingyuan Wu
* @author Bogdan Ilchyshyn
* @deprecated since 1.2.
*/
@Deprecated
class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private final Log logger = LogFactory.getLog(getClass());
@@ -105,6 +108,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
this.builder = builder;
}
@Override
public ConnectionFactory getConnectionFactory() {
return this.connector;
}
@Override
public Builder mutate() {
return this.builder;
@@ -1171,7 +1179,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
StatementMapper.InsertSpec insert = mapper.createInsert(this.table);
for (SqlIdentifier column : outboundRow.keySet()) {
SettableValue settableValue = outboundRow.get(column);
Parameter settableValue = outboundRow.get(column);
if (settableValue.hasValue()) {
insert = insert.withColumn(column, settableValue);
}
@@ -1348,7 +1356,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
private UpdatedRowsFetchSpec exchange(SqlIdentifier table) {
StatementMapper mapper = dataAccessStrategy.getStatementMapper();
Map<SqlIdentifier, SettableValue> columns = dataAccessStrategy.getOutboundRow(this.objectToUpdate);
Map<SqlIdentifier, Parameter> columns = dataAccessStrategy.getOutboundRow(this.objectToUpdate);
List<SqlIdentifier> ids = dataAccessStrategy.getIdentifierColumns(this.typeToUpdate);
if (ids.isEmpty()) {

16
src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java
View File

@@ -207,10 +207,10 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
for (RelationalPersistentProperty property : entity) {
SettableValue value = row.get(property.getColumnName());
Parameter value = row.get(property.getColumnName());
if (value != null && shouldConvertArrayValue(property, value)) {
SettableValue writeValue = getArrayValue(value, property);
Parameter writeValue = getArrayValue(value, property);
row.put(property.getColumnName(), writeValue);
}
}
@@ -218,7 +218,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return row;
}
private boolean shouldConvertArrayValue(RelationalPersistentProperty property, SettableValue value) {
private boolean shouldConvertArrayValue(RelationalPersistentProperty property, Parameter value) {
if (!property.isCollectionLike()) {
return false;
@@ -235,7 +235,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
return false;
}
private SettableValue getArrayValue(SettableValue value, RelationalPersistentProperty property) {
private Parameter getArrayValue(Parameter value, RelationalPersistentProperty property) {
if (value.getType().equals(byte[].class)) {
return value;
@@ -250,8 +250,8 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
}
Class<?> actualType = null;
if (value instanceof Collection) {
actualType = CollectionUtils.findCommonElementType((Collection<?>) value);
if (value.getValue() instanceof Collection) {
actualType = CollectionUtils.findCommonElementType((Collection<?>) value.getValue());
} else if (value.getClass().isArray()) {
actualType = value.getClass().getComponentType();
}
@@ -265,10 +265,10 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
Class<?> targetType = arrayColumns.getArrayType(actualType);
int depth = actualType.isArray() ? ArrayUtils.getDimensionDepth(actualType) : 1;
Class<?> targetArrayType = ArrayUtils.getArrayClass(targetType, depth);
return SettableValue.empty(targetArrayType);
return Parameter.empty(targetArrayType);
}
return SettableValue.fromOrEmpty(this.converter.getArrayValue(arrayColumns, property, value.getValue()),
return Parameter.fromOrEmpty(this.converter.getArrayValue(arrayColumns, property, value.getValue()),
actualType);
}

1
src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityOperations.java
View File

@@ -22,6 +22,7 @@ import org.springframework.dao.DataAccessException;
import org.springframework.dao.TransientDataAccessResourceException;
import org.springframework.data.relational.core.query.Query;
import org.springframework.data.relational.core.query.Update;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Interface specifying a basic set of reactive R2DBC operations using entities. Implemented by

203
src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java
View File

@@ -15,6 +15,8 @@
*/
package org.springframework.data.r2dbc.core;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import reactor.core.publisher.Flux;
@@ -27,6 +29,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
@@ -47,8 +50,8 @@ import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.projection.ProjectionInformation;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.r2dbc.dialect.R2dbcDialect;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.r2dbc.mapping.event.AfterConvertCallback;
import org.springframework.data.r2dbc.mapping.event.AfterSaveCallback;
import org.springframework.data.r2dbc.mapping.event.BeforeConvertCallback;
@@ -64,8 +67,13 @@ import org.springframework.data.relational.core.sql.Functions;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.data.relational.core.sql.Table;
import org.springframework.data.util.ProxyUtils;
import org.springframework.r2dbc.core.PreparedOperation;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.FetchSpec;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.r2dbc.core.PreparedOperation;
import org.springframework.r2dbc.core.RowsFetchSpec;
import org.springframework.r2dbc.core.StatementFilterFunction;
import org.springframework.util.Assert;
/**
@@ -96,8 +104,21 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient}.
*
* @param databaseClient must not be {@literal null}.
* @param dialect the dialect to use, must not be {@literal null}.
* @since 1.2
*/
public R2dbcEntityTemplate(DatabaseClient databaseClient) {
public R2dbcEntityTemplate(org.springframework.r2dbc.core.DatabaseClient databaseClient, R2dbcDialect dialect) {
this(databaseClient, new DefaultReactiveDataAccessStrategy(dialect));
}
/**
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient}.
*
* @param databaseClient must not be {@literal null}.
* @deprecated since 1.2, use {@link #R2dbcEntityTemplate(DatabaseClient, R2dbcDialect)} instead.
*/
@Deprecated
public R2dbcEntityTemplate(org.springframework.data.r2dbc.core.DatabaseClient databaseClient) {
this(databaseClient, getDataAccessStrategy(databaseClient));
}
@@ -105,8 +126,10 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient} and {@link ReactiveDataAccessStrategy}.
*
* @param databaseClient must not be {@literal null}.
* @since 1.2
*/
public R2dbcEntityTemplate(DatabaseClient databaseClient, ReactiveDataAccessStrategy strategy) {
public R2dbcEntityTemplate(org.springframework.r2dbc.core.DatabaseClient databaseClient,
ReactiveDataAccessStrategy strategy) {
Assert.notNull(databaseClient, "DatabaseClient must not be null");
Assert.notNull(strategy, "ReactiveDataAccessStrategy must not be null");
@@ -117,6 +140,18 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
this.projectionFactory = new SpelAwareProxyProjectionFactory();
}
/**
* Create a new {@link R2dbcEntityTemplate} given {@link DatabaseClient} and {@link ReactiveDataAccessStrategy}.
*
* @param databaseClient must not be {@literal null}.
* @deprecated since 1.2, use {@link #R2dbcEntityTemplate(DatabaseClient, ReactiveDataAccessStrategy)} instead.
*/
@Deprecated
public R2dbcEntityTemplate(org.springframework.data.r2dbc.core.DatabaseClient databaseClient,
ReactiveDataAccessStrategy strategy) {
this(new DatabaseClientAdapter(databaseClient), strategy);
}
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.core.R2dbcEntityOperations#getDatabaseClient()
@@ -251,7 +286,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
return this.databaseClient.execute(operation) //
return this.databaseClient.sql(operation) //
.map((r, md) -> r.get(0, Long.class)) //
.first() //
.defaultIfEmpty(0L);
@@ -290,7 +325,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
return this.databaseClient.execute(operation) //
return this.databaseClient.sql(operation) //
.map((r, md) -> r) //
.first() //
.hasElement();
@@ -361,7 +396,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
rowMapper = dataAccessStrategy.getRowMapper(returnType);
}
return this.databaseClient.execute(operation).map(rowMapper);
return this.databaseClient.sql(operation).map(rowMapper);
}
/*
@@ -400,7 +435,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
}
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
return this.databaseClient.execute(operation).fetch().rowsUpdated();
return this.databaseClient.sql(operation).fetch().rowsUpdated();
}
/*
@@ -429,7 +464,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
}
PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
return this.databaseClient.execute(operation).fetch().rowsUpdated().defaultIfEmpty(0);
return this.databaseClient.sql(operation).fetch().rowsUpdated().defaultIfEmpty(0);
}
// -------------------------------------------------------------------------
@@ -454,14 +489,13 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
T entityWithVersion = setVersionIfNecessary(persistentEntity, entity);
return maybeCallBeforeConvert(entityWithVersion, tableName)
.flatMap(beforeConvert -> {
return maybeCallBeforeConvert(entityWithVersion, tableName).flatMap(beforeConvert -> {
OutboundRow outboundRow = dataAccessStrategy.getOutboundRow(beforeConvert);
OutboundRow outboundRow = dataAccessStrategy.getOutboundRow(beforeConvert);
return maybeCallBeforeSave(beforeConvert, outboundRow, tableName) //
.flatMap(entityToSave -> doInsert(entityToSave, tableName, outboundRow));
});
return maybeCallBeforeSave(beforeConvert, outboundRow, tableName) //
.flatMap(entityToSave -> doInsert(entityToSave, tableName, outboundRow));
});
}
private <T> Mono<T> doInsert(T entity, SqlIdentifier tableName, OutboundRow outboundRow) {
@@ -470,7 +504,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
StatementMapper.InsertSpec insert = mapper.createInsert(tableName);
for (SqlIdentifier column : outboundRow.keySet()) {
SettableValue settableValue = outboundRow.get(column);
Parameter settableValue = outboundRow.get(column);
if (settableValue.hasValue()) {
insert = insert.withColumn(column, settableValue);
}
@@ -478,7 +512,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
PreparedOperation<?> operation = mapper.getMappedObject(insert);
return this.databaseClient.execute(operation) //
return this.databaseClient.sql(operation) //
.filter(statement -> statement.returnGeneratedValues())
.map(this.dataAccessStrategy.getConverter().populateIdIfNecessary(entity)) //
.first() //
@@ -540,7 +574,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
.flatMap(entityToSave -> {
SqlIdentifier idColumn = persistentEntity.getRequiredIdProperty().getColumnName();
SettableValue id = outboundRow.remove(idColumn);
Parameter id = outboundRow.remove(idColumn);
Criteria criteria = Criteria.where(dataAccessStrategy.toSql(idColumn)).is(id);
if (matchingVersionCriteria != null) {
@@ -563,7 +597,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
PreparedOperation<?> operation = mapper.getMappedObject(updateSpec);
return this.databaseClient.execute(operation) //
return this.databaseClient.sql(operation) //
.fetch() //
.rowsUpdated() //
.handle((rowsUpdated, sink) -> {
@@ -720,7 +754,8 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
return query.getColumns().stream().map(table::column).collect(Collectors.toList());
}
private static ReactiveDataAccessStrategy getDataAccessStrategy(DatabaseClient databaseClient) {
private static ReactiveDataAccessStrategy getDataAccessStrategy(
org.springframework.data.r2dbc.core.DatabaseClient databaseClient) {
Assert.notNull(databaseClient, "DatabaseClient must not be null");
@@ -733,4 +768,132 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
throw new IllegalStateException("Cannot obtain ReactiveDataAccessStrategy");
}
/**
* Adapter to adapt our deprecated {@link org.springframework.data.r2dbc.core.DatabaseClient} into Spring R2DBC
* {@link DatabaseClient}.
*/
private static class DatabaseClientAdapter implements DatabaseClient {
private final org.springframework.data.r2dbc.core.DatabaseClient delegate;
private DatabaseClientAdapter(org.springframework.data.r2dbc.core.DatabaseClient delegate) {
Assert.notNull(delegate, "DatabaseClient must not be null");
this.delegate = delegate;
}
@Override
public ConnectionFactory getConnectionFactory() {
return delegate.getConnectionFactory();
}
@Override
public GenericExecuteSpec sql(String sql) {
return new GenericExecuteSpecAdapter(delegate.execute(sql));
}
@Override
public GenericExecuteSpec sql(Supplier<String> sqlSupplier) {
return new GenericExecuteSpecAdapter(delegate.execute(sqlSupplier));
}
@Override
public <T> Mono<T> inConnection(Function<Connection, Mono<T>> action) throws DataAccessException {
return ((ConnectionAccessor) delegate).inConnection(action);
}
@Override
public <T> Flux<T> inConnectionMany(Function<Connection, Flux<T>> action) throws DataAccessException {
return ((ConnectionAccessor) delegate).inConnectionMany(action);
}
static class GenericExecuteSpecAdapter implements GenericExecuteSpec {
private final org.springframework.data.r2dbc.core.DatabaseClient.GenericExecuteSpec delegate;
public GenericExecuteSpecAdapter(org.springframework.data.r2dbc.core.DatabaseClient.GenericExecuteSpec delegate) {
this.delegate = delegate;
}
@Override
public GenericExecuteSpec bind(int index, Object value) {
return new GenericExecuteSpecAdapter(delegate.bind(index, value));
}
@Override
public GenericExecuteSpec bindNull(int index, Class<?> type) {
return new GenericExecuteSpecAdapter(delegate.bindNull(index, type));
}
@Override
public GenericExecuteSpec bind(String name, Object value) {
return new GenericExecuteSpecAdapter(delegate.bind(name, value));
}
@Override
public GenericExecuteSpec bindNull(String name, Class<?> type) {
return new GenericExecuteSpecAdapter(delegate.bindNull(name, type));
}
@Override
public GenericExecuteSpec filter(StatementFilterFunction filter) {
return new GenericExecuteSpecAdapter(delegate.filter(filter::filter));
}
@Override
public <R> RowsFetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
return new RowFetchSpecAdapter<>(delegate.map(mappingFunction));
}
@Override
public FetchSpec<Map<String, Object>> fetch() {
return new FetchSpecAdapter<>(delegate.fetch());
}
@Override
public Mono<Void> then() {
return delegate.then();
}
}
private static class RowFetchSpecAdapter<T> implements RowsFetchSpec<T> {
private final org.springframework.data.r2dbc.core.RowsFetchSpec<T> delegate;
RowFetchSpecAdapter(org.springframework.data.r2dbc.core.RowsFetchSpec<T> delegate) {
this.delegate = delegate;
}
@Override
public Mono<T> one() {
return delegate.one();
}
@Override
public Mono<T> first() {
return delegate.first();
}
@Override
public Flux<T> all() {
return delegate.all();
}
}
private static class FetchSpecAdapter<T> extends RowFetchSpecAdapter<T> implements FetchSpec<T> {
private final org.springframework.data.r2dbc.core.FetchSpec<T> delegate;
FetchSpecAdapter(org.springframework.data.r2dbc.core.FetchSpec<T> delegate) {
super(delegate);
this.delegate = delegate;
}
@Override
public Mono<Integer> rowsUpdated() {
return delegate.rowsUpdated();
}
}
}
}

5
src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java
View File

@@ -53,7 +53,7 @@ public interface ReactiveDataAccessStrategy {
List<SqlIdentifier> getIdentifierColumns(Class<?> entityType);
/**
* Returns a {@link OutboundRow} that maps column names to a {@link SettableValue} value.
* Returns a {@link OutboundRow} that maps column names to a {@link Parameter} value.
*
* @param object must not be {@literal null}.
* @return
@@ -101,7 +101,10 @@ public interface ReactiveDataAccessStrategy {
* @param parameterProvider indexed parameter bindings.
* @return the {@link PreparedOperation} encapsulating expanded SQL and namedBindings.
* @throws org.springframework.dao.InvalidDataAccessApiUsageException if a named parameter value cannot be resolved.
* @deprecated since 1.2. {@link org.springframework.r2dbc.core.DatabaseClient} encapsulates named parameter handling
* entirely.
*/
@Deprecated
PreparedOperation<?> processNamedParameters(String query, NamedParameterProvider parameterProvider);
/**

1
src/main/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationSupport.java
View File

@@ -21,6 +21,7 @@ import reactor.core.publisher.Mono;
import org.springframework.data.relational.core.query.Query;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.RowsFetchSpec;
import org.springframework.util.Assert;
/**

37
src/main/java/org/springframework/data/r2dbc/core/StatementMapper.java
View File

@@ -36,6 +36,7 @@ import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.data.relational.core.sql.Table;
import org.springframework.data.relational.core.sql.render.RenderContext;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.r2dbc.core.PreparedOperation;
/**
@@ -425,9 +426,9 @@ public interface StatementMapper {
class InsertSpec {
private final SqlIdentifier table;
private final Map<SqlIdentifier, SettableValue> assignments;
private final Map<SqlIdentifier, Parameter> assignments;
protected InsertSpec(SqlIdentifier table, Map<SqlIdentifier, SettableValue> assignments) {
protected InsertSpec(SqlIdentifier table, Map<SqlIdentifier, Parameter> assignments) {
this.table = table;
this.assignments = assignments;
}
@@ -459,21 +460,49 @@ public interface StatementMapper {
* @param column
* @param value
* @return the {@link InsertSpec}.
* @deprecated since 1.2, use {@link #withColumn(String, Parameter)} instead.
*/
@Deprecated
public InsertSpec withColumn(String column, SettableValue value) {
return withColumn(SqlIdentifier.unquoted(column), value);
}
/**
* Associate a column with a {@link Parameter} and create a new {@link InsertSpec}.
*
* @param column
* @param value
* @return the {@link InsertSpec}.
* @since 1.2
*/
public InsertSpec withColumn(String column, Parameter value) {
return withColumn(SqlIdentifier.unquoted(column), value);
}
/**
* Associate a column with a {@link SettableValue} and create a new {@link InsertSpec}.
*
* @param column
* @param value
* @return the {@link InsertSpec}.
* @deprecated since 1.2, use {@link #withColumn(SqlIdentifier, Parameter)} instead.
*/
@Deprecated
public InsertSpec withColumn(SqlIdentifier column, SettableValue value) {
return withColumn(column, value.toParameter());
}
Map<SqlIdentifier, SettableValue> values = new LinkedHashMap<>(this.assignments);
/**
* Associate a column with a {@link Parameter} and create a new {@link InsertSpec}.
*
* @param column
* @param value
* @return the {@link InsertSpec}.
* @since 1.2
*/
public InsertSpec withColumn(SqlIdentifier column, Parameter value) {
Map<SqlIdentifier, Parameter> values = new LinkedHashMap<>(this.assignments);
values.put(column, value);
return new InsertSpec(this.table, values);
@@ -483,7 +512,7 @@ public interface StatementMapper {
return this.table;
}
public Map<SqlIdentifier, SettableValue> getAssignments() {
public Map<SqlIdentifier, Parameter> getAssignments() {
return Collections.unmodifiableMap(this.assignments);
}
}

35
src/main/java/org/springframework/data/r2dbc/mapping/OutboundRow.java
View File

@@ -24,6 +24,7 @@ import java.util.Set;
import java.util.function.BiConsumer;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.util.Assert;
/**
@@ -33,11 +34,11 @@ import org.springframework.util.Assert;
*
* @author Mark Paluch
* @see SqlIdentifier
* @see SettableValue
* @see Parameter
*/
public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
public class OutboundRow implements Map<SqlIdentifier, Parameter> {
private final Map<SqlIdentifier, SettableValue> rowAsMap;
private final Map<SqlIdentifier, Parameter> rowAsMap;
/**
* Creates an empty {@link OutboundRow} instance.
@@ -51,13 +52,13 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
*
* @param map the map used to initialize the {@link OutboundRow}.
*/
public OutboundRow(Map<String, SettableValue> map) {
public OutboundRow(Map<String, Parameter> map) {
Assert.notNull(map, "Map must not be null");
this.rowAsMap = new LinkedHashMap<>(map.size());
map.forEach((s, settableValue) -> this.rowAsMap.put(SqlIdentifier.unquoted(s), settableValue));
map.forEach((s, Parameter) -> this.rowAsMap.put(SqlIdentifier.unquoted(s), Parameter));
}
/**
@@ -67,7 +68,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @param value value.
* @see SqlIdentifier#unquoted(String)
*/
public OutboundRow(String key, SettableValue value) {
public OutboundRow(String key, Parameter value) {
this(SqlIdentifier.unquoted(key), value);
}
@@ -78,7 +79,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @param value value.
* @since 1.1
*/
public OutboundRow(SqlIdentifier key, SettableValue value) {
public OutboundRow(SqlIdentifier key, Parameter value) {
this.rowAsMap = new LinkedHashMap<>();
this.rowAsMap.put(key, value);
}
@@ -96,7 +97,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @return this
* @see SqlIdentifier#unquoted(String)
*/
public OutboundRow append(String key, SettableValue value) {
public OutboundRow append(String key, Parameter value) {
return append(SqlIdentifier.unquoted(key), value);
}
@@ -113,7 +114,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @return this
* @since 1.1
*/
public OutboundRow append(SqlIdentifier key, SettableValue value) {
public OutboundRow append(SqlIdentifier key, Parameter value) {
this.rowAsMap.put(key, value);
return this;
}
@@ -159,7 +160,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#get(java.lang.Object)
*/
@Override
public SettableValue get(Object key) {
public Parameter get(Object key) {
return this.rowAsMap.get(convertKeyIfNecessary(key));
}
@@ -167,7 +168,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* (non-Javadoc)
* @see java.util.Map#put(java.lang.Object, java.lang.Object)
*/
public SettableValue put(String key, SettableValue value) {
public Parameter put(String key, Parameter value) {
return put(SqlIdentifier.unquoted(key), value);
}
@@ -176,7 +177,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#put(java.lang.Object, java.lang.Object)
*/
@Override
public SettableValue put(SqlIdentifier key, SettableValue value) {
public Parameter put(SqlIdentifier key, Parameter value) {
return this.rowAsMap.put(key, value);
}
@@ -185,7 +186,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#remove(java.lang.Object)
*/
@Override
public SettableValue remove(Object key) {
public Parameter remove(Object key) {
return this.rowAsMap.remove(key);
}
@@ -194,7 +195,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#putAll(java.util.Map)
*/
@Override
public void putAll(Map<? extends SqlIdentifier, ? extends SettableValue> m) {
public void putAll(Map<? extends SqlIdentifier, ? extends Parameter> m) {
this.rowAsMap.putAll(m);
}
@@ -221,7 +222,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#values()
*/
@Override
public Collection<SettableValue> values() {
public Collection<Parameter> values() {
return this.rowAsMap.values();
}
@@ -230,7 +231,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
* @see java.util.Map#entrySet()
*/
@Override
public Set<Entry<SqlIdentifier, SettableValue>> entrySet() {
public Set<Entry<SqlIdentifier, Parameter>> entrySet() {
return this.rowAsMap.entrySet();
}
@@ -273,7 +274,7 @@ public class OutboundRow implements Map<SqlIdentifier, SettableValue> {
}
@Override
public void forEach(BiConsumer<? super SqlIdentifier, ? super SettableValue> action) {
public void forEach(BiConsumer<? super SqlIdentifier, ? super Parameter> action) {
this.rowAsMap.forEach(action);
}

8
src/main/java/org/springframework/data/r2dbc/mapping/SettableValue.java
View File

@@ -128,4 +128,12 @@ public class SettableValue {
public String toString() {
return this.parameter.toString();
}
/**
* @return the {@link Parameter} representing this settable value.
* @since 1.2
*/
public Parameter toParameter() {
return isEmpty() ? Parameter.empty(getType()) : Parameter.fromOrEmpty(getValue(), getType());
}
}

51
src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java
View File

@@ -23,10 +23,8 @@ import reactor.core.publisher.Mono;
import org.reactivestreams.Publisher;
import org.springframework.core.KotlinDetector;
import org.springframework.data.mapping.model.EntityInstantiators;
import org.springframework.data.r2dbc.convert.EntityRowMapper;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DatabaseClient.GenericExecuteSpec;
import org.springframework.data.r2dbc.core.FetchSpec;
import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingConverter;
import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingExecution;
import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -36,6 +34,9 @@ import org.springframework.data.repository.query.ParameterAccessor;
import org.springframework.data.repository.query.RepositoryQuery;
import org.springframework.data.repository.query.ResultProcessor;
import org.springframework.data.repository.query.ReturnedType;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.FetchSpec;
import org.springframework.r2dbc.core.RowsFetchSpec;
import org.springframework.data.util.ReflectionUtils;
import org.springframework.util.Assert;
@@ -107,8 +108,15 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
BindableQuery query = createQuery(parameterAccessor);
ResultProcessor processor = method.getResultProcessor().withDynamicProjection(parameterAccessor);
GenericExecuteSpec boundQuery = query.bind(databaseClient.execute(query));
FetchSpec<?> fetchSpec = boundQuery.as(resolveResultType(processor)).fetch();
DatabaseClient.GenericExecuteSpec boundQuery = query.bind(databaseClient.sql(query));
FetchSpec<?> fetchSpec;
if (requiresMapping()) {
EntityRowMapper<?> rowMapper = new EntityRowMapper<>(resolveResultType(processor), converter);
fetchSpec = new FetchSpecAdapter<>(boundQuery.map(rowMapper));
} else {
fetchSpec = boundQuery.fetch();
}
SqlIdentifier tableName = method.getEntityInformation().getTableName();
@@ -118,6 +126,10 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
return execution.execute(fetchSpec, processor.getReturnedType().getDomainType(), tableName);
}
private boolean requiresMapping() {
return !isModifyingQuery();
}
private Class<?> resolveResultType(ResultProcessor resultProcessor) {
ReturnedType returnedType = resultProcessor.getReturnedType();
@@ -169,4 +181,33 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
* @return the {@link BindableQuery}.
*/
protected abstract BindableQuery createQuery(RelationalParameterAccessor accessor);
private static class FetchSpecAdapter<T> implements FetchSpec<T> {
private final RowsFetchSpec<T> delegate;
private FetchSpecAdapter(RowsFetchSpec<T> delegate) {
this.delegate = delegate;
}
@Override
public Mono<T> one() {
return delegate.one();
}
@Override
public Mono<T> first() {
return delegate.first();
}
@Override
public Flux<T> all() {
return delegate.all();
}
@Override
public Mono<Integer> rowsUpdated() {
throw new UnsupportedOperationException("Not supported after applying a row mapper");
}
}
}

8
src/main/java/org/springframework/data/r2dbc/repository/query/BindableQuery.java
View File

@@ -17,20 +17,20 @@ package org.springframework.data.r2dbc.repository.query;
import java.util.function.Supplier;
import org.springframework.data.r2dbc.core.DatabaseClient.BindSpec;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Interface declaring a query that supplies SQL and can bind parameters to a {@link BindSpec}.
* Interface declaring a query that supplies SQL and can bind parameters to a {@link DatabaseClient.GenericExecuteSpec}.
*
* @author Mark Paluch
*/
public interface BindableQuery extends Supplier<String> {
/**
* Bind parameters to the {@link BindSpec query}.
* Bind parameters to the {@link DatabaseClient.GenericExecuteSpec query}.
*
* @param bindSpec must not be {@literal null}.
* @return the bound query object.
*/
<T extends BindSpec<T>> T bind(T bindSpec);
DatabaseClient.GenericExecuteSpec bind(DatabaseClient.GenericExecuteSpec bindSpec);
}

28
src/main/java/org/springframework/data/r2dbc/repository/query/ExpressionEvaluatingParameterBinder.java
View File

@@ -22,8 +22,6 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
import org.springframework.data.repository.query.Parameter;
import org.springframework.data.repository.query.Parameters;
@@ -31,6 +29,7 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.util.Assert;
/**
@@ -76,25 +75,27 @@ class ExpressionEvaluatingParameterBinder {
* @param bindSpec must not be {@literal null}.
* @param parameterAccessor must not be {@literal null}.
*/
public <T extends DatabaseClient.BindSpec<T>> T bind(T bindSpec, RelationalParameterAccessor parameterAccessor) {
public DatabaseClient.GenericExecuteSpec bind(DatabaseClient.GenericExecuteSpec bindSpec,
RelationalParameterAccessor parameterAccessor) {
Object[] values = parameterAccessor.getValues();
Parameters<?, ?> bindableParameters = parameterAccessor.getBindableParameters();
T bindSpecToUse = bindExpressions(bindSpec, values, bindableParameters);
DatabaseClient.GenericExecuteSpec bindSpecToUse = bindExpressions(bindSpec, values, bindableParameters);
bindSpecToUse = bindParameters(bindSpecToUse, parameterAccessor.hasBindableNullValue(), values, bindableParameters);
return bindSpecToUse;
}
private <T extends DatabaseClient.BindSpec<T>> T bindExpressions(T bindSpec, Object[] values,
private DatabaseClient.GenericExecuteSpec bindExpressions(DatabaseClient.GenericExecuteSpec bindSpec, Object[] values,
Parameters<?, ?> bindableParameters) {
T bindSpecToUse = bindSpec;
DatabaseClient.GenericExecuteSpec bindSpecToUse = bindSpec;
for (ParameterBinding binding : expressionQuery.getBindings()) {
SettableValue valueForBinding = getParameterValueForBinding(bindableParameters, values, binding);
org.springframework.r2dbc.core.Parameter valueForBinding = getParameterValueForBinding(bindableParameters, values,
binding);
if (valueForBinding.isEmpty()) {
bindSpecToUse = bindSpecToUse.bindNull(binding.getParameterName(), valueForBinding.getType());
@@ -106,10 +107,11 @@ class ExpressionEvaluatingParameterBinder {
return bindSpecToUse;
}
private <T extends DatabaseClient.BindSpec<T>> T bindParameters(T bindSpec, boolean bindableNull, Object[] values,
private DatabaseClient.GenericExecuteSpec bindParameters(DatabaseClient.GenericExecuteSpec bindSpec,
boolean bindableNull, Object[] values,
Parameters<?, ?> bindableParameters) {
T bindSpecToUse = bindSpec;
DatabaseClient.GenericExecuteSpec bindSpecToUse = bindSpec;
int bindingIndex = 0;
@@ -166,7 +168,8 @@ class ExpressionEvaluatingParameterBinder {
* @param binding must not be {@literal null}.
* @return the value used for the given {@link ParameterBinding}.
*/
private SettableValue getParameterValueForBinding(Parameters<?, ?> parameters, Object[] values,
private org.springframework.r2dbc.core.Parameter getParameterValueForBinding(Parameters<?, ?> parameters,
Object[] values,
ParameterBinding binding) {
return evaluateExpression(binding.getExpression(), parameters, values);
}
@@ -179,7 +182,8 @@ class ExpressionEvaluatingParameterBinder {
* @param parameterValues must not be {@literal null}.
* @return the value of the {@code expressionString} evaluation.
*/
private SettableValue evaluateExpression(String expressionString, Parameters<?, ?> parameters,
private org.springframework.r2dbc.core.Parameter evaluateExpression(String expressionString,
Parameters<?, ?> parameters,
Object[] parameterValues) {
EvaluationContext evaluationContext = evaluationContextProvider.getEvaluationContext(parameters, parameterValues);
@@ -188,6 +192,6 @@ class ExpressionEvaluatingParameterBinder {
Object value = expression.getValue(evaluationContext, Object.class);
Class<?> valueType = expression.getValueType(evaluationContext);
return SettableValue.fromOrEmpty(value, valueType != null ? valueType : Object.class);
return org.springframework.r2dbc.core.Parameter.fromOrEmpty(value, valueType != null ? valueType : Object.class);
}
}

2
src/main/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQuery.java
View File

@@ -21,7 +21,6 @@ import java.util.List;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.relational.repository.query.RelationalEntityMetadata;
import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
@@ -29,6 +28,7 @@ import org.springframework.data.relational.repository.query.RelationalParameters
import org.springframework.data.repository.query.ResultProcessor;
import org.springframework.data.repository.query.ReturnedType;
import org.springframework.data.repository.query.parser.PartTree;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.PreparedOperation;
/**

23
src/main/java/org/springframework/data/r2dbc/repository/query/PreparedOperationBindableQuery.java
View File

@@ -15,15 +15,16 @@
*/
package org.springframework.data.r2dbc.repository.query;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.dialect.BindTarget;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.PreparedOperation;
import org.springframework.r2dbc.core.binding.BindTarget;
import org.springframework.util.Assert;
/**
* A {@link BindableQuery} implementation based on a {@link PreparedOperation}.
*
* @author Roman Chigvintsev
* @author Mark Paluch
*/
class PreparedOperationBindableQuery implements BindableQuery {
@@ -41,12 +42,11 @@ class PreparedOperationBindableQuery implements BindableQuery {
this.preparedQuery = preparedQuery;
}
@SuppressWarnings("unchecked")
@Override
public <T extends DatabaseClient.BindSpec<T>> T bind(T bindSpec) {
BindSpecBindTargetAdapter<T> bindTargetAdapter = new BindSpecBindTargetAdapter<>(bindSpec);
public DatabaseClient.GenericExecuteSpec bind(DatabaseClient.GenericExecuteSpec bindSpec) {
BindSpecBindTargetAdapter bindTargetAdapter = new BindSpecBindTargetAdapter(bindSpec);
preparedQuery.bindTo(bindTargetAdapter);
return (T) bindTargetAdapter.bindSpec;
return bindTargetAdapter.bindSpec;
}
@Override
@@ -55,13 +55,14 @@ class PreparedOperationBindableQuery implements BindableQuery {
}
/**
* This class adapts {@link org.springframework.data.r2dbc.core.DatabaseClient.BindSpec} to {@link BindTarget}
* allowing easy binding of query parameters using {@link PreparedOperation}.
* This class adapts {@link DatabaseClient.GenericExecuteSpec} to {@link BindTarget} allowing easy binding of query
* parameters using {@link PreparedOperation}.
*/
private static class BindSpecBindTargetAdapter<T extends DatabaseClient.BindSpec<T>> implements BindTarget {
DatabaseClient.BindSpec<T> bindSpec;
private static class BindSpecBindTargetAdapter implements BindTarget {
private BindSpecBindTargetAdapter(DatabaseClient.BindSpec<T> bindSpec) {
DatabaseClient.GenericExecuteSpec bindSpec;
private BindSpecBindTargetAdapter(DatabaseClient.GenericExecuteSpec bindSpec) {
this.bindSpec = bindSpec;
}

2
src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryExecution.java
View File

@@ -23,7 +23,6 @@ import org.reactivestreams.Publisher;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mapping.model.EntityInstantiators;
import org.springframework.data.r2dbc.core.FetchSpec;
import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -31,6 +30,7 @@ import org.springframework.data.relational.repository.query.DtoInstantiatingConv
import org.springframework.data.repository.query.ResultProcessor;
import org.springframework.data.repository.query.ReturnedType;
import org.springframework.data.util.ReflectionUtils;
import org.springframework.r2dbc.core.FetchSpec;
import org.springframework.util.ClassUtils;
/**

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

@@ -16,12 +16,11 @@
package org.springframework.data.r2dbc.repository.query;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DatabaseClient.BindSpec;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.util.Assert;
/**
@@ -76,7 +75,7 @@ public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
this.binder = new ExpressionEvaluatingParameterBinder(expressionParser, evaluationContextProvider, expressionQuery);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.repository.query.AbstractR2dbcQuery#isModifyingQuery()
*/
@@ -85,7 +84,7 @@ public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
return getQueryMethod().isModifyingQuery();
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.r2dbc.repository.query.AbstractR2dbcQuery#createQuery(org.springframework.data.relational.repository.query.RelationalParameterAccessor)
*/
@@ -95,7 +94,7 @@ public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
return new BindableQuery() {
@Override
public <T extends BindSpec<T>> T bind(T bindSpec) {
public DatabaseClient.GenericExecuteSpec bind(DatabaseClient.GenericExecuteSpec bindSpec) {
return binder.bind(bindSpec, accessor);
}

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

@@ -43,6 +43,7 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro
import org.springframework.data.repository.query.RepositoryQuery;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.util.Assert;
/**

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

@@ -25,6 +25,7 @@ import org.springframework.data.repository.Repository;
import org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport;
import org.springframework.data.repository.core.support.RepositoryFactorySupport;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.util.Assert;
/**

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

@@ -22,16 +22,16 @@ import org.reactivestreams.Publisher;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.query.Query;
import org.springframework.data.relational.repository.query.RelationalEntityInformation;
import org.springframework.data.repository.reactive.ReactiveSortingRepository;
import org.springframework.data.util.Lazy;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
@@ -76,6 +76,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveSortingRepository<T
* @param databaseClient
* @param converter
* @param accessStrategy
* @since 1.2
*/
public SimpleR2dbcRepository(RelationalEntityInformation<T, ID> entity, DatabaseClient databaseClient,
R2dbcConverter converter, ReactiveDataAccessStrategy accessStrategy) {
@@ -88,6 +89,28 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveSortingRepository<T
.getRequiredIdProperty());
}
/**
* Create a new {@link SimpleR2dbcRepository}.
*
* @param entity
* @param databaseClient
* @param converter
* @param accessStrategy
* @deprecated since 1.2.
*/
@Deprecated
public SimpleR2dbcRepository(RelationalEntityInformation<T, ID> entity,
org.springframework.data.r2dbc.core.DatabaseClient databaseClient, R2dbcConverter converter,
ReactiveDataAccessStrategy accessStrategy) {
this.entity = entity;
this.entityOperations = new R2dbcEntityTemplate(databaseClient, accessStrategy);
this.idProperty = Lazy.of(() -> converter //
.getMappingContext() //
.getRequiredPersistentEntity(this.entity.getJavaType()) //
.getRequiredIdProperty());
}
/* (non-Javadoc)
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#save(S)
*/

7
src/test/java/org/springframework/data/r2dbc/config/H2IntegrationTests.java
View File

@@ -32,13 +32,13 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.FilterType;
import org.springframework.dao.DataAccessException;
import org.springframework.data.annotation.Id;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
import org.springframework.data.r2dbc.testing.H2TestSupport;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
@@ -70,9 +70,8 @@ public class H2IntegrationTests {
jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
databaseClient.execute("SELECT COUNT(*) FROM legoset") //
.as(Long.class) //
.fetch() //
databaseClient.sql("SELECT COUNT(*) FROM legoset") //
.map(it -> it.get(0, Long.class)) //
.all() //
.as(StepVerifier::create) //
.expectNext(1L) //

2
src/test/java/org/springframework/data/r2dbc/config/R2dbcConfigurationIntegrationTests.java
View File

@@ -27,7 +27,7 @@ import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Tests for {@link AbstractR2dbcConfiguration}.

488
src/test/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManagerUnitTests.java
View File

@@ -1,488 +0,0 @@
/*
* Copyright 2019-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.connectionfactory;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.*;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.*;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.IsolationLevel;
import io.r2dbc.spi.R2dbcBadGrammarException;
import io.r2dbc.spi.Statement;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Before;
import org.junit.Test;
import org.springframework.data.r2dbc.BadSqlGrammarException;
import org.springframework.transaction.CannotCreateTransactionException;
import org.springframework.transaction.IllegalTransactionStateException;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.reactive.TransactionSynchronization;
import org.springframework.transaction.reactive.TransactionSynchronizationManager;
import org.springframework.transaction.reactive.TransactionalOperator;
import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
* Unit tests for {@link R2dbcTransactionManager}.
*
* @author Mark Paluch
*/
public class R2dbcTransactionManagerUnitTests {
ConnectionFactory connectionFactoryMock = mock(ConnectionFactory.class);
Connection connectionMock = mock(Connection.class);
private R2dbcTransactionManager tm;
@Before
public void before() {
when(connectionFactoryMock.create()).thenReturn((Mono) Mono.just(connectionMock));
when(connectionMock.beginTransaction()).thenReturn(Mono.empty());
when(connectionMock.close()).thenReturn(Mono.empty());
tm = new R2dbcTransactionManager(connectionFactoryMock);
}
@Test // gh-107
public void testSimpleTransaction() {
TestTransactionSynchronization sync = new TestTransactionSynchronization(
TransactionSynchronization.STATUS_COMMITTED);
AtomicInteger commits = new AtomicInteger();
when(connectionMock.commitTransaction()).thenReturn(Mono.fromRunnable(commits::incrementAndGet));
TransactionalOperator operator = TransactionalOperator.create(tm);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).flatMap(it -> {
return TransactionSynchronizationManager.forCurrentTransaction()
.doOnNext(synchronizationManager -> synchronizationManager.registerSynchronization(sync));
}) //
.as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
assertThat(commits).hasValue(1);
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
assertThat(sync.beforeCommitCalled).isTrue();
assertThat(sync.afterCommitCalled).isTrue();
assertThat(sync.beforeCompletionCalled).isTrue();
assertThat(sync.afterCompletionCalled).isTrue();
}
@Test // gh-329
public void testBeginFails() {
reset(connectionFactoryMock);
when(connectionFactoryMock.create()).thenReturn(Mono.error(new R2dbcBadGrammarException("fail")));
when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectErrorSatisfies(actual -> {
assertThat(actual).isInstanceOf(CannotCreateTransactionException.class)
.hasCauseInstanceOf(BadSqlGrammarException.class);
}).verify();
}
@Test // gh-107
public void appliesIsolationLevel() {
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
when(connectionMock.getTransactionIsolationLevel()).thenReturn(IsolationLevel.READ_COMMITTED);
when(connectionMock.setTransactionIsolationLevel(any())).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
verify(connectionMock).beginTransaction();
verify(connectionMock).setTransactionIsolationLevel(IsolationLevel.READ_COMMITTED);
verify(connectionMock).setTransactionIsolationLevel(IsolationLevel.SERIALIZABLE);
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
}
@Test // gh-184
public void doesNotSetIsolationLevelIfMatch() {
when(connectionMock.getTransactionIsolationLevel()).thenReturn(IsolationLevel.READ_COMMITTED);
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
verify(connectionMock).beginTransaction();
verify(connectionMock, never()).setTransactionIsolationLevel(any());
verify(connectionMock).commitTransaction();
}
@Test // gh-184
public void doesNotSetAutoCommitDisabled() {
when(connectionMock.isAutoCommit()).thenReturn(false);
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
verify(connectionMock).beginTransaction();
verify(connectionMock, never()).setAutoCommit(anyBoolean());
verify(connectionMock).commitTransaction();
}
@Test // gh-184
public void restoresAutoCommit() {
when(connectionMock.isAutoCommit()).thenReturn(true);
when(connectionMock.setAutoCommit(anyBoolean())).thenReturn(Mono.empty());
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
verify(connectionMock).beginTransaction();
verify(connectionMock).setAutoCommit(false);
verify(connectionMock).setAutoCommit(true);
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
}
@Test // gh-107
public void appliesReadOnly() {
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
when(connectionMock.setTransactionIsolationLevel(any())).thenReturn(Mono.empty());
Statement statement = mock(Statement.class);
when(connectionMock.createStatement(anyString())).thenReturn(statement);
when(statement.execute()).thenReturn(Mono.empty());
tm.setEnforceReadOnly(true);
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setReadOnly(true);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).createStatement("SET TRANSACTION READ ONLY");
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
}
@Test // gh-107
public void testCommitFails() {
when(connectionMock.commitTransaction()).thenReturn(Mono.defer(() -> {
return Mono.error(new R2dbcBadGrammarException("Commit should fail"));
}));
when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty());
TransactionalOperator operator = TransactionalOperator.create(tm);
ConnectionFactoryUtils.getConnection(connectionFactoryMock) //
.doOnNext(it -> {
it.createStatement("foo");
}).then() //
.as(operator::transactional) //
.as(StepVerifier::create) //
.verifyError(IllegalTransactionStateException.class);
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).createStatement("foo");
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
}
@Test // gh-107
public void testRollback() {
AtomicInteger commits = new AtomicInteger();
when(connectionMock.commitTransaction()).thenReturn(Mono.fromRunnable(commits::incrementAndGet));
AtomicInteger rollbacks = new AtomicInteger();
when(connectionMock.rollbackTransaction()).thenReturn(Mono.fromRunnable(rollbacks::incrementAndGet));
TransactionalOperator operator = TransactionalOperator.create(tm);
ConnectionFactoryUtils.getConnection(connectionFactoryMock).doOnNext(it -> {
throw new IllegalStateException();
}).as(operator::transactional) //
.as(StepVerifier::create) //
.verifyError(IllegalStateException.class);
assertThat(commits).hasValue(0);
assertThat(rollbacks).hasValue(1);
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).rollbackTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
}
@Test // gh-329
public void testRollbackFails() {
when(connectionMock.rollbackTransaction()).thenReturn(Mono.defer(() -> {
return Mono.error(new R2dbcBadGrammarException("Commit should fail"));
}), Mono.empty());
TransactionalOperator operator = TransactionalOperator.create(tm);
operator.execute(reactiveTransaction -> {
reactiveTransaction.setRollbackOnly();
return ConnectionFactoryUtils.getConnection(connectionFactoryMock) //
.doOnNext(it -> {
it.createStatement("foo");
}).then();
}).as(StepVerifier::create) //
.verifyError(IllegalTransactionStateException.class);
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).createStatement("foo");
verify(connectionMock, never()).commitTransaction();
verify(connectionMock).rollbackTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
}
@Test // gh-107
public void testTransactionSetRollbackOnly() {
when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty());
TestTransactionSynchronization sync = new TestTransactionSynchronization(
TransactionSynchronization.STATUS_ROLLED_BACK);
TransactionalOperator operator = TransactionalOperator.create(tm);
operator.execute(tx -> {
tx.setRollbackOnly();
assertThat(tx.isNewTransaction()).isTrue();
return TransactionSynchronizationManager.forCurrentTransaction().doOnNext(it -> {
assertThat(it.hasResource(connectionFactoryMock)).isTrue();
it.registerSynchronization(sync);
}).then();
}).as(StepVerifier::create) //
.verifyComplete();
verify(connectionMock).isAutoCommit();
verify(connectionMock).beginTransaction();
verify(connectionMock).rollbackTransaction();
verify(connectionMock).close();
verifyNoMoreInteractions(connectionMock);
assertThat(sync.beforeCommitCalled).isFalse();
assertThat(sync.afterCommitCalled).isFalse();
assertThat(sync.beforeCompletionCalled).isTrue();
assertThat(sync.afterCompletionCalled).isTrue();
}
@Test // gh-107
public void testPropagationNeverWithExistingTransaction() {
when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
operator.execute(tx1 -> {
assertThat(tx1.isNewTransaction()).isTrue();
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_NEVER);
return operator.execute(tx2 -> {
fail("Should have thrown IllegalTransactionStateException");
return Mono.empty();
});
}).as(StepVerifier::create) //
.verifyError(IllegalTransactionStateException.class);
verify(connectionMock).rollbackTransaction();
verify(connectionMock).close();
}
@Test // gh-107
public void testPropagationSupportsAndRequiresNew() {
when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);
TransactionalOperator operator = TransactionalOperator.create(tm, definition);
operator.execute(tx1 -> {
assertThat(tx1.isNewTransaction()).isFalse();
DefaultTransactionDefinition innerDef = new DefaultTransactionDefinition();
innerDef.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
TransactionalOperator inner = TransactionalOperator.create(tm, innerDef);
return inner.execute(tx2 -> {
assertThat(tx2.isNewTransaction()).isTrue();
return Mono.empty();
});
}).as(StepVerifier::create) //
.verifyComplete();
verify(connectionMock).commitTransaction();
verify(connectionMock).close();
}
private static class TestTransactionSynchronization implements TransactionSynchronization {
private int status;
public boolean beforeCommitCalled;
public boolean beforeCompletionCalled;
public boolean afterCommitCalled;
public boolean afterCompletionCalled;
public Throwable afterCompletionException;
public TestTransactionSynchronization(int status) {
this.status = status;
}
@Override
public Mono<Void> suspend() {
return Mono.empty();
}
@Override
public Mono<Void> resume() {
return Mono.empty();
}
@Override
public Mono<Void> beforeCommit(boolean readOnly) {
if (this.status != TransactionSynchronization.STATUS_COMMITTED) {
fail("Should never be called");
}
return Mono.fromRunnable(() -> {
assertFalse(this.beforeCommitCalled);
this.beforeCommitCalled = true;
});
}
@Override
public Mono<Void> beforeCompletion() {
return Mono.fromRunnable(() -> {
assertFalse(this.beforeCompletionCalled);
this.beforeCompletionCalled = true;
});
}
@Override
public Mono<Void> afterCommit() {
if (this.status != TransactionSynchronization.STATUS_COMMITTED) {
fail("Should never be called");
}
return Mono.fromRunnable(() -> {
assertFalse(this.afterCommitCalled);
this.afterCommitCalled = true;
});
}
@Override
public Mono<Void> afterCompletion(int status) {
try {
return Mono.fromRunnable(() -> doAfterCompletion(status));
} catch (Throwable ex) {
this.afterCompletionException = ex;
}
return Mono.empty();
}
protected void doAfterCompletion(int status) {
assertFalse(this.afterCompletionCalled);
this.afterCompletionCalled = true;
assertTrue(status == this.status);
}
}
}

17
src/test/java/org/springframework/data/r2dbc/connectionfactory/SingleConnectionConnectionFactoryUnitTests.java
View File

@@ -100,23 +100,6 @@ public class SingleConnectionConnectionFactoryUnitTests {
factory.destroy();
}
@Test // gh-204
public void releaseConnectionShouldNotCloseConnection() {
Connection connectionMock = mock(Connection.class);
ConnectionFactoryMetadata metadata = mock(ConnectionFactoryMetadata.class);
SingleConnectionConnectionFactory factory = new SingleConnectionConnectionFactory(connectionMock, metadata, false);
Connection connection = factory.create().block();
ConnectionFactoryUtils.releaseConnection(connection, factory) //
.as(StepVerifier::create) //
.verifyComplete();
verify(connectionMock, never()).close();
}
@Test // gh-204
public void releaseConnectionShouldCloseUnrelatedConnection() {

26
src/test/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverterUnitTests.java
View File

@@ -37,9 +37,9 @@ import org.springframework.data.convert.ReadingConverter;
import org.springframework.data.convert.WritingConverter;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.r2dbc.core.Parameter;
/**
* Unit tests for {@link MappingR2dbcConverter}.
@@ -72,12 +72,12 @@ public class MappingR2dbcConverterUnitTests {
LocalDateTime localDateTime = LocalDateTime.now();
converter.write(new Person("id", "Walter", "White", instant, localDateTime), row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("id"), SettableValue.fromOrEmpty("id", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("id"), Parameter.fromOrEmpty("id", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("firstname"),
SettableValue.fromOrEmpty("Walter", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("lastname"), SettableValue.fromOrEmpty("White", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("instant"), SettableValue.from(instant));
assertThat(row).containsEntry(SqlIdentifier.unquoted("local_date_time"), SettableValue.from(localDateTime));
Parameter.fromOrEmpty("Walter", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("lastname"), Parameter.fromOrEmpty("White", String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("instant"), Parameter.from(instant));
assertThat(row).containsEntry(SqlIdentifier.unquoted("local_date_time"), Parameter.from(localDateTime));
}
@Test // gh-41
@@ -117,7 +117,7 @@ public class MappingR2dbcConverterUnitTests {
OutboundRow row = new OutboundRow();
converter.write(withMap, row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("nested"), SettableValue.from("map"));
assertThat(row).containsEntry(SqlIdentifier.unquoted("nested"), Parameter.from("map"));
}
@Test // gh-59
@@ -138,7 +138,7 @@ public class MappingR2dbcConverterUnitTests {
OutboundRow row = new OutboundRow();
converter.write(withMap, row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("condition"), SettableValue.from("Mint"));
assertThat(row).containsEntry(SqlIdentifier.unquoted("condition"), Parameter.from("Mint"));
}
@Test // gh-59
@@ -148,7 +148,7 @@ public class MappingR2dbcConverterUnitTests {
OutboundRow row = new OutboundRow();
converter.write(withMap, row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("condition"), SettableValue.fromOrEmpty(null, String.class));
assertThat(row).containsEntry(SqlIdentifier.unquoted("condition"), Parameter.fromOrEmpty(null, String.class));
}
@Test // gh-59
@@ -172,8 +172,8 @@ public class MappingR2dbcConverterUnitTests {
OutboundRow row = new OutboundRow();
converter.write(person, row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("foo_column"), SettableValue.from("bar"))
.containsEntry(SqlIdentifier.unquoted("entity"), SettableValue.from("nested_entity"));
assertThat(row).containsEntry(SqlIdentifier.unquoted("foo_column"), Parameter.from("bar"))
.containsEntry(SqlIdentifier.unquoted("entity"), Parameter.from("nested_entity"));
}
@Test // gh-59
@@ -263,8 +263,8 @@ public class MappingR2dbcConverterUnitTests {
public OutboundRow convert(CustomConversionPerson source) {
OutboundRow row = new OutboundRow();
row.put("foo_column", SettableValue.from(source.foo));
row.put("entity", SettableValue.from("nested_entity"));
row.put("foo_column", Parameter.from(source.foo));
row.put("entity", Parameter.from("nested_entity"));
return row;
}

4
src/test/java/org/springframework/data/r2dbc/convert/PostgresMappingR2dbcConverterUnitTests.java
View File

@@ -32,9 +32,9 @@ import org.springframework.data.convert.CustomConversions;
import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.r2dbc.core.Parameter;
/**
* Postgres-specific unit tests for {@link MappingR2dbcConverter}.
@@ -69,7 +69,7 @@ public class PostgresMappingR2dbcConverterUnitTests {
OutboundRow row = new OutboundRow();
converter.write(person, row);
assertThat(row).containsEntry(SqlIdentifier.unquoted("json_value"), SettableValue.from(person.jsonValue));
assertThat(row).containsEntry(SqlIdentifier.unquoted("json_value"), Parameter.from(person.jsonValue));
}
@AllArgsConstructor

17
src/test/java/org/springframework/data/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java
View File

@@ -36,9 +36,10 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.dao.DataAccessException;
import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.connectionfactory.R2dbcTransactionManager;
import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.r2dbc.connection.R2dbcTransactionManager;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.transaction.ReactiveTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.annotation.Transactional;
@@ -146,7 +147,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
public void executeInsertInManagedTransaction() {
Flux<Integer> integerFlux = databaseClient //
.execute(getInsertIntoLegosetStatement()) //
.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
@@ -162,7 +163,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
@Test // gh-2
public void executeInsertInAutoCommitTransaction() {
Flux<Integer> integerFlux = databaseClient.execute(getInsertIntoLegosetStatement()) //
Flux<Integer> integerFlux = databaseClient.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
@@ -178,7 +179,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
@Test // gh-2
public void shouldRollbackTransaction() {
Mono<Object> integerFlux = databaseClient.execute(getInsertIntoLegosetStatement()) //
Mono<Object> integerFlux = databaseClient.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
@@ -196,7 +197,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
@Test // gh-2, gh-75, gh-107
public void emitTransactionIds() {
Flux<Object> txId = databaseClient.execute(getCurrentTransactionIdStatement()) //
Flux<Object> txId = databaseClient.sql(getCurrentTransactionIdStatement()) //
.map((row, md) -> row.get(0)) //
.all();
@@ -220,7 +221,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
TransactionalOperator transactionalOperator = TransactionalOperator
.create(new R2dbcTransactionManager(connectionFactory), new DefaultTransactionDefinition());
Flux<Integer> integerFlux = databaseClient.execute(getInsertIntoLegosetStatement()) //
Flux<Integer> integerFlux = databaseClient.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //
@@ -301,7 +302,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
@Transactional
public Flux<Object> emitTransactionIds(Mono<Void> prepareTransaction, String idStatement) {
Flux<Object> txId = databaseClient.execute(idStatement) //
Flux<Object> txId = databaseClient.sql(idStatement) //
.map((row, md) -> row.get(0)) //
.all();
@@ -311,7 +312,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
@Transactional
public Flux<Integer> shouldRollbackTransactionUsingTransactionalOperator(String insertStatement) {
return databaseClient.execute(insertStatement) //
return databaseClient.sql(insertStatement) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //

3
src/test/java/org/springframework/data/r2dbc/core/JasyncMySqlTransactionalDatabaseClientIntegrationTests.java
View File

@@ -26,6 +26,7 @@ import org.junit.ClassRule;
import org.springframework.data.r2dbc.testing.ExternalDatabase;
import org.springframework.data.r2dbc.testing.MySqlTestSupport;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Transactional integration tests for {@link DatabaseClient} against MySQL using Jasync MySQL.
@@ -62,7 +63,7 @@ public class JasyncMySqlTransactionalDatabaseClientIntegrationTests
* batches every now and then.
* @see: https://dev.mysql.com/doc/refman/5.7/en/innodb-information-schema-internal-data.html
*/
return client.execute(getInsertIntoLegosetStatement()) //
return client.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //

3
src/test/java/org/springframework/data/r2dbc/core/MariaDbTransactionalDatabaseClientIntegrationTests.java
View File

@@ -26,6 +26,7 @@ import org.junit.ClassRule;
import org.springframework.data.r2dbc.testing.ExternalDatabase;
import org.springframework.data.r2dbc.testing.MariaDbTestSupport;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Transactional integration tests for {@link DatabaseClient} against MariaDb.
@@ -61,7 +62,7 @@ public class MariaDbTransactionalDatabaseClientIntegrationTests
* And we need to delay emitting the result so that Mariadb has time to write the transaction id, which is done in
* batches every now and then.
*/
return client.execute(getInsertIntoLegosetStatement()) //
return client.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //

3
src/test/java/org/springframework/data/r2dbc/core/MySqlTransactionalDatabaseClientIntegrationTests.java
View File

@@ -26,6 +26,7 @@ import org.junit.ClassRule;
import org.springframework.data.r2dbc.testing.ExternalDatabase;
import org.springframework.data.r2dbc.testing.MySqlTestSupport;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Transactional integration tests for {@link DatabaseClient} against MySQL.
@@ -62,7 +63,7 @@ public class MySqlTransactionalDatabaseClientIntegrationTests
* batches every now and then.
* @see: https://dev.mysql.com/doc/refman/5.7/en/innodb-information-schema-internal-data.html
*/
return client.execute(getInsertIntoLegosetStatement()) //
return client.sql(getInsertIntoLegosetStatement()) //
.bind(0, 42055) //
.bind(1, "SCHAUFELRADBAGGER") //
.bindNull(2, Integer.class) //

13
src/test/java/org/springframework/data/r2dbc/core/PostgresReactiveDataAccessStrategyTests.java
View File

@@ -33,6 +33,7 @@ import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.r2dbc.core.Parameter;
/**
* {@link PostgresDialect} specific tests for {@link ReactiveDataAccessStrategy}.
@@ -88,8 +89,8 @@ public class PostgresReactiveDataAccessStrategyTests extends ReactiveDataAccessS
OutboundRow outboundRow = strategy.getOutboundRow(withArray);
assertThat(outboundRow) //
.containsEntry(SqlIdentifier.unquoted("string_array"), SettableValue.from(new String[] { "hello", "world" }))
.containsEntry(SqlIdentifier.unquoted("string_list"), SettableValue.from(new String[] { "hello", "world" }));
.containsEntry(SqlIdentifier.unquoted("string_array"), Parameter.from(new String[] { "hello", "world" }))
.containsEntry(SqlIdentifier.unquoted("string_list"), Parameter.from(new String[] { "hello", "world" }));
}
@Test // gh-139
@@ -104,7 +105,7 @@ public class PostgresReactiveDataAccessStrategyTests extends ReactiveDataAccessS
OutboundRow outboundRow = strategy.getOutboundRow(withConversion);
assertThat(outboundRow) //
.containsEntry(SqlIdentifier.unquoted("my_objects"), SettableValue.from("[one, two]"));
.containsEntry(SqlIdentifier.unquoted("my_objects"), Parameter.from("[one, two]"));
}
@Test // gh-139
@@ -121,7 +122,7 @@ public class PostgresReactiveDataAccessStrategyTests extends ReactiveDataAccessS
assertThat(outboundRow) //
.containsKey(SqlIdentifier.unquoted("my_objects"));
SettableValue value = outboundRow.get("my_objects");
Parameter value = outboundRow.get("my_objects");
assertThat(value.isEmpty()).isTrue();
assertThat(value.getType()).isEqualTo(String.class);
}
@@ -139,7 +140,7 @@ public class PostgresReactiveDataAccessStrategyTests extends ReactiveDataAccessS
assertThat(outboundRow).containsKey(SqlIdentifier.unquoted("enum_set"));
SettableValue value = outboundRow.get(SqlIdentifier.unquoted("enum_set"));
Parameter value = outboundRow.get(SqlIdentifier.unquoted("enum_set"));
assertThat(value.getValue()).isEqualTo(new String[] { "ONE", "TWO" });
}
@@ -156,7 +157,7 @@ public class PostgresReactiveDataAccessStrategyTests extends ReactiveDataAccessS
assertThat(outboundRow).containsKey(SqlIdentifier.unquoted("enum_array"));
SettableValue value = outboundRow.get(SqlIdentifier.unquoted("enum_array"));
Parameter value = outboundRow.get(SqlIdentifier.unquoted("enum_array"));
assertThat(value.getValue()).isEqualTo(new String[] { "ONE", "TWO" });
}

43
src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java
View File

@@ -40,7 +40,6 @@ import org.springframework.data.domain.Sort;
import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.r2dbc.mapping.event.AfterConvertCallback;
import org.springframework.data.r2dbc.mapping.event.AfterSaveCallback;
import org.springframework.data.r2dbc.mapping.event.BeforeConvertCallback;
@@ -52,6 +51,8 @@ import org.springframework.data.relational.core.query.Query;
import org.springframework.data.relational.core.query.Update;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.lang.Nullable;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.util.CollectionUtils;
/**
@@ -61,7 +62,7 @@ import org.springframework.util.CollectionUtils;
*/
public class R2dbcEntityTemplateUnitTests {
DatabaseClient client;
org.springframework.r2dbc.core.DatabaseClient client;
R2dbcEntityTemplate entityTemplate;
StatementRecorder recorder;
@@ -70,8 +71,8 @@ public class R2dbcEntityTemplateUnitTests {
recorder = StatementRecorder.newInstance();
client = DatabaseClient.builder().connectionFactory(recorder)
.dataAccessStrategy(new DefaultReactiveDataAccessStrategy(PostgresDialect.INSTANCE)).build();
entityTemplate = new R2dbcEntityTemplate(client);
.bindMarkers(PostgresDialect.INSTANCE.getBindMarkersFactory()).build();
entityTemplate = new R2dbcEntityTemplate(client, PostgresDialect.INSTANCE);
}
@Test // gh-220
@@ -92,7 +93,7 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220
@@ -113,7 +114,7 @@ public class R2dbcEntityTemplateUnitTests {
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
assertThat(statement.getSql()).isEqualTo("SELECT person.id FROM person WHERE person.THE_NAME = $1 LIMIT 1");
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220
@@ -129,7 +130,7 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-215
@@ -170,7 +171,7 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220
@@ -191,8 +192,8 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(2).containsEntry(0, Parameter.from("Heisenberg")).containsEntry(1,
Parameter.from("Walter"));
}
@Test // gh-220
@@ -212,7 +213,7 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220
@@ -229,7 +230,7 @@ public class R2dbcEntityTemplateUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-365
@@ -249,8 +250,8 @@ public class R2dbcEntityTemplateUnitTests {
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("INSERT"));
assertThat(statement.getSql()).isEqualTo("INSERT INTO versioned_person (id, version, name) VALUES ($1, $2, $3)");
assertThat(statement.getBindings()).hasSize(3).containsEntry(0, SettableValue.from("id")).containsEntry(1,
SettableValue.from(1L));
assertThat(statement.getBindings()).hasSize(3).containsEntry(0, Parameter.from("id")).containsEntry(1,
Parameter.from(1L));
}
@Test // gh-215
@@ -277,8 +278,8 @@ public class R2dbcEntityTemplateUnitTests {
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("INSERT"));
assertThat(statement.getSql()).isEqualTo("INSERT INTO person (THE_NAME, description) VALUES ($1, $2)");
assertThat(statement.getBindings()).hasSize(2).containsEntry(0, SettableValue.from("before-convert"))
.containsEntry(1, SettableValue.from("before-save"));
assertThat(statement.getBindings()).hasSize(2).containsEntry(0, Parameter.from("before-convert")).containsEntry(1,
Parameter.from("before-save"));
}
@Test // gh-365
@@ -299,8 +300,8 @@ public class R2dbcEntityTemplateUnitTests {
assertThat(statement.getSql()).isEqualTo(
"UPDATE versioned_person SET version = $1, name = $2 WHERE versioned_person.id = $3 AND (versioned_person.version = $4)");
assertThat(statement.getBindings()).hasSize(4).containsEntry(0, SettableValue.from(2L)).containsEntry(3,
SettableValue.from(1L));
assertThat(statement.getBindings()).hasSize(4).containsEntry(0, Parameter.from(2L)).containsEntry(3,
Parameter.from(1L));
}
@Test // gh-215
@@ -332,8 +333,8 @@ public class R2dbcEntityTemplateUnitTests {
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("UPDATE"));
assertThat(statement.getSql()).isEqualTo("UPDATE person SET THE_NAME = $1, description = $2 WHERE person.id = $3");
assertThat(statement.getBindings()).hasSize(3).containsEntry(0, SettableValue.from("before-convert"))
.containsEntry(1, SettableValue.from("before-save"));
assertThat(statement.getBindings()).hasSize(3).containsEntry(0, Parameter.from("before-convert")).containsEntry(1,
Parameter.from("before-save"));
}
@ToString
@@ -402,7 +403,7 @@ public class R2dbcEntityTemplateUnitTests {
public Mono<Person> onBeforeSave(Person entity, OutboundRow outboundRow, SqlIdentifier table) {
capture(entity);
outboundRow.put(SqlIdentifier.unquoted("description"), SettableValue.from("before-save"));
outboundRow.put(SqlIdentifier.unquoted("description"), Parameter.from("before-save"));
return Mono.just(entity);
}
}

4
src/test/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategyTestSupport.java
View File

@@ -38,8 +38,8 @@ import org.junit.Test;
import org.springframework.data.annotation.ReadOnlyProperty;
import org.springframework.data.r2dbc.dialect.R2dbcDialect;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.r2dbc.core.Parameter;
/**
* Abstract base class for {@link R2dbcDialect}-aware {@link DefaultReactiveDataAccessStrategy} tests.
@@ -205,7 +205,7 @@ public abstract class ReactiveDataAccessStrategyTestSupport {
setter.accept(toSave, testValue);
assertThat(strategy.getOutboundRow(toSave)).containsEntry(SqlIdentifier.unquoted(fieldname),
SettableValue.from(testValue));
Parameter.from(testValue));
when(rowMock.get(fieldname)).thenReturn(testValue);

3
src/test/java/org/springframework/data/r2dbc/core/ReactiveDeleteOperationUnitTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.r2dbc.testing.StatementRecorder;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.r2dbc.core.Parameter;
/**
* Unit test for {@link ReactiveDeleteOperation}.
@@ -104,7 +105,7 @@ public class ReactiveDeleteOperationUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220

3
src/test/java/org/springframework/data/r2dbc/core/ReactiveInsertOperationUnitTests.java
View File

@@ -31,6 +31,7 @@ import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.r2dbc.testing.StatementRecorder;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.r2dbc.core.Parameter;
/**
* Unit test for {@link ReactiveInsertOperation}.
@@ -77,7 +78,7 @@ public class ReactiveInsertOperationUnitTests {
StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("INSERT"));
assertThat(statement.getSql()).isEqualTo("INSERT INTO person (THE_NAME) VALUES ($1)");
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, SettableValue.from("Walter"));
assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
}
@Test // gh-220

6
src/test/java/org/springframework/data/r2dbc/core/ReactiveUpdateOperationUnitTests.java
View File

@@ -27,10 +27,10 @@ import org.junit.Test;
import org.springframework.data.annotation.Id;
import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.data.r2dbc.testing.StatementRecorder;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.data.relational.core.query.Update;
import org.springframework.r2dbc.core.Parameter;
/**
* Unit test for {@link ReactiveUpdateOperation}.
@@ -107,8 +107,8 @@ public class ReactiveUpdateOperationUnitTests {
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"));
assertThat(statement.getBindings()).hasSize(2).containsEntry(0, Parameter.from("Heisenberg")).containsEntry(1,
Parameter.from("Walter"));
}
@Test // gh-220

5
src/test/java/org/springframework/data/r2dbc/repository/ConvertingR2dbcRepositoryIntegrationTests.java
View File

@@ -49,6 +49,7 @@ import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
import org.springframework.data.r2dbc.testing.H2TestSupport;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.test.context.junit4.SpringRunner;
/**
@@ -149,10 +150,10 @@ public class ConvertingR2dbcRepositoryIntegrationTests {
OutboundRow outboundRow = new OutboundRow();
if (convertedEntity.getId() != null) {
outboundRow.put("id", SettableValue.from(convertedEntity.getId()));
outboundRow.put("id", Parameter.from(convertedEntity.getId()));
}
outboundRow.put("name", SettableValue.from("prefixed: " + convertedEntity.getName()));
outboundRow.put("name", Parameter.from("prefixed: " + convertedEntity.getName()));
return outboundRow;
}

2
src/test/java/org/springframework/data/r2dbc/repository/config/R2dbcRepositoriesRegistrarTests.java
View File

@@ -26,13 +26,13 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DefaultReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.dialect.MySqlDialect;
import org.springframework.data.r2dbc.dialect.PostgresDialect;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.dialect.SqlServerDialect;
import org.springframework.data.r2dbc.repository.config.mysql.MySqlPersonRepository;
import org.springframework.data.r2dbc.repository.config.sqlserver.SqlServerPersonRepository;

15
src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java
View File

@@ -38,7 +38,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.annotation.Id;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DefaultReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.dialect.DialectResolver;
@@ -88,8 +88,7 @@ public class PartTreeR2dbcQueryUnitTests {
R2dbcDialect dialect = DialectResolver.getDialect(connectionFactory);
dataAccessStrategy = new DefaultReactiveDataAccessStrategy(dialect, r2dbcConverter);
databaseClient = DatabaseClient.builder().connectionFactory(connectionFactory)
.dataAccessStrategy(dataAccessStrategy).build();
databaseClient = DatabaseClient.builder().connectionFactory(connectionFactory).build();
}
@Test // gh-282
@@ -166,7 +165,7 @@ public class PartTreeR2dbcQueryUnitTests {
assertThat(bindableQuery.get())
.isEqualTo("SELECT " + ALL_FIELDS + " FROM " + TABLE + " WHERE " + TABLE + ".date_of_birth BETWEEN $1 AND $2");
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
when(bindSpecMock.bind(anyInt(), any())).thenReturn(bindSpecMock);
bindableQuery.bind(bindSpecMock);
@@ -325,7 +324,7 @@ public class PartTreeR2dbcQueryUnitTests {
dataAccessStrategy);
RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { "Jo" });
BindableQuery bindableQuery = r2dbcQuery.createQuery(accessor);
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
bindableQuery.bind(bindSpecMock);
verify(bindSpecMock, times(1)).bind(0, "Jo%");
@@ -353,7 +352,7 @@ public class PartTreeR2dbcQueryUnitTests {
dataAccessStrategy);
RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { "hn" });
BindableQuery bindableQuery = r2dbcQuery.createQuery(accessor);
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
bindableQuery.bind(bindSpecMock);
verify(bindSpecMock, times(1)).bind(0, "%hn");
@@ -381,7 +380,7 @@ public class PartTreeR2dbcQueryUnitTests {
dataAccessStrategy);
RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { "oh" });
BindableQuery bindableQuery = r2dbcQuery.createQuery(accessor);
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
bindableQuery.bind(bindSpecMock);
verify(bindSpecMock, times(1)).bind(0, "%oh%");
@@ -409,7 +408,7 @@ public class PartTreeR2dbcQueryUnitTests {
dataAccessStrategy);
RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { "oh" });
BindableQuery bindableQuery = r2dbcQuery.createQuery(accessor);
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
bindableQuery.bind(bindSpecMock);
verify(bindSpecMock, times(1)).bind(0, "%oh%");

10
src/test/java/org/springframework/data/r2dbc/repository/query/PreparedOperationBindableQueryUnitTests.java
View File

@@ -23,11 +23,14 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.PreparedOperation;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.PreparedOperation;
/**
* Unit tests for {@link PreparedOperationBindableQuery}.
*
* @author Roman Chigvintsev
* @author Marl Paluch
*/
@RunWith(MockitoJUnitRunner.class)
@Ignore
@@ -35,11 +38,10 @@ public class PreparedOperationBindableQueryUnitTests {
@Mock PreparedOperation<?> preparedOperation;
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test // gh-282
public void bindsQueryParameterValues() {
DatabaseClient.BindSpec bindSpecMock = mock(DatabaseClient.BindSpec.class);
DatabaseClient.GenericExecuteSpec bindSpecMock = mock(DatabaseClient.GenericExecuteSpec.class);
PreparedOperationBindableQuery query = new PreparedOperationBindableQuery(preparedOperation);
query.bind(bindSpecMock);

4
src/test/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQueryUnitTests.java
View File

@@ -32,8 +32,8 @@ import org.springframework.data.geo.Point;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.DatabaseClient.GenericExecuteSpec;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.DatabaseClient.GenericExecuteSpec;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;

2
src/test/java/org/springframework/data/r2dbc/repository/support/AbstractSimpleR2dbcRepositoryIntegrationTests.java
View File

@@ -41,7 +41,6 @@ import org.springframework.data.annotation.Version;
import org.springframework.data.domain.Persistable;
import org.springframework.data.domain.Sort;
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
import org.springframework.data.relational.core.mapping.RelationalMappingContext;
@@ -50,6 +49,7 @@ import org.springframework.data.relational.core.mapping.Table;
import org.springframework.data.relational.repository.query.RelationalEntityInformation;
import org.springframework.data.relational.repository.support.MappingRelationalEntityInformation;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Abstract integration tests for {@link SimpleR2dbcRepository} to be ran against various databases.

2
src/test/java/org/springframework/data/r2dbc/repository/support/R2dbcRepositoryFactoryUnitTests.java
View File

@@ -27,12 +27,12 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.annotation.Id;
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.convert.R2dbcConverter;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.relational.repository.query.RelationalEntityInformation;
import org.springframework.data.relational.repository.support.MappingRelationalEntityInformation;
import org.springframework.data.repository.Repository;
import org.springframework.r2dbc.core.DatabaseClient;
/**
* Unit test for {@link R2dbcRepositoryFactory}.

14
src/test/java/org/springframework/data/r2dbc/testing/StatementRecorder.java
View File

@@ -40,7 +40,7 @@ import java.util.regex.Pattern;
import org.reactivestreams.Publisher;
import org.springframework.data.r2dbc.mapping.SettableValue;
import org.springframework.r2dbc.core.Parameter;
/**
* Recorder utility for R2DBC {@link Statement}s. Allows stubbing and introspection.
@@ -255,7 +255,7 @@ public class StatementRecorder implements ConnectionFactory {
private final List<Result> results;
private final Map<Object, SettableValue> bindings = new LinkedHashMap<>();
private final Map<Object, Parameter> bindings = new LinkedHashMap<>();
public RecordedStatement(String sql, Result result) {
this(sql, Collections.singletonList(result));
@@ -266,7 +266,7 @@ public class StatementRecorder implements ConnectionFactory {
this.results = results;
}
public Map<Object, SettableValue> getBindings() {
public Map<Object, Parameter> getBindings() {
return bindings;
}
@@ -281,25 +281,25 @@ public class StatementRecorder implements ConnectionFactory {
@Override
public Statement bind(int index, Object o) {
this.bindings.put(index, SettableValue.from(o));
this.bindings.put(index, Parameter.from(o));
return this;
}
@Override
public Statement bind(String identifier, Object o) {
this.bindings.put(identifier, SettableValue.from(o));
this.bindings.put(identifier, Parameter.from(o));
return this;
}
@Override
public Statement bindNull(int index, Class<?> type) {
this.bindings.put(index, SettableValue.empty(type));
this.bindings.put(index, Parameter.empty(type));
return this;
}
@Override
public Statement bindNull(String identifier, Class<?> type) {
this.bindings.put(identifier, SettableValue.empty(type));
this.bindings.put(identifier, Parameter.empty(type));
return this;
}