diff --git a/src/main/asciidoc/reference/redis-transactions.adoc b/src/main/asciidoc/reference/redis-transactions.adoc index 5f3b7b582..2ab8cb3e2 100644 --- a/src/main/asciidoc/reference/redis-transactions.adoc +++ b/src/main/asciidoc/reference/redis-transactions.adoc @@ -1,9 +1,9 @@ [[tx]] = Redis Transactions -Redis provides support for https://redis.io/topics/transactions[transactions] through the `multi`, `exec`, and `discard` commands. These operations are available on `RedisTemplate`. However, `RedisTemplate` is not guaranteed to run all the operations in the transaction with the same connection. +Redis provides support for https://redis.io/topics/transactions[transactions] through the `multi`, `exec`, and `discard` commands.These operations are available on `RedisTemplate`.However, `RedisTemplate` is not guaranteed to run all the operations in the transaction with the same connection. -Spring Data Redis provides the `SessionCallback` interface for use when multiple operations need to be performed with the same `connection`, such as when using Redis transactions. The following example uses the `multi` method: +Spring Data Redis provides the `SessionCallback` interface for use when multiple operations need to be performed with the same `connection`, such as when using Redis transactions.The following example uses the `multi` method: [source,java] ---- @@ -20,14 +20,23 @@ List txResults = redisTemplate.execute(new SessionCallback> System.out.println("Number of items added to set: " + txResults.get(0)); ---- -`RedisTemplate` uses its value, hash key, and hash value serializers to deserialize all results of `exec` before returning. There is an additional `exec` method that lets you pass a custom serializer for transaction results. +`RedisTemplate` uses its value, hash key, and hash value serializers to deserialize all results of `exec` before returning.There is an additional `exec` method that lets you pass a custom serializer for transaction results. include::version-note.adoc[] [[tx.spring]] == @Transactional Support -By default, transaction Support is disabled and has to be explicitly enabled for each `RedisTemplate` in use by setting `setEnableTransactionSupport(true)`. Doing so forces binding the current `RedisConnection` to the current `Thread` that is triggering `MULTI`. If the transaction finishes without errors, `EXEC` is called. Otherwise `DISCARD` is called. Once in `MULTI`, `RedisConnection` queues write operations. All `readonly` operations, such as `KEYS`, are piped to a fresh (non-thread-bound) `RedisConnection`. +By default, `RedisTemplate` does not participate in managed Spring transactions. +If you want `RedisTemplate` to make use of Redis transaction when using `@Transactional` or `TransactionTemplate`, you need to be explicitly enable transaction support for each `RedisTemplate` by setting `setEnableTransactionSupport(true)`. +Enabling transaction support binds `RedisConnection` to the current transaction backed by a `ThreadLocal`. +If the transaction finishes without errors, the Redis transaction gets commited with `EXEC`, otherwise rolled back with `DISCARD`. +Redis transactions are batch-oriented. +Commands issued during an ongoing transaction are queued and only applied when committing the transaction. + +Spring Data Redis distinguishes between read-only and write commands in an ongoing transaction. +Read-only commands, such as `KEYS`, are piped to a fresh (non-thread-bound) `RedisConnection` to allow reads. +Write commands are queued by `RedisTemplate` and applied upon commit. The following example shows how to configure transaction management: @@ -65,7 +74,7 @@ public class RedisTxContextConfiguration { ---- <1> Configures a Spring Context to enable https://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/data-access.html#transaction-declarative[declarative transaction management]. <2> Configures `RedisTemplate` to participate in transactions by binding connections to the current thread. -<3> Transaction management requires a `PlatformTransactionManager`. Spring Data Redis does not ship with a `PlatformTransactionManager` implementation. Assuming your application uses JDBC, Spring Data Redis can participate in transactions by using existing transaction managers. +<3> Transaction management requires a `PlatformTransactionManager`.Spring Data Redis does not ship with a `PlatformTransactionManager` implementation.Assuming your application uses JDBC, Spring Data Redis can participate in transactions by using existing transaction managers. ==== The following examples each demonstrate a usage constraint: diff --git a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java index 26844776c..379fe3f24 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java +++ b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java @@ -22,44 +22,55 @@ import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.aop.RawTargetAccess; import org.springframework.aop.framework.ProxyFactory; -import org.springframework.cglib.proxy.MethodProxy; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.lang.Nullable; -import org.springframework.transaction.support.ResourceHolder; +import org.springframework.transaction.support.ResourceHolderSupport; import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; /** - * Helper class featuring {@link RedisConnection} handling, allowing for reuse of instances within - * 'transactions'/scopes. + * Helper class that provides static methods for obtaining {@link RedisConnection} from a + * {@link RedisConnectionFactory}. Includes special support for Spring-managed transactional RedisConnections, e.g. + * managed by {@link org.springframework.transaction.support.AbstractPlatformTransactionManager}.. + *

