DATAREDIS-891 - Revise RedisConnection transaction/closure resource binding.
We now correctly consider the transactional state along with the requested bindings scope when binding RedisConnection for reuse. Previously, enabling transaction support on RedisTemplate registered the connection unconditionally with TransactionSynchronizationManager expecting a TransactionManager to close/unbind the connection upon transaction cleanup. This behavior could easily lead to lingering connections when a transactional RedisTemplate was used outside of a managed transaction. We now distinguish between closure-scope binding and transaction binding. Closure-scoped bindings (RedisTemplate.execute(SessionCallback)) do not interfere with transactional resources if there's no transactionally bound RedisConnection. Closure-scope however reuses a transactionally bound connection. Transaction-scoped binding binds only a connection to the transaction if there's an active transaction. Without an ongoing transaction, RedisTemplate does no longer bind connections to TransactionSynchronizationManager. Original Pull Request: #573
This commit is contained in:
Mark Paluch
committed by
Christoph Strobl
Christoph Strobl
parent
dc44425925
commit
1912d0c999
@@ -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}..
|
||||
* <p>
|
||||
* 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 <tt>true</tt>.
|
||||
* 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}.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* 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();
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -206,18 +206,11 @@ public class RedisTemplate<K, V> 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<K, V> 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<K, V> 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<K, V> 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;
|
||||
|
||||
Reference in New Issue
Block a user