diff --git a/pom.xml b/pom.xml index 2486f880c..0e9542456 100644 --- a/pom.xml +++ b/pom.xml @@ -22,7 +22,7 @@ 1.9.2 1.4.8 2.2 - 3.4.2.Final + 4.2.2.Final 2.9.0 0.7 06052013 diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java index 610d103d5..d1448ab00 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -372,6 +372,10 @@ public class ClusterCommandExecutor implements DisposableBean { if (executor instanceof DisposableBean) { ((DisposableBean) executor).destroy(); } + + if (resourceProvider instanceof DisposableBean) { + ((DisposableBean) resourceProvider).destroy(); + } } /** diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java b/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java index e37a5151d..1853d6da4 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java @@ -15,54 +15,44 @@ */ package org.springframework.data.redis.connection.lettuce; -import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; -import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisURI; +import com.lambdaworks.redis.api.StatefulRedisConnection; +import com.lambdaworks.redis.api.async.RedisAsyncCommands; import com.lambdaworks.redis.codec.RedisCodec; -import com.lambdaworks.redis.pubsub.RedisPubSubConnection; +import com.lambdaworks.redis.pubsub.StatefulRedisPubSubConnection; /** * Extension of {@link RedisClient} that calls auth on all new connections using the supplied credentials * * @author Jennifer Hickey - * @author Mar Paluch + * @author Mark Paluch * @author Christoph Strobl * @deprecated since 1.6 - Please use {@link RedisURI#setPassword(String)} */ @Deprecated public class AuthenticatingRedisClient extends RedisClient { - private String password; - public AuthenticatingRedisClient(String host, int port, String password) { - super(host, port); - this.password = password; + super(null, RedisURI.builder().withHost(host).withPort(port).withPassword(password).build()); } public AuthenticatingRedisClient(String host, String password) { - super(host); - this.password = password; + super(null, RedisURI.builder().withHost(host).withPassword(password).build()); } @Override - public RedisConnection connect(RedisCodec codec) { - RedisConnection conn = super.connect(codec); - conn.auth(password); - return conn; + public StatefulRedisConnection connect(RedisCodec codec) { + return super.connect(codec); } @Override - public RedisAsyncConnection connectAsync(RedisCodec codec) { - RedisAsyncConnection conn = super.connectAsync(codec); - conn.auth(password); - return conn; + public RedisAsyncCommands connectAsync(RedisCodec codec) { + return super.connectAsync(codec); } @Override - public RedisPubSubConnection connectPubSub(RedisCodec codec) { - RedisPubSubConnection conn = super.connectPubSub(codec); - conn.auth(password); - return conn; + public StatefulRedisPubSubConnection connectPubSub(RedisCodec codec) { + return super.connectPubSub(codec); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/BytesRedisCodec.java b/src/main/java/org/springframework/data/redis/connection/lettuce/BytesRedisCodec.java deleted file mode 100644 index 67b7d967e..000000000 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/BytesRedisCodec.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 2011-2013 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.nio.ByteBuffer; - -import com.lambdaworks.redis.codec.RedisCodec; - -/** - * Basic codec that returns the raw data as byte[]. - * - * @author Costin Leau - */ -class BytesRedisCodec extends RedisCodec { - - @Override - public byte[] decodeKey(ByteBuffer bytes) { - return getBytes(bytes); - } - - @Override - public byte[] decodeValue(ByteBuffer bytes) { - return getBytes(bytes); - } - - @Override - public byte[] encodeKey(byte[] key) { - return key; - } - - @Override - public byte[] encodeValue(byte[] value) { - return value; - } - - private static byte[] getBytes(ByteBuffer buffer) { - byte[] b = new byte[buffer.remaining()]; - buffer.get(b); - return b; - } -} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java index be52945a9..eb2ef0085 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java @@ -28,9 +28,10 @@ import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; import com.lambdaworks.redis.RedisURI; +import com.lambdaworks.redis.api.StatefulConnection; +import com.lambdaworks.redis.api.StatefulRedisConnection; import com.lambdaworks.redis.resource.ClientResources; /** @@ -43,7 +44,7 @@ import com.lambdaworks.redis.resource.ClientResources; public class DefaultLettucePool implements LettucePool, InitializingBean { @SuppressWarnings("rawtypes") // - private GenericObjectPool internalPool; + private GenericObjectPool> internalPool; private RedisClient client; private int dbIndex = 0; private GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); @@ -112,7 +113,8 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS); - this.internalPool = new GenericObjectPool(new LettuceFactory(client, dbIndex), poolConfig); + this.internalPool = new GenericObjectPool>(new LettuceFactory(client, dbIndex), + poolConfig); } /** @@ -135,7 +137,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } @SuppressWarnings("unchecked") - public RedisAsyncConnection getResource() { + public StatefulConnection getResource() { try { return internalPool.borrowObject(); } catch (Exception e) { @@ -143,7 +145,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } } - public void returnBrokenResource(final RedisAsyncConnection resource) { + public void returnBrokenResource(final StatefulConnection resource) { try { internalPool.invalidateObject(resource); } catch (Exception e) { @@ -151,7 +153,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } } - public void returnResource(final RedisAsyncConnection resource) { + public void returnResource(final StatefulConnection resource) { try { internalPool.returnObject(resource); } catch (Exception e) { @@ -302,7 +304,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } @SuppressWarnings("rawtypes") - private static class LettuceFactory extends BasePooledObjectFactory { + private static class LettuceFactory extends BasePooledObjectFactory> { private final RedisClient client; @@ -315,11 +317,14 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } @Override - public void activateObject(PooledObject pooledObject) throws Exception { - pooledObject.getObject().select(dbIndex); + public void activateObject(PooledObject> pooledObject) throws Exception { + + if (pooledObject.getObject() instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) pooledObject.getObject()).sync().select(dbIndex); + } } - public void destroyObject(final PooledObject obj) throws Exception { + public void destroyObject(final PooledObject> obj) throws Exception { try { obj.getObject().close(); } catch (Exception e) { @@ -327,9 +332,11 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } } - public boolean validateObject(final PooledObject obj) { + public boolean validateObject(final PooledObject> obj) { try { - obj.getObject().ping(); + if (obj.getObject() instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) obj.getObject()).sync().ping(); + } return true; } catch (Exception e) { return false; @@ -337,14 +344,13 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } @Override - public RedisAsyncConnection create() throws Exception { - return client.connectAsync(LettuceConnection.CODEC); + public StatefulConnection create() throws Exception { + return client.connect(LettuceConnection.CODEC); } @Override - public PooledObject wrap(RedisAsyncConnection obj) { - return new DefaultPooledObject(obj); + public PooledObject> wrap(StatefulConnection obj) { + return new DefaultPooledObject>(obj); } - } } 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 c84461524..107aed9ce 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 @@ -30,7 +30,7 @@ import java.util.Random; import java.util.Set; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.dao.DataAccessResourceFailureException; +import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; @@ -57,14 +57,14 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import com.lambdaworks.redis.KeyValue; -import com.lambdaworks.redis.RedisAsyncConnection; -import com.lambdaworks.redis.RedisAsyncConnectionImpl; -import com.lambdaworks.redis.RedisClusterConnection; -import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisException; +import com.lambdaworks.redis.api.StatefulConnection; import com.lambdaworks.redis.cluster.RedisClusterClient; import com.lambdaworks.redis.cluster.SlotHash; +import com.lambdaworks.redis.cluster.api.StatefulRedisClusterConnection; +import com.lambdaworks.redis.cluster.api.sync.RedisClusterCommands; import com.lambdaworks.redis.cluster.models.partitions.Partitions; +import com.lambdaworks.redis.codec.ByteArrayCodec; import com.lambdaworks.redis.codec.RedisCodec; /** @@ -77,7 +77,7 @@ public class LettuceClusterConnection extends LettuceConnection static final ExceptionTranslationStrategy exceptionConverter = new PassThroughExceptionTranslationStrategy( new LettuceExceptionConverter()); - static final RedisCodec CODEC = new BytesRedisCodec(); + static final RedisCodec CODEC = ByteArrayCodec.INSTANCE; private final RedisClusterClient clusterClient; private ClusterCommandExecutor clusterCommandExecutor; @@ -141,7 +141,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection connection) { + public List doInCluster(RedisClusterCommands connection) { return connection.keys(pattern); } }).resultsAsList(); @@ -164,7 +164,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.flushall(); } }); @@ -180,7 +180,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.flushdb(); } }); @@ -197,7 +197,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public Long doInCluster(RedisClusterConnection client) { + public Long doInCluster(RedisClusterCommands client) { return client.dbsize(); } @@ -227,7 +227,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public Properties doInCluster(RedisClusterConnection client) { + public Properties doInCluster(RedisClusterCommands client) { return LettuceConverters.toProperties(client.info()); } }).getResults(); @@ -249,7 +249,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public Properties doInCluster(RedisClusterConnection client) { + public Properties doInCluster(RedisClusterCommands client) { return LettuceConverters.toProperties(client.info(section)); } }).getResults(); @@ -274,7 +274,7 @@ public class LettuceClusterConnection extends LettuceConnection .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.info(section); } }, node).getValue()); @@ -298,16 +298,8 @@ public class LettuceClusterConnection extends LettuceConnection Assert.noNullElements(keys, "Keys must not be null or contain null key!"); - if (ClusterSlotHashUtil.isSameSlotForAllKeys(keys)) { - return super.del(keys); - } - - long total = 0; - for (byte[] key : keys) { - Long delted = super.del(key); - total += (delted != null ? delted.longValue() : 0); - } - return Long.valueOf(total); + // Routing for mget is handled by lettuce itself. + return super.del(keys); } /* @@ -325,7 +317,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override - public Set doInCluster(RedisClusterConnection client) { + public Set doInCluster(RedisClusterCommands client) { return LettuceConverters.toSetOfRedisClusterNodes(client.clusterSlaves(nodeToUse.getId())); } }, master).getValue(); @@ -371,7 +363,7 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback() { @Override - public ClusterInfo doInCluster(RedisClusterConnection client) { + public ClusterInfo doInCluster(RedisClusterCommands client) { return new ClusterInfo(LettuceConverters.toProperties(client.clusterInfo())); } }).getValue(); @@ -387,7 +379,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clusterAddSlots(slots); } }, node); @@ -416,7 +408,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clusterDelSlots(slots); } }, node); @@ -448,7 +440,7 @@ public class LettuceClusterConnection extends LettuceConnection this.clusterCommandExecutor.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clusterForget(nodeToRemove.getId()); } @@ -469,7 +461,7 @@ public class LettuceClusterConnection extends LettuceConnection this.clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clusterMeet(node.getHost(), node.getPort()); } }); @@ -491,7 +483,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { switch (mode) { case MIGRATING: return client.clusterSetSlotMigrating(slot, nodeId); @@ -547,7 +539,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clusterReplicate(masterNode.getId()); } }, slave); @@ -563,8 +555,8 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection connection) { - return doPing(connection); + public String doInCluster(RedisClusterCommands connection) { + return connection.ping(); } }).resultsAsList(); @@ -587,29 +579,12 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { - return doPing(client); + public String doInCluster(RedisClusterCommands client) { + return client.ping(); } }, node).getValue(); } - protected String doPing(RedisClusterConnection client) { - - if (client instanceof RedisConnection) { - return ((RedisConnection) client).ping(); - } - - if (client instanceof RedisAsyncConnectionImpl) { - try { - return (String) ((RedisAsyncConnectionImpl) client).ping().get(); - } catch (Exception e) { - throw exceptionConverter.translate(e); - } - } - - throw new DataAccessResourceFailureException("Cannot execute ping using " + client); - } - /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisClusterConnection#bgReWriteAof(org.springframework.data.redis.connection.RedisClusterNode) @@ -620,7 +595,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.bgrewriteaof(); } }, node); @@ -636,7 +611,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.bgsave(); } }, node); @@ -652,7 +627,7 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public Long doInCluster(RedisClusterConnection client) { + public Long doInCluster(RedisClusterCommands client) { return client.lastsave().getTime(); } }, node).getValue(); @@ -668,7 +643,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.save(); } }, node); @@ -685,7 +660,7 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public Long doInCluster(RedisClusterConnection client) { + public Long doInCluster(RedisClusterCommands client) { return client.dbsize(); } }, node).getValue(); @@ -701,7 +676,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.flushdb(); } }, node); @@ -717,7 +692,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.flushall(); } }, node); @@ -735,7 +710,7 @@ public class LettuceClusterConnection extends LettuceConnection .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.info(); } }, node).getValue()); @@ -752,7 +727,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection client) { + public List doInCluster(RedisClusterCommands client) { return client.keys(pattern); } }, node).getValue()); @@ -768,7 +743,7 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public byte[] doInCluster(RedisClusterConnection client) { + public byte[] doInCluster(RedisClusterCommands client) { return client.randomkey(); } }, node).getValue(); @@ -855,7 +830,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public Void doInCluster(RedisClusterConnection client) { + public Void doInCluster(RedisClusterCommands client) { client.shutdown(true); return null; } @@ -899,17 +874,10 @@ public class LettuceClusterConnection extends LettuceConnection @Override public List mGet(byte[]... keys) { - if (ClusterSlotHashUtil.isSameSlotForAllKeys(keys)) { - return super.mGet(keys); - } + Assert.notNull(keys, "Keys must not be null!"); - return this.clusterCommandExecutor.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback() { - - @Override - public byte[] doInCluster(RedisClusterConnection client, byte[] key) { - return client.get(key); - } - }, Arrays.asList(keys)).resultsAsListSortBy(keys); + // Routing for mget is handled by lettuce itself. + return super.mGet(keys); } /* @@ -919,16 +887,10 @@ public class LettuceClusterConnection extends LettuceConnection @Override public void mSet(Map tuples) { - Assert.notNull(tuples, "Tuple must not be null!"); + Assert.notNull(tuples, "Tuples must not be null!"); - if (ClusterSlotHashUtil.isSameSlotForAllKeys(tuples.keySet().toArray(new byte[tuples.keySet().size()][]))) { - super.mSet(tuples); - return; - } - - for (Map.Entry entry : tuples.entrySet()) { - set(entry.getKey(), entry.getValue()); - } + // Routing for msetnx is handled by lettuce itself. + super.mSet(tuples); } /* @@ -966,7 +928,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override - public KeyValue doInCluster(RedisClusterConnection client, byte[] key) { + public KeyValue doInCluster(RedisClusterCommands client, byte[] key) { return client.blpop(timeout, key); } }, Arrays.asList(keys)).resultsAsList(); @@ -995,7 +957,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override - public KeyValue doInCluster(RedisClusterConnection client, byte[] key) { + public KeyValue doInCluster(RedisClusterCommands client, byte[] key) { return client.brpop(timeout, key); } }, Arrays.asList(keys)).resultsAsList(); @@ -1045,6 +1007,18 @@ public class LettuceClusterConnection extends LettuceConnection return null; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisConnectionCommands#select(int) + */ + @Override + public void select(int dbIndex) { + + if (dbIndex != 0) { + throw new InvalidDataAccessApiUsageException("Cannot SELECT non zero index in cluster mode."); + } + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#sMove(byte[], byte[], byte[]) @@ -1079,7 +1053,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override - public Set doInCluster(RedisClusterConnection client, byte[] key) { + public Set doInCluster(RedisClusterCommands client, byte[] key) { return client.smembers(key); } }, Arrays.asList(keys)).resultsAsList(); @@ -1140,7 +1114,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override - public Set doInCluster(RedisClusterConnection client, byte[] key) { + public Set doInCluster(RedisClusterCommands client, byte[] key) { return client.smembers(key); } }, Arrays.asList(keys)).resultsAsList(); @@ -1196,7 +1170,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override - public Set doInCluster(RedisClusterConnection client, byte[] key) { + public Set doInCluster(RedisClusterCommands client, byte[] key) { return client.smembers(key); } }, Arrays.asList(others)).resultsAsList(); @@ -1238,8 +1212,8 @@ public class LettuceClusterConnection extends LettuceConnection * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#getAsyncDedicatedConnection() */ @Override - protected RedisAsyncConnection doGetAsyncDedicatedConnection() { - return (RedisAsyncConnection) clusterClient.connectClusterAsync(CODEC); + protected StatefulConnection doGetAsyncDedicatedConnection() { + return clusterClient.connect(CODEC); } // --> cluster node stuff @@ -1332,7 +1306,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandOnAllNodes(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection client) { + public List doInCluster(RedisClusterCommands client) { return client.configGet(pattern); } }).getResults(); @@ -1360,7 +1334,7 @@ public class LettuceClusterConnection extends LettuceConnection return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection client) { + public List doInCluster(RedisClusterCommands client) { return client.configGet(pattern); } }, node).getValue(); @@ -1376,7 +1350,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.configSet(param, value); } }); @@ -1392,7 +1366,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.configSet(param, value); } }, node); @@ -1409,7 +1383,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.configResetstat(); } }); @@ -1425,7 +1399,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.configResetstat(); } }, node); @@ -1442,7 +1416,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection client) { + public List doInCluster(RedisClusterCommands client) { return client.time(); } }).getValue()); @@ -1459,7 +1433,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override - public List doInCluster(RedisClusterConnection client) { + public List doInCluster(RedisClusterCommands client) { return client.time(); } }, node).getValue()); @@ -1485,7 +1459,7 @@ public class LettuceClusterConnection extends LettuceConnection List map = clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clientList(); } }).resultsAsList(); @@ -1508,7 +1482,7 @@ public class LettuceClusterConnection extends LettuceConnection clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override - public String doInCluster(RedisClusterConnection client) { + public String doInCluster(RedisClusterCommands client) { return client.clientList(); } }, node).getValue()); @@ -1525,7 +1499,7 @@ public class LettuceClusterConnection extends LettuceConnection .executeCommandAsyncOnNodes(new LettuceClusterCommandCallback>() { @Override - public Set doInCluster(RedisClusterConnection client) { + public Set doInCluster(RedisClusterCommands client) { return Converters.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId())); } }, topologyProvider.getTopology().getActiveMasterNodes()).getResults(); @@ -1547,7 +1521,7 @@ public class LettuceClusterConnection extends LettuceConnection * @since 1.7 */ protected interface LettuceClusterCommandCallback - extends ClusterCommandCallback, T> {} + extends ClusterCommandCallback, T> {} /** * Lettuce specific implementation of {@link MultiKeyClusterCommandCallback}. @@ -1557,7 +1531,7 @@ public class LettuceClusterConnection extends LettuceConnection * @since 1.7 */ protected interface LettuceMultiKeyClusterCommandCallback - extends MultiKeyClusterCommandCallback, T> { + extends MultiKeyClusterCommandCallback, T> { } @@ -1567,9 +1541,10 @@ public class LettuceClusterConnection extends LettuceConnection * @author Christoph Strobl * @since 1.7 */ - static class LettuceClusterNodeResourceProvider implements ClusterNodeResourceProvider { + static class LettuceClusterNodeResourceProvider implements ClusterNodeResourceProvider, DisposableBean { private final RedisClusterClient client; + private volatile StatefulRedisClusterConnection connection; public LettuceClusterNodeResourceProvider(RedisClusterClient client) { @@ -1578,14 +1553,20 @@ public class LettuceClusterConnection extends LettuceConnection @Override @SuppressWarnings("unchecked") - public RedisClusterConnection getResourceForSpecificNode(RedisClusterNode node) { + public RedisClusterCommands getResourceForSpecificNode(RedisClusterNode node) { Assert.notNull(node, "Node must not be null!"); + if (connection == null) { + synchronized (this) { + if (connection == null) { + this.connection = client.connect(CODEC); + } + } + } + try { - RedisClusterConnection connection = client.connectCluster(CODEC).getConnection(node.getHost(), - node.getPort()); - return connection; + return connection.getConnection(node.getHost(), node.getPort()).sync(); } catch (RedisException e) { // unwrap cause when cluster node not known in cluster @@ -1598,12 +1579,14 @@ public class LettuceClusterConnection extends LettuceConnection @Override @SuppressWarnings("unchecked") - public void returnResourceForSpecificNode(RedisClusterNode node, Object resource) { + public void returnResourceForSpecificNode(RedisClusterNode node, Object resource) {} - RedisClusterConnection connection = (RedisClusterConnection) resource; - connection.close(); + @Override + public void destroy() throws Exception { + if (connection != null) { + connection.close(); + } } - } /** 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 bc7f13ac9..453a50844 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 @@ -18,7 +18,6 @@ package org.springframework.data.redis.connection.lettuce; import static com.lambdaworks.redis.protocol.CommandType.*; import java.lang.reflect.Constructor; -import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -32,7 +31,6 @@ import java.util.Properties; import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -74,7 +72,6 @@ import org.springframework.data.redis.core.types.RedisClientInfo; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.ObjectUtils; -import org.springframework.util.ReflectionUtils; import com.lambdaworks.redis.AbstractRedisClient; import com.lambdaworks.redis.GeoArgs; @@ -83,14 +80,9 @@ import com.lambdaworks.redis.GeoWithin; import com.lambdaworks.redis.KeyScanCursor; import com.lambdaworks.redis.LettuceFutures; import com.lambdaworks.redis.MapScanCursor; -import com.lambdaworks.redis.RedisAsyncConnection; -import com.lambdaworks.redis.RedisAsyncConnectionImpl; -import com.lambdaworks.redis.RedisChannelHandler; import com.lambdaworks.redis.RedisClient; -import com.lambdaworks.redis.RedisClusterConnection; -import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisException; -import com.lambdaworks.redis.RedisSentinelAsyncConnection; +import com.lambdaworks.redis.RedisFuture; import com.lambdaworks.redis.RedisURI; import com.lambdaworks.redis.ScanArgs; import com.lambdaworks.redis.ScoredValue; @@ -98,9 +90,20 @@ import com.lambdaworks.redis.ScoredValueScanCursor; import com.lambdaworks.redis.SortArgs; import com.lambdaworks.redis.ValueScanCursor; import com.lambdaworks.redis.ZStoreArgs; +import com.lambdaworks.redis.api.StatefulConnection; +import com.lambdaworks.redis.api.StatefulRedisConnection; +import com.lambdaworks.redis.api.async.RedisAsyncCommands; +import com.lambdaworks.redis.api.async.RedisHLLAsyncCommands; +import com.lambdaworks.redis.api.sync.RedisCommands; +import com.lambdaworks.redis.api.sync.RedisHLLCommands; +import com.lambdaworks.redis.cluster.api.StatefulRedisClusterConnection; +import com.lambdaworks.redis.cluster.api.async.RedisClusterAsyncCommands; +import com.lambdaworks.redis.cluster.api.sync.RedisClusterCommands; +import com.lambdaworks.redis.codec.ByteArrayCodec; import com.lambdaworks.redis.codec.RedisCodec; import com.lambdaworks.redis.output.BooleanOutput; import com.lambdaworks.redis.output.ByteArrayOutput; +import com.lambdaworks.redis.output.CommandOutput; import com.lambdaworks.redis.output.DateOutput; import com.lambdaworks.redis.output.DoubleOutput; import com.lambdaworks.redis.output.IntegerOutput; @@ -114,9 +117,9 @@ import com.lambdaworks.redis.output.ValueOutput; import com.lambdaworks.redis.output.ValueSetOutput; import com.lambdaworks.redis.protocol.Command; import com.lambdaworks.redis.protocol.CommandArgs; -import com.lambdaworks.redis.protocol.CommandOutput; import com.lambdaworks.redis.protocol.CommandType; -import com.lambdaworks.redis.pubsub.RedisPubSubConnection; +import com.lambdaworks.redis.pubsub.StatefulRedisPubSubConnection; +import com.lambdaworks.redis.sentinel.api.StatefulRedisSentinelConnection; /** * {@code RedisConnection} implementation on top of Lettuce Redis @@ -132,9 +135,8 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection; */ public class LettuceConnection extends AbstractRedisConnection { - static final RedisCodec CODEC = new BytesRedisCodec(); + static final RedisCodec CODEC = ByteArrayCodec.INSTANCE; - private static final Method SYNC_HANDLER; private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy( LettuceConverters.exceptionConverter()); private static final TypeHints typeHints = new TypeHints(); @@ -142,16 +144,8 @@ public class LettuceConnection extends AbstractRedisConnection { private final int defaultDbIndex; private int dbIndex; - static { - SYNC_HANDLER = ReflectionUtils.findMethod(AbstractRedisClient.class, "syncHandler", RedisChannelHandler.class, - Class[].class); - ReflectionUtils.makeAccessible(SYNC_HANDLER); - } - - private final com.lambdaworks.redis.RedisAsyncConnection asyncSharedConn; - private final com.lambdaworks.redis.RedisConnection sharedConn; - private com.lambdaworks.redis.RedisAsyncConnection asyncDedicatedConn; - private com.lambdaworks.redis.RedisConnection dedicatedConn; + private final StatefulConnection asyncSharedConn; + private StatefulConnection asyncDedicatedConn; private final long timeout; @@ -169,13 +163,13 @@ public class LettuceConnection extends AbstractRedisConnection { private boolean convertPipelineAndTxResults = true; @SuppressWarnings("rawtypes") - private class LettuceResult extends FutureResult> { + private class LettuceResult extends FutureResult> { public LettuceResult(Future resultHolder, Converter converter) { - super((Command) resultHolder, converter); + super((com.lambdaworks.redis.protocol.RedisCommand) resultHolder, converter); } public LettuceResult(Future resultHolder) { - super((Command) resultHolder); + super((com.lambdaworks.redis.protocol.RedisCommand) resultHolder); } @SuppressWarnings("unchecked") @@ -183,10 +177,10 @@ public class LettuceConnection extends AbstractRedisConnection { public Object get() { try { if (convertPipelineAndTxResults && converter != null) { - return converter.convert(resultHolder.get()); + return converter.convert(resultHolder.getOutput().get()); } - return resultHolder.get(); - } catch (ExecutionException e) { + return resultHolder.getOutput().get(); + } catch (Exception e) { throw EXCEPTION_TRANSLATION.translate(e); } } @@ -292,8 +286,7 @@ public class LettuceConnection extends AbstractRedisConnection { * @param timeout The connection timeout (in milliseconds) * @param client The {@link RedisClient} to use when making pub/sub, blocking, and tx connections */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, long timeout, - RedisClient client) { + public LettuceConnection(StatefulRedisConnection sharedConnection, long timeout, RedisClient client) { this(sharedConnection, timeout, client, null); } @@ -306,8 +299,8 @@ public class LettuceConnection extends AbstractRedisConnection { * @param client The {@link RedisClient} to use when making pub/sub connections * @param pool The connection pool to use for blocking and tx operations */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, long timeout, - RedisClient client, LettucePool pool) { + public LettuceConnection(StatefulRedisConnection sharedConnection, long timeout, RedisClient client, + LettucePool pool) { this(sharedConnection, timeout, client, pool, 0); } @@ -321,13 +314,12 @@ public class LettuceConnection extends AbstractRedisConnection { * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. * @since 1.7 */ - public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection sharedConnection, long timeout, + public LettuceConnection(StatefulRedisConnection sharedConnection, long timeout, AbstractRedisClient client, LettucePool pool, int defaultDbIndex) { this.asyncSharedConn = sharedConnection; this.timeout = timeout; this.client = client; - this.sharedConn = sharedConnection != null ? syncConnection(asyncSharedConn) : null; this.pool = pool; this.defaultDbIndex = defaultDbIndex; this.dbIndex = this.defaultDbIndex; @@ -344,11 +336,13 @@ public class LettuceConnection extends AbstractRedisConnection { } @SuppressWarnings({ "rawtypes", "unchecked" }) - private Object await(com.lambdaworks.redis.protocol.RedisCommand cmd) { - if (isMulti && cmd instanceof Command && ((Command) cmd).isMulti()) { + private Object await(RedisFuture cmd) { + + if (isMulti) { return null; } - return LettuceFutures.await(cmd, timeout, TimeUnit.MILLISECONDS); + + return LettuceFutures.awaitOrCancel(cmd, timeout, TimeUnit.MILLISECONDS); } @Override @@ -380,21 +374,21 @@ public class LettuceConnection extends AbstractRedisConnection { cmdArg.addKeys(args); } - RedisAsyncConnectionImpl connectionImpl = (RedisAsyncConnectionImpl) getAsyncConnection(); + RedisClusterAsyncCommands connectionImpl = getAsyncConnection(); CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint : typeHints.getTypeHint(commandType); Command cmd = new Command(commandType, expectedOutput, cmdArg); if (isPipelined()) { - pipeline(new LettuceResult(connectionImpl.dispatch(cmd))); + pipeline(new LettuceResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs()))); return null; } else if (isQueueing()) { - transaction(new LettuceTxResult(connectionImpl.dispatch(cmd))); + transaction(new LettuceTxResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs()))); return null; } else { - return await(connectionImpl.dispatch(cmd)); + return await(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())); } } catch (RedisException ex) { throw convertLettuceAccessException(ex); @@ -445,8 +439,8 @@ public class LettuceConnection extends AbstractRedisConnection { return isClosed && !isSubscribed(); } - public RedisAsyncConnection getNativeConnection() { - return (subscription != null ? subscription.pubsub : getAsyncConnection()); + public RedisClusterAsyncCommands getNativeConnection() { + return (subscription != null ? subscription.pubsub.async() : getAsyncConnection()); } public boolean isQueueing() { @@ -467,47 +461,56 @@ public class LettuceConnection extends AbstractRedisConnection { public List closePipeline() { if (isPipelined) { isPipelined = false; - List> futures = new ArrayList>(); + List> futures = new ArrayList>(); for (LettuceResult result : ppline) { futures.add(result.getResultHolder()); } - boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, - futures.toArray(new Command[futures.size()])); - List results = new ArrayList(futures.size()); - Exception problem = null; + // boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, + // futures.toArray(new Command[futures.size()])); - if (done) { - for (LettuceResult result : ppline) { - if (result.getResultHolder().getOutput().hasError()) { - Exception err = new InvalidDataAccessApiUsageException(result.getResultHolder().getOutput().getError()); - // remember only the first error - if (problem == null) { - problem = err; - } - results.add(err); - } else if (!convertPipelineAndTxResults || !(result.isStatus())) { - try { - results.add(result.get()); - } catch (DataAccessException e) { + try { + boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, + futures.toArray(new RedisFuture[futures.size()])); + + List results = new ArrayList(futures.size()); + + Exception problem = null; + + if (done) { + for (LettuceResult result : ppline) { + if (result.getResultHolder().getOutput().hasError()) { + Exception err = new InvalidDataAccessApiUsageException(result.getResultHolder().getOutput().getError()); + // remember only the first error if (problem == null) { - problem = e; + problem = err; + } + results.add(err); + } else if (!convertPipelineAndTxResults || !(result.isStatus())) { + try { + results.add(result.get()); + } catch (DataAccessException e) { + if (problem == null) { + problem = e; + } + results.add(e); } - results.add(e); } } } - } - ppline.clear(); + ppline.clear(); - if (problem != null) { - throw new RedisPipelineException(problem, results); - } - if (done) { - return results; - } + if (problem != null) { + throw new RedisPipelineException(problem, results); + } + if (done) { + return results; + } - throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out")); + throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out")); + } catch (Exception e) { + throw new RedisPipelineException(e); + } } return Collections.emptyList(); @@ -839,10 +842,10 @@ public class LettuceConnection extends AbstractRedisConnection { isMulti = false; try { if (isPipelined()) { - pipeline(new LettuceStatusResult(getAsyncDedicatedConnection().discard())); + pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).discard())); return; } - getDedicatedConnection().discard(); + ((RedisCommands) getDedicatedConnection()).discard(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } finally { @@ -855,11 +858,12 @@ public class LettuceConnection extends AbstractRedisConnection { isMulti = false; try { if (isPipelined()) { - pipeline(new LettuceResult(getAsyncDedicatedConnection().exec(), new LettuceTransactionResultConverter( - new LinkedList>(txResults), LettuceConverters.exceptionConverter()))); + pipeline(new LettuceResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).exec(), + new LettuceTransactionResultConverter(new LinkedList>(txResults), + LettuceConverters.exceptionConverter()))); return null; } - List results = getDedicatedConnection().exec(); + List results = ((RedisCommands) getDedicatedConnection()).exec(); return convertPipelineAndTxResults ? new LettuceTransactionResultConverter(txResults, LettuceConverters.exceptionConverter()).convert(results) : results; @@ -873,14 +877,16 @@ public class LettuceConnection extends AbstractRedisConnection { public Boolean exists(byte[] key) { try { if (isPipelined()) { - pipeline(new LettuceResult(getAsyncConnection().exists(key))); + pipeline(new LettuceResult(getAsyncConnection().exists(new byte[][] { key }), + LettuceConverters.longToBooleanConverter())); return null; } if (isQueueing()) { - transaction(new LettuceTxResult(getConnection().exists(key))); + transaction(new LettuceResult(getAsyncConnection().exists(new byte[][] { key }), + LettuceConverters.longToBooleanConverter())); return null; } - return getConnection().exists(key); + return LettuceConverters.longToBooleanConverter().convert(getConnection().exists(new byte[][] { key })); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -1055,10 +1061,10 @@ public class LettuceConnection extends AbstractRedisConnection { isMulti = true; try { if (isPipelined()) { - getAsyncDedicatedConnection().multi(); + ((RedisAsyncCommands) getAsyncDedicatedConnection()).multi(); return; } - getDedicatedConnection().multi(); + ((RedisCommands) getDedicatedConnection()).multi(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -1157,10 +1163,10 @@ public class LettuceConnection extends AbstractRedisConnection { this.dbIndex = dbIndex; if (isQueueing()) { - transaction(new LettuceTxStatusResult(getConnection().select(dbIndex))); + transaction(new LettuceTxStatusResult(((RedisCommands) getAsyncConnection()).select(dbIndex))); return; } - getConnection().select(dbIndex); + ((RedisCommands) getConnection()).select(dbIndex); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -1235,14 +1241,14 @@ public class LettuceConnection extends AbstractRedisConnection { public void unwatch() { try { if (isPipelined()) { - pipeline(new LettuceStatusResult(getAsyncDedicatedConnection().unwatch())); + pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).unwatch())); return; } if (isQueueing()) { - transaction(new LettuceTxStatusResult(getDedicatedConnection().unwatch())); + transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).unwatch())); return; } - getDedicatedConnection().unwatch(); + ((RedisCommands) getDedicatedConnection()).unwatch(); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -1254,10 +1260,14 @@ public class LettuceConnection extends AbstractRedisConnection { } try { if (isPipelined()) { - pipeline(new LettuceStatusResult(getAsyncDedicatedConnection().watch(keys))); + pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).watch(keys))); return; } - getDedicatedConnection().watch(keys); + if (isQueueing()) { + transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).watch())); + return; + } + ((RedisCommands) getDedicatedConnection()).watch(keys); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -2135,13 +2145,13 @@ public class LettuceConnection extends AbstractRedisConnection { public List sRandMember(byte[] key, long count) { try { if (isPipelined()) { - pipeline( - new LettuceResult(getAsyncConnection().srandmember(key, count), LettuceConverters.bytesSetToBytesList())); + pipeline(new LettuceResult((RedisFuture) getAsyncConnection().srandmember(key, count), + LettuceConverters.bytesCollectionToBytesList())); return null; } if (isQueueing()) { - transaction( - new LettuceTxResult(getConnection().srandmember(key, count), LettuceConverters.bytesSetToBytesList())); + transaction(new LettuceTxResult(getConnection().srandmember(key, count), + LettuceConverters.bytesCollectionToBytesList())); return null; } return LettuceConverters.toBytesList(getConnection().srandmember(key, count)); @@ -3336,7 +3346,19 @@ public class LettuceConnection extends AbstractRedisConnection { Assert.notNull(members, "Members must not be null!"); Assert.noNullElements(members, "Members must not contain null!"); - throw new UnsupportedOperationException("Lettuce does currently not supprt GEOHASH."); + try { + if (isPipelined()) { + pipeline(new LettuceResult(getAsyncConnection().geohash(key, members))); + return null; + } + if (isQueueing()) { + transaction(new LettuceTxResult(getConnection().geohash(key, members))); + return null; + } + return getConnection().geohash(key, members); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } } /* @@ -3702,16 +3724,6 @@ public class LettuceConnection extends AbstractRedisConnection { com.lambdaworks.redis.ScanCursor scanCursor = getScanCursor(cursorId); ScanArgs scanArgs = getScanArgs(options); - if (isPipelined()) { - pipeline(new LettuceResult(getAsyncConnection().scan(scanCursor, scanArgs))); - return null; - } - - if (isQueueing()) { - transaction(new LettuceTxResult(getAsyncConnection().scan(scanCursor, scanArgs))); - return null; - } - KeyScanCursor keyScanCursor = getConnection().scan(scanCursor, scanArgs); String nextCursorId = keyScanCursor.getCursor(); @@ -3889,10 +3901,11 @@ public class LettuceConnection extends AbstractRedisConnection { } } - private RedisPubSubConnection switchToPubSub() { + private StatefulRedisPubSubConnection switchToPubSub() { close(); // open a pubsub one return ((RedisClient) client).connectPubSub(CODEC); + // return ((RedisClient) client).connectPubSub(CODEC); } private void pipeline(LettuceResult result) { @@ -3907,63 +3920,91 @@ public class LettuceConnection extends AbstractRedisConnection { txResults.add(result); } - private RedisAsyncConnection getAsyncConnection() { + private RedisClusterAsyncCommands getAsyncConnection() { if (isQueueing()) { return getAsyncDedicatedConnection(); } if (asyncSharedConn != null) { - return asyncSharedConn; + + if (asyncSharedConn instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) asyncSharedConn).async(); + } } return getAsyncDedicatedConnection(); } - protected com.lambdaworks.redis.RedisConnection getConnection() { + protected RedisClusterCommands getConnection() { + if (isQueueing()) { return getDedicatedConnection(); } - if (sharedConn != null) { - return sharedConn; + if (asyncSharedConn != null) { + + if (asyncSharedConn instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) asyncSharedConn).sync(); + } + if (asyncSharedConn instanceof StatefulRedisClusterConnection) { + return ((StatefulRedisClusterConnection) asyncSharedConn).sync(); + } } return getDedicatedConnection(); } - protected RedisAsyncConnection getAsyncDedicatedConnection() { + protected RedisClusterAsyncCommands getAsyncDedicatedConnection() { if (asyncDedicatedConn == null) { asyncDedicatedConn = doGetAsyncDedicatedConnection(); if (this.pool == null) { - this.asyncDedicatedConn.select(dbIndex); - } + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); + } + } } - return asyncDedicatedConn; + + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) asyncDedicatedConn).async(); + } + if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { + return ((StatefulRedisClusterConnection) asyncDedicatedConn).async(); + } + + throw new IllegalStateException(String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); } - protected RedisAsyncConnection doGetAsyncDedicatedConnection() { + protected StatefulConnection doGetAsyncDedicatedConnection() { if (this.pool != null) { return pool.getResource(); } else { - return ((RedisClient) client).connectAsync(CODEC); + return ((RedisClient) client).connect(CODEC); } } - private com.lambdaworks.redis.RedisConnection getDedicatedConnection() { - if (dedicatedConn == null) { - this.dedicatedConn = syncConnection(getAsyncDedicatedConnection()); - } - return dedicatedConn; - } + private RedisClusterCommands getDedicatedConnection() { + + if (asyncDedicatedConn == null) { + + asyncDedicatedConn = doGetAsyncDedicatedConnection(); + + if (this.pool == null) { + + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); + } + } - private com.lambdaworks.redis.RedisConnection syncConnection( - RedisAsyncConnection asyncConnection) { - try { - return (com.lambdaworks.redis.RedisConnection) SYNC_HANDLER.invoke(null, asyncConnection, - new Class[] { com.lambdaworks.redis.RedisConnection.class, RedisClusterConnection.class }); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); } + + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) asyncDedicatedConn).sync(); + } + if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { + return ((StatefulRedisClusterConnection) asyncDedicatedConn).sync(); + } + + throw new IllegalStateException(String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); } private Future asyncBitOp(BitOperation op, byte[] destination, byte[]... keys) { @@ -4025,7 +4066,12 @@ public class LettuceConnection extends AbstractRedisConnection { return null; } - ScanArgs scanArgs = ScanArgs.Builder.matches(options.getPattern()); + ScanArgs scanArgs = new ScanArgs(); + + if (options.getPattern() != null) { + scanArgs.match(options.getPattern()); + } + if (options.getCount() != null) { scanArgs.limit(options.getCount()); } @@ -4082,10 +4128,10 @@ public class LettuceConnection extends AbstractRedisConnection { return false; } - RedisConnection connection = null; + StatefulRedisConnection connection = null; try { connection = ((RedisClient) client).connect(getRedisURI(node)); - return connection.ping().equalsIgnoreCase("pong"); + return connection.sync().ping().equalsIgnoreCase("pong"); } catch (Exception e) { return false; } finally { @@ -4105,8 +4151,8 @@ public class LettuceConnection extends AbstractRedisConnection { */ @Override protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) { - RedisSentinelAsyncConnection connection = ((RedisClient) client) - .connectSentinelAsync(getRedisURI(sentinel)); + StatefulRedisSentinelConnection connection = ((RedisClient) client) + .connectSentinel(getRedisURI(sentinel)); return new LettuceSentinelConnection(connection); } @@ -4363,30 +4409,19 @@ public class LettuceConnection extends AbstractRedisConnection { try { if (isPipelined()) { - if (values.length == 1) { - pipeline(new LettuceResult(getAsyncConnection().pfadd(key, values[0]))); - } else { - pipeline( - new LettuceResult(getAsyncConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1)))); - } + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + pipeline(new LettuceResult(asyncConnection.pfadd(key, values))); return null; } + if (isQueueing()) { - if (values.length == 1) { - transaction(new LettuceTxResult(getConnection().pfadd(key, values[0]))); - } else { - transaction( - new LettuceTxResult(getConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1)))); - } - + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + transaction(new LettuceResult(asyncConnection.pfadd(key, values))); return null; } - if (values.length == 1) { - return getConnection().pfadd(key, values[0]); - } - - return getConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1)); + RedisHLLCommands connection = getConnection(); + return connection.pfadd(key, values); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -4403,28 +4438,19 @@ public class LettuceConnection extends AbstractRedisConnection { Assert.noNullElements(keys, "Keys for PFCOUNT must not contain 'null'."); try { if (isPipelined()) { - if (keys.length == 1) { - pipeline(new LettuceResult(getAsyncConnection().pfcount(keys[0]))); - } else { - pipeline(new LettuceResult(getAsyncConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1)))); - } + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + pipeline(new LettuceResult(asyncConnection.pfcount(keys))); return null; } + if (isQueueing()) { - if (keys.length == 1) { - transaction(new LettuceTxResult(getConnection().pfcount(keys[0]))); - } else { - transaction(new LettuceTxResult(getConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1)))); - } - + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + transaction(new LettuceResult(asyncConnection.pfcount(keys))); return null; } - if (keys.length == 1) { - return getConnection().pfcount(keys[0]); - } - - return getConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1)); + RedisHLLCommands connection = getConnection(); + return connection.pfcount(keys); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -4442,27 +4468,19 @@ public class LettuceConnection extends AbstractRedisConnection { try { if (isPipelined()) { - if (sourceKeys.length == 1) { - pipeline(new LettuceResult(getAsyncConnection().pfmerge(destinationKey, sourceKeys[0]))); - } else { - pipeline(new LettuceResult( - getAsyncConnection().pfmerge(destinationKey, sourceKeys[0], LettuceConverters.subarray(sourceKeys, 1)))); - } - } - if (isQueueing()) { - if (sourceKeys.length == 1) { - transaction(new LettuceTxResult(getConnection().pfmerge(destinationKey, sourceKeys[0]))); - } else { - transaction(new LettuceTxResult( - getConnection().pfmerge(destinationKey, sourceKeys[0], LettuceConverters.subarray(sourceKeys, 1)))); - } + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + pipeline(new LettuceResult(asyncConnection.pfmerge(destinationKey, sourceKeys))); + return; } - if (sourceKeys.length == 1) { - getConnection().pfmerge(destinationKey, sourceKeys[0]); - } else { - getConnection().pfmerge(destinationKey, sourceKeys[0], LettuceConverters.subarray(sourceKeys, 1)); + if (isQueueing()) { + RedisHLLAsyncCommands asyncConnection = getAsyncConnection(); + transaction(new LettuceResult(asyncConnection.pfmerge(destinationKey, sourceKeys))); + return; } + + RedisHLLCommands connection = getConnection(); + connection.pfmerge(destinationKey, sourceKeys); } catch (Exception ex) { throw convertLettuceAccessException(ex); } 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 8d5387bd5..1390834a5 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 @@ -44,12 +44,10 @@ import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; import com.lambdaworks.redis.AbstractRedisClient; -import com.lambdaworks.redis.LettuceFutures; -import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; import com.lambdaworks.redis.RedisException; -import com.lambdaworks.redis.RedisFuture; import com.lambdaworks.redis.RedisURI; +import com.lambdaworks.redis.api.StatefulRedisConnection; import com.lambdaworks.redis.cluster.RedisClusterClient; import com.lambdaworks.redis.resource.ClientResources; @@ -88,7 +86,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea private long shutdownTimeout = TimeUnit.MILLISECONDS.convert(2, TimeUnit.SECONDS); private boolean validateConnection = false; private boolean shareNativeConnection = true; - private RedisAsyncConnection connection; + private StatefulRedisConnection connection; private LettucePool pool; private int dbIndex = 0; /** Synchronization monitor for the shared Connection */ @@ -207,6 +205,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } public void initConnection() { + synchronized (this.connectionMonitor) { if (this.connection != null) { resetConnection(); @@ -237,11 +236,8 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea if (connection.isOpen()) { try { - RedisFuture ping = connection.ping(); - LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, ping); - if (PING_REPLY.equalsIgnoreCase(ping.get())) { - valid = true; - } + connection.sync().ping(); + valid = true; } catch (Exception e) { log.debug("Validation failed", e); } @@ -511,7 +507,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea this.convertPipelineAndTxResults = convertPipelineAndTxResults; } - protected RedisAsyncConnection getSharedConnection() { + protected StatefulRedisConnection getSharedConnection() { if (shareNativeConnection) { synchronized (this.connectionMonitor) { if (this.connection == null) { @@ -527,18 +523,17 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea } } - protected RedisAsyncConnection createLettuceConnector() { + protected StatefulRedisConnection createLettuceConnector() { try { - RedisAsyncConnection connection = null; + StatefulRedisConnection connection = null; if (client instanceof RedisClient) { - connection = ((RedisClient) client).connectAsync(LettuceConnection.CODEC); + connection = ((RedisClient) client).connect(LettuceConnection.CODEC); if (dbIndex > 0) { - connection.select(dbIndex); + connection.sync().select(dbIndex); } } else { - connection = (RedisAsyncConnection) ((RedisClusterClient) client) - .connectClusterAsync(LettuceConnection.CODEC); + connection = null; } return connection; } catch (RedisException e) { @@ -630,6 +625,6 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea if (!(client instanceof RedisClient)) { throw new InvalidDataAccessResourceUsageException("Unable to connect to sentinels using " + client.getClass()); } - return new LettuceSentinelConnection(((RedisClient) client).connectSentinelAsync()); + return new LettuceSentinelConnection(((RedisClient) client).connectSentinel()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java index 09e073951..1c62fac19 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.lettuce; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.Date; import java.util.Iterator; @@ -73,11 +74,11 @@ import com.lambdaworks.redis.KeyValue; import com.lambdaworks.redis.RedisURI; import com.lambdaworks.redis.ScoredValue; import com.lambdaworks.redis.ScriptOutputType; +import com.lambdaworks.redis.SetArgs; import com.lambdaworks.redis.SortArgs; import com.lambdaworks.redis.cluster.models.partitions.Partitions; import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag; import com.lambdaworks.redis.protocol.LettuceCharsets; -import com.lambdaworks.redis.protocol.SetArgs; /** * Lettuce type converters @@ -95,6 +96,7 @@ abstract public class LettuceConverters extends Converters { private static final Converter BYTES_TO_STRING; private static final Converter STRING_TO_BYTES; private static final Converter, List> BYTES_SET_TO_BYTES_LIST; + private static final Converter, List> BYTES_COLLECTION_TO_BYTES_LIST; private static final Converter, List> KEY_VALUE_TO_BYTES_LIST; private static final Converter>, Set> SCORED_VALUES_TO_TUPLE_SET; private static final Converter>, List> SCORED_VALUES_TO_TUPLE_LIST; @@ -151,6 +153,14 @@ abstract public class LettuceConverters extends Converters { return results != null ? new ArrayList(results) : null; } }; + BYTES_COLLECTION_TO_BYTES_LIST = new Converter, List>() { + public List convert(Collection results) { + if (results instanceof List) { + return (List) results; + } + return results != null ? new ArrayList(results) : null; + } + }; KEY_VALUE_TO_BYTES_LIST = new Converter, List>() { public List convert(KeyValue source) { if (source == null) { @@ -365,8 +375,12 @@ abstract public class LettuceConverters extends Converters { return KEY_VALUE_TO_BYTES_LIST; } - public static Converter, List> bytesSetToBytesList() { - return BYTES_SET_TO_BYTES_LIST; + public static Converter, List> bytesSetToBytesList() { + return BYTES_COLLECTION_TO_BYTES_LIST; + } + + public static Converter, List> bytesCollectionToBytesList() { + return BYTES_COLLECTION_TO_BYTES_LIST; } public static Converter>, Set> scoredValuesToTupleSet() { @@ -405,8 +419,8 @@ abstract public class LettuceConverters extends Converters { return KEY_VALUE_TO_BYTES_LIST.convert(source); } - public static List toBytesList(Set source) { - return BYTES_SET_TO_BYTES_LIST.convert(source); + public static List toBytesList(Collection source) { + return BYTES_COLLECTION_TO_BYTES_LIST.convert(source); } public static Set toTupleSet(List> source) { @@ -769,7 +783,7 @@ abstract public class LettuceConverters extends Converters { public GeoResults> convert(Set source) { if (CollectionUtils.isEmpty(source)) { - return new GeoResults>(Collections.>> emptyList()); + return new GeoResults>(Collections.>>emptyList()); } List>> results = new ArrayList>>(source.size()); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java index e7ab27977..16bcd264c 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java @@ -18,20 +18,20 @@ package org.springframework.data.redis.connection.lettuce; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; -import com.lambdaworks.redis.RedisCommandInterruptedException; -import com.lambdaworks.redis.RedisCommandTimeoutException; -import com.lambdaworks.redis.RedisConnectionException; -import com.lambdaworks.redis.RedisException; -import io.netty.channel.ChannelException; import org.springframework.core.convert.converter.Converter; import org.springframework.dao.DataAccessException; import org.springframework.dao.QueryTimeoutException; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; +import com.lambdaworks.redis.RedisCommandExecutionException; import com.lambdaworks.redis.RedisCommandInterruptedException; +import com.lambdaworks.redis.RedisCommandTimeoutException; +import com.lambdaworks.redis.RedisConnectionException; import com.lambdaworks.redis.RedisException; +import io.netty.channel.ChannelException; + /** * Converts Lettuce Exceptions to {@link DataAccessException}s * @@ -42,9 +42,9 @@ public class LettuceExceptionConverter implements Converter> { +public interface LettucePool extends Pool> { /** * @return The {@link RedisClient} used to create pooled connections 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 a4d3fafd4..6295940c3 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 @@ -27,9 +27,9 @@ import org.springframework.data.redis.connection.RedisServer; import org.springframework.util.Assert; import com.lambdaworks.redis.RedisClient; -import com.lambdaworks.redis.RedisSentinelAsyncConnection; import com.lambdaworks.redis.RedisURI.Builder; import com.lambdaworks.redis.resource.ClientResources; +import com.lambdaworks.redis.sentinel.api.StatefulRedisSentinelConnection; /** * @author Christoph Strobl @@ -42,7 +42,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { LettuceConverters.exceptionConverter()); private RedisClient redisClient; - private RedisSentinelAsyncConnection connection; + private StatefulRedisSentinelConnection connection; /** * Creates a {@link LettuceSentinelConnection} with a dedicated client for a supplied {@link RedisNode}. @@ -101,7 +101,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { * * @param connection native Lettuce connection, must not be {@literal null} */ - protected LettuceSentinelConnection(RedisSentinelAsyncConnection connection) { + protected LettuceSentinelConnection(StatefulRedisSentinelConnection connection) { Assert.notNull(connection, "Cannot create LettuceSentinelConnection using 'null' as connection."); this.connection = connection; @@ -116,7 +116,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { Assert.notNull(master, "Redis node master must not be 'null' for failover."); Assert.hasText(master.getName(), "Redis master name must not be 'null' or empty for failover."); - connection.failover(master.getName()); + connection.sync().failover(master.getName()); } /* @@ -126,7 +126,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { @Override public List masters() { try { - return LettuceConverters.toListOfRedisServer(connection.masters().get()); + return LettuceConverters.toListOfRedisServer(connection.sync().masters()); } catch (Exception e) { throw EXCEPTION_TRANSLATION.translate(e); } @@ -152,7 +152,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when loading slaves."); try { - return LettuceConverters.toListOfRedisServer(connection.slaves(masterName).get()); + return LettuceConverters.toListOfRedisServer(connection.sync().slaves(masterName)); } catch (Exception e) { throw EXCEPTION_TRANSLATION.translate(e); } @@ -176,7 +176,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { public void remove(String masterName) { Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when trying to remove."); - connection.remove(masterName); + connection.sync().remove(masterName); } /* @@ -191,7 +191,8 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { Assert.hasText(server.getHost(), "Host must not be 'null' for server to monitor."); Assert.notNull(server.getPort(), "Port must not be 'null' for server to monitor."); Assert.notNull(server.getQuorum(), "Quorum must not be 'null' for server to monitor."); - connection.monitor(server.getName(), server.getHost(), server.getPort().intValue(), server.getQuorum().intValue()); + connection.sync().monitor(server.getName(), server.getHost(), server.getPort().intValue(), + server.getQuorum().intValue()); } /* @@ -214,13 +215,12 @@ public class LettuceSentinelConnection implements RedisSentinelConnection { } } - private RedisSentinelAsyncConnection connectSentinel() { - return redisClient.connectSentinelAsync(); + private StatefulRedisSentinelConnection connectSentinel() { + return redisClient.connectSentinel(); } @Override public boolean isOpen() { return connection != null && connection.isOpen(); } - } 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 5407d1964..047053ed8 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 @@ -19,7 +19,7 @@ package org.springframework.data.redis.connection.lettuce; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.util.AbstractSubscription; -import com.lambdaworks.redis.pubsub.RedisPubSubConnection; +import com.lambdaworks.redis.pubsub.StatefulRedisPubSubConnection; /** * Message subscription on top of Lettuce. @@ -28,10 +28,10 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection; */ class LettuceSubscription extends AbstractSubscription { - final RedisPubSubConnection pubsub; + final StatefulRedisPubSubConnection pubsub; private LettuceMessageListener listener; - LettuceSubscription(MessageListener listener, RedisPubSubConnection pubsubConnection) { + LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection) { super(listener); this.pubsub = pubsubConnection; this.listener = new LettuceMessageListener(listener); @@ -41,30 +41,30 @@ class LettuceSubscription extends AbstractSubscription { protected void doClose() { if (!getChannels().isEmpty()) { - pubsub.unsubscribe(new byte[0]); + pubsub.sync().unsubscribe(new byte[0]); } if (!getPatterns().isEmpty()) { - pubsub.punsubscribe(new byte[0]); + pubsub.sync().punsubscribe(new byte[0]); } pubsub.removeListener(this.listener); pubsub.close(); } protected void doPsubscribe(byte[]... patterns) { - pubsub.psubscribe(patterns); + pubsub.sync().psubscribe(patterns); } protected void doPUnsubscribe(boolean all, byte[]... patterns) { // lettuce doesn't automatically subscribe from all channels - pubsub.punsubscribe(patterns); + pubsub.sync().punsubscribe(patterns); } protected void doSubscribe(byte[]... channels) { - pubsub.subscribe(channels); + pubsub.sync().subscribe(channels); } protected void doUnsubscribe(boolean all, byte[]... channels) { // lettuce doesn't automatically subscribe from all patterns - pubsub.unsubscribe(channels); + pubsub.sync().unsubscribe(channels); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java deleted file mode 100644 index 8cc24fd45..000000000 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java +++ /dev/null @@ -1,209 +0,0 @@ -/* - * Copyright 2011-2013 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.io.StringReader; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Properties; -import java.util.Set; - -import com.lambdaworks.redis.protocol.LettuceCharsets; -import org.springframework.dao.DataAccessException; -import org.springframework.data.redis.RedisSystemException; -import org.springframework.data.redis.connection.DefaultTuple; -import org.springframework.data.redis.connection.ReturnType; -import org.springframework.data.redis.connection.RedisListCommands.Position; -import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; -import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; -import org.springframework.data.redis.connection.SortParameters; -import org.springframework.data.redis.connection.SortParameters.Order; -import org.springframework.util.Assert; - -import com.lambdaworks.redis.KeyValue; -import com.lambdaworks.redis.RedisCommandInterruptedException; -import com.lambdaworks.redis.RedisException; -import com.lambdaworks.redis.ScoredValue; -import com.lambdaworks.redis.ScriptOutputType; -import com.lambdaworks.redis.SortArgs; -import com.lambdaworks.redis.ZStoreArgs; -import com.lambdaworks.redis.codec.RedisCodec; - -/** - * Helper class featuring methods for Lettuce connection handling, providing support for exception translation. - * Deprecated in favor of {@link LettuceConverters} - * - * @author Costin Leau - */ -@Deprecated -abstract class LettuceUtils { - - static final RedisCodec CODEC = new BytesRedisCodec(); - - static DataAccessException convertRedisAccessException(RuntimeException ex) { - if (ex instanceof RedisCommandInterruptedException) { - return new RedisSystemException("Redis command interrupted", ex); - } - if (ex instanceof RedisException) { - return new RedisSystemException("Redis exception", ex); - } - return null; - } - - static Properties info(String reply) { - if (reply == null) { - return null; - } - Properties info = new Properties(); - StringReader stringReader = new StringReader(reply); - try { - info.load(stringReader); - } catch (Exception ex) { - throw new RedisSystemException("Cannot read Redis info", ex); - } finally { - stringReader.close(); - } - return info; - } - - static int asBit(boolean value) { - return (value ? 1 : 0); - } - - static boolean convertPosition(Position where) { - Assert.notNull("list positions are mandatory"); - return (Position.AFTER.equals(where) ? false : true); - } - - static Set convertTuple(List> zrange) { - if (zrange == null) { - return null; - } - Set tuples = new LinkedHashSet(zrange.size()); - - for (int i = 0; i < zrange.size(); i++) { - tuples.add(new DefaultTuple(zrange.get(i).value, Double.valueOf(zrange.get(i).score))); - } - return tuples; - } - - static SortArgs sort(SortParameters params) { - SortArgs args = new SortArgs(); - - if (params == null) { - return args; - } - - if (params.getByPattern() != null) { - args.by(new String(params.getByPattern(), LettuceCharsets.ASCII)); - } - - if (params.getLimit() != null) { - args.limit(params.getLimit().getStart(), params.getLimit().getCount()); - } - - if (params.getGetPattern() != null) { - byte[][] pattern = params.getGetPattern(); - for (byte[] bs : pattern) { - args.get(new String(bs, LettuceCharsets.ASCII)); - } - } - - if (params.getOrder() != null) { - if (params.getOrder() == Order.ASC) { - args.asc(); - } else { - args.desc(); - } - } - - Boolean isAlpha = params.isAlphabetic(); - if (isAlpha != null && isAlpha) { - args.alpha(); - } - return args; - } - - static ZStoreArgs zArgs(Aggregate aggregate, int[] weights) { - ZStoreArgs args = new ZStoreArgs(); - - if (aggregate != null) { - switch (aggregate) { - case MIN: - args.min(); - break; - case MAX: - args.max(); - break; - default: - args.sum(); - break; - } - } - - long[] lg = new long[weights.length]; - for (int i = 0; i < lg.length; i++) { - lg[i] = (long) weights[i]; - } - args.weights(lg); - return args; - } - - static List toList(KeyValue blpop) { - if (blpop == null) { - return null; - } - List list = new ArrayList(2); - list.add(blpop.key); - list.add(blpop.value); - return list; - } - - static ScriptOutputType toScriptOutputType(ReturnType returnType) { - switch (returnType) { - case BOOLEAN: - return ScriptOutputType.BOOLEAN; - case MULTI: - return ScriptOutputType.MULTI; - case VALUE: - return ScriptOutputType.VALUE; - case INTEGER: - return ScriptOutputType.INTEGER; - case STATUS: - return ScriptOutputType.STATUS; - default: - throw new IllegalArgumentException("Return type " + returnType + " is not a supported script output type"); - } - } - - static byte[][] extractScriptKeys(int numKeys, byte[]... keysAndArgs) { - if (numKeys > 0) { - return Arrays.copyOfRange(keysAndArgs, 0, numKeys); - } - return new byte[0][0]; - } - - static byte[][] extractScriptArgs(int numKeys, byte[]... keysAndArgs) { - if (keysAndArgs.length > numKeys) { - return Arrays.copyOfRange(keysAndArgs, numKeys, keysAndArgs.length); - } - return new byte[0][0]; - } - -} diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 702ba2c38..209a5bc62 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -48,6 +48,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.hamcrest.core.IsNot; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.internal.AssumptionViolatedException; @@ -575,6 +576,7 @@ public abstract class AbstractConnectionIntegrationTests { } @Test + @Ignore("DATAREDIS-525") public void testNullKey() throws Exception { try { connection.decr((String) null); @@ -585,7 +587,9 @@ public abstract class AbstractConnectionIntegrationTests { } @Test + @Ignore("DATAREDIS-525") public void testNullValue() throws Exception { + byte[] key = UUID.randomUUID().toString().getBytes(); connection.append(key, EMPTY_ARRAY); try { @@ -597,7 +601,9 @@ public abstract class AbstractConnectionIntegrationTests { } @Test + @Ignore("DATAREDIS-525") public void testHashNullKey() throws Exception { + byte[] key = UUID.randomUUID().toString().getBytes(); try { connection.hExists(key, null); @@ -608,6 +614,7 @@ public abstract class AbstractConnectionIntegrationTests { } @Test + @Ignore("DATAREDIS-525") public void testHashNullValue() throws Exception { byte[] key = UUID.randomUUID().toString().getBytes(); byte[] field = "random".getBytes(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClientTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClientTests.java index 4864b69ad..349b9a11c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClientTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClientTests.java @@ -17,9 +17,10 @@ package org.springframework.data.redis.connection.lettuce; import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; -import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisException; -import com.lambdaworks.redis.pubsub.RedisPubSubConnection; +import com.lambdaworks.redis.api.StatefulRedisConnection; +import com.lambdaworks.redis.pubsub.StatefulRedisPubSubConnection; + import org.junit.After; import org.junit.Before; import org.junit.Ignore; @@ -50,8 +51,8 @@ public class AuthenticatingRedisClientTests { @Test public void connect() { - RedisConnection conn = client.connect(); - conn.ping(); + StatefulRedisConnection conn = client.connect(); + conn.sync().ping(); conn.close(); } @@ -68,8 +69,8 @@ public class AuthenticatingRedisClientTests { @Test public void codecConnect() { - RedisConnection conn = client.connect(LettuceConnection.CODEC); - conn.ping(); + StatefulRedisConnection conn = client.connect(LettuceConnection.CODEC); + conn.sync().ping(); conn.close(); } @@ -89,15 +90,15 @@ public class AuthenticatingRedisClientTests { @Test public void connectPubSub() { - RedisPubSubConnection conn = client.connectPubSub(); - conn.ping(); + StatefulRedisPubSubConnection conn = client.connectPubSub(); + conn.sync().ping(); conn.close(); } @Test public void codecConnectPubSub() { - RedisPubSubConnection conn = client.connectPubSub(LettuceConnection.CODEC); - conn.ping(); + StatefulRedisPubSubConnection conn = client.connectPubSub(LettuceConnection.CODEC); + conn.sync().ping(); conn.close(); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java index dc76348ff..d301bbfee 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java @@ -30,9 +30,9 @@ import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.PoolException; import org.springframework.data.redis.connection.RedisSentinelConfiguration; -import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisException; import com.lambdaworks.redis.RedisURI; +import com.lambdaworks.redis.api.StatefulRedisConnection; /** * Unit test of {@link DefaultLettucePool} @@ -65,9 +65,9 @@ public class DefaultLettucePoolTests { pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.afterPropertiesSet(); - RedisAsyncConnection client = pool.getResource(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); assertNotNull(client); - client.ping(); + client.sync().ping(); client.close(); } @@ -80,7 +80,7 @@ public class DefaultLettucePoolTests { pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig); pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.afterPropertiesSet(); - RedisAsyncConnection client = pool.getResource(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); assertNotNull(client); try { pool.getResource(); @@ -98,7 +98,7 @@ public class DefaultLettucePoolTests { pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig); pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.afterPropertiesSet(); - RedisAsyncConnection client = pool.getResource(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); assertNotNull(client); client.close(); } @@ -121,7 +121,7 @@ public class DefaultLettucePoolTests { pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig); pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.afterPropertiesSet(); - RedisAsyncConnection client = pool.getResource(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); assertNotNull(client); pool.returnResource(client); assertNotNull(pool.getResource()); @@ -137,13 +137,13 @@ public class DefaultLettucePoolTests { pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort(), poolConfig); pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.afterPropertiesSet(); - RedisAsyncConnection client = pool.getResource(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); assertNotNull(client); pool.returnBrokenResource(client); - RedisAsyncConnection client2 = pool.getResource(); + StatefulRedisConnection client2 = (StatefulRedisConnection) pool.getResource(); assertNotSame(client, client2); try { - client.ping(); + client.sync().ping(); fail("Broken resouce connection should be closed"); } catch (RedisException e) {} finally { client.close(); @@ -189,9 +189,9 @@ public class DefaultLettucePoolTests { pool.setClientResources(LettuceTestClientResources.getSharedClientResources()); pool.setPassword("foo"); pool.afterPropertiesSet(); - RedisAsyncConnection conn = pool.getResource(); - conn.ping(); - conn.close(); + StatefulRedisConnection client = (StatefulRedisConnection) pool.getResource(); + client.sync().ping(); + client.sync().getStatefulConnection().close(); } @Ignore("Redis must have requirepass set to run this test") diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index fa0b5ffec..05a43cca8 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -42,7 +42,6 @@ import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.ClassRule; -import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.springframework.dao.DataAccessException; @@ -72,8 +71,9 @@ import org.springframework.data.redis.test.util.RedisClusterRule; import org.springframework.test.annotation.IfProfileValue; import com.lambdaworks.redis.RedisURI.Builder; -import com.lambdaworks.redis.cluster.RedisAdvancedClusterConnection; +import com.lambdaworks.redis.api.sync.RedisHLLCommands; import com.lambdaworks.redis.cluster.RedisClusterClient; +import com.lambdaworks.redis.cluster.api.sync.RedisAdvancedClusterCommands; /** * @author Christoph Strobl @@ -105,7 +105,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { POINT_PALERMO); RedisClusterClient client; - RedisAdvancedClusterConnection nativeConnection; + RedisAdvancedClusterCommands nativeConnection; LettuceClusterConnection clusterConnection; public static @ClassRule RedisClusterRule clusterAvailable = new RedisClusterRule(); @@ -120,7 +120,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { client = RedisClusterClient.create(LettuceTestClientResources.getSharedClientResources(), Builder.redis(CLUSTER_HOST, MASTER_NODE_1_PORT).withTimeout(100, TimeUnit.MILLISECONDS).build()); - nativeConnection = client.connectCluster(); + nativeConnection = client.connect().sync(); clusterConnection = new LettuceClusterConnection(client); } @@ -128,7 +128,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { public void tearDown() throws InterruptedException { clusterConnection.flushDb(); - nativeConnection.close(); + nativeConnection.getStatefulConnection().close(); clusterConnection.close(); client.shutdown(0, 0, TimeUnit.MILLISECONDS); } @@ -2243,7 +2243,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { clusterConnection.pfAdd(KEY_1_BYTES, VALUE_1_BYTES, VALUE_2_BYTES, VALUE_3_BYTES); - assertThat(nativeConnection.pfcount(KEY_1), is(3L)); + assertThat(((RedisHLLCommands) nativeConnection).pfcount(KEY_1), is(3L)); } /** @@ -2252,7 +2252,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { @Test public void pfCountShouldAllowCountingOnSingleKey() { - nativeConnection.pfadd(KEY_1, VALUE_1, VALUE_2, VALUE_3); + ((RedisHLLCommands) nativeConnection).pfadd(KEY_1, VALUE_1, VALUE_2, VALUE_3); assertThat(clusterConnection.pfCount(KEY_1_BYTES), is(3L)); } @@ -2263,8 +2263,8 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { @Test public void pfCountShouldAllowCountingOnSameSlotKeys() { - nativeConnection.pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2); - nativeConnection.pfadd(SAME_SLOT_KEY_2, VALUE_2, VALUE_3); + ((RedisHLLCommands) nativeConnection).pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2); + ((RedisHLLCommands) nativeConnection).pfadd(SAME_SLOT_KEY_2, VALUE_2, VALUE_3); assertThat(clusterConnection.pfCount(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES), is(3L)); } @@ -2275,8 +2275,8 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { @Test(expected = DataAccessException.class) public void pfCountShouldThrowErrorCountingOnDifferentSlotKeys() { - nativeConnection.pfadd(KEY_1, VALUE_1, VALUE_2); - nativeConnection.pfadd(KEY_2, VALUE_2, VALUE_3); + ((RedisHLLCommands) nativeConnection).pfadd(KEY_1, VALUE_1, VALUE_2); + ((RedisHLLCommands) nativeConnection).pfadd(KEY_2, VALUE_2, VALUE_3); clusterConnection.pfCount(KEY_1_BYTES, KEY_2_BYTES); } @@ -2287,12 +2287,12 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { @Test public void pfMergeShouldWorkWhenAllKeysMapToSameSlot() { - nativeConnection.pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2); - nativeConnection.pfadd(SAME_SLOT_KEY_2, VALUE_2, VALUE_3); + ((RedisHLLCommands) nativeConnection).pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2); + ((RedisHLLCommands) nativeConnection).pfadd(SAME_SLOT_KEY_2, VALUE_2, VALUE_3); - nativeConnection.pfmerge(SAME_SLOT_KEY_3, SAME_SLOT_KEY_1, SAME_SLOT_KEY_2); + ((RedisHLLCommands) nativeConnection).pfmerge(SAME_SLOT_KEY_3, SAME_SLOT_KEY_1, SAME_SLOT_KEY_2); - assertThat(nativeConnection.pfcount(SAME_SLOT_KEY_3), is(3L)); + assertThat(((RedisHLLCommands) nativeConnection).pfcount(SAME_SLOT_KEY_3), is(3L)); } /** @@ -2558,7 +2558,6 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { */ @Test @IfProfileValue(name = "redisVersion", value = "3.2+") - @Ignore("see mp911de/lettuce#241") public void geoHash() { nativeConnection.geoadd(KEY_1, PALERMO.getPoint().getX(), PALERMO.getPoint().getY(), PALERMO.getName()); @@ -2573,7 +2572,6 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { */ @Test @IfProfileValue(name = "redisVersion", value = "3.2+") - @Ignore("see mp911de/lettuce#241") public void geoHashNonExisting() { nativeConnection.geoadd(KEY_1, PALERMO.getPoint().getX(), PALERMO.getPoint().getY(), PALERMO.getName()); 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 d9391d57c..6be8da2d7 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 @@ -30,7 +30,6 @@ import java.util.Collections; import java.util.List; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; @@ -40,10 +39,10 @@ import org.springframework.data.redis.connection.ClusterNodeResourceProvider; import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots; import org.springframework.data.redis.connection.RedisClusterNode; -import com.lambdaworks.redis.RedisAsyncConnection; -import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisURI; import com.lambdaworks.redis.cluster.RedisClusterClient; +import com.lambdaworks.redis.cluster.api.async.RedisClusterAsyncCommands; +import com.lambdaworks.redis.cluster.api.sync.RedisClusterCommands; import com.lambdaworks.redis.cluster.models.partitions.Partitions; import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag; @@ -65,10 +64,10 @@ public class LettuceClusterConnectionUnitTests { @Mock RedisClusterClient clusterMock; @Mock ClusterNodeResourceProvider resourceProvider; - @Mock RedisAsyncConnection dedicatedConnectionMock; - @Mock RedisConnection clusterConnection1Mock; - @Mock RedisConnection clusterConnection2Mock; - @Mock RedisConnection clusterConnection3Mock; + @Mock RedisClusterAsyncCommands dedicatedConnectionMock; + @Mock RedisClusterCommands clusterConnection1Mock; + @Mock RedisClusterCommands clusterConnection2Mock; + @Mock RedisClusterCommands clusterConnection3Mock; LettuceClusterConnection connection; @@ -112,7 +111,7 @@ public class LettuceClusterConnectionUnitTests { connection = new LettuceClusterConnection(clusterMock, executor) { @Override - protected RedisAsyncConnection getAsyncDedicatedConnection() { + protected RedisClusterAsyncCommands getAsyncDedicatedConnection() { return dedicatedConnectionMock; } @@ -139,12 +138,12 @@ public class LettuceClusterConnectionUnitTests { connection.clusterMeet(UNKNOWN_CLUSTER_NODE); - verify(clusterConnection1Mock, times(1)) - .clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), UNKNOWN_CLUSTER_NODE.getPort()); - verify(clusterConnection2Mock, times(1)) - .clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), UNKNOWN_CLUSTER_NODE.getPort()); - verify(clusterConnection3Mock, times(1)) - .clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), UNKNOWN_CLUSTER_NODE.getPort()); + verify(clusterConnection1Mock, times(1)).clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), + UNKNOWN_CLUSTER_NODE.getPort()); + verify(clusterConnection2Mock, times(1)).clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), + UNKNOWN_CLUSTER_NODE.getPort()); + verify(clusterConnection3Mock, times(1)).clusterMeet(UNKNOWN_CLUSTER_NODE.getHost(), + UNKNOWN_CLUSTER_NODE.getPort()); } /** @@ -210,9 +209,9 @@ public class LettuceClusterConnectionUnitTests { @Test public void keysShouldBeRunOnAllClusterNodes() { - when(clusterConnection1Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); - when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); - when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); + when(clusterConnection1Mock.keys(any(byte[].class))).thenReturn(Collections.emptyList()); + when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.emptyList()); + when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections.emptyList()); byte[] pattern = LettuceConverters.toBytes("*"); @@ -229,7 +228,7 @@ public class LettuceClusterConnectionUnitTests { @Test public void keysShouldOnlyBeRunOnDedicatedNodeWhenPinned() { - when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); + when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.emptyList()); byte[] pattern = LettuceConverters.toBytes("*"); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index c6589637e..e2bd7924f 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -33,8 +33,8 @@ import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.StringRedisConnection; -import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisException; +import com.lambdaworks.redis.api.async.RedisAsyncCommands; /** * Integration test of {@link LettuceConnectionFactory} @@ -79,8 +79,8 @@ public class LettuceConnectionFactoryTests { public void testGetNewConnectionOnError() throws Exception { factory.setValidateConnection(true); connection.lPush("alist", "baz"); - RedisAsyncConnection nativeConn = (RedisAsyncConnection) connection.getNativeConnection(); - nativeConn.close(); + RedisAsyncCommands nativeConn = (RedisAsyncCommands) connection.getNativeConnection(); + nativeConn.getStatefulConnection().close(); // Give some time for async channel close Thread.sleep(500); connection.bLPop(1, "alist".getBytes()); @@ -101,7 +101,7 @@ public class LettuceConnectionFactoryTests { @Test public void testConnectionErrorNoValidate() throws Exception { connection.lPush("ablist", "baz"); - ((RedisAsyncConnection) connection.getNativeConnection()).close(); + ((RedisAsyncCommands) connection.getNativeConnection()).getStatefulConnection().close(); // Give some time for async channel close Thread.sleep(500); DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(factory.getConnection()); @@ -159,7 +159,7 @@ public class LettuceConnectionFactoryTests { // Give some time for native connection to asynchronously close Thread.sleep(100); try { - ((RedisAsyncConnection) conn2.getNativeConnection()).ping(); + ((RedisAsyncCommands) conn2.getNativeConnection()).ping(); fail("The native connection should be closed"); } catch (RedisException e) { // expected @@ -169,17 +169,17 @@ public class LettuceConnectionFactoryTests { @SuppressWarnings("unchecked") @Test public void testResetConnection() { - RedisAsyncConnection nativeConn = (RedisAsyncConnection) connection + RedisAsyncCommands nativeConn = (RedisAsyncCommands) connection .getNativeConnection(); factory.resetConnection(); assertNotSame(nativeConn, factory.getConnection().getNativeConnection()); - nativeConn.close(); + nativeConn.getStatefulConnection().close(); } @SuppressWarnings("unchecked") @Test public void testInitConnection() { - RedisAsyncConnection nativeConn = (RedisAsyncConnection) connection + RedisAsyncCommands nativeConn = (RedisAsyncCommands) connection .getNativeConnection(); factory.initConnection(); RedisConnection newConnection = factory.getConnection(); @@ -190,7 +190,7 @@ public class LettuceConnectionFactoryTests { @SuppressWarnings("unchecked") @Test public void testResetAndInitConnection() { - RedisAsyncConnection nativeConn = (RedisAsyncConnection) connection + RedisAsyncCommands nativeConn = (RedisAsyncCommands) connection .getNativeConnection(); factory.resetConnection(); factory.initConnection(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index e7990e05f..571fb952c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -49,7 +49,7 @@ import org.springframework.data.redis.test.util.RequiresRedisSentinel; import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; -import com.lambdaworks.redis.RedisAsyncConnection; +import com.lambdaworks.redis.api.async.RedisAsyncCommands; /** * Integration test of {@link LettuceConnection} @@ -210,7 +210,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra RedisConnection connection = factory2.getConnection(); // Use the connection to make sure the channel is initialized, else nothing happens on close connection.ping(); - ((RedisAsyncConnection) connection.getNativeConnection()).close(); + ((RedisAsyncCommands) connection.getNativeConnection()).getStatefulConnection().close(); try { connection.ping(); fail("Exception should be thrown trying to use a closed connection"); @@ -351,8 +351,8 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra @RequiresRedisSentinel(RedisSentinelRule.SentinelsAvailable.ONE_ACTIVE) public void shouldReturnSentinelCommandsWhenWhenActiveSentinelFound() { - ((LettuceConnection) byteConnection).setSentinelConfiguration(new RedisSentinelConfiguration().master("mymaster") - .sentinel("127.0.0.1", 26379).sentinel("127.0.0.1", 26380)); + ((LettuceConnection) byteConnection).setSentinelConfiguration( + new RedisSentinelConfiguration().master("mymaster").sentinel("127.0.0.1", 26379).sentinel("127.0.0.1", 26380)); assertThat(connection.getSentinelConnection(), notNullValue()); } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java index 7066f6ea0..88409576a 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java @@ -19,7 +19,6 @@ import static org.junit.Assert.*; import java.util.Arrays; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests; @@ -67,84 +66,4 @@ public class LettuceConnectionTransactionIntegrationTests extends AbstractConnec public void testSelect() { super.testSelect(); } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoPosition() { - super.geoPosition(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoPositionNonExisting() { - super.geoPositionNonExisting(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoRadiusByMemberShouldApplyLimit() { - super.geoRadiusByMemberShouldApplyLimit(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoRadiusByMemberShouldReturnDistanceCorrectly() { - super.geoRadiusByMemberShouldReturnDistanceCorrectly(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoRadiusByMemberShouldReturnMembersCorrectly() { - super.geoRadiusByMemberShouldReturnMembersCorrectly(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoRadiusShouldApplyLimit() { - super.geoRadiusShouldApplyLimit(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#") - public void geoRadiusShouldReturnDistanceCorrectly() { - super.geoRadiusShouldReturnDistanceCorrectly(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoRadiusShouldReturnMembersCorrectly() { - super.geoRadiusShouldReturnMembersCorrectly(); - } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java index 6025c23a4..14213372b 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java @@ -15,7 +15,8 @@ */ package org.springframework.data.redis.connection.lettuce; -import static org.mockito.Matchers.*; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; import static org.mockito.Mockito.*; import java.lang.reflect.InvocationTargetException; @@ -30,8 +31,10 @@ import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOpt import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettuceConnectionUnitTests; import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettucePipelineConnectionUnitTests; -import com.lambdaworks.redis.RedisAsyncConnectionImpl; import com.lambdaworks.redis.RedisClient; +import com.lambdaworks.redis.api.StatefulRedisConnection; +import com.lambdaworks.redis.api.async.RedisAsyncCommands; +import com.lambdaworks.redis.api.sync.RedisCommands; import com.lambdaworks.redis.codec.RedisCodec; /** @@ -43,17 +46,27 @@ import com.lambdaworks.redis.codec.RedisCodec; public class LettuceConnectionUnitTestSuite { @SuppressWarnings("rawtypes") - public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase { + public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase { protected LettuceConnection connection; private RedisClient clientMock; + protected StatefulRedisConnection statefulConnectionMock; + protected RedisAsyncCommands asyncCommandsMock; + protected RedisCommands syncCommandsMock; @SuppressWarnings({ "unchecked" }) @Before public void setUp() throws InvocationTargetException, IllegalAccessException { clientMock = mock(RedisClient.class); - when(clientMock.connectAsync((RedisCodec) any())).thenReturn(getNativeRedisConnectionMock()); + statefulConnectionMock = mock(StatefulRedisConnection.class); + when(clientMock.connect((RedisCodec) any())).thenReturn(statefulConnectionMock); + + asyncCommandsMock = getNativeRedisConnectionMock(); + syncCommandsMock = mock(RedisCommands.class); + + when(statefulConnectionMock.async()).thenReturn(getNativeRedisConnectionMock()); + when(statefulConnectionMock.sync()).thenReturn(syncCommandsMock); connection = new LettuceConnection(0, clientMock); } @@ -61,10 +74,10 @@ public class LettuceConnectionUnitTestSuite { * @see DATAREDIS-184 */ @Test - public void shutdownWithNullOpionsIsCalledCorrectly() { + public void shutdownWithNullOptionsIsCalledCorrectly() { connection.shutdown(null); - verifyNativeConnectionInvocation().shutdown(true); + verify(syncCommandsMock, times(1)).shutdown(true); } /** @@ -74,7 +87,7 @@ public class LettuceConnectionUnitTestSuite { public void shutdownWithNosaveOptionIsCalledCorrectly() { connection.shutdown(ShutdownOption.NOSAVE); - verifyNativeConnectionInvocation().shutdown(false); + verify(syncCommandsMock, times(1)).shutdown(false); } /** @@ -84,7 +97,7 @@ public class LettuceConnectionUnitTestSuite { public void shutdownWithSaveOptionIsCalledCorrectly() { connection.shutdown(ShutdownOption.SAVE); - verifyNativeConnectionInvocation().shutdown(true); + verify(syncCommandsMock, times(1)).shutdown(true); } /** @@ -95,7 +108,7 @@ public class LettuceConnectionUnitTestSuite { String ipPort = "127.0.0.1:1001"; connection.killClient("127.0.0.1", 1001); - verifyNativeConnectionInvocation().clientKill(eq(ipPort)); + verify(syncCommandsMock, times(1)).clientKill(eq(ipPort)); } /** @@ -105,7 +118,7 @@ public class LettuceConnectionUnitTestSuite { public void getClientNameShouldSendRequestCorrectly() { connection.getClientName(); - verifyNativeConnectionInvocation().clientGetname(); + verify(syncCommandsMock, times(1)).clientGetname(); } /** @@ -123,7 +136,7 @@ public class LettuceConnectionUnitTestSuite { public void slaveOfShouldBeSentCorrectly() { connection.slaveOf("127.0.0.1", 1001); - verifyNativeConnectionInvocation().slaveof(eq("127.0.0.1"), eq(1001)); + verify(syncCommandsMock, times(1)).slaveof(eq("127.0.0.1"), eq(1001)); } /** @@ -133,7 +146,7 @@ public class LettuceConnectionUnitTestSuite { public void slaveOfNoOneShouldBeSentCorrectly() { connection.slaveOfNoOne(); - verifyNativeConnectionInvocation().slaveofNoOne(); + verify(syncCommandsMock, times(1)).slaveofNoOne(); } /** @@ -148,12 +161,12 @@ public class LettuceConnectionUnitTestSuite { * @see DATAREDIS-431 */ @Test - public void dbIndexShouldBeSetWhenOptainingConnection() { + public void dbIndexShouldBeSetWhenObtainingConnection() { connection = new LettuceConnection(null, 0, clientMock, null, 1); connection.getNativeConnection(); - verify(getNativeRedisConnectionMock(), times(1)).select(1); + verify(syncCommandsMock, times(1)).select(1); } } @@ -165,6 +178,76 @@ public class LettuceConnectionUnitTestSuite { super.setUp(); this.connection.openPipeline(); } - } + /** + * @see DATAREDIS-528 + */ + @Test + public void shutdownWithSaveOptionIsCalledCorrectly() { + + connection.shutdown(ShutdownOption.SAVE); + verify(asyncCommandsMock, times(1)).shutdown(true); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void shutdownWithNosaveOptionIsCalledCorrectly() { + + connection.shutdown(ShutdownOption.NOSAVE); + verify(asyncCommandsMock, times(1)).shutdown(false); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void slaveOfShouldBeSentCorrectly() { + + connection.slaveOf("127.0.0.1", 1001); + verify(asyncCommandsMock, times(1)).slaveof(eq("127.0.0.1"), eq(1001)); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void shutdownWithNullOptionsIsCalledCorrectly() { + + connection.shutdown(null); + verify(asyncCommandsMock, times(1)).shutdown(true); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void killClientShouldDelegateCallCorrectly() { + + String ipPort = "127.0.0.1:1001"; + connection.killClient("127.0.0.1", 1001); + verify(asyncCommandsMock, times(1)).clientKill(eq(ipPort)); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void slaveOfNoOneShouldBeSentCorrectly() { + + connection.slaveOfNoOne(); + verify(asyncCommandsMock, times(1)).slaveofNoOne(); + } + + /** + * @see DATAREDIS-528 + */ + @Test + public void getClientNameShouldSendRequestCorrectly() { + + connection.getClientName(); + verify(asyncCommandsMock, times(1)).clientGetname(); + } + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConvertersUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConvertersUnitTests.java index b4f1b3e2a..8a3136342 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConvertersUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConvertersUnitTests.java @@ -37,9 +37,9 @@ import org.springframework.data.redis.core.types.Expiration; import org.springframework.data.redis.core.types.RedisClientInfo; import com.lambdaworks.redis.RedisURI; +import com.lambdaworks.redis.SetArgs; import com.lambdaworks.redis.cluster.models.partitions.Partitions; import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag; -import com.lambdaworks.redis.protocol.SetArgs; /** * @author Christoph Strobl @@ -53,7 +53,7 @@ public class LettuceConvertersUnitTests { */ @Test public void convertingEmptyStringToListOfRedisClientInfoShouldReturnEmptyList() { - assertThat(LettuceConverters.toListOfRedisClientInformation(""), equalTo(Collections. emptyList())); + assertThat(LettuceConverters.toListOfRedisClientInformation(""), equalTo(Collections.emptyList())); } /** @@ -62,7 +62,7 @@ public class LettuceConvertersUnitTests { @Test public void convertingNullToListOfRedisClientInfoShouldReturnEmptyList() { assertThat(LettuceConverters.toListOfRedisClientInformation(null), - equalTo(Collections. emptyList())); + equalTo(Collections.emptyList())); } /** @@ -100,7 +100,7 @@ public class LettuceConvertersUnitTests { partition.setConnected(true); partition.setFlags(new HashSet(Arrays.asList(NodeFlag.MASTER, NodeFlag.MYSELF))); partition.setUri(RedisURI.create("redis://" + CLUSTER_HOST + ":" + MASTER_NODE_1_PORT)); - partition.setSlots(Arrays. asList(1, 2, 3, 4, 5)); + partition.setSlots(Arrays.asList(1, 2, 3, 4, 5)); partitions.addPartition(partition); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java index 86fc1afb3..b38b00ddb 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.lettuce; import static org.mockito.Matchers.*; import static org.mockito.Mockito.*; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -32,7 +33,8 @@ import org.springframework.data.redis.connection.RedisServer; import com.lambdaworks.redis.RedisClient; import com.lambdaworks.redis.RedisFuture; -import com.lambdaworks.redis.RedisSentinelAsyncConnection; +import com.lambdaworks.redis.sentinel.api.StatefulRedisSentinelConnection; +import com.lambdaworks.redis.sentinel.api.sync.RedisSentinelCommands; /** * @author Christoph Strobl @@ -45,7 +47,8 @@ public class LettuceSentinelConnectionUnitTests { private @Mock RedisClient redisClientMock; - private @Mock RedisSentinelAsyncConnection connectionMock; + private @Mock StatefulRedisSentinelConnection connectionMock; + private @Mock RedisSentinelCommands sentinelCommandsMock; private @Mock RedisFuture>> redisFutureMock; @@ -54,7 +57,8 @@ public class LettuceSentinelConnectionUnitTests { @Before public void setUp() { - when(redisClientMock.connectSentinelAsync()).thenReturn(connectionMock); + when(redisClientMock.connectSentinel()).thenReturn(connectionMock); + when(connectionMock.sync()).thenReturn(sentinelCommandsMock); this.connection = new LettuceSentinelConnection(redisClientMock); } @@ -63,7 +67,7 @@ public class LettuceSentinelConnectionUnitTests { */ @Test public void shouldConnectAfterCreation() { - verify(redisClientMock, times(1)).connectSentinelAsync(); + verify(redisClientMock, times(1)).connectSentinel(); } /** @@ -73,7 +77,7 @@ public class LettuceSentinelConnectionUnitTests { public void failoverShouldBeSentCorrectly() { connection.failover(new RedisNodeBuilder().withName(MASTER_ID).build()); - verify(connectionMock, times(1)).failover(eq(MASTER_ID)); + verify(sentinelCommandsMock, times(1)).failover(eq(MASTER_ID)); } /** @@ -98,9 +102,9 @@ public class LettuceSentinelConnectionUnitTests { @Test public void mastersShouldReadMastersCorrectly() { - when(connectionMock.masters()).thenReturn(redisFutureMock); + when(sentinelCommandsMock.masters()).thenReturn(Collections.>emptyList()); connection.masters(); - verify(connectionMock, times(1)).masters(); + verify(sentinelCommandsMock, times(1)).masters(); } /** @@ -109,9 +113,9 @@ public class LettuceSentinelConnectionUnitTests { @Test public void shouldReadSlavesCorrectly() { - when(connectionMock.slaves(MASTER_ID)).thenReturn(redisFutureMock); + when(sentinelCommandsMock.slaves(MASTER_ID)).thenReturn(Collections.>emptyList()); connection.slaves(MASTER_ID); - verify(connectionMock, times(1)).slaves(eq(MASTER_ID)); + verify(sentinelCommandsMock, times(1)).slaves(eq(MASTER_ID)); } /** @@ -120,9 +124,9 @@ public class LettuceSentinelConnectionUnitTests { @Test public void shouldReadSlavesCorrectlyWhenGivenNamedNode() { - when(connectionMock.slaves(MASTER_ID)).thenReturn(redisFutureMock); + when(sentinelCommandsMock.slaves(MASTER_ID)).thenReturn(Collections.>emptyList()); connection.slaves(new RedisNodeBuilder().withName(MASTER_ID).build()); - verify(connectionMock, times(1)).slaves(eq(MASTER_ID)); + verify(sentinelCommandsMock, times(1)).slaves(eq(MASTER_ID)); } /** @@ -156,7 +160,7 @@ public class LettuceSentinelConnectionUnitTests { public void shouldRemoveMasterCorrectlyWhenGivenNamedNode() { connection.remove(new RedisNodeBuilder().withName(MASTER_ID).build()); - verify(connectionMock, times(1)).remove(eq(MASTER_ID)); + verify(sentinelCommandsMock, times(1)).remove(eq(MASTER_ID)); } /** @@ -194,7 +198,6 @@ public class LettuceSentinelConnectionUnitTests { server.setQuorum(3L); connection.monitor(server); - verify(connectionMock, times(1)).monitor(eq("anothermaster"), eq("127.0.0.1"), eq(6382), eq(3)); + verify(sentinelCommandsMock, times(1)).monitor(eq("anothermaster"), eq("127.0.0.1"), eq(6382), eq(3)); } - } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java index deb70242b..ab02cd35c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java @@ -124,24 +124,4 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati connection.close(); } } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoHash() { - super.geoHash(); - } - - /** - * @see DATAREDIS-438 - */ - @Test - @Override - @Ignore("see mp911de/lettuce#241") - public void geoHashNonExisting() { - super.geoHashNonExisting(); - } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java index 08e3f8615..d2856df84 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java @@ -15,14 +15,10 @@ */ package org.springframework.data.redis.connection.lettuce; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.any; +import static org.junit.Assert.*; +import static org.mockito.Matchers.*; +import static org.mockito.Mockito.*; + import java.util.Collection; import org.junit.Before; @@ -31,7 +27,8 @@ import org.mockito.Mockito; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisInvalidSubscriptionException; -import com.lambdaworks.redis.pubsub.RedisPubSubConnection; +import com.lambdaworks.redis.pubsub.StatefulRedisPubSubConnection; +import com.lambdaworks.redis.pubsub.api.sync.RedisPubSubCommands; /** * Unit test of {@link LettuceSubscription} @@ -42,15 +39,21 @@ public class LettuceSubscriptionTests { private LettuceSubscription subscription; - RedisPubSubConnection pubsub; + StatefulRedisPubSubConnection pubsub; private MessageListener listener; + private RedisPubSubCommands asyncCommands; + @SuppressWarnings("unchecked") @Before public void setUp() { - pubsub = Mockito.mock(RedisPubSubConnection.class); + + pubsub = Mockito.mock(StatefulRedisPubSubConnection.class); listener = Mockito.mock(MessageListener.class); + asyncCommands = Mockito.mock(RedisPubSubCommands.class); + + Mockito.when(pubsub.sync()).thenReturn(asyncCommands); subscription = new LettuceSubscription(listener, pubsub); } @@ -58,9 +61,9 @@ public class LettuceSubscriptionTests { public void testUnsubscribeAllAndClose() { subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.unsubscribe(); - verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); verify(pubsub).close(); verify(pubsub).removeListener(any(LettuceMessageListener.class)); assertFalse(subscription.isAlive()); @@ -73,9 +76,9 @@ public class LettuceSubscriptionTests { subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(); - verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); assertTrue(subscription.getChannels().isEmpty()); Collection patterns = subscription.getPatterns(); @@ -88,9 +91,9 @@ public class LettuceSubscriptionTests { byte[][] channel = new byte[][] { "a".getBytes() }; subscription.subscribe(channel); subscription.unsubscribe(channel); - verify(pubsub, times(1)).unsubscribe(channel); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(channel); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); verify(pubsub).close(); verify(pubsub).removeListener(any(LettuceMessageListener.class)); assertFalse(subscription.isAlive()); @@ -103,9 +106,9 @@ public class LettuceSubscriptionTests { byte[][] channels = new byte[][] { "a".getBytes(), "b".getBytes() }; subscription.subscribe(channels); subscription.unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); Collection subChannels = subscription.getChannels(); assertEquals(1, subChannels.size()); @@ -119,9 +122,9 @@ public class LettuceSubscriptionTests { subscription.subscribe(channel); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(channel); - verify(pubsub, times(1)).unsubscribe(channel); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(channel); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); assertTrue(subscription.getChannels().isEmpty()); Collection patterns = subscription.getPatterns(); @@ -135,9 +138,9 @@ public class LettuceSubscriptionTests { subscription.subscribe(new byte[][] { "a".getBytes(), "b".getBytes() }); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(channel); - verify(pubsub, times(1)).unsubscribe(channel); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(channel); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); Collection channels = subscription.getChannels(); assertEquals(1, channels.size()); @@ -151,8 +154,8 @@ public class LettuceSubscriptionTests { public void testUnsubscribeAllNoChannels() { subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.unsubscribe(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); assertTrue(subscription.getChannels().isEmpty()); Collection patterns = subscription.getPatterns(); @@ -168,9 +171,9 @@ public class LettuceSubscriptionTests { verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class)); assertFalse(subscription.isAlive()); subscription.unsubscribe(); - verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); } @Test(expected = RedisInvalidSubscriptionException.class) @@ -185,9 +188,9 @@ public class LettuceSubscriptionTests { public void testPUnsubscribeAllAndClose() { subscription.pSubscribe(new byte[][] { "a*".getBytes() }); subscription.pUnsubscribe(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); - verify(pubsub, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); assertFalse(subscription.isAlive()); verify(pubsub).close(); verify(pubsub).removeListener(any(LettuceMessageListener.class)); @@ -200,9 +203,9 @@ public class LettuceSubscriptionTests { subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(new byte[][] { "s*".getBytes() }); subscription.pUnsubscribe(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); - verify(pubsub, times(1)).punsubscribe(new byte[][] { "s*".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(new byte[][] { "s*".getBytes() }); assertTrue(subscription.isAlive()); assertTrue(subscription.getPatterns().isEmpty()); Collection channels = subscription.getChannels(); @@ -215,9 +218,9 @@ public class LettuceSubscriptionTests { byte[][] pattern = new byte[][] { "a*".getBytes() }; subscription.pSubscribe(pattern); subscription.pUnsubscribe(pattern); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); - verify(pubsub, times(1)).punsubscribe(pattern); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(pattern); verify(pubsub).close(); verify(pubsub).removeListener(any(LettuceMessageListener.class)); assertFalse(subscription.isAlive()); @@ -230,9 +233,9 @@ public class LettuceSubscriptionTests { byte[][] patterns = new byte[][] { "a*".getBytes(), "b*".getBytes() }; subscription.pSubscribe(patterns); subscription.pUnsubscribe(new byte[][] { "a*".getBytes() }); - verify(pubsub, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(new byte[][] { "a*".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); Collection subPatterns = subscription.getPatterns(); assertEquals(1, subPatterns.size()); @@ -246,9 +249,9 @@ public class LettuceSubscriptionTests { subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pSubscribe(pattern); subscription.pUnsubscribe(pattern); - verify(pubsub, times(1)).punsubscribe(pattern); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(pattern); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); assertTrue(subscription.getPatterns().isEmpty()); Collection channels = subscription.getChannels(); @@ -262,9 +265,9 @@ public class LettuceSubscriptionTests { subscription.pSubscribe(new byte[][] { "a*".getBytes(), "b*".getBytes() }); subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.pUnsubscribe(pattern); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); - verify(pubsub, times(1)).punsubscribe(pattern); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(pattern); assertTrue(subscription.isAlive()); Collection channels = subscription.getChannels(); assertEquals(1, channels.size()); @@ -278,8 +281,8 @@ public class LettuceSubscriptionTests { public void testPUnsubscribeAllNoPatterns() { subscription.subscribe(new byte[][] { "s".getBytes() }); subscription.pUnsubscribe(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); assertTrue(subscription.isAlive()); assertTrue(subscription.getPatterns().isEmpty()); Collection channels = subscription.getChannels(); @@ -295,9 +298,9 @@ public class LettuceSubscriptionTests { subscription.pUnsubscribe(); verify(pubsub, times(1)).close(); verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class)); - verify(pubsub, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() }); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); } @Test(expected = RedisInvalidSubscriptionException.class) @@ -311,24 +314,23 @@ public class LettuceSubscriptionTests { @Test public void testDoCloseNotSubscribed() { subscription.doClose(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); } @Test public void testDoCloseSubscribedChannels() { subscription.subscribe(new byte[][] { "a".getBytes() }); subscription.doClose(); - verify(pubsub, times(1)).unsubscribe(new byte[0]); - verify(pubsub, never()).punsubscribe(new byte[0]); + verify(asyncCommands, times(1)).unsubscribe(new byte[0]); + verify(asyncCommands, never()).punsubscribe(new byte[0]); } @Test public void testDoCloseSubscribedPatterns() { subscription.pSubscribe(new byte[][] { "a*".getBytes() }); subscription.doClose(); - verify(pubsub, never()).unsubscribe(new byte[0]); - verify(pubsub, times(1)).punsubscribe(new byte[0]); + verify(asyncCommands, never()).unsubscribe(new byte[0]); + verify(asyncCommands, times(1)).punsubscribe(new byte[0]); } - } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientResources.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientResources.java index 3570d4a4a..65aec9a15 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientResources.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientResources.java @@ -35,8 +35,8 @@ public class LettuceTestClientResources { static { - SHARED_CLIENT_RESOURCES = new DefaultClientResources.Builder() - .eventLoopGroupProvider(new TestEventLoopGroupProvider()).build(); + SHARED_CLIENT_RESOURCES = DefaultClientResources.builder().eventLoopGroupProvider(new TestEventLoopGroupProvider()) + .build(); appendShutdownHook(); }