+ * Used internally by Spring's {@link RedisTemplate}. Can also be used directly in application code. * * @author Costin Leau * @author Christoph Strobl * @author Thomas Darimont * @author Mark Paluch + * @see #getConnection + * @see #releaseConnection + * @see org.springframework.transaction.support.TransactionSynchronizationManager */ public abstract class RedisConnectionUtils { private static final Log log = LogFactory.getLog(RedisConnectionUtils.class); /** - * Binds a new Redis connection (from the given factory) to the current thread, if none is already bound. + * Obtain a {@link RedisConnection} from the given {@link RedisConnectionFactory} and binds the connection to the + * current thread to be used in closure-scope, if none is already bound. Considers ongoing transactions by reusing the + * transaction-bound connection and allows reentrant connection retrieval. Does not bind the connection to potentially + * ongoing transactions. * * @param factory connection factory * @return a new Redis connection without transaction support. */ public static RedisConnection bindConnection(RedisConnectionFactory factory) { - return bindConnection(factory, false); + return doGetConnection(factory, true, true, false); } /** - * Binds a new Redis connection (from the given factory) to the current thread, if none is already bound and enables - * transaction support if {@code transactionSupport} is set to {@literal true}. + * Obtain a {@link RedisConnection} from the given {@link RedisConnectionFactory} and binds the connection to the + * current thread to be used in closure-scope, if none is already bound. Considers ongoing transactions by reusing the + * transaction-bound connection and allows reentrant connection retrieval. Binds also the connection to the ongoing + * transaction if no connection is already bound if {@code transactionSupport} is enabled. * * @param factory connection factory. * @param transactionSupport whether transaction support is enabled. @@ -70,9 +81,9 @@ public abstract class RedisConnectionUtils { } /** - * Gets a Redis connection from the given factory. Is aware of and will return any existing corresponding connections - * bound to the current thread, for example when using a transaction manager. Will always create a new connection - * otherwise. + * Obtain a {@link RedisConnection} from the given {@link RedisConnectionFactory}. Is aware of existing connections + * bound to the current transaction (when using a transaction manager) or the current thread (when binding a + * connection to a closure-scope). Does not bind newly created connections to ongoing transactions. * * @param factory connection factory for creating the connection. * @return an active Redis connection without transaction management. @@ -82,9 +93,9 @@ public abstract class RedisConnectionUtils { } /** - * Gets a Redis connection from the given factory. Is aware of and will return any existing corresponding connections - * bound to the current thread, for example when using a transaction manager. Will always create a new connection - * otherwise. + * Obtain a {@link RedisConnection} from the given {@link RedisConnectionFactory}. Is aware of existing connections + * bound to the current transaction (when using a transaction manager) or the current thread (when binding a + * connection to a closure-scope). * * @param factory connection factory for creating the connection. * @param transactionSupport whether transaction support is enabled. @@ -95,9 +106,10 @@ public abstract class RedisConnectionUtils { } /** - * Gets a Redis connection. Is aware of and will return any existing corresponding connections bound to the current - * thread, for example when using a transaction manager. Will create a new Connection otherwise, if - * {@code allowCreate} is true. + * Actually obtain a {@link RedisConnection} from the given {@link RedisConnectionFactory}. Is aware of existing + * connections bound to the current transaction (when using a transaction manager) or the current thread (when binding + * a connection to a closure-scope). Will create a new {@link RedisConnection} otherwise, if {@code allowCreate} is + * {@literal true}. This method allows for re-entrance as {@link RedisConnectionHolder} keeps track of ref-count. * * @param factory connection factory for creating the connection. * @param allowCreate whether a new (unbound) connection should be created when no connection can be found for the @@ -111,59 +123,95 @@ public abstract class RedisConnectionUtils { Assert.notNull(factory, "No RedisConnectionFactory specified"); - RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); + RedisConnectionHolder conHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); - if (connHolder != null) { - if (transactionSupport) { - potentiallyRegisterTransactionSynchronisation(connHolder, factory); + if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) { + conHolder.requested(); + if (!conHolder.hasConnection()) { + log.debug("Fetching resumed Redis Connection from RedisConnectionFactory"); + conHolder.setConnection(fetchConnection(factory)); } - return connHolder.getConnection(); + return conHolder.getRequiredConnection(); } + // Else we either got no holder or an empty thread-bound holder here. + if (!allowCreate) { throw new IllegalArgumentException("No connection found and allowCreate = false"); } - if (log.isDebugEnabled()) { - log.debug("Opening RedisConnection"); - } + log.debug("Fetching Redis Connection from RedisConnectionFactory"); + RedisConnection connection = fetchConnection(factory); - RedisConnection conn = factory.getConnection(); + boolean bindSynchronization = TransactionSynchronizationManager.isActualTransactionActive() && transactionSupport; - if (bind) { + if (bind || bindSynchronization) { - RedisConnection connectionToBind = conn; - if (transactionSupport && isActualNonReadonlyTransactionActive()) { - connectionToBind = createConnectionProxy(conn, factory); + if (bindSynchronization && isActualNonReadonlyTransactionActive()) { + connection = createConnectionSplittingProxy(connection, factory); } - connHolder = new RedisConnectionHolder(connectionToBind); + try { + // Use same RedisConnection for further Redis actions within the transaction. + // Thread-bound object will get removed by synchronization at transaction completion. + RedisConnectionHolder holderToUse = conHolder; + if (holderToUse == null) { + holderToUse = new RedisConnectionHolder(connection); + } else { + holderToUse.setConnection(connection); + } + holderToUse.requested(); - TransactionSynchronizationManager.bindResource(factory, connHolder); - if (transactionSupport) { - potentiallyRegisterTransactionSynchronisation(connHolder, factory); + // Consider callback-scope connection binding vs. transaction scope binding + if (bindSynchronization) { + potentiallyRegisterTransactionSynchronisation(holderToUse, factory); + } + + if (holderToUse != conHolder) { + TransactionSynchronizationManager.bindResource(factory, holderToUse); + } + } catch (RuntimeException ex) { + // Unexpected exception from external delegation call -> close Connection and rethrow. + releaseConnection(connection, factory); + throw ex; } - return connHolder.getConnection(); + return connection; } - return conn; + return connection; + } + + /** + * Actually create a {@link RedisConnection} from the given {@link RedisConnectionFactory}. + * + * @param factory the {@link RedisConnectionFactory} to obtain RedisConnections from. + * @return a Redis Connection from the given {@link RedisConnectionFactory} (never {@literal null}). + * @see RedisConnectionFactory#getConnection() + */ + private static RedisConnection fetchConnection(RedisConnectionFactory factory) { + return factory.getConnection(); } private static void potentiallyRegisterTransactionSynchronisation(RedisConnectionHolder connHolder, final RedisConnectionFactory factory) { - if (isActualNonReadonlyTransactionActive()) { + // Should go actually into RedisTransactionManager - if (!connHolder.isTransactionSyncronisationActive()) { - connHolder.setTransactionSyncronisationActive(true); + if (!connHolder.isTransactionActive()) { - RedisConnection conn = connHolder.getConnection(); + connHolder.setTransactionActive(true); + connHolder.setSynchronizedWithTransaction(true); + connHolder.requested(); + + RedisConnection conn = connHolder.getRequiredConnection(); + boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly(); + if (!readOnly) { conn.multi(); - - TransactionSynchronizationManager - .registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn, factory)); } + + TransactionSynchronizationManager + .registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn, factory, readOnly)); } } @@ -172,94 +220,144 @@ public abstract class RedisConnectionUtils { && !TransactionSynchronizationManager.isCurrentTransactionReadOnly(); } - private static RedisConnection createConnectionProxy(RedisConnection connection, RedisConnectionFactory factory) { + private static RedisConnection createConnectionSplittingProxy(RedisConnection connection, + RedisConnectionFactory factory) { ProxyFactory proxyFactory = new ProxyFactory(connection); proxyFactory.addAdvice(new ConnectionSplittingInterceptor(factory)); + proxyFactory.addInterface(RedisConnectionProxy.class); return RedisConnection.class.cast(proxyFactory.getProxy()); } /** - * Closes the given connection, created via the given factory if not managed externally (i.e. not bound to the - * thread). + * Closes the given {@link RedisConnection}, created via the given factory if not managed externally (i.e. not bound + * to the transaction). * * @param conn the Redis connection to close. * @param factory the Redis factory that the connection was created with. - * @deprecated since 2.1.10, use {@link #releaseConnection(RedisConnection, RedisConnectionFactory, boolean)} instead. */ - @Deprecated public static void releaseConnection(@Nullable RedisConnection conn, RedisConnectionFactory factory) { - releaseConnection(conn, factory, false); + if (conn == null) { + return; + } + + RedisConnectionHolder conHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); + if (conHolder != null) { + + if (conHolder.isTransactionActive()) { + if (connectionEquals(conHolder, conn)) { + if (log.isDebugEnabled()) { + log.debug("RedisConnection will be closed when transaction finished."); + } + + // It's the transactional Connection: Don't close it. + conHolder.released(); + } + return; + } + + // release transactional/read-only and non-transactional/non-bound connections. + // transactional connections for read-only transactions get no synchronizer registered + + unbindConnection(factory); + return; + } + + doCloseConnection(conn); } /** - * Closes the given connection, created via the given factory if not managed externally (i.e. not bound to the - * thread). + * Closes the given {@link RedisConnection}, created via the given factory if not managed externally (i.e. not bound + * to the transaction). * * @param conn the Redis connection to close. * @param factory the Redis factory that the connection was created with. * @param transactionSupport whether transaction support is enabled. * @since 2.1.9 + * @deprecated since 2.4.0, use {@link #releaseConnection(RedisConnection, RedisConnectionFactory)} */ + @Deprecated public static void releaseConnection(@Nullable RedisConnection conn, RedisConnectionFactory factory, boolean transactionSupport) { - - if (conn == null) { - return; - } - - RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); - - if (connHolder != null && connHolder.isTransactionSyncronisationActive()) { - if (log.isDebugEnabled()) { - log.debug("Redis Connection will be closed when transaction finished."); - } - return; - } - - if (isConnectionTransactional(conn, factory)) { - - // release transactional/read-only and non-transactional/non-bound connections. - // transactional connections for read-only transactions get no synchronizer registered - if (transactionSupport && TransactionSynchronizationManager.isCurrentTransactionReadOnly()) { - if (log.isDebugEnabled()) { - log.debug("Unbinding Redis Connection."); - } - unbindConnection(factory); - } else { - - // Not participating in transaction management. - // Connection could have been attached via session callback. - if (log.isDebugEnabled()) { - log.debug("Leaving bound Redis Connection attached."); - } - } - } else { - doCloseConnection(conn); - } + releaseConnection(conn, factory); } /** - * Unbinds and closes the connection (if any) associated with the given factory. + * Determine whether the given two RedisConnections are equal, asking the target {@link RedisConnection} 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 RedisConnectionHolder} for the held Connection (potentially a proxy) + * @param passedInCon the {@link RedisConnection} passed-in by the user (potentially a target Connection without + * proxy) + * @return whether the given Connections are equal + * @see #getTargetConnection + */ + private static boolean connectionEquals(RedisConnectionHolder conHolder, RedisConnection passedInCon) { + + if (!conHolder.hasConnection()) { + return false; + } + + RedisConnection heldCon = conHolder.getRequiredConnection(); + + return heldCon.equals(passedInCon) || getTargetConnection(heldCon).equals(passedInCon); + } + + /** + * Return the innermost target {@link RedisConnection} of the given {@link RedisConnection}. If the given + * {@link RedisConnection} is a proxy, it will be unwrapped until a non-proxy {@link RedisConnection} is found. + * Otherwise, the passed-in {@link RedisConnection} will be returned as-is. + * + * @param con the {@link RedisConnection} proxy to unwrap + * @return the innermost target Connection, or the passed-in one if no proxy + * @see RedisConnectionProxy#getTargetConnection() + */ + private static RedisConnection getTargetConnection(RedisConnection con) { + + RedisConnection conToUse = con; + + while (conToUse instanceof RedisConnectionProxy) { + conToUse = ((RedisConnectionProxy) conToUse).getTargetConnection(); + } + + return conToUse; + } + + /** + * Unbinds and closes the connection (if any) associated with the given factory from closure-scope. Considers ongoing + * transactions so transaction-bound connections aren't closed and reentrant closure-scope bound connections. Only the + * outer-most call to leads to releasing and closing the connection. * * @param factory Redis factory */ public static void unbindConnection(RedisConnectionFactory factory) { - RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager - .unbindResourceIfPossible(factory); + RedisConnectionHolder conHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); - if (connHolder == null) { + if (conHolder == null) { return; } - if (connHolder.isTransactionSyncronisationActive()) { + if (log.isDebugEnabled()) { + log.debug("Unbinding Redis Connection."); + } + + if (conHolder.isTransactionActive()) { if (log.isDebugEnabled()) { log.debug("Redis Connection will be closed when outer transaction finished."); } } else { - doCloseConnection(connHolder.getConnection()); + + RedisConnection connection = conHolder.getConnection(); + conHolder.released(); + + if (!conHolder.isOpen()) { + + TransactionSynchronizationManager.unbindResourceIfPossible(factory); + + doCloseConnection(connection); + } } } @@ -278,10 +376,14 @@ public abstract class RedisConnectionUtils { RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager .getResource(connFactory); - return (connHolder != null && conn == connHolder.getConnection()); + return connHolder != null && connectionEquals(connHolder, conn); } - private static void doCloseConnection(RedisConnection connection) { + private static void doCloseConnection(@Nullable RedisConnection connection) { + + if (connection == null) { + return; + } if (log.isDebugEnabled()) { log.debug("Closing Redis Connection."); @@ -297,40 +399,45 @@ public abstract class RedisConnectionUtils { } /** - * A {@link TransactionSynchronization} that makes sure that the associated RedisConnection is released after the - * transaction completes. + * A {@link TransactionSynchronization} that makes sure that the associated {@link RedisConnection} is released after + * the transaction completes. * * @author Christoph Strobl * @author Thomas Darimont + * @author Mark Paluch */ private static class RedisTransactionSynchronizer implements TransactionSynchronization { private final RedisConnectionHolder connHolder; private final RedisConnection connection; private final RedisConnectionFactory factory; + private final boolean readOnly; RedisTransactionSynchronizer(RedisConnectionHolder connHolder, RedisConnection connection, - RedisConnectionFactory factory) { + RedisConnectionFactory factory, boolean readOnly) { this.connHolder = connHolder; this.connection = connection; this.factory = factory; + this.readOnly = readOnly; } @Override public void afterCompletion(int status) { try { - switch (status) { + if (!readOnly) { + switch (status) { - case TransactionSynchronization.STATUS_COMMITTED: - connection.exec(); - break; + case TransactionSynchronization.STATUS_COMMITTED: + connection.exec(); + break; - case TransactionSynchronization.STATUS_ROLLED_BACK: - case TransactionSynchronization.STATUS_UNKNOWN: - default: - connection.discard(); + case TransactionSynchronization.STATUS_ROLLED_BACK: + case TransactionSynchronization.STATUS_UNKNOWN: + default: + connection.discard(); + } } } finally { @@ -338,19 +445,24 @@ public abstract class RedisConnectionUtils { log.debug("Closing bound connection after transaction completed with " + status); } - connHolder.setTransactionSyncronisationActive(false); + connHolder.setTransactionActive(false); doCloseConnection(connection); TransactionSynchronizationManager.unbindResource(factory); + connHolder.reset(); } } } /** + * {@link MethodInterceptor} that invokes read-only commands on a new {@link RedisConnection} while read-write + * commands are queued on the bound connection. + * * @author Christoph Strobl + * @author Mark Paluch * @since 1.3 */ static class ConnectionSplittingInterceptor - implements MethodInterceptor, org.springframework.cglib.proxy.MethodInterceptor { + implements MethodInterceptor { private final RedisConnectionFactory factory; @@ -359,7 +471,16 @@ public abstract class RedisConnectionUtils { } @Override - public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable { + public Object invoke(MethodInvocation invocation) throws Throwable { + return intercept(invocation.getThis(), invocation.getMethod(), invocation.getArguments()); + } + + public Object intercept(Object obj, Method method, Object[] args) throws Throwable { + + if (method.getName().equals("getTargetConnection")) { + // Handle getTargetConnection method: return underlying RedisConnection. + return obj; + } RedisCommand commandToExecute = RedisCommand.failsafeCommandLookup(method.getName()); @@ -397,10 +518,6 @@ public abstract class RedisConnectionUtils { } } - @Override - public Object invoke(MethodInvocation invocation) throws Throwable { - return intercept(invocation.getThis(), invocation.getMethod(), invocation.getArguments(), null); - } private boolean isPotentiallyThreadBoundCommand(RedisCommand command) { return RedisCommand.UNKNOWN.equals(command) || !command.isReadonly(); @@ -408,48 +525,117 @@ public abstract class RedisConnectionUtils { } /** + * Resource holder wrapping a {@link RedisConnection}. {@link RedisConnectionUtils} binds instances of this class to + * the thread, for a specific {@link RedisConnectionFactory}. + * * @author Christoph Strobl + * @author Mark Paluch */ - private static class RedisConnectionHolder implements ResourceHolder { + private static class RedisConnectionHolder extends ResourceHolderSupport { - private boolean unbound; - private final RedisConnection conn; - private boolean transactionSyncronisationActive; + @Nullable private RedisConnection connection; - public RedisConnectionHolder(RedisConnection conn) { - this.conn = conn; + private boolean transactionActive = false; + + /** + * Create a new RedisConnectionHolder for the given Redis Connection assuming that there is no ongoing transaction. + * + * @param connection the Redis Connection to hold. + * @see #RedisConnectionHolder(RedisConnection, boolean) + */ + public RedisConnectionHolder(RedisConnection connection) { + this.connection = connection; } - public boolean isVoid() { - return unbound; + /** + * Return whether this holder currently has a {@link RedisConnection}. + */ + protected boolean hasConnection() { + return (this.connection != null); } + @Nullable public RedisConnection getConnection() { - return conn; + return connection; } - public void reset() { - // no-op - } + public RedisConnection getRequiredConnection() { - public void unbound() { - this.unbound = true; + RedisConnection connection = getConnection(); + + if (connection == null) { + throw new IllegalStateException("No active RedisConnection"); + } + + return connection; } /** - * @return - * @since 1.3 + * Override the existing {@link RedisConnection} handle with the given {@link RedisConnection}. Reset the handle if + * given {@literal null}. + *

+ * Used for releasing the Connection on suspend (with a {@code null} argument) and setting a fresh Connection on + * resume. */ - public boolean isTransactionSyncronisationActive() { - return transactionSyncronisationActive; + protected void setConnection(@Nullable RedisConnection connection) { + this.connection = connection; } /** - * @param transactionSyncronisationActive - * @since 1.3 + * Set whether this holder represents an active, managed transaction. + * + * @see org.springframework.transaction.PlatformTransactionManager */ - public void setTransactionSyncronisationActive(boolean transactionSyncronisationActive) { - this.transactionSyncronisationActive = transactionSyncronisationActive; + protected void setTransactionActive(boolean transactionActive) { + this.transactionActive = transactionActive; + } + + /** + * Return whether this holder represents an active, managed transaction. + */ + protected boolean isTransactionActive() { + return this.transactionActive; + } + + /** + * Releases the current Connection held by this ConnectionHolder. + *

+ * This is necessary for ConnectionHandles that expect "Connection borrowing", where each returned Connection is + * only temporarily leased and needs to be returned once the data operation is done, to make the Connection + * available for other operations within the same transaction. + */ + @Override + public void released() { + super.released(); + if (!isOpen()) { + setConnection(null); + } + } + + @Override + public void clear() { + super.clear(); + this.transactionActive = false; } } + + /** + * Subinterface of {@link RedisConnection} to be implemented by {@link RedisConnection} proxies. Allows access to the + * underlying target {@link RedisConnection}. + * + * @since 2.4 + * @see RedisConnectionUtils#getTargetConnection(RedisConnection) + */ + interface RedisConnectionProxy extends RedisConnection, RawTargetAccess { + + /** + * Return the target {@link RedisConnection} of this proxy. + *

+ * This will typically be the native driver {@link RedisConnection} or a wrapper from a connection pool. + * + * @return the underlying {@link RedisConnection} (never {@link null}). + */ + RedisConnection getTargetConnection(); + + } } diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index d7baa8d3e..181573ea9 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -206,18 +206,11 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation Assert.notNull(action, "Callback object must not be null"); RedisConnectionFactory factory = getRequiredConnectionFactory(); - RedisConnection conn = null; + RedisConnection conn = RedisConnectionUtils.getConnection(factory, enableTransactionSupport); + try { - if (enableTransactionSupport) { - // only bind resources in case of potential transaction synchronization - conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); - } else { - conn = RedisConnectionUtils.getConnection(factory); - } - boolean existingConnection = TransactionSynchronizationManager.hasResource(factory); - RedisConnection connToUse = preProcessConnection(conn, existingConnection); boolean pipelineStatus = connToUse.isPipelined(); @@ -233,7 +226,6 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation connToUse.closePipeline(); } - // TODO: any other connection processing? return postProcessResult(result, connToUse, existingConnection); } finally { RedisConnectionUtils.releaseConnection(conn, factory, enableTransactionSupport); @@ -383,10 +375,12 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation return session.execute(this); } - protected RedisConnection createRedisConnectionProxy(RedisConnection pm) { - Class[] ifcs = ClassUtils.getAllInterfacesForClass(pm.getClass(), getClass().getClassLoader()); - return (RedisConnection) Proxy.newProxyInstance(pm.getClass().getClassLoader(), ifcs, - new CloseSuppressingInvocationHandler(pm)); + protected RedisConnection createRedisConnectionProxy(RedisConnection connection) { + + Class[] ifcs = ClassUtils.getAllInterfacesForClass(connection.getClass(), getClass().getClassLoader()); + + return (RedisConnection) Proxy.newProxyInstance(connection.getClass().getClassLoader(), ifcs, + new CloseSuppressingInvocationHandler(connection)); } /** @@ -1354,10 +1348,13 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation } /** - * If set to {@code true} {@link RedisTemplate} will use {@literal MULTI...EXEC|DISCARD} to keep track of operations. + * If set to {@code true} {@link RedisTemplate} will participate in ongoing transactions using + * {@literal MULTI...EXEC|DISCARD} to keep track of operations. * - * @param enableTransactionSupport + * @param enableTransactionSupport whether to participate in ongoing transactions. * @since 1.3 + * @see RedisConnectionUtils#getConnection(RedisConnectionFactory, boolean) + * @see TransactionSynchronizationManager#isActualTransactionActive() */ public void setEnableTransactionSupport(boolean enableTransactionSupport) { this.enableTransactionSupport = enableTransactionSupport; diff --git a/src/test/java/org/springframework/data/redis/core/ConnectionSplittingInterceptorUnitTests.java b/src/test/java/org/springframework/data/redis/core/ConnectionSplittingInterceptorUnitTests.java index 7c085b662..079d8f0e1 100644 --- a/src/test/java/org/springframework/data/redis/core/ConnectionSplittingInterceptorUnitTests.java +++ b/src/test/java/org/springframework/data/redis/core/ConnectionSplittingInterceptorUnitTests.java @@ -69,7 +69,7 @@ class ConnectionSplittingInterceptorUnitTests { @Test // DATAREDIS-73 void interceptorShouldRequestFreshConnectionForReadonlyCommand() throws Throwable { - interceptor.intercept(boundConnectionMock, READONLY_METHOD, new Object[] { new byte[] {} }, null); + interceptor.intercept(boundConnectionMock, READONLY_METHOD, new Object[] { new byte[] {} }); verify(connectionFactoryMock, times(1)).getConnection(); verifyZeroInteractions(boundConnectionMock); } @@ -77,7 +77,7 @@ class ConnectionSplittingInterceptorUnitTests { @Test // DATAREDIS-73 void interceptorShouldUseBoundConnectionForWriteOperations() throws Throwable { - interceptor.intercept(boundConnectionMock, WRITE_METHOD, new Object[] { new byte[] {}, 0L }, null); + interceptor.intercept(boundConnectionMock, WRITE_METHOD, new Object[] { new byte[] {}, 0L }); verify(boundConnectionMock, times(1)).expire(any(byte[].class), anyLong()); verifyZeroInteractions(connectionFactoryMock); } @@ -90,7 +90,7 @@ class ConnectionSplittingInterceptorUnitTests { InvalidDataAccessApiUsageException.class); Assertions.assertThatExceptionOfType(InvalidDataAccessApiUsageException.class).isThrownBy( - () -> interceptor.intercept(boundConnectionMock, READONLY_METHOD, new Object[] { new byte[] {} }, null)); + () -> interceptor.intercept(boundConnectionMock, READONLY_METHOD, new Object[] { new byte[] {} })); } } diff --git a/src/test/java/org/springframework/data/redis/core/RedisConnectionUtilsUnitTests.java b/src/test/java/org/springframework/data/redis/core/RedisConnectionUtilsUnitTests.java index 72718445c..985741938 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisConnectionUtilsUnitTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisConnectionUtilsUnitTests.java @@ -15,12 +15,21 @@ */ package org.springframework.data.redis.core; +import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionException; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; /** * Unit tests for {@link RedisConnectionUtils}. @@ -29,16 +38,188 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; */ class RedisConnectionUtilsUnitTests { + RedisConnection connectionMock1 = mock(RedisConnection.class); + RedisConnection connectionMock2 = mock(RedisConnection.class); + RedisConnectionFactory factoryMock = mock(RedisConnectionFactory.class); + + @BeforeEach + void setUp() { + + // cleanup to avoid lingering resources + TransactionSynchronizationManager.clear(); + + if (TransactionSynchronizationManager.isSynchronizationActive()) { + TransactionSynchronizationManager.clearSynchronization(); + } + + when(factoryMock.getConnection()).thenReturn(connectionMock1, connectionMock2); + } + @Test // DATAREDIS-1104 void shouldSilentlyCloseRedisConnection() { - RedisConnection connectionMock = mock(RedisConnection.class); - RedisConnectionFactory factoryMock = mock(RedisConnectionFactory.class); + Mockito.reset(connectionMock1); + doThrow(new IllegalStateException()).when(connectionMock1).close(); + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock, false); - doThrow(new IllegalStateException()).when(connectionMock).close(); + verify(connectionMock1).close(); + } - RedisConnectionUtils.releaseConnection(connectionMock, factoryMock, false); + @Test // DATAREDIS-891 + void bindConnectionShouldBindConnectionToClosureScope() { - verify(connectionMock).close(); + assertThat(RedisConnectionUtils.bindConnection(factoryMock)).isSameAs(connectionMock1); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + + assertThat(RedisConnectionUtils.getConnection(factoryMock)).isSameAs(connectionMock1); + + RedisConnectionUtils.unbindConnection(factoryMock); + verifyNoInteractions(connectionMock1); + + RedisConnectionUtils.unbindConnection(factoryMock); + verify(connectionMock1).close(); + + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + @Test // DATAREDIS-891 + void getConnectionShouldBindConnectionToTransactionScopeWithReadOnlyTransaction() { + + TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager()); + template.setReadOnly(true); + + template.executeWithoutResult(status -> { + + assertThat(RedisConnectionUtils.getConnection(factoryMock, true)).isSameAs(connectionMock1); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + assertThat(RedisConnectionUtils.getConnection(factoryMock)).isSameAs(connectionMock1); + + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + + verifyNoInteractions(connectionMock1); + }); + + verify(connectionMock1).close(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + @Test // DATAREDIS-891 + void bindConnectionShouldBindConnectionToOngoingTransactionScope() { + + TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager()); + + template.executeWithoutResult(status -> { + + assertThat(RedisConnectionUtils.bindConnection(factoryMock, true)).isNotNull(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + assertThat(RedisConnectionUtils.getConnection(factoryMock)).isNotNull(); + + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + + verify(connectionMock1).multi(); + verifyNoMoreInteractions(connectionMock1); + }); + + verify(connectionMock1).close(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + @Test // DATAREDIS-891 + void bindConnectionShouldNotBindConnectionToTransactionWithoutTransaction() { + + assertThat(RedisConnectionUtils.bindConnection(factoryMock, true)).isNotNull(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + assertThat(RedisConnectionUtils.getConnection(factoryMock)).isNotNull(); + + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + + verify(connectionMock1).close(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + @Test // DATAREDIS-891 + void getConnectionShouldBindConnectionToTransactionScopeWithReadWriteTransaction() { + + TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager()); + + template.executeWithoutResult(status -> { + + assertThat(RedisConnectionUtils.getConnection(factoryMock, true)).isNotNull(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + + verify(connectionMock1).multi(); + verify(factoryMock, times(1)).getConnection(); + }); + + verify(connectionMock1).close(); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + @Test // DATAREDIS-891 + void getConnectionShouldNotBindConnectionToTransaction() { + + TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager()); + + template.executeWithoutResult(status -> { + + assertThat(RedisConnectionUtils.getConnection(factoryMock)).isSameAs(connectionMock1); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + + RedisConnectionUtils.releaseConnection(connectionMock1, factoryMock); + + verify(factoryMock).getConnection(); + verify(connectionMock1).close(); + }); + + verifyNoMoreInteractions(factoryMock); + verifyNoMoreInteractions(connectionMock1); + } + + @Test // DATAREDIS-891 + void bindConnectionShouldNotBindConnectionToTransaction() { + + TransactionTemplate template = new TransactionTemplate(new DummyTransactionManager()); + + template.executeWithoutResult(status -> { + + assertThat(RedisConnectionUtils.bindConnection(factoryMock)).isSameAs(connectionMock1); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isTrue(); + + RedisConnectionUtils.unbindConnection(factoryMock); + + verify(connectionMock1).close(); + verify(factoryMock, times(1)).getConnection(); + }); + + verifyNoMoreInteractions(factoryMock); + assertThat(TransactionSynchronizationManager.hasResource(factoryMock)).isFalse(); + } + + static class DummyTransactionManager extends AbstractPlatformTransactionManager { + + @Override + protected Object doGetTransaction() throws TransactionException { + return new Object(); + } + + @Override + protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException { + + } + + @Override + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { + + } + + @Override + protected void doRollback(DefaultTransactionStatus status) throws TransactionException { + + } } } diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java index 0093199fc..a05a1725e 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java @@ -122,8 +122,6 @@ class RedisTemplateUnitTests { @Test // DATAREDIS-988 void executeSessionInTransactionShouldReuseConnection() { - TransactionSynchronizationManager.setCurrentTransactionReadOnly(true); - template.execute(new SessionCallback() { @Override public Object execute(RedisOperations operations) throws DataAccessException { @@ -142,12 +140,11 @@ class RedisTemplateUnitTests { void transactionAwareTemplateShouldReleaseConnection() { template.setEnableTransactionSupport(true); - TransactionSynchronizationManager.setCurrentTransactionReadOnly(true); - - template.execute(new SessionCallback() { + template.execute(new RedisCallback() { + @Nullable @Override - public Object execute(RedisOperations operations) throws DataAccessException { + public Object doInRedis(RedisConnection connection) throws DataAccessException { template.multi(); template.multi(); @@ -155,8 +152,8 @@ class RedisTemplateUnitTests { } }); - verify(connectionFactoryMock, times(2)).getConnection(); - verify(redisConnectionMock, times(2)).close(); + verify(connectionFactoryMock, times(3)).getConnection(); + verify(redisConnectionMock, times(3)).close(); } private static class SomeArbitrarySerializableObject implements Serializable {