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 index f087afd1a..ae2a0555c 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/ClusterConnectionProvider.java @@ -23,6 +23,7 @@ import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -40,6 +41,10 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien private final RedisCodec codec; private final Optional readFrom; + private final Object monitor = new Object(); + + private volatile boolean initialized; + /** * Create new {@link ClusterConnectionProvider}. * @@ -70,25 +75,39 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) */ @Override - public > T getConnection(Class connectionType) { + public > CompletableFuture getConnectionAsync(Class connectionType) { + + if (!initialized) { + + // partitions have to be initialized before asynchronous usage. + // Needs to happen only once. Initialize eagerly if + // blocking is not an options. + synchronized (monitor) { + if (!initialized) { + client.getPartitions(); + initialized = true; + } + } + } if (connectionType.equals(StatefulRedisPubSubConnection.class)) { - return connectionType.cast(client.connectPubSub(codec)); + return client.connectPubSubAsync(codec).thenApply(connectionType::cast); } if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType) || connectionType.equals(StatefulConnection.class)) { - StatefulRedisClusterConnection connection = client.connect(codec); - readFrom.ifPresent(connection::setReadFrom); - - return connectionType.cast(connection); + return client.connectAsync(codec).thenApply(conn -> { + readFrom.ifPresent(conn::setReadFrom); + return connectionType.cast(conn); + }); } - throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!"); + return LettuceFutureUtils + .failed(new UnsupportedOperationException("Connection type " + connectionType + " not supported!")); } /* 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 973c7fde5..0bf4ef96b 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 @@ -49,6 +49,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Queue; +import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -1253,11 +1254,28 @@ public class LettuceConnection extends AbstractRedisConnection { private final LettucePool pool; + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + */ @Override public > T getConnection(Class connectionType) { return connectionType.cast(pool.getResource()); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) + */ + @Override + public > CompletionStage getConnectionAsync(Class connectionType) { + throw new UnsupportedOperationException("Async operations not supported!"); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection) + */ @Override @SuppressWarnings("unchecked") public void release(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 957f294c0..3c12bdf5a 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 @@ -101,6 +101,7 @@ public class LettuceConnectionFactory private @Nullable LettuceConnectionProvider reactiveConnectionProvider; private boolean validateConnection = false; private boolean shareNativeConnection = true; + private boolean eagerInitialization = false; private @Nullable SharedConnection connection; private @Nullable SharedConnection reactiveConnection; private @Nullable LettucePool pool; @@ -280,6 +281,10 @@ public class LettuceConnectionFactory new LettuceClusterConnection.LettuceClusterNodeResourceProvider(this.connectionProvider), EXCEPTION_TRANSLATION); } + + if (getEagerInitialization() && getShareNativeConnection()) { + initConnection(); + } } /* @@ -629,6 +634,30 @@ public class LettuceConnectionFactory this.shareNativeConnection = shareNativeConnection; } + /** + * Indicates {@link #setShareNativeConnection(boolean) shared connections} should be eagerly initialized. Eager + * initialization requires a running Redis instance during application startup to allow early validation of connection + * factory configuration. Eager initialization also prevents blocking connect while using reactive API and is + * recommended for reactive API usage. + * + * @return {@link true} if the shared connection is initialized upon {@link #afterPropertiesSet()}. + * @since 2.2 + */ + public boolean getEagerInitialization() { + return eagerInitialization; + } + + /** + * Enables eager initialization of {@link #setShareNativeConnection(boolean) shared connections}. + * + * @param eagerInitialization enable eager connection shared connection initialization upon + * {@link #afterPropertiesSet()}. + * @since 2.2 + */ + public void setEagerInitialization(boolean eagerInitialization) { + this.eagerInitialization = eagerInitialization; + } + /** * Returns the index of the database. * 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 index c1012609c..4c80b0d60 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionProvider.java @@ -18,6 +18,9 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + /** * Defines a provider for Lettuce connections. *

@@ -46,7 +49,20 @@ public interface LettuceConnectionProvider { * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no * longer in use. */ - > T getConnection(Class connectionType); + default > T getConnection(Class connectionType) { + return LettuceFutureUtils.join(getConnectionAsync(connectionType)); + } + + /** + * Request asynchronously 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 a {@link CompletionStage} that is notified with the connection progress. Must be + * {@link #releaseAsync(StatefulConnection) released} if the connection is no longer in use. + * @since 2.2 + */ + > CompletionStage getConnectionAsync(Class connectionType); /** * Release the {@link StatefulConnection connection}. Closes connection {@link StatefulConnection#close()} by default. @@ -55,12 +71,26 @@ public interface LettuceConnectionProvider { * @param connection must not be {@literal null}. */ default void release(StatefulConnection connection) { - connection.close(); + LettuceFutureUtils.join(releaseAsync(connection)); + } + + /** + * Release asynchronously the {@link StatefulConnection connection}. Closes connection + * {@link StatefulConnection#closeAsync()} by default. Implementations may choose whether they override this method + * and return the connection to a pool. + * + * @param connection must not be {@literal null}. + * @return Close {@link CompletableFuture future} notified once the connection is released. + * @since 2.2 + */ + default CompletableFuture releaseAsync(StatefulConnection connection) { + return connection.closeAsync(); } /** * Extension to {@link LettuceConnectionProvider} for providers that allow connection creation to specific nodes. */ + @FunctionalInterface interface TargetAware { /** @@ -71,6 +101,20 @@ public interface LettuceConnectionProvider { * @param redisURI must not be {@literal null}. * @return the requested connection. */ - > T getConnection(Class connectionType, RedisURI redisURI); + default > T getConnection(Class connectionType, RedisURI redisURI) { + return LettuceFutureUtils.join(getConnectionAsync(connectionType, redisURI)); + } + + /** + * Request asynchronously 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 a {@link CompletionStage} that is notified with the connection progress. + * @since 2.2 + */ + > CompletionStage getConnectionAsync(Class connectionType, + RedisURI redisURI); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceFutureUtils.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceFutureUtils.java new file mode 100644 index 000000000..c9f96dc98 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceFutureUtils.java @@ -0,0 +1,98 @@ +/* + * Copyright 2018 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 java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.function.Function; + +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +/** + * Utility methods to interact with {@link CompletableFuture} and {@link CompletionStage}. + * + * @author Mark Paluch + * @since 2.2 + */ +class LettuceFutureUtils { + + /** + * Creates a {@link CompletableFuture} that is completed {@link CompletableFuture#exceptionally(Function) + * exceptionally} given {@link Throwable}. This utility method allows exceptionally future creation with a single + * invocation. + * + * @param t must not be {@literal null}. + * @return the completed {@link CompletableFuture future}. + */ + static CompletableFuture failed(Throwable t) { + + Assert.notNull(t, "Throwable must not be null!"); + + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(t); + + return future; + } + + /** + * Synchronizes a {@link CompletableFuture} result by {@link CompletableFuture#join() waiting until the future is + * complete}. This method preserves {@link RuntimeException}s that may get thrown as result of future completion. + * Checked exceptions are thrown encapsulated within {@link java.util.concurrent.CompletionException}. + * + * @param future must not be {@literal null}. + * @throws RuntimeException thrown if the future is completed with a {@link RuntimeException}. + * @throws CompletionException thrown if the future is completed with a checked exception. + * @return the future result if completed normally. + */ + @Nullable + static T join(CompletionStage future) throws RuntimeException, CompletionException { + + Assert.notNull(future, "CompletableFuture must not be null!"); + + try { + return future.toCompletableFuture().join(); + } catch (Exception e) { + + Throwable exceptionToUse = e; + + if (e instanceof CompletionException) { + exceptionToUse = new LettuceExceptionConverter().convert((Exception) e.getCause()); + if (exceptionToUse == null) { + exceptionToUse = e.getCause(); + } + } + + if (exceptionToUse instanceof RuntimeException) { + throw (RuntimeException) exceptionToUse; + } + + throw new CompletionException(exceptionToUse); + } + } + + /** + * Returns a {@link Function} that ignores {@link CompletionStage#exceptionally(Function) exceptional completion} by + * recovering to {@code null}. This allows to progress with a previously failed {@link CompletionStage} without regard + * to the actual success/exception state. + * + * @return + */ + static Function ignoreErrors() { + return ignored -> null; + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java index f60b8dd1a..99108c3f2 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java @@ -17,9 +17,17 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.AbstractRedisClient; import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.support.AsyncConnectionPoolSupport; +import io.lettuce.core.support.AsyncPool; +import io.lettuce.core.support.BoundedPoolConfig; +import io.lettuce.core.support.CommonsPool2ConfigConverter; import io.lettuce.core.support.ConnectionPoolSupport; +import java.util.ArrayList; +import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.logging.Log; @@ -32,11 +40,17 @@ import org.springframework.util.Assert; /** * {@link LettuceConnectionProvider} with connection pooling support. This connection provider holds multiple pools (one - * per connection type) for contextualized connection allocation. + * per connection type and allocation type (synchronous/asynchronous)) for contextualized connection allocation. *

* Each allocated connection is tracked and to be returned into the pool which created the connection. Instances of this * class require {@link #destroy() disposal} to de-allocate lingering connections that were not returned to the pool and * to close the pools. + *

+ * This provider maintains separate pools due to the allocation nature (synchronous/asynchronous). Asynchronous + * connection pooling requires a non-blocking allocation API. Connections requested asynchronously can be returned + * synchronously and vice versa. A connection obtained synchronously is returned to the synchronous pool even if + * {@link #releaseAsync(StatefulConnection) released asynchronously}. This is an undesired case as the synchronous pool + * will block the asynchronous flow for the time of release. * * @author Mark Paluch * @author Christoph Strobl @@ -51,7 +65,14 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red private final GenericObjectPoolConfig poolConfig; private final Map, GenericObjectPool>> poolRef = new ConcurrentHashMap<>( 32); + + private final Map, AsyncPool>> asyncPoolRef = new ConcurrentHashMap<>( + 32); + private final Map>, AsyncPool>> inProgressAsyncPoolRef = new ConcurrentHashMap<>( + 32); private final Map, GenericObjectPool>> pools = new ConcurrentHashMap<>(32); + private final Map, AsyncPool>> asyncPools = new ConcurrentHashMap<>(32); + private final BoundedPoolConfig asyncPoolConfig; LettucePoolingConnectionProvider(LettuceConnectionProvider connectionProvider, LettucePoolingClientConfiguration clientConfiguration) { @@ -61,6 +82,7 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red this.connectionProvider = connectionProvider; this.poolConfig = clientConfiguration.getPoolConfig(); + this.asyncPoolConfig = CommonsPool2ConfigConverter.bounded(this.poolConfig); } /* @@ -87,6 +109,33 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) + */ + @Override + public > CompletionStage getConnectionAsync(Class connectionType) { + + AsyncPool> pool = asyncPools.computeIfAbsent(connectionType, poolType -> { + + return AsyncConnectionPoolSupport.createBoundedObjectPool( + () -> connectionProvider.getConnectionAsync(connectionType).thenApply(connectionType::cast), asyncPoolConfig, + false); + }); + + CompletableFuture> acquire = pool.acquire(); + + inProgressAsyncPoolRef.put(acquire, pool); + return acquire.whenComplete((conn, e) -> { + + inProgressAsyncPoolRef.remove(acquire); + + if (conn != null) { + asyncPoolRef.put(conn, pool); + } + }).thenApply(connectionType::cast); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.RedisClientProvider#getRedisClient() @@ -113,13 +162,47 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red GenericObjectPool> pool = poolRef.remove(connection); if (pool == null) { - throw new PoolException("Returned connection " + connection - + " was either previously returned or does not belong to this connection provider"); + + AsyncPool> asyncPool = asyncPoolRef.remove(connection); + + if (asyncPool == null) { + throw new PoolException("Returned connection " + connection + + " was either previously returned or does not belong to this connection provider"); + } + + asyncPool.release(connection).join(); + return; } pool.returnObject(connection); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#releaseAsync(io.lettuce.core.api.StatefulConnection) + */ + @Override + public CompletableFuture releaseAsync(StatefulConnection connection) { + + GenericObjectPool> blockingPool = poolRef.remove(connection); + + if (blockingPool != null) { + + log.warn("Releasing asynchronously a connection that was obtained from a non-blocking pool"); + blockingPool.returnObject(connection); + return CompletableFuture.completedFuture(null); + } + + AsyncPool> pool = asyncPoolRef.remove(connection); + + if (pool == null) { + return LettuceFutureUtils.failed(new PoolException("Returned connection " + connection + + " was either previously returned or does not belong to this connection provider")); + } + + return pool.release(connection); + } + /* * (non-Javadoc) * @see org.springframework.beans.factory.DisposableBean#destroy() @@ -127,15 +210,47 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red @Override public void destroy() throws Exception { - if (!poolRef.isEmpty()) { - + List> futures = new ArrayList<>(); + if (!poolRef.isEmpty() || !asyncPoolRef.isEmpty()) { log.warn("LettucePoolingConnectionProvider contains unreleased connections"); + } + + if (!inProgressAsyncPoolRef.isEmpty()) { + + log.warn("LettucePoolingConnectionProvider has active connection retrievals"); + inProgressAsyncPoolRef.forEach((k, v) -> futures.add(k.thenApply(StatefulConnection::closeAsync))); + } + + if (!poolRef.isEmpty()) { poolRef.forEach((connection, pool) -> pool.returnObject(connection)); poolRef.clear(); } + if (!asyncPoolRef.isEmpty()) { + + asyncPoolRef.forEach((connection, pool) -> futures.add(pool.release(connection))); + asyncPoolRef.clear(); + } + pools.forEach((type, pool) -> pool.close()); + + CompletableFuture + .allOf(futures.stream().map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors())) + .toArray(CompletableFuture[]::new)) // + .thenCompose(ignored -> { + + CompletableFuture[] poolClose = asyncPools.values().stream().map(AsyncPool::closeAsync) + .map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors())).toArray(CompletableFuture[]::new); + + return CompletableFuture.allOf(poolClose); + }) // + .thenRun(() -> { + asyncPoolRef.clear(); + inProgressAsyncPoolRef.clear(); + }) // + .join(); + pools.clear(); } } 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 44873ddbb..b3998a0f1 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 @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; import io.lettuce.core.api.reactive.RedisReactiveCommands; import io.lettuce.core.cluster.RedisClusterClient; @@ -240,7 +241,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti .map(it -> it.getConnection(node.getId()).reactive()); } - return getConnection().cast(StatefulRedisClusterConnection.class) - .map(it -> it.getConnection(node.getHost(), node.getPort()).reactive()); + return getConnection().flatMap(it -> Mono.fromCompletionStage(it.getConnectionAsync(node.getHost(), node.getPort())) + .map(StatefulRedisConnection::reactive)); } } 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 8dac9f250..9b597d3ba 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,6 +15,8 @@ */ package org.springframework.data.redis.connection.lettuce; +import static org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection.AsyncConnect.State.*; + import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; @@ -24,10 +26,8 @@ import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; import java.nio.ByteBuffer; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; @@ -237,7 +237,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { * @see org.springframework.data.redis.connection.ReactiveRedisConnection#closeLater() */ public Mono closeLater() { - return Mono.fromRunnable(() -> dedicatedConnection.close()); + return Mono.fromRunnable(dedicatedConnection::close); } protected Mono> getConnection() { @@ -335,8 +335,9 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { * {@link #close()} methods: *

    *
  1. Initial state: Not connected. Calling {@link #getConnection()} will transition to the next state.
    2. - *
  2. Connection requested: First caller to {@link #getConnection()} initiates an asynchronous connect. Connection is - * accessible through the resulting {@link Mono}.
    3. + *
  3. Connection requested: First subscriber to {@link #getConnection()} initiates an asynchronous connect. + * Connection is accessible through the resulting {@link Mono} and will be cached. Connection will be closed if + * {@link AsyncConnect} gets closed before the connection publisher emits the actual connection.
    4. *
  4. Closing: Call to {@link #close()} initiates connection closing.
    5. *
  5. Closed: Connection is closed.
    6. *
@@ -345,24 +346,42 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { * @author Christoph Strobl * @since 2.0.1 */ - static class AsyncConnect> { + static class AsyncConnect> { private final Mono connectionPublisher; private final LettuceConnectionProvider connectionProvider; private AtomicReference state = new AtomicReference<>(State.INITIAL); - private volatile @Nullable CompletableFuture connection; + private volatile @Nullable StatefulConnection connection; @SuppressWarnings("unchecked") - AsyncConnect(LettuceConnectionProvider connectionProvider, Class connectionType) { + AsyncConnect(LettuceConnectionProvider connectionProvider, Class connectionType) { Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); + Assert.notNull(connectionType, "Connection type must not be null!"); this.connectionProvider = connectionProvider; - Mono defer = Mono.defer(() -> Mono. just(connectionProvider.getConnection(connectionType))); + Mono defer = Mono + .defer(() -> Mono.fromCompletionStage(connectionProvider.getConnectionAsync(connectionType))); - this.connectionPublisher = defer.subscribeOn(Schedulers.elastic()); + this.connectionPublisher = defer.doOnNext(it -> { + + if (isClosing(this.state.get())) { + it.closeAsync(); + } else { + connection = it; + } + }) // + .cache() // + .handle((connection, sink) -> { + + if (isClosing(this.state.get())) { + sink.error(new IllegalStateException("Unable to connect. Connection is closed!")); + } else { + sink.next((T) connection); + } + }); } /** @@ -373,33 +392,14 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { */ Mono getConnection() { - if (state.get() == State.CLOSED) { - throw new IllegalStateException("Unable to connect. Connection is closed!"); + State state = this.state.get(); + if (isClosing(state)) { + return Mono.error(new IllegalStateException("Unable to connect. Connection is closed!")); } - CompletableFuture connection = this.connection; + this.state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED); - if (connection != null) { - return Mono.fromCompletionStage(connection); - } - - if (state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED)) { - this.connection = connectionPublisher.toFuture(); - } - - while (true) { - - connection = this.connection; - if (connection != null) { - return Mono.fromCompletionStage(connection); - } - - Thread thread = Thread.currentThread(); - if (thread.isInterrupted()) { - thread.interrupt(); - return Mono.error(new InterruptedException()); - } - } + return connectionPublisher; } /** @@ -407,11 +407,21 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { */ void close() { - if (state.compareAndSet(State.CONNECTION_REQUESTED, State.CLOSING)) { - getConnection().doOnSuccess(connectionProvider::release).doOnSuccess(it -> state.set(State.CLOSED)).block(); - } + if (state.compareAndSet(State.INITIAL, CLOSING) || state.compareAndSet(State.CONNECTION_REQUESTED, CLOSING)) { - state.compareAndSet(State.INITIAL, State.CLOSED); + StatefulConnection connection = this.connection; + this.connection = null; + + if (connection != null) { + LettuceFutureUtils.join(connectionProvider.releaseAsync(connection)); + } + + state.set(State.CLOSED); + } + } + + private static boolean isClosing(State state) { + return state == State.CLOSING || state == State.CLOSED; } enum State { 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 d424426de..c7e07b549 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 @@ -16,14 +16,17 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; import io.lettuce.core.RedisURI.Builder; import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.codec.StringCodec; import io.lettuce.core.resource.ClientResources; import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection; import io.lettuce.core.sentinel.api.sync.RedisSentinelCommands; import java.io.IOException; import java.util.List; +import java.util.concurrent.CompletableFuture; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.FallbackExceptionTranslationStrategy; @@ -99,6 +102,11 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { public > T getConnection(Class t) { return t.cast(redisClient.connectSentinel()); } + + @Override + public > CompletableFuture getConnectionAsync(Class t) { + return CompletableFuture.completedFuture(t.cast(connection)); + } }; init(); } @@ -116,6 +124,11 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { public > T getConnection(Class t) { return t.cast(connection); } + + @Override + public > CompletableFuture getConnectionAsync(Class t) { + return CompletableFuture.completedFuture(t.cast(connection)); + } }; init(); } @@ -253,12 +266,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { private static class DedicatedClientConnectionProvider implements LettuceConnectionProvider { private final RedisClient redisClient; + private final RedisURI uri; DedicatedClientConnectionProvider(String host, int port) { Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host."); - redisClient = RedisClient.create(Builder.redis(host, port).build()); + uri = Builder.redis(host, port).build(); + redisClient = RedisClient.create(uri); } DedicatedClientConnectionProvider(String host, int port, ClientResources clientResources) { @@ -266,16 +281,17 @@ 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.uri = Builder.redis(host, port).build(); + redisClient = RedisClient.create(clientResources, uri); } - /* + /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) */ @Override - public > T getConnection(Class connectionType) { - return connectionType.cast(redisClient.connectSentinel()); + public > CompletableFuture getConnectionAsync(Class connectionType) { + return redisClient.connectSentinelAsync(StringCodec.UTF8, uri).thenApply(connectionType::cast); } /* @@ -288,5 +304,15 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { connection.close(); redisClient.shutdown(); } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#releaseAsync(io.lettuce.core.api.StatefulConnection) + */ + @Override + public CompletableFuture releaseAsync(StatefulConnection connection) { + return connection.closeAsync().exceptionally(LettuceFutureUtils.ignoreErrors()) + .thenCompose(it -> redisClient.shutdownAsync()); + } } } 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 index 72422669d..a4953d824 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/StandaloneConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StandaloneConnectionProvider.java @@ -21,12 +21,15 @@ import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.RedisCodec; +import io.lettuce.core.codec.StringCodec; import io.lettuce.core.masterslave.MasterSlave; import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection; import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -74,7 +77,7 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA redisURISupplier = new Supplier() { - AtomicReference uriFieldReference = new AtomicReference(); + AtomicReference uriFieldReference = new AtomicReference<>(); @Override public RedisURI get() { @@ -95,7 +98,6 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class) */ - @SuppressWarnings("null") @Override public > T getConnection(Class connectionType) { @@ -118,27 +120,38 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnection(java.lang.Class, io.lettuce.core.RedisURI) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) */ - @SuppressWarnings("null") @Override - public > T getConnection(Class connectionType, RedisURI redisURI) { + public > CompletionStage getConnectionAsync(Class connectionType) { + return getConnectionAsync(connectionType, redisURISupplier.get()); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnectionAsync(java.lang.Class, io.lettuce.core.RedisURI) + */ + @SuppressWarnings({ "null", "unchecked", "rawtypes" }) + @Override + public > CompletionStage getConnectionAsync(Class connectionType, + RedisURI redisURI) { if (connectionType.equals(StatefulRedisSentinelConnection.class)) { - return connectionType.cast(client.connectSentinel(redisURI)); + return client.connectSentinelAsync(StringCodec.UTF8, redisURI).thenApply(connectionType::cast); } if (connectionType.equals(StatefulRedisPubSubConnection.class)) { - return connectionType.cast(client.connectPubSub(codec, redisURI)); + return client.connectPubSubAsync(codec, redisURI).thenApply(connectionType::cast); } if (StatefulConnection.class.isAssignableFrom(connectionType)) { - - return connectionType - .cast(readFrom.map(it -> this.masterReplicaConnection(redisURI, it)).orElseGet(() -> client.connect(codec))); + return readFrom.map(it -> this.masterReplicaConnectionAsync(redisURI, it)) // + .orElseGet(() -> (CompletionStage) client.connectAsync(codec, redisURI)) // + .thenApply(connectionType::cast); } - throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!"); + return LettuceFutureUtils + .failed(new UnsupportedOperationException("Connection type " + connectionType + " not supported!")); } private StatefulRedisConnection masterReplicaConnection(RedisURI redisUri, ReadFrom readFrom) { @@ -150,4 +163,17 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA return connection; } + private CompletionStage> masterReplicaConnectionAsync(RedisURI redisUri, + ReadFrom readFrom) { + + CompletableFuture> connection = MasterSlave.connectAsync(client, + codec, redisUri); + + return connection.thenApply(conn -> { + + conn.setReadFrom(readFrom); + + return conn; + }); + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/StaticMasterReplicaConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/StaticMasterReplicaConnectionProvider.java index 0269dee4c..4eb9b9ba3 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/StaticMasterReplicaConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StaticMasterReplicaConnectionProvider.java @@ -25,6 +25,8 @@ import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection; import java.util.Collection; import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import org.springframework.lang.Nullable; @@ -79,4 +81,27 @@ class StaticMasterReplicaConnectionProvider implements LettuceConnectionProvider throw new UnsupportedOperationException(String.format("Connection type %s not supported!", connectionType)); } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class) + */ + @Override + public > CompletionStage getConnectionAsync(Class connectionType) { + + if (StatefulConnection.class.isAssignableFrom(connectionType)) { + + // See https://github.com/lettuce-io/lettuce-core/issues/845 for MasterSlave -> MasterReplica change. + CompletableFuture> connection = MasterSlave + .connectAsync(client, codec, nodes); + + connection.thenApply(it -> { + + readFrom.ifPresent(readFrom -> it.setReadFrom(readFrom)); + return connectionType.cast(connection); + }); + } + + throw new UnsupportedOperationException(String.format("Connection type %s not supported!", connectionType)); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java index 5cc7fa32e..0a4191318 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java @@ -29,6 +29,8 @@ import io.lettuce.core.AbstractRedisClient; import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.codec.ByteArrayCodec; @@ -38,6 +40,7 @@ import io.lettuce.core.resource.ClientResources; import java.security.NoSuchAlgorithmException; import java.time.Duration; import java.util.Collections; +import java.util.concurrent.CompletableFuture; import org.junit.After; import org.junit.Before; @@ -590,7 +593,7 @@ public class LettuceConnectionFactoryUnitTests { RedisClusterClient clientMock = mock(RedisClusterClient.class); StatefulRedisClusterConnection connectionMock = mock(StatefulRedisClusterConnection.class); - when(clientMock.connect(ByteArrayCodec.INSTANCE)).thenReturn(connectionMock); + when(clientMock.connectAsync(ByteArrayCodec.INSTANCE)).thenReturn(CompletableFuture.completedFuture(connectionMock)); LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, LettuceClientConfiguration.defaultConfiguration()) { @@ -608,7 +611,31 @@ public class LettuceConnectionFactoryUnitTests { connectionFactory.getClusterConnection().close(); connectionFactory.getClusterConnection().close(); - verify(clientMock).connect(ArgumentMatchers.any(RedisCodec.class)); + verify(clientMock).connectAsync(ArgumentMatchers.any(RedisCodec.class)); + } + + @Test // DATAREDIS-721 + @SuppressWarnings("unchecked") + public void shouldEagerlyInitializeSharedConnection() { + + LettuceConnectionProvider connectionProviderMock = mock(LettuceConnectionProvider.class); + StatefulRedisConnection connectionMock = mock(StatefulRedisConnection.class); + + when(connectionProviderMock.getConnectionAsync(any())) + .thenReturn(CompletableFuture.completedFuture(connectionMock)); + + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory() { + @Override + protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client, + RedisCodec codec) { + return connectionProviderMock; + } + }; + connectionFactory.setEagerInitialization(true); + + connectionFactory.afterPropertiesSet(); + + verify(connectionProviderMock, times(2)).getConnection(StatefulConnection.class); } @Test // DATAREDIS-842 diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnectionUnitTests.java index ffd58401a..383eec60e 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnectionUnitTests.java @@ -21,6 +21,7 @@ import static org.mockito.Mockito.anyInt; import static org.mockito.Mockito.anyString; import static org.springframework.data.redis.connection.ClusterTestVariables.*; +import io.lettuce.core.ConnectionFuture; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.reactive.RedisReactiveCommands; import io.lettuce.core.cluster.RedisClusterClient; @@ -30,6 +31,7 @@ import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.nio.ByteBuffer; +import java.util.concurrent.CompletableFuture; import org.junit.Before; import org.junit.Test; @@ -60,8 +62,8 @@ public class LettuceReactiveRedisClusterConnectionUnitTests { @Before public void before() { - when(connectionProvider.getConnection(any())).thenReturn(sharedConnection); - when(sharedConnection.getConnection(anyString(), anyInt())).thenReturn(nodeConnection); + when(connectionProvider.getConnectionAsync(any())).thenReturn(CompletableFuture.completedFuture(sharedConnection)); + when(sharedConnection.getConnectionAsync(anyString(), anyInt())).thenReturn(CompletableFuture.completedFuture(nodeConnection)); when(nodeConnection.reactive()).thenReturn(reactiveNodeCommands); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java index 011f87800..1dda376eb 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java @@ -27,7 +27,6 @@ import reactor.test.StepVerifier; import java.nio.ByteBuffer; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Before; @@ -53,7 +52,8 @@ public class LettuceReactiveRedisConnectionUnitTests { @Before public void before() { - when(connectionProvider.getConnection(any())).thenReturn(sharedConnection); + when(connectionProvider.getConnectionAsync(any())).thenReturn(CompletableFuture.completedFuture(sharedConnection)); + when(connectionProvider.releaseAsync(any())).thenReturn(CompletableFuture.completedFuture(null)); when(sharedConnection.reactive()).thenReturn(reactiveCommands); } @@ -65,24 +65,24 @@ public class LettuceReactiveRedisConnectionUnitTests { verifyZeroInteractions(connectionProvider); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldExecuteUsingConnectionProvider() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); StepVerifier.create(connection.execute(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); - verify(connectionProvider).getConnection(StatefulConnection.class); + verify(connectionProvider).getConnectionAsync(StatefulConnection.class); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldExecuteDedicatedUsingConnectionProvider() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); StepVerifier.create(connection.executeDedicated(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); - verify(connectionProvider).getConnection(StatefulConnection.class); + verify(connectionProvider).getConnectionAsync(StatefulConnection.class); } @Test // DATAREDIS-720 @@ -96,7 +96,7 @@ public class LettuceReactiveRedisConnectionUnitTests { verifyZeroInteractions(connectionProvider); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldExecuteDedicatedWithSharedConnection() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(sharedConnection, @@ -104,20 +104,20 @@ public class LettuceReactiveRedisConnectionUnitTests { StepVerifier.create(connection.executeDedicated(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); - verify(connectionProvider).getConnection(StatefulConnection.class); + verify(connectionProvider).getConnectionAsync(StatefulConnection.class); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldOperateOnDedicatedConnection() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); StepVerifier.create(connection.getConnection()).expectNextCount(1).verifyComplete(); - verify(connectionProvider).getConnection(StatefulConnection.class); + verify(connectionProvider).getConnectionAsync(StatefulConnection.class); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldCloseOnlyDedicatedConnection() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(sharedConnection, @@ -128,11 +128,11 @@ public class LettuceReactiveRedisConnectionUnitTests { connection.close(); - verify(sharedConnection, never()).close(); - verify(connectionProvider, times(1)).release(sharedConnection); + verify(sharedConnection, never()).closeAsync(); + verify(connectionProvider, times(1)).releaseAsync(sharedConnection); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldCloseConnectionOnlyOnce() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); @@ -142,21 +142,16 @@ public class LettuceReactiveRedisConnectionUnitTests { connection.close(); connection.close(); - verify(connectionProvider, times(1)).release(sharedConnection); + verify(connectionProvider, times(1)).releaseAsync(sharedConnection); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 @SuppressWarnings("unchecked") public void multipleCallsInProgressShouldConnectOnlyOnce() throws Exception { - CountDownLatch latch = new CountDownLatch(1); - + CompletableFuture> connectionFuture = new CompletableFuture<>(); reset(connectionProvider); - when(connectionProvider.getConnection(any())).thenAnswer(invocation -> { - - latch.await(); - return sharedConnection; - }); + when(connectionProvider.getConnectionAsync(any())).thenReturn(connectionFuture); LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); @@ -168,9 +163,9 @@ public class LettuceReactiveRedisConnectionUnitTests { assertThat(first).isNotDone(); assertThat(second).isNotDone(); - verify(connectionProvider, times(1)).getConnection(StatefulConnection.class); + verify(connectionProvider, times(1)).getConnectionAsync(StatefulConnection.class); - latch.countDown(); + connectionFuture.complete(sharedConnection); first.get(10, TimeUnit.SECONDS); second.get(10, TimeUnit.SECONDS); @@ -179,24 +174,25 @@ public class LettuceReactiveRedisConnectionUnitTests { assertThat(second).isCompletedWithValue(sharedConnection); } - @Test // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldPropagateConnectionFailures() { reset(connectionProvider); - when(connectionProvider.getConnection(any())).thenThrow(new RedisConnectionException("something went wrong")); + when(connectionProvider.getConnectionAsync(any())) + .thenReturn(LettuceFutureUtils.failed(new RedisConnectionException("something went wrong"))); LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); StepVerifier.create(connection.getConnection()).expectError(RedisConnectionFailureException.class).verify(); } - @Test(expected = IllegalStateException.class) // DATAREDIS-720 + @Test // DATAREDIS-720, DATAREDIS-721 public void shouldRejectCommandsAfterClose() { LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); connection.close(); - connection.getConnection(); + StepVerifier.create(connection.getConnection()).expectError(IllegalStateException.class).verify(); } @Test // DATAREDIS-659, DATAREDIS-708