diff --git a/src/main/asciidoc/reference/r2dbc-transactions.adoc b/src/main/asciidoc/reference/r2dbc-transactions.adoc index 8917eb8..0f0fdba 100644 --- a/src/main/asciidoc/reference/r2dbc-transactions.adoc +++ b/src/main/asciidoc/reference/r2dbc-transactions.adoc @@ -4,7 +4,8 @@ A common pattern when using relational databases is grouping multiple queries within a unit of work that is guarded by a transaction. Relational databases typically associate a transaction with a single transport connection. Using different connections hence results in utilizing different transactions. -Spring Data R2DBC includes transaction-awareness in `DatabaseClient` that allows you to group multiple statements within the same transaction using https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction[Spring's Transaction Management]. +Spring Data R2DBC includes transaction-awareness in `DatabaseClient` that allows you to group multiple statements within +the same transaction using https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction[Spring's Transaction Management]. Spring Data R2DBC provides a implementation for `ReactiveTransactionManager` with `ConnectionFactoryTransactionManager`. See <> for further details. @@ -12,54 +13,75 @@ See <> for further detail ==== [source,java] ---- -ConnectionFactoryTransactionManager tm = new ConnectionFactoryTransactionManager(connectionFactory); -TransactionalOperator operator = TransactionalOperator.create(tm); -DatabaseClient db = DatabaseClient.create(connectionFactory); +ReactiveTransactionManager tm = new ConnectionFactoryTransactionManager(connectionFactory); +TransactionalOperator operator = TransactionalOperator.create(tm); <1> -Mono atomicOperation = db.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") - .bind("id", "joe") - .bind("name", "Joe") - .bind("age", 34) - .fetch().rowsUpdated() - .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") - .bind("id", "joe") - .bind("name", "Joe") - .fetch().rowsUpdated()) - .then() - .as(operator::transactional); +DatabaseClient client = DatabaseClient.create(connectionFactory); + +Mono atomicOperation = client.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") + .bind("id", "joe") + .bind("name", "Joe") + .bind("age", 34) + .fetch().rowsUpdated() + .then(client.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") + .bind("id", "joe") + .bind("name", "Joe") + .fetch().rowsUpdated()) + .then() + .as(operator::transactional); <2> }); ---- +<1> Associate the `TransactionalOperator` with the `ReactiveTransactionManager`. +<2> Bind the operation to the `TransactionalOperator`. ==== -https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction-declarative[Spring's declarative Transaction Management] is a less invasive, annotation-based approach to transaction demarcation. +https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction-declarative[Spring's declarative Transaction Management] +is a less invasive, annotation-based approach to transaction demarcation. .Declarative Transaction Management ==== [source,java] ---- -class MyService { +@Configuration +@EnableTransactionManagement <1> +class Config extends AbstractR2dbcConfiguration { - private final DatabaseClient db; - - MyService(DatabaseClient db) { - this.db = db; + @Override + public ConnectionFactory connectionFactory() { + return // ... } + @Bean + ReactiveTransactionManager txMgr(ConnectionFactory connectionFactory) { <2> + return new ConnectionFactoryTransactionManager(connectionFactory); + } +} + +@Service +class MyService { + + private final DatabaseClient client; + + MyService(DatabaseClient client) { + this.client = client; + } @Transactional public Mono insertPerson() { - return db.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") + return client.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") .bind("id", "joe") .bind("name", "Joe") .bind("age", 34) .fetch().rowsUpdated() - .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") - .bind("id", "joe") - .bind("name", "Joe") - .fetch().rowsUpdated()) - .then(); + .then(client.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") + .bind("id", "joe") + .bind("name", "Joe") + .fetch().rowsUpdated()) + .then(); } } ---- +<1> Enable declarative transaction management. +<2> Provide a `ReactiveTransactionManager` implementation to back reactive tansaction features. ==== diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java index a5a0ea1..4bb0caf 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java index c017222..f54db09 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java @@ -17,14 +17,9 @@ package org.springframework.data.r2dbc.function.connectionfactory; import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; -import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; - import org.springframework.core.Ordered; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.lang.Nullable; @@ -32,6 +27,9 @@ import org.springframework.transaction.NoTransactionException; import org.springframework.transaction.reactive.TransactionSynchronization; import org.springframework.transaction.reactive.TransactionSynchronizationManager; import org.springframework.util.Assert; +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; /** * Helper class that provides static methods for obtaining R2DBC Connections from a @@ -41,6 +39,7 @@ import org.springframework.util.Assert; * objects. Can also be used directly in application code. * * @author Mark Paluch + * @author Christoph Strobl */ public abstract class ConnectionFactoryUtils { @@ -52,7 +51,6 @@ public abstract class ConnectionFactoryUtils { private static final Log logger = LogFactory.getLog(ConnectionFactoryUtils.class); private ConnectionFactoryUtils() { - } /** @@ -93,14 +91,20 @@ public abstract class ConnectionFactoryUtils { if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) { conHolder.requested(); if (!conHolder.hasConnection()) { - logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); + + 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. - logger.debug("Fetching R2DBC Connection from ConnectionFactory"); + if (logger.isDebugEnabled()) { + logger.debug("Fetching R2DBC Connection from ConnectionFactory"); + } + Mono con = fetchConnection(connectionFactory); if (synchronizationManager.isSynchronizationActive()) { @@ -124,11 +128,9 @@ public abstract class ConnectionFactoryUtils { 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)); - }); }); } @@ -158,7 +160,9 @@ public abstract class ConnectionFactoryUtils { if (synchronization.isSynchronizationActive()) { - logger.debug("Registering transaction synchronization for R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Registering transaction synchronization for R2DBC Connection"); + } TransactionResources txContext = synchronization.getCurrentTransaction(); ConnectionFactory resource = txContext.getResource(ConnectionFactory.class); @@ -166,7 +170,9 @@ public abstract class ConnectionFactoryUtils { Mono> attachNewConnection = Mono .defer(() -> Mono.from(connectionFactory.create()).map(it -> { - logger.debug("Fetching new R2DBC Connection from ConnectionFactory"); + if (logger.isDebugEnabled()) { + logger.debug("Fetching new R2DBC Connection from ConnectionFactory"); + } SingletonConnectionFactory s = new SingletonConnectionFactory(connectionFactory.getMetadata(), it); txContext.registerResource(ConnectionFactory.class, s); @@ -174,14 +180,9 @@ public abstract class ConnectionFactoryUtils { return Tuples.of(it, connectionFactory); })); - return Mono.justOrEmpty(resource).flatMap(factory -> { - - logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); - - return Mono.from(factory.create()) - .map(connection -> Tuples. of(connection, factory)); - - }).switchIfEmpty(attachNewConnection); + return Mono.justOrEmpty(resource) + .flatMap(ConnectionFactoryUtils::createConnection) + .switchIfEmpty(attachNewConnection); } return Mono.empty(); @@ -199,7 +200,7 @@ public abstract class ConnectionFactoryUtils { private static Mono fetchConnection(ConnectionFactory connectionFactory) { Publisher con = connectionFactory.create(); - if (con == null) { + if (con == null) { // TODO: seriously why would it do that? throw new IllegalStateException("ConnectionFactory returned null from getConnection(): " + connectionFactory); } return Mono.from(con); @@ -211,7 +212,7 @@ public abstract class ConnectionFactoryUtils { * * @param con the {@link io.r2dbc.spi.Connection} to close if necessary. * @param connectionFactory the {@link ConnectionFactory} that the Connection was obtained from (may be - * {@literal null}). + * {@literal null}). * @see #getConnection */ public static Mono releaseConnection(@Nullable io.r2dbc.spi.Connection con, @@ -227,7 +228,7 @@ public abstract class ConnectionFactoryUtils { * * @param con the {@link io.r2dbc.spi.Connection} to close if necessary. * @param connectionFactory the {@link ConnectionFactory} that the Connection was obtained from (may be - * {@literal null}). + * {@literal null}). * @see #doGetConnection */ public static Mono doReleaseConnection(@Nullable io.r2dbc.spi.Connection con, @@ -247,12 +248,16 @@ public abstract class ConnectionFactoryUtils { SingletonConnectionFactory factory = (SingletonConnectionFactory) connectionFactory; - logger.debug("Releasing R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Releasing R2DBC Connection"); + } return factory.close(con); } - logger.debug("Closing R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Closing R2DBC Connection"); + } return Mono.from(con.close()); }); @@ -267,6 +272,7 @@ public abstract class ConnectionFactoryUtils { * @throws DataAccessResourceFailureException if the attempt to get a {@link io.r2dbc.spi.Connection} failed */ public static Mono closeConnection(Connection connection, ConnectionFactory connectionFactory) { + return doCloseConnection(connection, connectionFactory) .onErrorMap(e -> new DataAccessResourceFailureException("Failed to obtain R2DBC Connection", e)); } @@ -295,10 +301,10 @@ public abstract class ConnectionFactoryUtils { * Obtain the currently {@link ReactiveTransactionSynchronization} from the current subscriber * {@link reactor.util.context.Context}. * + * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the + * current subscription. * @see Mono#subscriberContext() * @see ReactiveTransactionSynchronization - * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the - * current subscription. */ public static Mono currentReactiveTransactionSynchronization() { @@ -312,10 +318,10 @@ public abstract class ConnectionFactoryUtils { * Obtain the currently active {@link ReactiveTransactionSynchronization} from the current subscriber * {@link reactor.util.context.Context}. * + * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the + * current subscription. * @see Mono#subscriberContext() * @see ReactiveTransactionSynchronization - * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the - * current subscription. */ public static Mono currentActiveReactiveTransactionSynchronization() { @@ -341,16 +347,8 @@ public abstract class ConnectionFactoryUtils { return true; } return false; - }).map(it -> connectionFactory).onErrorResume(NoTransactionException.class, e -> { - - return currentActiveReactiveTransactionSynchronization().map(synchronization -> { - - TransactionResources currentSynchronization = synchronization.getCurrentTransaction(); - return currentSynchronization.getResource(ConnectionFactory.class); - }).switchIfEmpty(Mono.error(new DataAccessResourceFailureException( - "Cannot extract ConnectionFactory from current TransactionContext!"))); - - }); + }).map(it -> connectionFactory) // + .onErrorResume(NoTransactionException.class, ConnectionFactoryUtils::obtainDefaultConnectionFactory); } /** @@ -359,7 +357,7 @@ public abstract class ConnectionFactoryUtils { * * @param conHolder the {@link ConnectionHolder} for the held Connection (potentially a proxy) * @param passedInCon the {@link Connection} passed-in by the user (potentially a target {@link Connection} without - * proxy) + * proxy) * @return whether the given Connections are equal * @see #getTargetConnection */ @@ -402,6 +400,7 @@ public abstract class ConnectionFactoryUtils { * @see #CONNECTION_SYNCHRONIZATION_ORDER */ private static int getConnectionSynchronizationOrder(ConnectionFactory connectionFactory) { + int order = CONNECTION_SYNCHRONIZATION_ORDER; ConnectionFactory current = connectionFactory; while (current instanceof DelegatingConnectionFactory) { @@ -411,6 +410,37 @@ public abstract class ConnectionFactoryUtils { return order; } + /** + * @param e + * @return an {@link Mono#error(Throwable) error} if not transaction present. + */ + private static Mono obtainDefaultConnectionFactory(NoTransactionException e) { + + return currentActiveReactiveTransactionSynchronization().map(synchronization -> { + + TransactionResources currentSynchronization = synchronization.getCurrentTransaction(); + return currentSynchronization.getResource(ConnectionFactory.class); + }).switchIfEmpty(Mono.error(new DataAccessResourceFailureException( + "Cannot extract ConnectionFactory from current TransactionContext!"))); + } + + /** + * Create a {@link Connection} via the given {@link ConnectionFactory#create() factory} and return a {@link Tuple2} associating the + * {@link Connection} with its creating {@link ConnectionFactory}. + * + * @param factory must not be {@literal null}. + * @return never {@literal null} + */ + private static Mono> createConnection(ConnectionFactory factory) { + + if (logger.isDebugEnabled()) { + logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); + } + + return Mono.from(factory.create()) + .map(connection -> Tuples.of(connection, factory)); + } + /** * Callback for resource cleanup at the end of a non-native R2DBC transaction. */ @@ -465,7 +495,6 @@ public abstract class ConnectionFactoryUtils { }).then(); } return Mono.empty(); - } @Override @@ -510,17 +539,14 @@ public abstract class ConnectionFactoryUtils { // Reset the ConnectionHolder: It might remain bound to the context. this.connectionHolder.setConnection(null); }); - } return Mono.empty(); }); - } this.connectionHolder.reset(); return Mono.empty(); } } - } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java index 155a79c..ed5a4e6 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, @@ -38,8 +38,8 @@ public interface ConnectionHandle { *

* The default implementation is empty, assuming that the lifecycle of the connection is managed externally. * - * @param con the R2DBC Connection to release + * @param connection the R2DBC Connection to release */ - default void releaseConnection(Connection con) {} - + default void releaseConnection(Connection connection) { + } } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java index fcc3f3b..f84f6c7 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, @@ -17,7 +17,6 @@ package org.springframework.data.r2dbc.function.connectionfactory; import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; - import org.springframework.lang.Nullable; import org.springframework.transaction.support.ResourceHolderSupport; import org.springframework.util.Assert; @@ -30,6 +29,8 @@ import org.springframework.util.Assert; *

* Note: This is an SPI class, not intended to be used by applications. * + * @author Mark Paluch + * @author Christoph Strobl * @see ConnectionFactoryTransactionManager * @see ConnectionFactoryUtils */ @@ -39,7 +40,7 @@ public class ConnectionHolder extends ResourceHolderSupport { @Nullable private Connection currentConnection; - private boolean transactionActive = false; + private boolean transactionActive; /** * Create a new ConnectionHolder for the given R2DBC {@link Connection}, wrapping it with a @@ -50,7 +51,7 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see #ConnectionHolder(Connection, boolean) */ public ConnectionHolder(Connection connection) { - this.connectionHandle = new SimpleConnectionHandle(connection); + this(connection, false); } /** @@ -62,7 +63,8 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see SimpleConnectionHandle */ public ConnectionHolder(Connection connection, boolean transactionActive) { - this(connection); + + this.connectionHandle = new SimpleConnectionHandle(connection); this.transactionActive = transactionActive; } @@ -127,6 +129,7 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see #released() */ public Connection getConnection() { + Assert.notNull(this.connectionHandle, "Active Connection is required"); if (this.currentConnection == null) { this.currentConnection = this.connectionHandle.getConnection(); @@ -152,7 +155,7 @@ public class ConnectionHolder extends ResourceHolderSupport { } } - /* + /* * (non-Javadoc) * @see org.springframework.transaction.support.ResourceHolderSupport#clear() */ diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java index 37f600c..38bd461 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2019 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. @@ -19,12 +19,13 @@ import io.r2dbc.spi.Connection; import io.r2dbc.spi.Wrapped; /** - * Subinterface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target + * Sub interface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target * Connection. *

* This interface can be checked when there is a need to cast to a native R2DBC {@link Connection}. * * @author Mark Paluch + * @author Christoph Strobl */ public interface ConnectionProxy extends Connection, Wrapped { @@ -34,6 +35,7 @@ public interface ConnectionProxy extends Connection, Wrapped { * This will typically be the native driver {@link Connection} or a wrapper from a connection pool. * * @return the underlying Connection (never {@literal null}) + * @throws IllegalStateException in case the connection has already been closed. */ Connection getTargetConnection(); } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java index f28b2c8..3b57769 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, @@ -19,7 +19,6 @@ import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; import io.r2dbc.spi.ConnectionFactoryMetadata; import io.r2dbc.spi.Wrapped; - import org.reactivestreams.Publisher; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -43,7 +42,7 @@ public class DelegatingConnectionFactory implements ConnectionFactory, Wrapped getTransactionAwareConnectionProxy(ConnectionFactory targetConnectionFactory) { + return ConnectionFactoryUtils.getConnection(targetConnectionFactory).map(TransactionAwareConnectionFactoryProxy::proxyConnection); + } - return ConnectionFactoryUtils.getConnection(targetConnectionFactory).map(tuple -> { - return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(), - new Class[] { ConnectionProxy.class }, - new TransactionAwareInvocationHandler(tuple.getT1(), targetConnectionFactory)); - }); + private static Connection proxyConnection(Tuple2 connectionConnectionFactoryTuple) { + + return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(), + new Class[]{ConnectionProxy.class}, + new TransactionAwareInvocationHandler(connectionConnectionFactoryTuple.getT1(), connectionConnectionFactoryTuple.getT2())); } /** * Invocation handler that delegates close calls on R2DBC Connections to {@link ConnectionFactoryUtils} for being * aware of context-bound transactions. */ - private class TransactionAwareInvocationHandler implements InvocationHandler { + private static class TransactionAwareInvocationHandler implements InvocationHandler { private final Connection connection; @@ -116,11 +123,12 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection private boolean closed = false; TransactionAwareInvocationHandler(Connection connection, ConnectionFactory targetConnectionFactory) { + this.connection = connection; this.targetConnectionFactory = targetConnectionFactory; } - /* + /* * (non-Javadoc) * @see java.lang.reflect.InvocationHandler#invoke(java.lang.Object, java.lang.reflect.Method, java.lang.Object[]) */ @@ -128,23 +136,24 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection @Nullable public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if (ReflectionUtils.isObjectMethod(method)) { + + if (ReflectionUtils.isToStringMethod(method)) { + return proxyToString(proxy); + } + + if (ReflectionUtils.isEqualsMethod(method)) { + return (proxy == args[0]); + } + + if (ReflectionUtils.isHashCodeMethod(method)) { + return System.identityHashCode(proxy); + } + } + // Invocation on ConnectionProxy interface coming in... switch (method.getName()) { - case "equals": - // Only considered as equal when proxies are identical. - return (proxy == args[0]); - case "hashCode": - // Use hashCode of Connection proxy. - return System.identityHashCode(proxy); - case "toString": - // Allow for differentiating between the proxy and the raw Connection. - StringBuilder sb = new StringBuilder("Transaction-aware proxy for target Connection "); - if (this.connection != null) { - sb.append("[").append(this.connection.toString()).append("]"); - } else { - sb.append(" from ConnectionFactory [").append(this.targetConnectionFactory).append("]"); - } - return sb.toString(); + case "unwrap": return this.connection; case "close": @@ -171,5 +180,17 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection throw ex.getTargetException(); } } + + private String proxyToString(@Nullable Object proxy) { + + // Allow for differentiating between the proxy and the raw Connection. + StringBuilder sb = new StringBuilder("Transaction-aware proxy for target Connection "); + if (this.connection != null) { + sb.append("[").append(this.connection.toString()).append("]"); + } else { + sb.append(" from ConnectionFactory [").append(this.targetConnectionFactory).append("]"); + } + return sb.toString(); + } } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java index 24a84cc..d47d438 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java @@ -17,33 +17,41 @@ package org.springframework.data.r2dbc.function; import static org.assertj.core.api.Assertions.*; -import io.r2dbc.spi.ConnectionFactory; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; - +import javax.sql.DataSource; import java.util.ArrayList; import java.util.List; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; -import javax.sql.DataSource; - +import io.r2dbc.spi.ConnectionFactory; +import org.junit.After; import org.junit.Before; import org.junit.Test; - +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +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.function.connectionfactory.ConnectionFactoryTransactionManager; import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.transaction.NoTransactionException; +import org.springframework.transaction.ReactiveTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.reactive.TransactionalOperator; import org.springframework.transaction.support.DefaultTransactionDefinition; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * Abstract base class for integration tests for {@link TransactionalDatabaseClient}. * * @author Mark Paluch + * @author Christoph Strobl */ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extends R2dbcIntegrationTestSupport { @@ -51,19 +59,35 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend private JdbcTemplate jdbc; + AnnotationConfigApplicationContext context; + TransactionalService service; + @Before public void before() { connectionFactory = createConnectionFactory(); + context = new AnnotationConfigApplicationContext(); + context.registerBean("theConnectionFactory", ConnectionFactory.class, () -> connectionFactory); + context.register(Config.class, TransactionalService.class); + context.refresh(); + + service = context.getBean(TransactionalService.class); + jdbc = createJdbcTemplate(createDataSource()); try { jdbc.execute("DROP TABLE legoset"); - } catch (DataAccessException e) {} + } catch (DataAccessException e) { + } jdbc.execute(getCreateTableStatement()); jdbc.execute("DELETE FROM legoset"); } + @After + public void tearDown() { + context.close(); + } + /** * Creates a {@link DataSource} to be used in this test. * @@ -97,6 +121,17 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend return "INSERT INTO legoset (id, name, manual) VALUES(:id, :name, :manual)"; } + /** + * Some Databases require special treatment to convince them to start a transaction. Some even start a transaction but + * store its id async so that it might show up a little late. + * + * @param client the client to use + * @return an empty {@link Mono} by default. + */ + protected Mono prepareForTransaction(DatabaseClient client) { + return Mono.empty(); + } + /** * Get a statement that returns the current transactionId. */ @@ -191,7 +226,8 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend .bind(0, 42055) // .bind(1, "SCHAUFELRADBAGGER") // .bindNull(2, Integer.class) // - .fetch().rowsUpdated().then(Mono.error(new IllegalStateException("failed"))); + .fetch().rowsUpdated() // + .then(Mono.error(new IllegalStateException("failed"))); }); integerFlux.as(StepVerifier::create) // @@ -202,7 +238,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend assertThat(count).isEqualTo(0); } - @Test // gh-2, gh-75 + @Test // gh-2, gh-75, gh-107 public void emitTransactionIds() { DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); @@ -210,22 +246,13 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend TransactionalOperator transactionalOperator = TransactionalOperator .create(new ConnectionFactoryTransactionManager(connectionFactory), new DefaultTransactionDefinition()); - // We have to execute a sql statement first. - // Otherwise some databases (MySql) don't have a transaction id. - Mono insert = databaseClient.execute().sql(getInsertIntoLegosetStatement()) // - .bind(0, 42055) // - .bind(1, "SCHAUFELRADBAGGER") // - .bindNull(2, Integer.class) // - .fetch().rowsUpdated(); - Flux txId = databaseClient.execute() // .sql(getCurrentTransactionIdStatement()) // .map((row, md) -> row.get(0)) // .all(); - // insert.thenMany fails because of a cancel signal. Probably a consequence of dematerialize - // in TransactionalOperator.execute. - Flux transactionIds = txId.concatWith(txId).as(transactionalOperator::transactional); + Flux transactionIds = prepareForTransaction(databaseClient).thenMany(txId.concatWith(txId)) // + .as(transactionalOperator::transactional); transactionIds.collectList().as(StepVerifier::create) // .consumeNextWith(actual -> { @@ -235,4 +262,114 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend }) // .verifyComplete(); } + + @Test // gh-107 + public void shouldRollbackTransactionUsingTransactionalOperator() { + + DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); + + TransactionalOperator transactionalOperator = TransactionalOperator + .create(new ConnectionFactoryTransactionManager(connectionFactory), new DefaultTransactionDefinition()); + + Flux integerFlux = databaseClient.execute() // + .sql(getInsertIntoLegosetStatement()) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .thenMany(Mono.fromSupplier(() -> { + throw new IllegalStateException("failed"); + })); + + integerFlux.as(transactionalOperator::transactional) // + .as(StepVerifier::create) // + .expectError(IllegalStateException.class) // + .verify(); + + Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM legoset", Integer.class); + assertThat(count).isEqualTo(0); + } + + @Test //gh-107 + public void emitTransactionIdsUsingManagedTransactions() { + + service.emitTransactionIds(prepareForTransaction(service.getDatabaseClient()), getCurrentTransactionIdStatement()).collectList().as(StepVerifier::create) // + .consumeNextWith(actual -> { + + assertThat(actual).hasSize(2); + assertThat(actual.get(0)).isEqualTo(actual.get(1)); + }) // + .verifyComplete(); + } + + @Test // gh-107 + public void shouldRollbackTransactionUsingManagedTransactions() { + + service.shouldRollbackTransactionUsingTransactionalOperator(getInsertIntoLegosetStatement()) + .as(StepVerifier::create) // + .expectError(IllegalStateException.class) // + .verify(); + + Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM legoset", Integer.class); + assertThat(count).isEqualTo(0); + } + + @Configuration + @EnableTransactionManagement + static class Config extends AbstractR2dbcConfiguration { + + @Autowired GenericApplicationContext context; + + @Override + public ConnectionFactory connectionFactory() { + return lookup(); + } + + ConnectionFactory lookup() { + return context.getBean("theConnectionFactory", ConnectionFactory.class); + } + + @Bean + ReactiveTransactionManager txMgr(ConnectionFactory connectionFactory) { + return new ConnectionFactoryTransactionManager(connectionFactory); + } + } + + static class TransactionalService { + + private DatabaseClient databaseClient; + + public TransactionalService(DatabaseClient databaseClient) { + this.databaseClient = databaseClient; + } + + @Transactional + public Flux emitTransactionIds(Mono prepareTransaction, String idStatement) { + + Flux txId = databaseClient.execute() // + .sql(idStatement) // + .map((row, md) -> row.get(0)) // + .all(); + + return prepareTransaction.thenMany(txId.concatWith(txId)); + } + + + @Transactional + public Flux shouldRollbackTransactionUsingTransactionalOperator(String insertStatement) { + + return databaseClient.execute().sql(insertStatement) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .thenMany(Mono.fromSupplier(() -> { + throw new IllegalStateException("failed"); + })); + } + + public DatabaseClient getDatabaseClient() { + return databaseClient; + } + } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java index 3efbc57..7519423 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java @@ -15,16 +15,16 @@ */ package org.springframework.data.r2dbc.function; -import io.r2dbc.spi.ConnectionFactory; - import javax.sql.DataSource; +import java.time.Duration; +import io.r2dbc.spi.ConnectionFactory; import org.junit.ClassRule; import org.junit.Ignore; import org.junit.Test; - import org.springframework.data.r2dbc.testing.ExternalDatabase; import org.springframework.data.r2dbc.testing.MySqlTestSupport; +import reactor.core.publisher.Mono; /** * Integration tests for {@link TransactionalDatabaseClient} against MySQL. @@ -51,6 +51,25 @@ public class MySqlTransactionalDatabaseClientIntegrationTests return MySqlTestSupport.CREATE_TABLE_LEGOSET; } + @Override + protected Mono prepareForTransaction(DatabaseClient client) { + + /* + * We have to execute a sql statement first. + * Otherwise MySql don't have a transaction id. + * And we need to delay emitting the result so that MySql has time to write the transaction id, which is done in + * batches every now and then. + * @see: https://dev.mysql.com/doc/refman/5.7/en/innodb-information-schema-internal-data.html + */ + return client.execute().sql(getInsertIntoLegosetStatement()) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .delayElement(Duration.ofMillis(50)) // + .then(); + } + @Override protected String getCurrentTransactionIdStatement() { return "SELECT tx.trx_id FROM information_schema.innodb_trx tx WHERE tx.trx_mysql_thread_id = connection_id()"; @@ -59,12 +78,6 @@ public class MySqlTransactionalDatabaseClientIntegrationTests @Override @Test @Ignore("MySQL creates transactions only on interaction with transactional tables. BEGIN does not create a txid") - public void shouldManageUserTransaction() {} - - @Override - @Test - @Ignore("Third element is cancelled, looks like a bug") - public void emitTransactionIds() { - super.emitTransactionIds(); + public void shouldManageUserTransaction() { } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java index ede09c9..b13fb55 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java index 002de90..b77a1a2 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java @@ -17,17 +17,20 @@ package org.springframework.data.r2dbc.function.connectionfactory; import static org.mockito.Mockito.*; +import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; -import reactor.test.StepVerifier; - +import org.assertj.core.api.Assertions; import org.junit.Test; - +import org.reactivestreams.Publisher; import org.springframework.transaction.NoTransactionException; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * Unit tests for {@link ConnectionFactoryUtils}. * * @author Mark Paluch + * @author Christoph Strobl */ public class ConnectionFactoryUtilsUnitTests { @@ -91,4 +94,21 @@ public class ConnectionFactoryUtilsUnitTests { .expectNext(factoryMock) // .verifyComplete(); } + + @Test // gh-107 + public void connectionFactoryRetunsConnectionWhenNoSyncronisationActive() { + + ConnectionFactory factoryMock = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + Publisher p = Mono.just(connection); + doReturn(p).when(factoryMock).create(); + + ConnectionFactoryUtils.getConnection(factoryMock) // + .as(StepVerifier::create) // + .consumeNextWith(it -> { + Assertions.assertThat(it.getT1()).isEqualTo(connection); + Assertions.assertThat(it.getT2()).isEqualTo(factoryMock); + }) + .verifyComplete(); + } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java index 1ea4390..d1ad075 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java index ecbdac0..6a578ee 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * 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, @@ -18,23 +18,22 @@ package org.springframework.data.r2dbc.function.connectionfactory; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; +import java.util.concurrent.atomic.AtomicReference; + import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; +import org.junit.Before; +import org.junit.Test; +import org.springframework.transaction.reactive.TransactionalOperator; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import reactor.util.function.Tuple2; -import java.util.concurrent.atomic.AtomicReference; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.transaction.reactive.TransactionalOperator; - /** * Unit tests for {@link TransactionAwareConnectionFactoryProxy}. * * @author Mark Paluch + * @author Christoph Strobl */ public class TransactionAwareConnectionFactoryProxyUnitTests { @@ -53,6 +52,96 @@ public class TransactionAwareConnectionFactoryProxyUnitTests { tm = new ConnectionFactoryTransactionManager(connectionFactoryMock); } + @Test // gh-107 + public void createShouldProxyConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .as(StepVerifier::create) // + .consumeNextWith(connection -> { + assertThat(connection).isInstanceOf(ConnectionProxy.class); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void unwrapShouldReturnTargetConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.unwrap()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void unwrapShouldReturnTargetConnectionEvenWhenClosed() { + + when(connectionMock1.close()).thenReturn(Mono.empty()); + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .flatMap(it -> Mono.from(it.close()).then(Mono.just(it))) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.unwrap()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void getTargetConnectionShouldReturnTargetConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.getTargetConnection()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void getTargetConnectionShouldThrowsErrorEvenWhenClosed() { + + when(connectionMock1.close()).thenReturn(Mono.empty()); + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .flatMap(it -> Mono.from(it.close()).then(Mono.just(it))) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> proxy.getTargetConnection()); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void hashCodeShouldReturnProxyHash() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.hashCode()).isEqualTo(System.identityHashCode(proxy)); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void equalsShouldCompareCorrectly() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.equals(proxy)).isTrue(); + assertThat(proxy.equals(connectionMock1)).isFalse(); + }) + .verifyComplete(); + } + @Test // gh-107 public void shouldEmitBoundConnection() { @@ -68,14 +157,13 @@ public class TransactionAwareConnectionFactoryProxyUnitTests { ConnectionFactoryUtils.getConnection(connectionFactoryMock).map(Tuple2::getT1) // .doOnNext(transactionalConnection::set).flatMap(it -> { - return proxyCf.create().doOnNext(connectionFromProxy -> { + return proxyCf.create().doOnNext(connectionFromProxy -> { - ConnectionProxy connectionProxy = (ConnectionProxy) connectionFromProxy; - assertThat(connectionProxy.getTargetConnection()).isSameAs(it); - assertThat(connectionProxy.unwrap()).isSameAs(it); - }); - - }).as(rxtx::transactional) // + ConnectionProxy connectionProxy = (ConnectionProxy) connectionFromProxy; + assertThat(connectionProxy.getTargetConnection()).isSameAs(it); + assertThat(connectionProxy.unwrap()).isSameAs(it); + }); + }).as(rxtx::transactional) // .flatMapMany(Connection::close) // .as(StepVerifier::create) // .verifyComplete(); diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml index c9be4b4..32090ee 100644 --- a/src/test/resources/logback.xml +++ b/src/test/resources/logback.xml @@ -10,7 +10,7 @@ - +