diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java new file mode 100644 index 000000000..ae0507f3b --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java @@ -0,0 +1,64 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.codec.RedisCodec; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; + +/** + * Connection provider for Cluster connections. + * + * @author Mark Paluch + * @since 2.0 + */ +class ClusterConnectionProvider implements LettuceConnectionProvider { + + private final RedisClusterClient client; + private final RedisCodec codec; + + ClusterConnectionProvider(RedisClusterClient client, RedisCodec codec) { + + this.client = client; + this.codec = codec; + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + */ + @SuppressWarnings("rawtypes") + @Override + public StatefulConnection getConnection(Class connectionType) { + + if (connectionType.equals(StatefulRedisPubSubConnection.class)) { + return client.connectPubSub(codec); + } + + if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType) + || connectionType.equals(StatefulConnection.class)) { + return client.connect(codec); + } + + throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!"); + } + + public RedisClusterClient getClient() { + return client; + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java index 1e7aacfa0..757f97a90 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java @@ -17,15 +17,12 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.RedisException; import io.lettuce.core.RedisURI; -import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.sync.BaseRedisCommands; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.SlotHash; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; -import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.codec.RedisCodec; -import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; +import lombok.RequiredArgsConstructor; import java.time.Duration; import java.util.ArrayList; @@ -60,11 +57,9 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau static final ExceptionTranslationStrategy exceptionConverter = new PassThroughExceptionTranslationStrategy( new LettuceExceptionConverter()); - static final RedisCodec CODEC = ByteArrayCodec.INSTANCE; private final Log log = LogFactory.getLog(getClass()); - private final RedisClusterClient clusterClient; private ClusterCommandExecutor clusterCommandExecutor; private ClusterTopologyProvider topologyProvider; private boolean disposeClusterCommandExecutorOnClose; @@ -77,12 +72,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau * @param clusterClient must not be {@literal null}. */ public LettuceClusterConnection(RedisClusterClient clusterClient) { - - this(clusterClient, RedisURI.DEFAULT_TIMEOUT_DURATION, - new ClusterCommandExecutor(new LettuceClusterTopologyProvider(clusterClient), - new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter)); - - this.disposeClusterCommandExecutorOnClose = true; + this(new ClusterConnectionProvider(clusterClient, CODEC)); } /** @@ -93,7 +83,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau * @param executor must not be {@literal null}. */ public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommandExecutor executor) { - this(clusterClient, RedisURI.DEFAULT_TIMEOUT_DURATION, executor); + this(clusterClient, executor, RedisURI.DEFAULT_TIMEOUT_DURATION); } /** @@ -105,16 +95,64 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau * @param executor must not be {@literal null}. * @since 2.0 */ - public LettuceClusterConnection(RedisClusterClient clusterClient, Duration timeout, ClusterCommandExecutor executor) { + public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommandExecutor executor, Duration timeout) { + this(new ClusterConnectionProvider(clusterClient, CODEC), executor, timeout); + } - super(null, timeout.toMillis(), clusterClient, null, 0); + /** + * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the + * cluster via given {@link ClusterCommandExecutor}. + * + * @param connectionProvider must not be {@literal null}. + * @since 2.0 + */ + public LettuceClusterConnection(LettuceConnectionProvider connectionProvider) { + + super(null, connectionProvider, RedisURI.DEFAULT_TIMEOUT_DURATION.toMillis(), 0); + + topologyProvider = new LettuceClusterTopologyProvider(getClient()); + clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + new LettuceClusterNodeResourceProvider(getConnectionProvider()), exceptionConverter); + disposeClusterCommandExecutorOnClose = true; + } + + /** + * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the + * cluster via given {@link ClusterCommandExecutor}. + * + * @param connectionProvider must not be {@literal null}. + * @param executor must not be {@literal null}. + * @since 2.0 + */ + public LettuceClusterConnection(LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor) { + this(connectionProvider, executor, RedisURI.DEFAULT_TIMEOUT_DURATION); + } + + /** + * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the + * cluster via given {@link ClusterCommandExecutor}. + * + * @param connectionProvider must not be {@literal null}. + * @param executor must not be {@literal null}. + * @since 2.0 + */ + public LettuceClusterConnection(LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor, + Duration timeout) { + + super(null, connectionProvider, timeout.toMillis(), 0); - Assert.notNull(clusterClient, "RedisClusterClient must not be null."); Assert.notNull(executor, "ClusterCommandExecutor must not be null."); - this.clusterClient = clusterClient; - this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient); + this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); this.clusterCommandExecutor = executor; + this.disposeClusterCommandExecutorOnClose = false; + } + + /** + * @return access to {@link RedisClusterClient} for non-connection access. + */ + private RedisClusterClient getClient() { + return ((ClusterConnectionProvider) getConnectionProvider()).getClient(); } /* @@ -235,7 +273,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau @Override public RedisClusterNode clusterGetNodeForSlot(int slot) { - return LettuceConverters.toRedisClusterNode(clusterClient.getPartitions().getPartitionBySlot(slot)); + return LettuceConverters.toRedisClusterNode(getClient().getPartitions().getPartitionBySlot(slot)); } /* @@ -459,15 +497,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau } } - /* - * (non-Javadoc) - * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#getAsyncDedicatedConnection() - */ - @Override - protected StatefulConnection doGetAsyncDedicatedConnection() { - return clusterClient.connect(CODEC); - } - // --> cluster node stuff /* @@ -476,7 +505,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau */ @Override public List clusterGetNodes() { - return LettuceConverters.partitionsToClusterNodes(clusterClient.getPartitions()); + return LettuceConverters.partitionsToClusterNodes(getClient().getPartitions()); } /* @@ -527,19 +556,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau return result; } - /* - * (non-Javadoc) - * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#switchToPubSub() - */ - @Override - protected StatefulRedisPubSubConnection switchToPubSub() { - - close(); - - // open a pubsub one - return clusterClient.connectPubSub(CODEC); - } - public ClusterCommandExecutor getClusterCommandExecutor() { return clusterCommandExecutor; } @@ -588,16 +604,12 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau * @author Christoph Strobl * @since 1.7 */ + @RequiredArgsConstructor static class LettuceClusterNodeResourceProvider implements ClusterNodeResourceProvider, DisposableBean { - private final RedisClusterClient client; + private final LettuceConnectionProvider connectionProvider; private volatile StatefulRedisClusterConnection connection; - public LettuceClusterNodeResourceProvider(RedisClusterClient client) { - - this.client = client; - } - @Override @SuppressWarnings("unchecked") public RedisClusterCommands getResourceForSpecificNode(RedisClusterNode node) { @@ -607,7 +619,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau if (connection == null) { synchronized (this) { if (connection == null) { - this.connection = client.connect(CODEC); + this.connection = connectionProvider.getConnection(); } } } @@ -631,7 +643,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau @Override public void destroy() throws Exception { if (connection != null) { - connection.close(); + connectionProvider.release(connection); } } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 2481334e8..624218216 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -40,6 +40,7 @@ import io.lettuce.core.protocol.CommandArgs; import io.lettuce.core.protocol.CommandType; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection; +import lombok.RequiredArgsConstructor; import java.lang.reflect.Constructor; import java.util.ArrayList; @@ -60,9 +61,9 @@ import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.dao.QueryTimeoutException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.FallbackExceptionTranslationStrategy; -import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.connection.*; import org.springframework.data.redis.connection.convert.TransactionResultConverter; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware; import org.springframework.data.redis.core.RedisCommand; import org.springframework.data.redis.core.ScanOptions; import org.springframework.util.Assert; @@ -92,6 +93,7 @@ public class LettuceConnection extends AbstractRedisConnection { private final int defaultDbIndex; private int dbIndex; + private final LettuceConnectionProvider connectionProvider; private final StatefulConnection asyncSharedConn; private StatefulConnection asyncDedicatedConn; @@ -103,11 +105,8 @@ public class LettuceConnection extends AbstractRedisConnection { private boolean isPipelined = false; private List ppline; private Queue> txResults = new LinkedList<>(); - private AbstractRedisClient client; private volatile LettuceSubscription subscription; - private LettucePool pool; /** flag indicating whether the connection needs to be dropped or not */ - private boolean broken = false; private boolean convertPipelineAndTxResults = true; @SuppressWarnings("rawtypes") @@ -258,7 +257,7 @@ public class LettuceConnection extends AbstractRedisConnection { /** * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be - * used for transactions or blocking operations + * used for transactions or blocking operations. * @param timeout The connection timeout (in milliseconds) * @param client The {@link RedisClient} to use when making pub/sub connections. * @param pool The connection pool to use for blocking and tx operations. @@ -268,22 +267,40 @@ public class LettuceConnection extends AbstractRedisConnection { public LettuceConnection(StatefulRedisConnection sharedConnection, long timeout, AbstractRedisClient client, LettucePool pool, int defaultDbIndex) { + if (pool != null) { + this.connectionProvider = new LettucePoolConnectionProvider(pool); + } else { + this.connectionProvider = new StandaloneConnectionProvider((RedisClient) client, CODEC); + } + this.asyncSharedConn = sharedConnection; this.timeout = timeout; - this.client = client; - this.pool = pool; + this.defaultDbIndex = defaultDbIndex; + this.dbIndex = this.defaultDbIndex; + } + + /** + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be + * used for transactions or blocking operations. + * @param connectionProvider connection provider to obtain and release native connections. + * @param timeout The connection timeout (in milliseconds) + * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. + * @since 2.0 + */ + public LettuceConnection(StatefulRedisConnection sharedConnection, + LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) { + + Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null."); + + this.asyncSharedConn = sharedConnection; + this.connectionProvider = connectionProvider; + this.timeout = timeout; this.defaultDbIndex = defaultDbIndex; this.dbIndex = this.defaultDbIndex; } protected DataAccessException convertLettuceAccessException(Exception ex) { - - DataAccessException exception = EXCEPTION_TRANSLATION.translate(ex); - - if (exception instanceof RedisConnectionFailureException) { - broken = true; - } - return exception; + return EXCEPTION_TRANSLATION.translate(ex); } /* @@ -435,34 +452,22 @@ public class LettuceConnection extends AbstractRedisConnection { } } - private void returnDedicatedAsyncConnection() { - - if (pool != null) { - - if (!broken) { - pool.returnResource(this.asyncDedicatedConn); - } else { - pool.returnBrokenResource(this.asyncDedicatedConn); - } - this.asyncDedicatedConn = null; - - } else { - - try { - asyncDedicatedConn.close(); - } catch (RuntimeException ex) { - throw convertLettuceAccessException(ex); - } - } - } - public void close() throws DataAccessException { + super.close(); + if (isClosed) { + return; + } + isClosed = true; if (asyncDedicatedConn != null) { - returnDedicatedAsyncConnection(); + try { + connectionProvider.release(asyncDedicatedConn); + } catch (RuntimeException ex) { + throw convertLettuceAccessException(ex); + } } if (subscription != null) { @@ -803,10 +808,11 @@ public class LettuceConnection extends AbstractRedisConnection { * * @return never {@literal null}. */ + @SuppressWarnings("unchecked") protected StatefulRedisPubSubConnection switchToPubSub() { close(); - return ((RedisClient) client).connectPubSub(CODEC); + return (StatefulRedisPubSubConnection) connectionProvider.getConnection(StatefulRedisPubSubConnection.class); } void pipeline(LettuceResult result) { @@ -852,15 +858,13 @@ public class LettuceConnection extends AbstractRedisConnection { } protected RedisClusterAsyncCommands getAsyncDedicatedConnection() { + if (asyncDedicatedConn == null) { asyncDedicatedConn = doGetAsyncDedicatedConnection(); - if (this.pool == null) { - - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); - } + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); } } @@ -875,26 +879,14 @@ public class LettuceConnection extends AbstractRedisConnection { String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); } - protected StatefulConnection doGetAsyncDedicatedConnection() { - - if (this.pool != null) { - return pool.getResource(); - } else { - return ((RedisClient) client).connect(CODEC); - } - } - RedisClusterCommands getDedicatedConnection() { if (asyncDedicatedConn == null) { asyncDedicatedConn = doGetAsyncDedicatedConnection(); - if (this.pool == null) { - - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); - } + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); } } @@ -909,6 +901,10 @@ public class LettuceConnection extends AbstractRedisConnection { String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); } + protected StatefulConnection doGetAsyncDedicatedConnection() { + return connectionProvider.getConnection(); + } + io.lettuce.core.ScanCursor getScanCursor(long cursorId) { return io.lettuce.core.ScanCursor.of(Long.toString(cursorId)); } @@ -957,15 +953,15 @@ public class LettuceConnection extends AbstractRedisConnection { return false; } - StatefulRedisConnection connection = null; + StatefulRedisSentinelConnection connection = null; try { - connection = ((RedisClient) client).connect(getRedisURI(node)); + connection = getConnection(node); return connection.sync().ping().equalsIgnoreCase("pong"); } catch (Exception e) { return false; } finally { if (connection != null) { - connection.close(); + connectionProvider.release(connection); } } } @@ -980,11 +976,21 @@ public class LettuceConnection extends AbstractRedisConnection { */ @Override protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) { - StatefulRedisSentinelConnection connection = ((RedisClient) client) - .connectSentinel(getRedisURI(sentinel)); + + StatefulRedisSentinelConnection connection = getConnection(sentinel); return new LettuceSentinelConnection(connection); } + @SuppressWarnings("unchecked") + private StatefulRedisSentinelConnection getConnection(RedisNode sentinel) { + return (StatefulRedisSentinelConnection) ((TargetAware) connectionProvider) + .getConnection(StatefulRedisSentinelConnection.class, getRedisURI(sentinel)); + } + + LettuceConnectionProvider getConnectionProvider() { + return connectionProvider; + } + /** * {@link TypeHints} provide {@link CommandOutput} information for a given {@link CommandType}. * @@ -1185,4 +1191,27 @@ public class LettuceConnection extends AbstractRedisConnection { return BeanUtils.instantiateClass(constructor, CODEC); } } + + @RequiredArgsConstructor + private class LettucePoolConnectionProvider implements LettuceConnectionProvider { + + private final LettucePool pool; + + @Override + @SuppressWarnings("unchecked") + public StatefulConnection getConnection(Class connectionType) { + return pool.getResource(); + } + + @Override + @SuppressWarnings("unchecked") + public void release(StatefulConnection connection) { + + if (connection.isOpen()) { + pool.returnResource((StatefulConnection) connection); + } else { + pool.returnBrokenResource((StatefulConnection) connection); + } + } + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 4050aef85..c9c6023db 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -23,6 +23,7 @@ import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.cluster.ClusterClientOptions; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.resource.ClientResources; import java.time.Duration; @@ -82,6 +83,8 @@ public class LettuceConnectionFactory private final LettuceClientConfiguration clientConfiguration; private AbstractRedisClient client; + private LettuceConnectionProvider connectionProvider; + private LettuceConnectionProvider reactiveConnectionProvider; private boolean validateConnection = false; private boolean shareNativeConnection = true; private StatefulRedisConnection connection; @@ -213,7 +216,18 @@ public class LettuceConnectionFactory * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() { - this.client = createRedisClient(); + + this.client = createClient(); + + this.connectionProvider = createConnectionProvider(client, LettuceConnection.CODEC); + this.reactiveConnectionProvider = createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC); + + if (isClusterAware()) { + this.clusterCommandExecutor = new ClusterCommandExecutor( + new LettuceClusterTopologyProvider((RedisClusterClient) client), + new LettuceClusterConnection.LettuceClusterNodeResourceProvider(this.connectionProvider), + EXCEPTION_TRANSLATION); + } } /* @@ -224,6 +238,17 @@ public class LettuceConnectionFactory resetConnection(); + if (connectionProvider instanceof DisposableBean) { + try { + ((DisposableBean) connectionProvider).destroy(); + } catch (Exception e) { + + if (log.isWarnEnabled()) { + log.warn(connectionProvider + " did not shut down gracefully.", e); + } + } + } + try { Duration timeout = clientConfiguration.getShutdownTimeout(); client.shutdown(timeout.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS); @@ -255,8 +280,14 @@ public class LettuceConnectionFactory return getClusterConnection(); } - LettuceConnection connection = new LettuceConnection(getSharedConnection(), getTimeout(), client, pool, - getDatabase()); + LettuceConnection connection; + if (pool != null) { + + connection = new LettuceConnection(getSharedConnection(), getTimeout(), null, pool, getDatabase()); + } else { + connection = new LettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), getDatabase()); + } + connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults); return connection; } @@ -272,8 +303,8 @@ public class LettuceConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } - return new LettuceClusterConnection((RedisClusterClient) client, clientConfiguration.getCommandTimeout(), - clusterCommandExecutor); + return new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, + clientConfiguration.getCommandTimeout()); } /* @@ -282,7 +313,7 @@ public class LettuceConnectionFactory */ @Override public LettuceReactiveRedisConnection getReactiveConnection() { - return new LettuceReactiveRedisConnection(client); + return new LettuceReactiveRedisConnection(reactiveConnectionProvider); } /* @@ -295,7 +326,7 @@ public class LettuceConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } - return new LettuceReactiveRedisClusterConnection((RedisClusterClient) client); + return new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider); } public void initConnection() { @@ -314,7 +345,7 @@ public class LettuceConnectionFactory public void resetConnection() { synchronized (this.connectionMonitor) { if (this.connection != null) { - this.connection.close(); + this.connectionProvider.release(this.connection); } this.connection = null; } @@ -338,6 +369,8 @@ public class LettuceConnectionFactory } if (!valid) { + + connectionProvider.release(connection); log.warn("Validation of shared connection failed. Creating a new connection."); initConnection(); } @@ -729,9 +762,9 @@ public class LettuceConnectionFactory protected StatefulRedisConnection createLettuceConnector() { try { - StatefulRedisConnection connection = null; - if (client instanceof RedisClient) { - connection = ((RedisClient) client).connect(LettuceConnection.CODEC); + StatefulRedisConnection connection; + if (!isClusterAware()) { + connection = connectionProvider.getConnection(); if (getDatabase() > 0) { connection.sync().select(getDatabase()); } @@ -744,7 +777,16 @@ public class LettuceConnectionFactory } } - private AbstractRedisClient createRedisClient() { + private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec codec) { + + if (isClusterAware()) { + return new ClusterConnectionProvider((RedisClusterClient) client, codec); + } + + return new StandaloneConnectionProvider((RedisClient) client, codec); + } + + private AbstractRedisClient createClient() { if (isRedisSentinelAware()) { @@ -768,9 +810,6 @@ public class LettuceConnectionFactory .map(clientResources -> RedisClusterClient.create(clientResources, initialUris)) // .orElseGet(() -> RedisClusterClient.create(initialUris)); - this.clusterCommandExecutor = new ClusterCommandExecutor(new LettuceClusterTopologyProvider(clusterClient), - new LettuceClusterConnection.LettuceClusterNodeResourceProvider(clusterClient), EXCEPTION_TRANSLATION); - clientConfiguration.getClientOptions() // .filter(clientOptions -> clientOptions instanceof ClusterClientOptions) // .ifPresent(clientOptions -> clusterClient.setOptions((ClusterClientOptions) clientOptions)); @@ -778,15 +817,12 @@ public class LettuceConnectionFactory return clusterClient; } - if (pool != null) { - return pool.getClient(); - } - RedisURI uri = createRedisURIAndApplySettings(getHostName(), getPort()); RedisClient redisClient = clientConfiguration.getClientResources() // .map(clientResources -> RedisClient.create(clientResources, uri)) // .orElseGet(() -> RedisClient.create(uri)); clientConfiguration.getClientOptions().ifPresent(redisClient::setOptions); + return redisClient; } @@ -817,10 +853,11 @@ public class LettuceConnectionFactory @Override public RedisSentinelConnection getSentinelConnection() { - if (!(client instanceof RedisClient)) { - throw new InvalidDataAccessResourceUsageException("Unable to connect to sentinels using " + client.getClass()); + if (!(connectionProvider instanceof StandaloneConnectionProvider)) { + throw new InvalidDataAccessResourceUsageException( + "Unable to connect to sentinels using " + connectionProvider.getClass()); } - return new LettuceSentinelConnection(((RedisClient) client).connectSentinel()); + return new LettuceSentinelConnection(connectionProvider); } private MutableLettuceClientConfiguration getMutableConfiguration() { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionProvider.java new file mode 100644 index 000000000..e81ce1d21 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionProvider.java @@ -0,0 +1,88 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; + +/** + * Defines a provider for Lettuce connections. + *

+ * This interface is typically used to encapsulate a native factory which returns a {@link StatefulConnection + * connection} of on each invocation. + *

+ * Connection providers may create a new connection on each invocation or return pooled instances. Each obtained + * connection must be released through its connection provider to allow disposal or release back to the pool. + *

+ * Connection providers are usually associated with a {@link io.lettuce.core.codec.RedisCodec} to create connections + * with an appropriate codec. + * + * @author Mark Paluch + * @since 2.0 + * @see StatefulConnection + */ +@FunctionalInterface +public interface LettuceConnectionProvider { + + /** + * Request a connection given {@code connectionType}. Providing a connection type allows specialization to provide a + * more specific connection type. + * + * @param connectionType must not be {@literal null}. + * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no + * longer in use. + */ + @SuppressWarnings("rawtypes") + StatefulConnection getConnection(Class connectionType); + + /** + * Request a connection. + * + * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no + * longer in use. + */ + @SuppressWarnings("unchecked") + default T getConnection() { + return (T) getConnection(StatefulConnection.class); + } + + /** + * Release the {@link StatefulConnection connection}. Closes connection {@link StatefulConnection#close()} by default. + * Implementations may choose whether they override this method and return the connection to a pool. + * + * @param connection must not be {@literal null}. + */ + default void release(StatefulConnection connection) { + connection.close(); + } + + /** + * Extension to {@link LettuceConnectionProvider} for providers that allow connection creation to specific nodes. + */ + interface TargetAware { + + /** + * Request a connection given {@code connectionType} for a specific {@link RedisURI}. Providing a connection type + * allows specialization to provide a more specific connection type. + * + * @param connectionType must not be {@literal null}. + * @param redisURI must not be {@literal null}. + * @return the requested connection. + */ + @SuppressWarnings("rawtypes") + StatefulConnection getConnection(Class connectionType, RedisURI redisURI); + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java index c00dd596b..72ac35a94 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java @@ -52,11 +52,12 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti * @throws org.springframework.dao.InvalidDataAccessResourceUsageException when {@code client} is not suitable for * cluster environment. */ - LettuceReactiveRedisClusterConnection(RedisClusterClient client) { + LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider) { - super(client); + super(connectionProvider); - this.topologyProvider = new LettuceClusterTopologyProvider(client); + this.topologyProvider = new LettuceClusterTopologyProvider( + ((ClusterConnectionProvider) connectionProvider).getClient()); } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java index 19fa06227..05274debf 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java @@ -15,12 +15,9 @@ */ package org.springframework.data.redis.connection.lettuce; -import io.lettuce.core.AbstractRedisClient; -import io.lettuce.core.RedisClient; import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; -import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; import io.lettuce.core.codec.RedisCodec; @@ -43,29 +40,25 @@ import org.springframework.util.Assert; */ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { - private static final RedisCodec CODEC = ByteBufferCodec.INSTANCE; + static final RedisCodec CODEC = ByteBufferCodec.INSTANCE; + + private final LettuceConnectionProvider connectionProvider; private StatefulConnection connection; /** * Creates new {@link LettuceReactiveRedisConnection}. * - * @param client must not be {@literal null}. + * @param connectionProvider must not be {@literal null}. * @throws IllegalArgumentException when {@code client} is {@literal null}. * @throws InvalidDataAccessResourceUsageException when {@code client} is not suitable for connection. */ - LettuceReactiveRedisConnection(AbstractRedisClient client) { + LettuceReactiveRedisConnection(LettuceConnectionProvider connectionProvider) { - Assert.notNull(client, "RedisClient must not be null!"); + Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); - if (client instanceof RedisClient) { - connection = ((RedisClient) client).connect(CODEC); - } else if (client instanceof RedisClusterClient) { - connection = ((RedisClusterClient) client).connect(CODEC); - } else { - throw new InvalidDataAccessResourceUsageException( - String.format("Cannot use client of type %s", client.getClass())); - } + this.connectionProvider = connectionProvider; + this.connection = connectionProvider.getConnection(); } /* @@ -189,7 +182,11 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { */ @Override public void close() { - connection.close(); + + synchronized (connectionProvider) { + connectionProvider.release(connection); + connection = null; + } } protected StatefulConnection getConnection() { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java index 576d1271a..9cb622f65 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java @@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI.Builder; +import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.resource.ClientResources; import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection; import io.lettuce.core.sentinel.api.sync.RedisSentinelCommands; @@ -42,12 +43,12 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy( LettuceConverters.exceptionConverter()); - private RedisClient redisClient; + private final LettuceConnectionProvider provider; private StatefulRedisSentinelConnection connection; /** * Creates a {@link LettuceSentinelConnection} with a dedicated client for a supplied {@link RedisNode}. - * + * * @param sentinel The sentinel to connect to. */ public LettuceSentinelConnection(RedisNode sentinel) { @@ -56,7 +57,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { /** * Creates a {@link LettuceSentinelConnection} with a client for the supplied {@code host} and {@code port}. - * + * * @param host must not be {@literal null}. * @param port sentinel port. */ @@ -64,14 +65,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host."); - redisClient = RedisClient.create(Builder.redis(host, port).build()); + this.provider = new DedicatedClientConnectionProvider(host, port); init(); } /** * Creates a {@link LettuceSentinelConnection} with a client for the supplied {@code host} and {@code port} and reuse * existing {@link ClientResources}. - * + * * @param host must not be {@literal null}. * @param port sentinel port. * @param clientResources must not be {@literal null}. @@ -81,31 +82,45 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { Assert.notNull(clientResources, "Cannot create LettuceSentinelConnection using 'null' as ClientResources."); Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host."); - redisClient = RedisClient.create(clientResources, Builder.redis(host, port).build()); + this.provider = new DedicatedClientConnectionProvider(host, port, clientResources); init(); } /** * Creates a {@link LettuceSentinelConnection} using a supplied {@link RedisClient}. - * - * @param redisClient + * + * @param redisClient must not be {@literal null}. */ public LettuceSentinelConnection(RedisClient redisClient) { Assert.notNull(redisClient, "Cannot create LettuceSentinelConnection using 'null' as client."); - this.redisClient = redisClient; + this.provider = t -> redisClient.connectSentinel(); init(); } /** * Creates a {@link LettuceSentinelConnection} using a supplied redis connection. - * + * * @param connection native Lettuce connection, must not be {@literal null} */ protected LettuceSentinelConnection(StatefulRedisSentinelConnection connection) { Assert.notNull(connection, "Cannot create LettuceSentinelConnection using 'null' as connection."); - this.connection = connection; + this.provider = t -> connection; + init(); + } + + /** + * Creates a {@link LettuceSentinelConnection} using a {@link LettuceConnectionProvider}. + * + * @param connectionProvider must not be {@literal null}. + * @since 2.0 + */ + public LettuceSentinelConnection(LettuceConnectionProvider connectionProvider) { + + Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); + this.provider = connectionProvider; + init(); } /* @@ -202,17 +217,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { */ @Override public void close() throws IOException { - connection.close(); - connection = null; - - if (redisClient != null) { - redisClient.shutdown(); - } + provider.release(connection); } + @SuppressWarnings({ "unchecked", "rawtypes" }) private void init() { + if (connection == null) { - connection = connectSentinel(); + connection = (StatefulRedisSentinelConnection) provider.getConnection(StatefulRedisSentinelConnection.class); } } @@ -220,12 +232,52 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { return connection.sync(); } - private StatefulRedisSentinelConnection connectSentinel() { - return redisClient.connectSentinel(); - } - @Override public boolean isOpen() { return connection != null && connection.isOpen(); } + + /** + * {@link LettuceConnectionProvider} for a dedicated client instance. + */ + private static class DedicatedClientConnectionProvider implements LettuceConnectionProvider { + + private final RedisClient redisClient; + + DedicatedClientConnectionProvider(String host, int port) { + + Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host."); + + redisClient = RedisClient.create(Builder.redis(host, port).build()); + } + + DedicatedClientConnectionProvider(String host, int port, ClientResources clientResources) { + + Assert.notNull(clientResources, "Cannot create LettuceSentinelConnection using 'null' as ClientResources."); + Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host."); + + redisClient = RedisClient.create(clientResources, Builder.redis(host, port).build()); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + */ + @SuppressWarnings("rawtypes") + @Override + public StatefulConnection getConnection(Class connectionType) { + return redisClient.connectSentinel(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection) + */ + @Override + public void release(StatefulConnection connection) { + + connection.close(); + redisClient.shutdown(); + } + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/StandaloneConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/StandaloneConnectionProvider.java new file mode 100644 index 000000000..502a1860f --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StandaloneConnectionProvider.java @@ -0,0 +1,87 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.codec.RedisCodec; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; +import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection; + +import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware; + +/** + * @author Mark Paluch + * @since 2.0 + */ +class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetAware { + + private final RedisClient client; + private final RedisCodec codec; + + StandaloneConnectionProvider(RedisClient client, RedisCodec codec) { + + this.client = client; + this.codec = codec; + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + */ + @SuppressWarnings("rawtypes") + @Override + public StatefulConnection getConnection(Class connectionType) { + + if (connectionType.equals(StatefulRedisSentinelConnection.class)) { + return client.connectSentinel(); + } + + if (connectionType.equals(StatefulRedisPubSubConnection.class)) { + return client.connectPubSub(codec); + } + + if (StatefulConnection.class.isAssignableFrom(connectionType)) { + return client.connect(codec); + } + + throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!"); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnection(java.lang.Class, io.lettuce.core.RedisURI) + */ + @SuppressWarnings("rawtypes") + @Override + public StatefulConnection getConnection(Class connectionType, RedisURI redisURI) { + + if (connectionType.equals(StatefulRedisSentinelConnection.class)) { + return client.connectSentinel(redisURI); + } + + if (connectionType.equals(StatefulRedisPubSubConnection.class)) { + return client.connectPubSub(codec, redisURI); + } + + if (StatefulConnection.class.isAssignableFrom(connectionType)) { + return client.connect(codec, redisURI); + } + + throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!"); + } +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTestsBase.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTestsBase.java index a71b1f09d..dd4f2b8d3 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTestsBase.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterCommandsTestsBase.java @@ -29,6 +29,7 @@ import org.springframework.data.redis.test.util.LettuceRedisClusterClientProvide /** * @author Christoph Strobl + * @author Mark Paluch */ public abstract class LettuceReactiveClusterCommandsTestsBase { @@ -41,7 +42,8 @@ public abstract class LettuceReactiveClusterCommandsTestsBase { public void before() { assumeThat(clientProvider.test(), is(true)); nativeCommands = clientProvider.getClient().connect().sync(); - connection = new LettuceReactiveRedisClusterConnection(clientProvider.getClient()); + connection = new LettuceReactiveRedisClusterConnection( + new ClusterConnectionProvider(clientProvider.getClient(), LettuceReactiveRedisConnection.CODEC)); } @After @@ -63,5 +65,4 @@ public abstract class LettuceReactiveClusterCommandsTestsBase { connection.close(); } } - } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java index bcd0430b6..0849846b3 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java @@ -39,6 +39,7 @@ import org.springframework.data.redis.test.util.LettuceRedisClusterClientProvide /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) public abstract class LettuceReactiveCommandsTestsBase { @@ -100,11 +101,13 @@ public abstract class LettuceReactiveCommandsTestsBase { if (abstractRedisClient instanceof RedisClient) { nativeCommands = ((RedisClient) abstractRedisClient).connect().sync(); - connection = new LettuceReactiveRedisConnection(abstractRedisClient); + connection = new LettuceReactiveRedisConnection( + new StandaloneConnectionProvider(((RedisClient) abstractRedisClient), LettuceReactiveRedisConnection.CODEC)); } else if (abstractRedisClient instanceof RedisClusterClient) { nativeCommands = ((RedisClusterClient) abstractRedisClient).connect().sync(); - connection = new LettuceReactiveRedisClusterConnection((RedisClusterClient) abstractRedisClient); + connection = new LettuceReactiveRedisClusterConnection(new ClusterConnectionProvider( + ((RedisClusterClient) abstractRedisClient), LettuceReactiveRedisConnection.CODEC)); } }