#373 - Use Spring R2DBC.

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

Author: mp911de

src/main/asciidoc/new-features.adoc

@@ -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>>.
 
 [[new-features.1-1-0]]

src/main/asciidoc/reference/r2dbc-upgrading.adoc

@@ -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
 

src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java

@@ -30,19 +30,18 @@
 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 R2dbcDialect getDialect(ConnectionFactory connectionFactory) {
 	 * @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}.
 	 *

src/main/java/org/springframework/data/r2dbc/connectionfactory/ConnectionFactoryUtils.java

@@ -22,11 +22,8 @@
 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 @@ private 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 static Mono<Connection> getConnection(ConnectionFactory connectionFactory
 	 * @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 static Mono<Void> releaseConnection(io.r2dbc.spi.Connection con, Connecti
 	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 static Mono<Void> doCloseConnection(Connection connection, @Nullable Conn
 	 */
 	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 @@ private static int getConnectionSynchronizationOrder(ConnectionFactory connectio
 		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();
-		}
-	}
 }