From 05798fb637dc2fe2ff2823aa2de770544b36f10c Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 17 Jun 2019 14:51:33 +0200 Subject: [PATCH] DATAREDIS-976 - Allow extension of Lettuce Connection and Subscription classes. LettuceConnection, LettuceClusterConnection, and LettuceSubscription can now be properly subclassed so they can be extended and created by LettuceConnectionFactory. LettuceConnectionFactory provides template methods doCreateLettuceConnection and doCreateLettuceClusterConnection. Original pull request: #450 & #457 --- .../lettuce/LettuceClusterConnection.java | 35 ++-- .../connection/lettuce/LettuceConnection.java | 18 +- .../lettuce/LettuceConnectionFactory.java | 68 ++++-- .../lettuce/LettuceSubscription.java | 28 ++- .../LettuceClusterConnectionUnitTests.java | 6 +- ...tuceClusterKeyspaceNotificationsTests.java | 194 ++++++++++++++++++ .../redis/test/util/RedisClusterRule.java | 5 + 7 files changed, 305 insertions(+), 49 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyspaceNotificationsTests.java 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 f6e5e19f3..bec205497 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 @@ -34,6 +34,7 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.DataAccessException; import org.springframework.dao.DataAccessResourceFailureException; @@ -66,7 +67,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau new LettuceExceptionConverter()); private final Log log = LogFactory.getLog(getClass()); - private final RedisClusterClient clusterClient; private ClusterCommandExecutor clusterCommandExecutor; private ClusterTopologyProvider topologyProvider; @@ -121,8 +121,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider, "LettuceConnectionProvider must be a ClusterConnectionProvider."); - this.clusterClient = getClient(); - this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient); + this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider, new LettuceClusterNodeResourceProvider(getConnectionProvider()), exceptionConverter); this.disposeClusterCommandExecutorOnClose = true; @@ -158,8 +157,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider, "LettuceConnectionProvider must be a ClusterConnectionProvider."); - this.clusterClient = getClient(); - this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient); + this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); this.clusterCommandExecutor = executor; this.disposeClusterCommandExecutorOnClose = false; } @@ -170,22 +168,20 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau * * @param sharedConnection may be {@literal null} if no shared connection used. * @param connectionProvider must not be {@literal null}. - * @param clusterClient must not be {@literal null}. + * @param clusterTopologyProvider must not be {@literal null}. * @param executor must not be {@literal null}. * @param timeout must not be {@literal null}. * @since 2.1 */ - LettuceClusterConnection(@Nullable StatefulRedisClusterConnection sharedConnection, - LettuceConnectionProvider connectionProvider, RedisClusterClient clusterClient, ClusterCommandExecutor executor, - Duration timeout) { + protected LettuceClusterConnection(@Nullable StatefulRedisClusterConnection sharedConnection, + LettuceConnectionProvider connectionProvider, ClusterTopologyProvider clusterTopologyProvider, + ClusterCommandExecutor executor, Duration timeout) { super(sharedConnection, connectionProvider, timeout.toMillis(), 0); Assert.notNull(executor, "ClusterCommandExecutor must not be null."); - Assert.notNull(clusterClient, "RedisClusterClient must not be null."); - this.clusterClient = clusterClient; - this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient); + this.topologyProvider = clusterTopologyProvider; this.clusterCommandExecutor = executor; this.disposeClusterCommandExecutorOnClose = false; } @@ -205,13 +201,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau connectionProvider.getClass().getName())); } - /** - * @return access to {@link RedisClusterClient} for non-connection access. - */ - private Partitions getPartitions() { - return clusterClient.getPartitions(); - } - /* * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#geoCommands() @@ -330,7 +319,11 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau @Override public RedisClusterNode clusterGetNodeForSlot(int slot) { - return LettuceConverters.toRedisClusterNode(getPartitions().getPartitionBySlot(slot)); + Set nodes = topologyProvider.getTopology().getSlotServingNodes(slot); + if (nodes.isEmpty()) { + return null; + } + return nodes.iterator().next(); } /* @@ -574,7 +567,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau */ @Override public List clusterGetNodes() { - return LettuceConverters.partitionsToClusterNodes(getPartitions()); + return new ArrayList<>(topologyProvider.getTopology().getNodes()); } /* 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 e41fdbcd9..830e9a014 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 @@ -879,7 +879,21 @@ public class LettuceConnection extends AbstractRedisConnection { } private LettuceSubscription initSubscription(MessageListener listener) { - return new LettuceSubscription(listener, switchToPubSub(), connectionProvider); + return doCreateSubscription(listener, switchToPubSub(), connectionProvider); + } + + /** + * Customization hook to create a {@link LettuceSubscription}. + * + * @param listener the {@link MessageListener} to notify. + * @param connection Pub/Sub connection. + * @param connectionProvider the {@link LettuceConnectionProvider} for connection release. + * @return a {@link LettuceSubscription}. + * @since 2.2 + */ + protected LettuceSubscription doCreateSubscription(MessageListener listener, + StatefulRedisPubSubConnection connection, LettuceConnectionProvider connectionProvider) { + return new LettuceSubscription(listener, connection, connectionProvider); } void pipeline(LettuceResult result) { @@ -1250,7 +1264,7 @@ public class LettuceConnection extends AbstractRedisConnection { } @RequiredArgsConstructor - private class LettucePoolConnectionProvider implements LettuceConnectionProvider { + static class LettucePoolConnectionProvider implements LettuceConnectionProvider { private final LettucePool pool; 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 74e24db75..1b6516387 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 @@ -15,6 +15,8 @@ */ package org.springframework.data.redis.connection.lettuce; +import static org.springframework.data.redis.connection.lettuce.LettuceConnection.*; + import io.lettuce.core.AbstractRedisClient; import io.lettuce.core.ClientOptions; import io.lettuce.core.ReadFrom; @@ -40,6 +42,7 @@ import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; @@ -273,7 +276,7 @@ public class LettuceConnectionFactory this.client = createClient(); - this.connectionProvider = createConnectionProvider(client, LettuceConnection.CODEC); + this.connectionProvider = createConnectionProvider(client, CODEC); this.reactiveConnectionProvider = createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC); if (isClusterAware()) { @@ -341,13 +344,7 @@ public class LettuceConnectionFactory } LettuceConnection connection; - - if (pool != null) { - connection = new LettuceConnection(getSharedConnection(), getTimeout(), null, pool, getDatabase()); - } else { - connection = new LettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), getDatabase()); - } - + connection = doCreateLettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), getDatabase()); connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults); return connection; } @@ -365,12 +362,51 @@ public class LettuceConnectionFactory RedisClusterClient clusterClient = (RedisClusterClient) client; - return getShareNativeConnection() - ? new LettuceClusterConnection( - (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection(), - connectionProvider, clusterClient, clusterCommandExecutor, clientConfiguration.getCommandTimeout()) - : new LettuceClusterConnection(null, connectionProvider, clusterClient, clusterCommandExecutor, - clientConfiguration.getCommandTimeout()); + StatefulRedisClusterConnection sharedConnection = getShareNativeConnection() + ? (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection() + : null; + + LettuceClusterTopologyProvider topologyProvider = new LettuceClusterTopologyProvider(clusterClient); + return doCreateLettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, + clusterCommandExecutor, clientConfiguration.getCommandTimeout()); + } + + /** + * Customization hook for {@link LettuceConnection} creation. + * + * @param sharedConnection the shared {@link StatefulRedisConnection} if {@link #getShareNativeConnection()} is + * {@literal true}; {@literal null} otherwise. + * @param connectionProvider the {@link LettuceConnectionProvider} to release connections. + * @param timeout command timeout in {@link TimeUnit#MILLISECONDS}. + * @param database database index to operate on. + * @return the {@link LettuceConnection}. + * @since 2.2 + */ + protected LettuceConnection doCreateLettuceConnection(StatefulRedisConnection sharedConnection, + LettuceConnectionProvider connectionProvider, long timeout, int database) { + + return new LettuceConnection(sharedConnection, connectionProvider, timeout, database); + } + + /** + * Customization hook for {@link LettuceClusterConnection} creation. + * + * @param sharedConnection the shared {@link StatefulRedisConnection} if {@link #getShareNativeConnection()} is + * {@literal true}; {@literal null} otherwise. + * @param connectionProvider the {@link LettuceConnectionProvider} to release connections. + * @param topologyProvider the {@link ClusterTopologyProvider}. + * @param clusterCommandExecutor the {@link ClusterCommandExecutor} to release connections. + * @param commandTimeout command timeout {@link Duration}. + * @return the {@link LettuceConnection}. + * @since 2.2 + */ + protected LettuceClusterConnection doCreateLettuceClusterConnection( + StatefulRedisClusterConnection sharedConnection, LettuceConnectionProvider connectionProvider, + ClusterTopologyProvider topologyProvider, ClusterCommandExecutor clusterCommandExecutor, + Duration commandTimeout) { + + return new LettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, clusterCommandExecutor, + commandTimeout); } /* @@ -909,6 +945,10 @@ public class LettuceConnectionFactory private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec codec) { + if (this.pool != null) { + return new LettucePoolConnectionProvider(this.pool); + } + LettuceConnectionProvider connectionProvider = doCreateConnectionProvider(client, codec); if (this.clientConfiguration instanceof LettucePoolingClientConfiguration) { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java index c9b785953..c0a0c9e8f 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java @@ -28,15 +28,23 @@ import org.springframework.data.redis.connection.util.AbstractSubscription; * @author Mark Paluch * @author Christoph Strobl */ -class LettuceSubscription extends AbstractSubscription { +public class LettuceSubscription extends AbstractSubscription { private final StatefulRedisPubSubConnection connection; private final LettuceMessageListener listener; private final LettuceConnectionProvider connectionProvider; private final RedisPubSubCommands pubsub; - LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection, - LettuceConnectionProvider connectionProvider) { + /** + * Creates a new {@link LettuceSubscription} given {@link MessageListener}, {@link StatefulRedisPubSubConnection}, and + * {@link LettuceConnectionProvider}. + * + * @param listener the listener to notify, must not be {@literal null}. + * @param pubsubConnection must not be {@literal null}. + * @param connectionProvider must not be {@literal null}. + */ + protected LettuceSubscription(MessageListener listener, + StatefulRedisPubSubConnection pubsubConnection, LettuceConnectionProvider connectionProvider) { super(listener); @@ -52,25 +60,25 @@ class LettuceSubscription extends AbstractSubscription { return connection; } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doClose() */ protected void doClose() { if (!getChannels().isEmpty()) { - pubsub.unsubscribe(new byte[0]); + doUnsubscribe(true, new byte[0]); } if (!getPatterns().isEmpty()) { - pubsub.punsubscribe(new byte[0]); + doPUnsubscribe(true, new byte[0]); } connection.removeListener(this.listener); connectionProvider.release(connection); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPsubscribe(byte[][]) */ @@ -78,7 +86,7 @@ class LettuceSubscription extends AbstractSubscription { pubsub.psubscribe(patterns); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doPUnsubscribe(boolean, byte[][]) */ @@ -88,7 +96,7 @@ class LettuceSubscription extends AbstractSubscription { pubsub.punsubscribe(patterns); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doSubscribe(byte[][]) */ @@ -96,7 +104,7 @@ class LettuceSubscription extends AbstractSubscription { pubsub.subscribe(channels); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.util.AbstractSubscription#doUnsubscribe(boolean, byte[][]) */ diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java index 8e14db4ed..94770850c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java @@ -48,6 +48,7 @@ import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.data.redis.connection.ClusterCommandExecutor; import org.springframework.data.redis.connection.ClusterNodeResourceProvider; +import org.springframework.data.redis.connection.ClusterTopologyProvider; import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots; import org.springframework.data.redis.connection.RedisClusterNode; @@ -67,6 +68,7 @@ public class LettuceClusterConnectionUnitTests { static final byte[] KEY_3_BYTES = KEY_3.getBytes(); @Mock RedisClusterClient clusterMock; + @Mock ClusterTopologyProvider topologyProviderMock; @Mock LettuceConnectionProvider connectionProviderMock; @Mock ClusterCommandExecutor executorMock; @@ -363,7 +365,7 @@ public class LettuceClusterConnectionUnitTests { when(sharedConnectionMock.sync()).thenReturn(sync); LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, - clusterMock, executorMock, Duration.ZERO); + topologyProviderMock, executorMock, Duration.ZERO); connection.keyCommands().del(KEY_1_BYTES); @@ -381,7 +383,7 @@ public class LettuceClusterConnectionUnitTests { when(dedicatedConnection.sync()).thenReturn(sync); LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, - clusterMock, executorMock, Duration.ZERO); + topologyProviderMock, executorMock, Duration.ZERO); connection.listCommands().bLPop(1, KEY_1_BYTES); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyspaceNotificationsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyspaceNotificationsTests.java new file mode 100644 index 000000000..32b79fafa --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyspaceNotificationsTests.java @@ -0,0 +1,194 @@ +/* + * Copyright 2019 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 + * + * https://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 io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.SetArgs; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; +import io.lettuce.core.cluster.SlotHash; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +import org.junit.After; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.data.redis.ConnectionFactoryTracker; +import org.springframework.data.redis.connection.ClusterCommandExecutor; +import org.springframework.data.redis.connection.ClusterTopologyProvider; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.RedisClusterConnection; +import org.springframework.data.redis.connection.RedisConfiguration; +import org.springframework.data.redis.test.util.RedisClusterRule; +import org.springframework.lang.Nullable; + +/** + * Integration tests to listen for keyspace notifications. + * + * @author Mark Paluch + */ +public class LettuceClusterKeyspaceNotificationsTests { + + @ClassRule public static RedisClusterRule clusterRule = new RedisClusterRule(); + + CustomLettuceConnectionFactory factory; + String keyspaceConfig; + + // maps to 127.0.0.1:7381/slot hash 13477 + String key = "10923"; + + @Before + public void before() { + + factory = new CustomLettuceConnectionFactory(clusterRule.getConfiguration()); + factory.setClientResources(LettuceTestClientResources.getSharedClientResources()); + ConnectionFactoryTracker.add(factory); + factory.afterPropertiesSet(); + + // enable keyspace events on a specific node. + withConnection("127.0.0.1", 7381, commands -> { + + keyspaceConfig = commands.configGet("*").get("notify-keyspace-events"); + commands.configSet("notify-keyspace-events", "KEx"); + }); + + assertThat(SlotHash.getSlot(key)).isEqualTo(13477); + } + + @After + public void tearDown() { + + // Restore previous settings. + withConnection("127.0.0.1", 7381, commands -> { + commands.configSet("notify-keyspace-events", keyspaceConfig); + }); + } + + @Test // DATAREDIS-976 + public void shouldListenForKeyspaceNotifications() throws Exception { + + CompletableFuture expiry = new CompletableFuture<>(); + + RedisClusterConnection connection = factory.getClusterConnection(); + + connection.pSubscribe((message, pattern) -> { + expiry.complete(new String(message.getBody()) + ":" + new String(message.getChannel())); + }, "__keyspace*@*".getBytes()); + + withConnection("127.0.0.1", 7381, commands -> { + commands.set(key, "foo", SetArgs.Builder.px(1)); + }); + + assertThat(expiry.get(2, TimeUnit.SECONDS)).isEqualTo("expired:__keyspace@0__:10923"); + + connection.getSubscription().close(); + connection.close(); + } + + private void withConnection(String hostname, int port, Consumer> commandsConsumer) { + + RedisClient client = RedisClient.create(LettuceTestClientResources.getSharedClientResources(), + RedisURI.create(hostname, port)); + + StatefulRedisConnection connection = client.connect(); + commandsConsumer.accept(connection.sync()); + + connection.close(); + client.shutdownAsync(); + } + + static class CustomLettuceConnectionFactory extends LettuceConnectionFactory { + + CustomLettuceConnectionFactory(RedisConfiguration redisConfiguration) { + super(redisConfiguration); + } + + @Override + protected LettuceClusterConnection doCreateLettuceClusterConnection( + StatefulRedisClusterConnection sharedConnection, LettuceConnectionProvider connectionProvider, + ClusterTopologyProvider topologyProvider, ClusterCommandExecutor clusterCommandExecutor, + Duration commandTimeout) { + return new CustomLettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, + clusterCommandExecutor, commandTimeout); + } + } + + static class CustomLettuceClusterConnection extends LettuceClusterConnection { + + CustomLettuceClusterConnection(@Nullable StatefulRedisClusterConnection sharedConnection, + LettuceConnectionProvider connectionProvider, ClusterTopologyProvider clusterTopologyProvider, + ClusterCommandExecutor executor, Duration timeout) { + super(sharedConnection, connectionProvider, clusterTopologyProvider, executor, timeout); + } + + @Override + protected LettuceSubscription doCreateSubscription(MessageListener listener, + StatefulRedisPubSubConnection connection, LettuceConnectionProvider connectionProvider) { + return new CustomLettuceSubscription(listener, (StatefulRedisClusterPubSubConnection) connection, + connectionProvider); + } + } + + /** + * Customized {@link LettuceSubscription}. Enables + * {@link StatefulRedisClusterPubSubConnection#setNodeMessagePropagation(boolean)} and uses + * {@link io.lettuce.core.cluster.api.sync.NodeSelection} to subscribe to all master nodes. + */ + static class CustomLettuceSubscription extends LettuceSubscription { + + private final StatefulRedisClusterPubSubConnection connection; + + CustomLettuceSubscription(MessageListener listener, StatefulRedisClusterPubSubConnection connection, + LettuceConnectionProvider connectionProvider) { + super(listener, connection, connectionProvider); + this.connection = connection; + + // Must be enabled for keyspace notification propagation + this.connection.setNodeMessagePropagation(true); + } + + @Override + protected void doPsubscribe(byte[]... patterns) { + connection.sync().all().commands().psubscribe(patterns); + } + + @Override + protected void doPUnsubscribe(boolean all, byte[]... patterns) { + connection.sync().all().commands().punsubscribe(); + } + + @Override + protected void doSubscribe(byte[]... channels) { + connection.sync().all().commands().subscribe(channels); + } + + @Override + protected void doUnsubscribe(boolean all, byte[]... channels) { + connection.sync().all().commands().unsubscribe(); + } + } +} diff --git a/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java b/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java index 5fd9a0416..6c85f60ab 100644 --- a/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java +++ b/src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java @@ -23,6 +23,7 @@ import org.junit.Assume; import org.junit.rules.ExternalResource; import org.junit.rules.TestRule; import org.springframework.data.redis.connection.RedisClusterConfiguration; +import org.springframework.data.redis.connection.RedisConfiguration; import org.springframework.data.redis.connection.RedisNode; import org.springframework.data.redis.connection.jedis.JedisConverters; @@ -63,6 +64,10 @@ public class RedisClusterRule extends ExternalResource { Assume.assumeThat(mode, is("cluster")); } + public RedisConfiguration getConfiguration() { + return this.clusterConfig; + } + private void init() { if (clusterConfig == null) {