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 ca4535202..6d81aa298 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 @@ -20,12 +20,15 @@ import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisException; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.cluster.ClusterClientOptions; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.resource.ClientResources; +import java.nio.ByteBuffer; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -88,6 +91,7 @@ public class LettuceConnectionFactory private boolean validateConnection = false; private boolean shareNativeConnection = true; private @Nullable StatefulRedisConnection connection; + private @Nullable StatefulConnection reactiveConnection; private @Nullable LettucePool pool; /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); @@ -242,6 +246,7 @@ public class LettuceConnectionFactory public void destroy() { resetConnection(); + resetReactiveConnection(); if (connectionProvider instanceof DisposableBean) { try { @@ -318,7 +323,10 @@ public class LettuceConnectionFactory */ @Override public LettuceReactiveRedisConnection getReactiveConnection() { - return new LettuceReactiveRedisConnection(reactiveConnectionProvider); + + return getShareNativeConnection() + ? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider) + : new LettuceReactiveRedisConnection(reactiveConnectionProvider); } /* @@ -327,11 +335,16 @@ public class LettuceConnectionFactory */ @Override public LettuceReactiveRedisClusterConnection getReactiveClusterConnection() { + if (!isClusterAware()) { throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } - return new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider); + RedisClusterClient client = (RedisClusterClient) this.client; + + return getShareNativeConnection() + ? new LettuceReactiveRedisClusterConnection(getSharedReactiveConnection(), reactiveConnectionProvider, client) + : new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider, client); } public void initConnection() { @@ -344,11 +357,23 @@ public class LettuceConnectionFactory } } + public void initReactiveConnection() { + + synchronized (this.connectionMonitor) { + if (this.reactiveConnection != null) { + resetReactiveConnection(); + } + this.reactiveConnection = createReactiveLettuceConnector(); + } + } + /** * Reset the underlying shared Connection, to be reinitialized on next access. */ public void resetConnection() { + synchronized (this.connectionMonitor) { + if (this.connection != null) { this.connectionProvider.release(this.connection); } @@ -356,17 +381,49 @@ public class LettuceConnectionFactory } } + /** + * Reset the underlying shared Connection, to be reinitialized on next access. + * + * @since 2.0.1 + */ + public void resetReactiveConnection() { + + synchronized (this.connectionMonitor) { + + if (this.reactiveConnection != null) { + this.reactiveConnectionProvider.release(this.reactiveConnection); + } + this.reactiveConnection = null; + } + } + /** * Validate the shared Connection and reinitialize if invalid */ public void validateConnection() { + + validateConnection(connection, connectionProvider, this::initConnection); + + validateConnection(reactiveConnection, reactiveConnectionProvider, this::initReactiveConnection); + } + + private void validateConnection(@Nullable StatefulConnection connection, + LettuceConnectionProvider connectionProvider, Runnable initConnection) { + synchronized (this.connectionMonitor) { boolean valid = false; - if (connection.isOpen()) { + if (connection != null && connection.isOpen()) { try { - connection.sync().ping(); + + if (connection instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) connection).sync().ping(); + } + + if (connection instanceof StatefulRedisClusterConnection) { + ((StatefulRedisConnection) connection).sync().ping(); + } valid = true; } catch (Exception e) { log.debug("Validation failed", e); @@ -375,9 +432,13 @@ public class LettuceConnectionFactory if (!valid) { - connectionProvider.release(connection); + if (connection != null) { + connectionProvider.release(connection); + } + log.warn("Validation of shared connection failed. Creating a new connection."); - initConnection(); + + initConnection.run(); } } } @@ -775,7 +836,12 @@ public class LettuceConnectionFactory return clusterConfiguration != null; } + /** + * @return the shared connection using {@literal byte} array encoding for imperative API use. {@literal null} if + * {@link #getShareNativeConnection() connection sharing} is disabled. + */ protected StatefulRedisConnection getSharedConnection() { + if (shareNativeConnection) { synchronized (this.connectionMonitor) { if (this.connection == null) { @@ -791,6 +857,28 @@ public class LettuceConnectionFactory } } + /** + * @return the shared connection using {@link ByteBuffer} encoding for reactive API use. {@literal null} if + * {@link #getShareNativeConnection() connection sharing} is disabled. + * @since 2.0.1 + */ + protected StatefulConnection getSharedReactiveConnection() { + + if (shareNativeConnection) { + synchronized (this.connectionMonitor) { + if (this.reactiveConnection == null) { + initReactiveConnection(); + } + if (validateConnection) { + validateConnection(); + } + return this.reactiveConnection; + } + } else { + return null; + } + } + protected StatefulRedisConnection createLettuceConnector() { try { @@ -809,6 +897,23 @@ public class LettuceConnectionFactory } } + protected StatefulConnection createReactiveLettuceConnector() { + + try { + + StatefulConnection connection; + connection = reactiveConnectionProvider.getConnection(StatefulConnection.class); + + if (connection instanceof StatefulRedisConnection && getDatabase() > 0) { + ((StatefulRedisConnection) connection).sync().select(getDatabase()); + } + + return connection; + } catch (RedisException e) { + throw new RedisConnectionFailureException("Unable to connect to Redis on " + getHostName() + ":" + getPort(), e); + } + } + private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec codec) { LettuceConnectionProvider connectionProvider = doConnectionProvider(client, codec); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommands.java index cffc92974..b9bf83f4e 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommands.java @@ -241,7 +241,7 @@ class LettuceReactiveListCommands implements ReactiveListCommands { @Override public Flux bPop(Publisher commands) { - return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + return connection.executeDedicated(cmd -> Flux.from(commands).concatMap(command -> { Assert.notNull(command.getKeys(), "Keys must not be null!"); Assert.notNull(command.getDirection(), "Direction must not be null!"); @@ -281,7 +281,7 @@ class LettuceReactiveListCommands implements ReactiveListCommands { @Override public Flux> bRPopLPush(Publisher commands) { - return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + return connection.executeDedicated(cmd -> Flux.from(commands).concatMap(command -> { Assert.notNull(command.getKey(), "Key must not be null!"); Assert.notNull(command.getDestination(), "Destination key must not be null!"); 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 72ac35a94..6b1df8872 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 @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.connection.lettuce; +import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; import io.lettuce.core.api.reactive.RedisReactiveCommands; import io.lettuce.core.cluster.RedisClusterClient; @@ -45,19 +46,37 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti private final ClusterTopologyProvider topologyProvider; /** - * Creates new {@link LettuceReactiveRedisClusterConnection}. + * Creates new {@link LettuceReactiveRedisClusterConnection} given {@link LettuceConnectionProvider} and + * {@link RedisClusterClient}. * + * @param connectionProvider must not be {@literal null}. * @param client must not be {@literal null}. * @throws IllegalArgumentException when {@code client} is {@literal null}. - * @throws org.springframework.dao.InvalidDataAccessResourceUsageException when {@code client} is not suitable for - * cluster environment. */ - LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider) { + LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider, RedisClusterClient client) { super(connectionProvider); - this.topologyProvider = new LettuceClusterTopologyProvider( - ((ClusterConnectionProvider) connectionProvider).getClient()); + this.topologyProvider = new LettuceClusterTopologyProvider(client); + } + + /** + * Creates new {@link LettuceReactiveRedisClusterConnection} given a shared {@link StatefulConnection connection}, + * {@link LettuceConnectionProvider} and {@link RedisClusterClient}. + * + * @param sharedConnection must not be {@literal null}. + * @param connectionProvider must not be {@literal null}. + * @param client must not be {@literal null}. + * @throws IllegalArgumentException when {@code client} is {@literal null}. + * @since 2.0.1 + */ + @SuppressWarnings("unchecked") + LettuceReactiveRedisClusterConnection(StatefulConnection sharedConnection, + LettuceConnectionProvider connectionProvider, RedisClusterClient client) { + + super(sharedConnection, connectionProvider); + + this.topologyProvider = new LettuceClusterTopologyProvider(client); } /* @@ -183,7 +202,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti return Flux.error(e); } - return Flux.defer(() -> callback.doWithCommands(getCommands(node))).onErrorMap(translateException()); + return getCommands(node).flatMapMany(callback::doWithCommands).onErrorMap(translateException()); } /* @@ -192,33 +211,27 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti */ @SuppressWarnings({ "unchecked", "rawtypes" }) @Override - protected StatefulRedisClusterConnection getConnection() { - - Assert.isInstanceOf(StatefulRedisClusterConnection.class, super.getConnection(), - "Connection needs to be instance of StatefulRedisClusterConnection"); - - return (StatefulRedisClusterConnection) super.getConnection(); + protected Mono> getConnection() { + return (Mono) super.getConnection(); } /* * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#getCommands() */ - protected RedisClusterReactiveCommands getCommands() { - return getConnection().reactive(); + protected Mono> getCommands() { + return getConnection().map(StatefulRedisClusterConnection::reactive); } @SuppressWarnings({ "unchecked", "rawtypes" }) - protected RedisReactiveCommands getCommands(RedisNode node) { - - if (!(getConnection() instanceof StatefulRedisClusterConnection)) { - throw new IllegalArgumentException("o.O connection needs to be cluster compatible " + getConnection()); - } + protected Mono> getCommands(RedisNode node) { if (StringUtils.hasText(node.getId())) { - return ((StatefulRedisClusterConnection) getConnection()).getConnection(node.getId()).reactive(); + return getConnection().cast(StatefulRedisClusterConnection.class) + .map(it -> it.getConnection(node.getId()).reactive()); } - return ((StatefulRedisClusterConnection) getConnection()).getConnection(node.getHost(), node.getPort()).reactive(); + return getConnection().cast(StatefulRedisClusterConnection.class) + .map(it -> it.getConnection(node.getHost(), node.getPort()).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 2be8cedad..9444e106f 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 @@ -23,8 +23,11 @@ import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; import io.lettuce.core.codec.RedisCodec; 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; import org.reactivestreams.Publisher; @@ -43,9 +46,9 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { static final RedisCodec CODEC = ByteBufferCodec.INSTANCE; - private final LettuceConnectionProvider connectionProvider; + private final AsyncConnect dedicatedConnection; - private @Nullable StatefulConnection connection; + private @Nullable Mono> sharedConnection; /** * Creates new {@link LettuceReactiveRedisConnection}. @@ -59,8 +62,27 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); - this.connectionProvider = connectionProvider; - this.connection = connectionProvider.getConnection(StatefulConnection.class); + this.dedicatedConnection = new AsyncConnect(connectionProvider); + } + + /** + * Creates new {@link LettuceReactiveRedisConnection} given a shared {@link StatefulConnection connection}. + * + * @param sharedConnection 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. + * @since 2.0.1 + */ + @SuppressWarnings("unchecked") + LettuceReactiveRedisConnection(StatefulConnection sharedConnection, + LettuceConnectionProvider connectionProvider) { + + Assert.notNull(sharedConnection, "Shared StatefulConnection must not be null!"); + Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); + + this.dedicatedConnection = new AsyncConnect(connectionProvider); + this.sharedConnection = Mono.just(sharedConnection); } /* @@ -176,33 +198,55 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { * @return */ public Flux execute(LettuceReactiveCallback callback) { - return Flux.defer(() -> callback.doWithCommands(getCommands())).onErrorMap(translateException()); + return getCommands().flatMapMany(callback::doWithCommands).onErrorMap(translateException()); } - /* (non-Javadoc) + /** + * @param callback + * @return + * @since 2.0.1 + */ + public Flux executeDedicated(LettuceReactiveCallback callback) { + return getDedicatedCommands().flatMapMany(callback::doWithCommands).onErrorMap(translateException()); + } + + /* + * (non-Javadoc) * @see java.io.Closeable#close() */ @Override public void close() { - - if (connection != null) { - synchronized (connectionProvider) { - connectionProvider.release(connection); - connection = null; - } - } + dedicatedConnection.close(); } - protected StatefulConnection getConnection() { + protected Mono> getConnection() { - if (connection != null) { - return connection; + if (sharedConnection != null) { + return sharedConnection; } - throw new IllegalStateException("Connection is closed"); + return getDedicatedConnection(); } - protected RedisClusterReactiveCommands getCommands() { + protected Mono> getDedicatedConnection() { + return dedicatedConnection.getConnection().onErrorMap(translateException()); + } + + protected Mono> getCommands() { + + if (sharedConnection != null) { + return sharedConnection.map(LettuceReactiveRedisConnection::getRedisClusterReactiveCommands); + } + + return getDedicatedCommands(); + } + + protected Mono> getDedicatedCommands() { + return dedicatedConnection.getConnection().map(LettuceReactiveRedisConnection::getRedisClusterReactiveCommands); + } + + private static RedisClusterReactiveCommands getRedisClusterReactiveCommands( + StatefulConnection connection) { if (connection instanceof StatefulRedisConnection) { return ((StatefulRedisConnection) connection).reactive(); @@ -210,7 +254,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { return ((StatefulRedisClusterConnection) connection).reactive(); } - throw new RuntimeException("o.O unknown connection type " + connection); + throw new IllegalStateException("o.O unknown connection type " + connection); } Function translateException() { @@ -260,4 +304,94 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { return value.duplicate(); } } + + /** + * Asynchronous connection utility. This utility has a lifecycle controlled by {@link #getConnection()} and + * {@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. Closing: Call to {@link #close()} initiates connection closing.
    4. + *
  4. Closed: Connection is closed.
    5. + *
+ * + * @author Mark Paluch + * @since 2.0.1 + */ + static class AsyncConnect { + + private final Mono> connectionPublisher; + private final LettuceConnectionProvider connectionProvider; + + private AtomicReference state = new AtomicReference<>(State.INITIAL); + private volatile @Nullable CompletableFuture> connection; + + @SuppressWarnings("unchecked") + AsyncConnect(LettuceConnectionProvider connectionProvider) { + + Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!"); + + this.connectionProvider = connectionProvider; + + Mono> defer = Mono + .defer(() -> Mono.> just( + connectionProvider.getConnection(StatefulConnection.class))); + + this.connectionPublisher = defer.subscribeOn(Schedulers.elastic()); + } + + /** + * Obtain a connection publisher. This method connects asynchronously by requesting a connection from + * {@link LettuceConnectionProvider} with non-blocking synchronization if called concurrently. + * + * @return + */ + Mono> getConnection() { + + if (state.get() == State.CLOSED) { + throw new IllegalStateException("Connection is closed!"); + } + + CompletableFuture> connection = this.connection; + + if (connection != null) { + return Mono.fromCompletionStage(connection); + } + + if (state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED)) { + this.connection = connectionPublisher.toFuture(); + } + + for (;;) { + + connection = this.connection; + if (connection != null) { + return Mono.fromCompletionStage(connection); + } + + Thread thread = Thread.currentThread(); + if (thread.isInterrupted()) { + thread.interrupt(); + return Mono.error(new InterruptedException()); + } + } + } + + /** + * Close connection (blocking call). + */ + void close() { + + if (state.compareAndSet(State.CONNECTION_REQUESTED, State.CLOSING)) { + getConnection().doOnSuccess(connectionProvider::release).doOnSuccess(it -> state.set(State.CLOSED)).block(); + } + + state.compareAndSet(State.INITIAL, State.CLOSED); + } + + enum State { + INITIAL, CONNECTION_REQUESTED, CLOSING, CLOSED; + } + } } 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 dd4f2b8d3..9b121f38b 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 @@ -43,7 +43,8 @@ public abstract class LettuceReactiveClusterCommandsTestsBase { assumeThat(clientProvider.test(), is(true)); nativeCommands = clientProvider.getClient().connect().sync(); connection = new LettuceReactiveRedisClusterConnection( - new ClusterConnectionProvider(clientProvider.getClient(), LettuceReactiveRedisConnection.CODEC)); + new ClusterConnectionProvider(clientProvider.getClient(), LettuceReactiveRedisConnection.CODEC), + clientProvider.getClient()); } @After 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 5f211a949..69dd6379a 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 @@ -119,8 +119,10 @@ public abstract class LettuceReactiveCommandsTestsBase { this.connection = new LettuceReactiveRedisConnection(connectionProvider); } else { + ClusterConnectionProvider clusterConnectionProvider = (ClusterConnectionProvider) nativeConnectionProvider; nativeCommands = nativeConnectionProvider.getConnection(StatefulRedisClusterConnection.class).sync(); - this.connection = new LettuceReactiveRedisClusterConnection(connectionProvider); + this.connection = new LettuceReactiveRedisClusterConnection(connectionProvider, + clusterConnectionProvider.getClient()); } } 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 new file mode 100644 index 000000000..cbd9e05ea --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnectionUnitTests.java @@ -0,0 +1,198 @@ +/* + * 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 static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +import io.lettuce.core.RedisConnectionException; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; +import reactor.core.publisher.Mono; +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; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Answers; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.data.redis.RedisConnectionFailureException; + +/** + * Unit tests for {@link LettuceReactiveRedisConnection}. + * + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class LettuceReactiveRedisConnectionUnitTests { + + @Mock(answer = Answers.RETURNS_MOCKS) StatefulRedisConnection sharedConnection; + + @Mock LettuceConnectionProvider connectionProvider; + + @Before + public void before() { + when(connectionProvider.getConnection(any())).thenReturn(sharedConnection); + } + + @Test // DATAREDIS-720 + public void shouldLazilyInitializeConnection() { + + new LettuceReactiveRedisConnection(connectionProvider); + + verifyZeroInteractions(connectionProvider); + } + + @Test // DATAREDIS-720 + public void shouldExecuteUsingConnectionProvider() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + StepVerifier.create(connection.execute(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); + + verify(connectionProvider).getConnection(StatefulConnection.class); + } + + @Test // DATAREDIS-720 + public void shouldExecuteDedicatedUsingConnectionProvider() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + StepVerifier.create(connection.executeDedicated(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); + + verify(connectionProvider).getConnection(StatefulConnection.class); + } + + @Test // DATAREDIS-720 + public void shouldExecuteOnSharedConnection() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(sharedConnection, + connectionProvider); + + StepVerifier.create(connection.execute(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); + + verifyZeroInteractions(connectionProvider); + } + + @Test // DATAREDIS-720 + public void shouldExecuteDedicatedWithSharedConnection() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(sharedConnection, + connectionProvider); + + StepVerifier.create(connection.executeDedicated(cmd -> Mono.just("foo"))).expectNext("foo").verifyComplete(); + + verify(connectionProvider).getConnection(StatefulConnection.class); + } + + @Test // DATAREDIS-720 + public void shouldOperateOnDedicatedConnection() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + StepVerifier.create(connection.getConnection()).expectNextCount(1).verifyComplete(); + + verify(connectionProvider).getConnection(StatefulConnection.class); + } + + @Test // DATAREDIS-720 + public void shouldCloseOnlyDedicatedConnection() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(sharedConnection, + connectionProvider); + + StepVerifier.create(connection.getConnection()).expectNextCount(1).verifyComplete(); + StepVerifier.create(connection.getDedicatedConnection()).expectNextCount(1).verifyComplete(); + + connection.close(); + + verify(sharedConnection, never()).close(); + verify(connectionProvider, times(1)).release(sharedConnection); + } + + @Test // DATAREDIS-720 + public void shouldCloseConnectionOnlyOnce() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + StepVerifier.create(connection.getConnection()).expectNextCount(1).verifyComplete(); + + connection.close(); + connection.close(); + + verify(connectionProvider, times(1)).release(sharedConnection); + } + + @Test // DATAREDIS-720 + @SuppressWarnings("unchecked") + public void multipleCallsInProgressShouldConnectOnlyOnce() throws Exception { + + CountDownLatch latch = new CountDownLatch(1); + + reset(connectionProvider); + when(connectionProvider.getConnection(any())).thenAnswer(invocation -> { + + latch.await(); + return sharedConnection; + }); + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + CompletableFuture> first = (CompletableFuture) connection.getConnection() + .toFuture(); + CompletableFuture> second = (CompletableFuture) connection + .getConnection().toFuture(); + + assertThat(first).isNotDone(); + assertThat(second).isNotDone(); + + verify(connectionProvider, times(1)).getConnection(StatefulConnection.class); + + latch.countDown(); + + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + + assertThat(first).isCompletedWithValue(sharedConnection); + assertThat(second).isCompletedWithValue(sharedConnection); + } + + @Test // DATAREDIS-720 + public void shouldPropagateConnectionFailures() { + + reset(connectionProvider); + when(connectionProvider.getConnection(any())).thenThrow(new RedisConnectionException("something went wrong")); + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + + StepVerifier.create(connection.getConnection()).expectError(RedisConnectionFailureException.class).verify(); + } + + @Test(expected = IllegalStateException.class) // DATAREDIS-720 + public void shouldRejectCommandsAfterClose() { + + LettuceReactiveRedisConnection connection = new LettuceReactiveRedisConnection(connectionProvider); + connection.close(); + + connection.getConnection(); + } +}