Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-528 - Upgrade to Lettuce 4.2.2.

Browse Source 
We now support Lettuce 4.2.2.Final. Using Lettuce requires Java 8.

Original pull request: #222.
This commit is contained in:
Christoph Strobl
2016-09-19 13:53:30 +02:00
committed by Mark Paluch
parent 863e5b8d68
commit 4be4ed0b28
28 changed files with 663 additions and 925 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -22,7 +22,7 @@
<beanutils>1.9.2</beanutils>
<xstream>1.4.8</xstream>
<pool>2.2</pool>
<lettuce>3.4.2.Final</lettuce>
<lettuce>4.2.2.Final</lettuce>
<jedis>2.9.0</jedis>
<srp>0.7</srp>
<jredis>06052013</jredis>

4
src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java
View File

@@ -372,6 +372,10 @@ public class ClusterCommandExecutor implements DisposableBean {
if (executor instanceof DisposableBean) {
((DisposableBean) executor).destroy();
}
if (resourceProvider instanceof DisposableBean) {
((DisposableBean) resourceProvider).destroy();
}
}
/**

34
src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java
View File

@@ -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 <K, V> RedisConnection<K, V> connect(RedisCodec<K, V> codec) {
RedisConnection<K, V> conn = super.connect(codec);
conn.auth(password);
return conn;
public <K, V> StatefulRedisConnection<K, V> connect(RedisCodec<K, V> codec) {
return super.connect(codec);
}
@Override
public <K, V> RedisAsyncConnection<K, V> connectAsync(RedisCodec<K, V> codec) {
RedisAsyncConnection<K, V> conn = super.connectAsync(codec);
conn.auth(password);
return conn;
public <K, V> RedisAsyncCommands<K, V> connectAsync(RedisCodec<K, V> codec) {
return super.connectAsync(codec);
}
@Override
public <K, V> RedisPubSubConnection<K, V> connectPubSub(RedisCodec<K, V> codec) {
RedisPubSubConnection<K, V> conn = super.connectPubSub(codec);
conn.auth(password);
return conn;
public <K, V> StatefulRedisPubSubConnection<K, V> connectPubSub(RedisCodec<K, V> codec) {
return super.connectPubSub(codec);
}
}

55
src/main/java/org/springframework/data/redis/connection/lettuce/BytesRedisCodec.java
View File

@@ -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<byte[], byte[]> {
@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;
}
}

40
src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java
View File

@@ -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<RedisAsyncConnection> internalPool;
private GenericObjectPool<StatefulConnection<byte[], byte[]>> 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<RedisAsyncConnection>(new LettuceFactory(client, dbIndex), poolConfig);
this.internalPool = new GenericObjectPool<StatefulConnection<byte[], byte[]>>(new LettuceFactory(client, dbIndex),
poolConfig);
}
/**
@@ -135,7 +137,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
}
@SuppressWarnings("unchecked")
public RedisAsyncConnection<byte[], byte[]> getResource() {
public StatefulConnection<byte[], byte[]> getResource() {
try {
return internalPool.borrowObject();
} catch (Exception e) {
@@ -143,7 +145,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
}
}
public void returnBrokenResource(final RedisAsyncConnection<byte[], byte[]> resource) {
public void returnBrokenResource(final StatefulConnection<byte[], byte[]> resource) {
try {
internalPool.invalidateObject(resource);
} catch (Exception e) {
@@ -151,7 +153,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
}
}
public void returnResource(final RedisAsyncConnection<byte[], byte[]> resource) {
public void returnResource(final StatefulConnection<byte[], byte[]> 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<RedisAsyncConnection> {
private static class LettuceFactory extends BasePooledObjectFactory<StatefulConnection<byte[], byte[]>> {
private final RedisClient client;
@@ -315,11 +317,14 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
}
@Override
public void activateObject(PooledObject<RedisAsyncConnection> pooledObject) throws Exception {
pooledObject.getObject().select(dbIndex);
public void activateObject(PooledObject<StatefulConnection<byte[], byte[]>> pooledObject) throws Exception {
if (pooledObject.getObject() instanceof StatefulRedisConnection) {
((StatefulRedisConnection) pooledObject.getObject()).sync().select(dbIndex);
}
}
public void destroyObject(final PooledObject<RedisAsyncConnection> obj) throws Exception {
public void destroyObject(final PooledObject<StatefulConnection<byte[], byte[]>> 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<RedisAsyncConnection> obj) {
public boolean validateObject(final PooledObject<StatefulConnection<byte[], byte[]>> 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<byte[], byte[]> create() throws Exception {
return client.connect(LettuceConnection.CODEC);
}
@Override
public PooledObject<RedisAsyncConnection> wrap(RedisAsyncConnection obj) {
return new DefaultPooledObject<RedisAsyncConnection>(obj);
public PooledObject<StatefulConnection<byte[], byte[]>> wrap(StatefulConnection<byte[], byte[]> obj) {
return new DefaultPooledObject<StatefulConnection<byte[], byte[]>>(obj);
}
}
}

205
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java
View File

@@ -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<byte[], byte[]> CODEC = new BytesRedisCodec();
static final RedisCodec<byte[], byte[]> CODEC = ByteArrayCodec.INSTANCE;
private final RedisClusterClient clusterClient;
private ClusterCommandExecutor clusterCommandExecutor;
@@ -141,7 +141,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> connection) {
public List<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> connection) {
return connection.keys(pattern);
}
}).resultsAsList();
@@ -164,7 +164,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.flushall();
}
});
@@ -180,7 +180,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.flushdb();
}
});
@@ -197,7 +197,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Long>() {
@Override
public Long doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Long doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.dbsize();
}
@@ -227,7 +227,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Properties doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info());
}
}).getResults();
@@ -249,7 +249,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<Properties>() {
@Override
public Properties doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Properties doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return LettuceConverters.toProperties(client.info(section));
}
}).getResults();
@@ -274,7 +274,7 @@ public class LettuceClusterConnection extends LettuceConnection
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> 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<Set<RedisClusterNode>>() {
@Override
public Set<RedisClusterNode> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Set<RedisClusterNode> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return LettuceConverters.toSetOfRedisClusterNodes(client.clusterSlaves(nodeToUse.getId()));
}
}, master).getValue();
@@ -371,7 +363,7 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback<ClusterInfo>() {
@Override
public ClusterInfo doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public ClusterInfo doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return new ClusterInfo(LettuceConverters.toProperties(client.clusterInfo()));
}
}).getValue();
@@ -387,7 +379,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clusterAddSlots(slots);
}
}, node);
@@ -416,7 +408,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clusterDelSlots(slots);
}
}, node);
@@ -448,7 +440,7 @@ public class LettuceClusterConnection extends LettuceConnection
this.clusterCommandExecutor.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clusterForget(nodeToRemove.getId());
}
@@ -469,7 +461,7 @@ public class LettuceClusterConnection extends LettuceConnection
this.clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clusterMeet(node.getHost(), node.getPort());
}
});
@@ -491,7 +483,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
switch (mode) {
case MIGRATING:
return client.clusterSetSlotMigrating(slot, nodeId);
@@ -547,7 +539,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clusterReplicate(masterNode.getId());
}
}, slave);
@@ -563,8 +555,8 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> connection) {
return doPing(connection);
public String doInCluster(RedisClusterCommands<byte[], byte[]> connection) {
return connection.ping();
}
}).resultsAsList();
@@ -587,29 +579,12 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
return doPing(client);
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.ping();
}
}, node).getValue();
}
protected String doPing(RedisClusterConnection<byte[], byte[]> 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<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.bgrewriteaof();
}
}, node);
@@ -636,7 +611,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.bgsave();
}
}, node);
@@ -652,7 +627,7 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<Long>() {
@Override
public Long doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Long doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.lastsave().getTime();
}
}, node).getValue();
@@ -668,7 +643,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.save();
}
}, node);
@@ -685,7 +660,7 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<Long>() {
@Override
public Long doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Long doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.dbsize();
}
}, node).getValue();
@@ -701,7 +676,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.flushdb();
}
}, node);
@@ -717,7 +692,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.flushall();
}
}, node);
@@ -735,7 +710,7 @@ public class LettuceClusterConnection extends LettuceConnection
.toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.info();
}
}, node).getValue());
@@ -752,7 +727,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public List<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.keys(pattern);
}
}, node).getValue());
@@ -768,7 +743,7 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<byte[]>() {
@Override
public byte[] doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public byte[] doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.randomkey();
}
}, node).getValue();
@@ -855,7 +830,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<Void>() {
@Override
public Void doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Void doInCluster(RedisClusterCommands<byte[], byte[]> client) {
client.shutdown(true);
return null;
}
@@ -899,17 +874,10 @@ public class LettuceClusterConnection extends LettuceConnection
@Override
public List<byte[]> 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<byte[]>() {
@Override
public byte[] doInCluster(RedisClusterConnection<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<KeyValue<byte[], byte[]>>() {
@Override
public KeyValue<byte[], byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
public KeyValue<byte[], byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client, byte[] key) {
return client.blpop(timeout, key);
}
}, Arrays.asList(keys)).resultsAsList();
@@ -995,7 +957,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<KeyValue<byte[], byte[]>>() {
@Override
public KeyValue<byte[], byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
public KeyValue<byte[], byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> 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<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
public Set<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys)).resultsAsList();
@@ -1140,7 +1114,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
public Set<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client, byte[] key) {
return client.smembers(key);
}
}, Arrays.asList(keys)).resultsAsList();
@@ -1196,7 +1170,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback<Set<byte[]>>() {
@Override
public Set<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client, byte[] key) {
public Set<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> 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<byte[], byte[]> doGetAsyncDedicatedConnection() {
return (RedisAsyncConnection<byte[], byte[]>) clusterClient.connectClusterAsync(CODEC);
protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
return clusterClient.connect(CODEC);
}
// --> cluster node stuff
@@ -1332,7 +1306,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandOnAllNodes(new LettuceClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public List<String> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configGet(pattern);
}
}).getResults();
@@ -1360,7 +1334,7 @@ public class LettuceClusterConnection extends LettuceConnection
return clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<List<String>>() {
@Override
public List<String> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public List<String> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configGet(pattern);
}
}, node).getValue();
@@ -1376,7 +1350,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configSet(param, value);
}
});
@@ -1392,7 +1366,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configSet(param, value);
}
}, node);
@@ -1409,7 +1383,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configResetstat();
}
});
@@ -1425,7 +1399,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.configResetstat();
}
}, node);
@@ -1442,7 +1416,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public List<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.time();
}
}).getValue());
@@ -1459,7 +1433,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<List<byte[]>>() {
@Override
public List<byte[]> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public List<byte[]> doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.time();
}
}, node).getValue());
@@ -1485,7 +1459,7 @@ public class LettuceClusterConnection extends LettuceConnection
List<String> map = clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clientList();
}
}).resultsAsList();
@@ -1508,7 +1482,7 @@ public class LettuceClusterConnection extends LettuceConnection
clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback<String>() {
@Override
public String doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public String doInCluster(RedisClusterCommands<byte[], byte[]> client) {
return client.clientList();
}
}, node).getValue());
@@ -1525,7 +1499,7 @@ public class LettuceClusterConnection extends LettuceConnection
.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback<Collection<RedisClusterNode>>() {
@Override
public Set<RedisClusterNode> doInCluster(RedisClusterConnection<byte[], byte[]> client) {
public Set<RedisClusterNode> doInCluster(RedisClusterCommands<byte[], byte[]> 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<T>
extends ClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {}
extends ClusterCommandCallback<RedisClusterCommands<byte[], byte[]>, T> {}
/**
* Lettuce specific implementation of {@link MultiKeyClusterCommandCallback}.
@@ -1557,7 +1531,7 @@ public class LettuceClusterConnection extends LettuceConnection
* @since 1.7
*/
protected interface LettuceMultiKeyClusterCommandCallback<T>
extends MultiKeyClusterCommandCallback<RedisClusterConnection<byte[], byte[]>, T> {
extends MultiKeyClusterCommandCallback<RedisClusterCommands<byte[], byte[]>, 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<byte[], byte[]> connection;
public LettuceClusterNodeResourceProvider(RedisClusterClient client) {
@@ -1578,14 +1553,20 @@ public class LettuceClusterConnection extends LettuceConnection
@Override
@SuppressWarnings("unchecked")
public RedisClusterConnection<byte[], byte[]> getResourceForSpecificNode(RedisClusterNode node) {
public RedisClusterCommands<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> connection = (RedisClusterConnection<byte[], byte[]>) resource;
connection.close();
@Override
public void destroy() throws Exception {
if (connection != null) {
connection.close();
}
}
}
/**

392
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java
View File

@@ -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 <a href="https://github.com/mp911de/lettuce">Lettuce</a> Redis
@@ -132,9 +135,8 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection;
*/
public class LettuceConnection extends AbstractRedisConnection {
static final RedisCodec<byte[], byte[]> CODEC = new BytesRedisCodec();
static final RedisCodec<byte[], byte[]> 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<byte[], byte[]> asyncSharedConn;
private final com.lambdaworks.redis.RedisConnection<byte[], byte[]> sharedConn;
private com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> asyncDedicatedConn;
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> dedicatedConn;
private final StatefulConnection<byte[], byte[]> asyncSharedConn;
private StatefulConnection<byte[], byte[]> 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<Command<?, ?, ?>> {
private class LettuceResult extends FutureResult<com.lambdaworks.redis.protocol.RedisCommand<?, ?, ?>> {
public <T> LettuceResult(Future<T> resultHolder, Converter<T, ?> 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<byte[], byte[]> sharedConnection, long timeout,
RedisClient client) {
public LettuceConnection(StatefulRedisConnection<byte[], byte[]> 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<byte[], byte[]> sharedConnection, long timeout,
RedisClient client, LettucePool pool) {
public LettuceConnection(StatefulRedisConnection<byte[], byte[]> 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<byte[], byte[]> sharedConnection, long timeout,
public LettuceConnection(StatefulRedisConnection<byte[], byte[]> 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<byte[], byte[]> connectionImpl = (RedisAsyncConnectionImpl<byte[], byte[]>) getAsyncConnection();
RedisClusterAsyncCommands<byte[], byte[]> 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<byte[], byte[]> getNativeConnection() {
return (subscription != null ? subscription.pubsub : getAsyncConnection());
public RedisClusterAsyncCommands<byte[], byte[]> getNativeConnection() {
return (subscription != null ? subscription.pubsub.async() : getAsyncConnection());
}
public boolean isQueueing() {
@@ -467,47 +461,56 @@ public class LettuceConnection extends AbstractRedisConnection {
public List<Object> closePipeline() {
if (isPipelined) {
isPipelined = false;
List<Command<?, ?, ?>> futures = new ArrayList<Command<?, ?, ?>>();
List<com.lambdaworks.redis.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<com.lambdaworks.redis.protocol.RedisCommand<?, ?, ?>>();
for (LettuceResult result : ppline) {
futures.add(result.getResultHolder());
}
boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS,
futures.toArray(new Command[futures.size()]));
List<Object> results = new ArrayList<Object>(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<Object> results = new ArrayList<Object>(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<FutureResult<?>>(txResults), LettuceConverters.exceptionConverter())));
pipeline(new LettuceResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).exec(),
new LettuceTransactionResultConverter(new LinkedList<FutureResult<?>>(txResults),
LettuceConverters.exceptionConverter())));
return null;
}
List<Object> results = getDedicatedConnection().exec();
List<Object> 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<byte[]> 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<byte[]> keyScanCursor = getConnection().scan(scanCursor, scanArgs);
String nextCursorId = keyScanCursor.getCursor();
@@ -3889,10 +3901,11 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
private RedisPubSubConnection<byte[], byte[]> switchToPubSub() {
private StatefulRedisPubSubConnection<byte[], byte[]> 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<byte[], byte[]> getAsyncConnection() {
private RedisClusterAsyncCommands<byte[], byte[]> getAsyncConnection() {
if (isQueueing()) {
return getAsyncDedicatedConnection();
}
if (asyncSharedConn != null) {
return asyncSharedConn;
if (asyncSharedConn instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) asyncSharedConn).async();
}
}
return getAsyncDedicatedConnection();
}
protected com.lambdaworks.redis.RedisConnection<byte[], byte[]> getConnection() {
protected RedisClusterCommands<byte[], byte[]> getConnection() {
if (isQueueing()) {
return getDedicatedConnection();
}
if (sharedConn != null) {
return sharedConn;
if (asyncSharedConn != null) {
if (asyncSharedConn instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) asyncSharedConn).sync();
}
if (asyncSharedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncSharedConn).sync();
}
}
return getDedicatedConnection();
}
protected RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
protected RedisClusterAsyncCommands<byte[], byte[]> getAsyncDedicatedConnection() {
if (asyncDedicatedConn == null) {
asyncDedicatedConn = doGetAsyncDedicatedConnection();
if (this.pool == null) {
this.asyncDedicatedConn.select(dbIndex);
}
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
}
}
}
return asyncDedicatedConn;
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).async();
}
if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncDedicatedConn).async();
}
throw new IllegalStateException(String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
protected RedisAsyncConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
if (this.pool != null) {
return pool.getResource();
} else {
return ((RedisClient) client).connectAsync(CODEC);
return ((RedisClient) client).connect(CODEC);
}
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getDedicatedConnection() {
if (dedicatedConn == null) {
this.dedicatedConn = syncConnection(getAsyncDedicatedConnection());
}
return dedicatedConn;
}
private RedisClusterCommands<byte[], byte[]> getDedicatedConnection() {
if (asyncDedicatedConn == null) {
asyncDedicatedConn = doGetAsyncDedicatedConnection();
if (this.pool == null) {
if (asyncDedicatedConn instanceof StatefulRedisConnection) {
((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
}
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> syncConnection(
RedisAsyncConnection<byte[], byte[]> asyncConnection) {
try {
return (com.lambdaworks.redis.RedisConnection<byte[], byte[]>) 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<byte[], byte[]>) asyncDedicatedConn).sync();
}
if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncDedicatedConn).sync();
}
throw new IllegalStateException(String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
private Future<Long> 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<String, String> connection = null;
StatefulRedisConnection<String, String> 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<String, String> connection = ((RedisClient) client)
.connectSentinelAsync(getRedisURI(sentinel));
StatefulRedisSentinelConnection<String, String> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> 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<byte[], byte[]> asyncConnection = getAsyncConnection();
transaction(new LettuceResult(asyncConnection.pfmerge(destinationKey, sourceKeys)));
return;
}
RedisHLLCommands<byte[], byte[]> connection = getConnection();
connection.pfmerge(destinationKey, sourceKeys);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}

29
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java
View File

@@ -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<byte[], byte[]> connection;
private StatefulRedisConnection<byte[], byte[]> 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<String> 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<byte[], byte[]> getSharedConnection() {
protected StatefulRedisConnection<byte[], byte[]> getSharedConnection() {
if (shareNativeConnection) {
synchronized (this.connectionMonitor) {
if (this.connection == null) {
@@ -527,18 +523,17 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
}
}
protected RedisAsyncConnection<byte[], byte[]> createLettuceConnector() {
protected StatefulRedisConnection<byte[], byte[]> createLettuceConnector() {
try {
RedisAsyncConnection connection = null;
StatefulRedisConnection<byte[], byte[]> 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<byte[], byte[]>) ((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());
}
}

26
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java
View File

@@ -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<byte[], String> BYTES_TO_STRING;
private static final Converter<String, byte[]> STRING_TO_BYTES;
private static final Converter<Set<byte[]>, List<byte[]>> BYTES_SET_TO_BYTES_LIST;
private static final Converter<Collection<byte[]>, List<byte[]>> BYTES_COLLECTION_TO_BYTES_LIST;
private static final Converter<KeyValue<byte[], byte[]>, List<byte[]>> KEY_VALUE_TO_BYTES_LIST;
private static final Converter<List<ScoredValue<byte[]>>, Set<Tuple>> SCORED_VALUES_TO_TUPLE_SET;
private static final Converter<List<ScoredValue<byte[]>>, List<Tuple>> SCORED_VALUES_TO_TUPLE_LIST;
@@ -151,6 +153,14 @@ abstract public class LettuceConverters extends Converters {
return results != null ? new ArrayList<byte[]>(results) : null;
}
};
BYTES_COLLECTION_TO_BYTES_LIST = new Converter<Collection<byte[]>, List<byte[]>>() {
public List<byte[]> convert(Collection<byte[]> results) {
if (results instanceof List) {
return (List<byte[]>) results;
}
return results != null ? new ArrayList<byte[]>(results) : null;
}
};
KEY_VALUE_TO_BYTES_LIST = new Converter<KeyValue<byte[], byte[]>, List<byte[]>>() {
public List<byte[]> convert(KeyValue<byte[], byte[]> source) {
if (source == null) {
@@ -365,8 +375,12 @@ abstract public class LettuceConverters extends Converters {
return KEY_VALUE_TO_BYTES_LIST;
}
public static Converter<Set<byte[]>, List<byte[]>> bytesSetToBytesList() {
return BYTES_SET_TO_BYTES_LIST;
public static Converter<Collection<byte[]>, List<byte[]>> bytesSetToBytesList() {
return BYTES_COLLECTION_TO_BYTES_LIST;
}
public static Converter<Collection<byte[]>, List<byte[]>> bytesCollectionToBytesList() {
return BYTES_COLLECTION_TO_BYTES_LIST;
}
public static Converter<List<ScoredValue<byte[]>>, Set<Tuple>> scoredValuesToTupleSet() {
@@ -405,8 +419,8 @@ abstract public class LettuceConverters extends Converters {
return KEY_VALUE_TO_BYTES_LIST.convert(source);
}
public static List<byte[]> toBytesList(Set<byte[]> source) {
return BYTES_SET_TO_BYTES_LIST.convert(source);
public static List<byte[]> toBytesList(Collection<byte[]> source) {
return BYTES_COLLECTION_TO_BYTES_LIST.convert(source);
}
public static Set<Tuple> toTupleSet(List<ScoredValue<byte[]>> source) {
@@ -769,7 +783,7 @@ abstract public class LettuceConverters extends Converters {
public GeoResults<GeoLocation<byte[]>> convert(Set<byte[]> source) {
if (CollectionUtils.isEmpty(source)) {
return new GeoResults<GeoLocation<byte[]>>(Collections.<GeoResult<GeoLocation<byte[]>>> emptyList());
return new GeoResults<GeoLocation<byte[]>>(Collections.<GeoResult<GeoLocation<byte[]>>>emptyList());
}
List<GeoResult<GeoLocation<byte[]>>> results = new ArrayList<GeoResult<GeoLocation<byte[]>>>(source.size());

14
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java
View File

@@ -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<Exception, DataAcces
public DataAccessException convert(Exception ex) {
if (ex instanceof ExecutionException) {
if (ex instanceof ExecutionException || ex instanceof RedisCommandExecutionException) {
if(ex.getCause() != ex && ex.getCause() instanceof Exception) {
if (ex.getCause() != ex && ex.getCause() instanceof Exception) {
return convert((Exception) ex.getCause());
}
return new RedisSystemException("Error in execution", ex);

6
src/main/java/org/springframework/data/redis/connection/lettuce/LettucePool.java
View File

@@ -18,15 +18,15 @@ package org.springframework.data.redis.connection.lettuce;
import org.springframework.data.redis.connection.Pool;
import com.lambdaworks.redis.RedisAsyncConnection;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.api.StatefulConnection;
/**
* Pool of Lettuce {@link RedisAsyncConnection}s
* Pool of Lettuce {@link StatefulConnection}s
*
* @author Jennifer Hickey
*/
public interface LettucePool extends Pool<RedisAsyncConnection<byte[], byte[]>> {
public interface LettucePool extends Pool<StatefulConnection<byte[], byte[]>> {
/**
* @return The {@link RedisClient} used to create pooled connections

22
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java
View File

@@ -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<String, String> connection;
private StatefulRedisSentinelConnection<String, String> 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<String, String> connection) {
protected LettuceSentinelConnection(StatefulRedisSentinelConnection<String, String> 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<RedisServer> 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<String, String> connectSentinel() {
return redisClient.connectSentinelAsync();
private StatefulRedisSentinelConnection<String, String> connectSentinel() {
return redisClient.connectSentinel();
}
@Override
public boolean isOpen() {
return connection != null && connection.isOpen();
}
}

18
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java
View File

@@ -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<byte[], byte[]> pubsub;
final StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
private LettuceMessageListener listener;
LettuceSubscription(MessageListener listener, RedisPubSubConnection<byte[], byte[]> pubsubConnection) {
LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection<byte[], byte[]> 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);
}
}

209
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java
View File

@@ -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<byte[], byte[]> 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<Tuple> convertTuple(List<ScoredValue<byte[]>> zrange) {
if (zrange == null) {
return null;
}
Set<Tuple> tuples = new LinkedHashSet<Tuple>(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<byte[]> toList(KeyValue<byte[], byte[]> blpop) {
if (blpop == null) {
return null;
}
List<byte[]> list = new ArrayList<byte[]>(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];
}
}

7
src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
View File

@@ -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();

21
src/test/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClientTests.java
View File

@@ -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<String, String> conn = client.connect();
conn.ping();
StatefulRedisConnection<String, String> conn = client.connect();
conn.sync().ping();
conn.close();
}
@@ -68,8 +69,8 @@ public class AuthenticatingRedisClientTests {
@Test
public void codecConnect() {
RedisConnection<byte[], byte[]> conn = client.connect(LettuceConnection.CODEC);
conn.ping();
StatefulRedisConnection<byte[], byte[]> conn = client.connect(LettuceConnection.CODEC);
conn.sync().ping();
conn.close();
}
@@ -89,15 +90,15 @@ public class AuthenticatingRedisClientTests {
@Test
public void connectPubSub() {
RedisPubSubConnection<String, String> conn = client.connectPubSub();
conn.ping();
StatefulRedisPubSubConnection<String, String> conn = client.connectPubSub();
conn.sync().ping();
conn.close();
}
@Test
public void codecConnectPubSub() {
RedisPubSubConnection<byte[], byte[]> conn = client.connectPubSub(LettuceConnection.CODEC);
conn.ping();
StatefulRedisPubSubConnection<byte[], byte[]> conn = client.connectPubSub(LettuceConnection.CODEC);
conn.sync().ping();
conn.close();
}

24
src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java
View File

@@ -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<byte[], byte[]> client = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) 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<byte[], byte[]> client = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) 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<byte[], byte[]> client = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) 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<byte[], byte[]> client = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) 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<byte[], byte[]> client = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) pool.getResource();
assertNotNull(client);
pool.returnBrokenResource(client);
RedisAsyncConnection<byte[], byte[]> client2 = pool.getResource();
StatefulRedisConnection<byte[], byte[]> client2 = (StatefulRedisConnection<byte[], byte[]>) 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<byte[], byte[]> conn = pool.getResource();
conn.ping();
conn.close();
StatefulRedisConnection<byte[], byte[]> client = (StatefulRedisConnection<byte[], byte[]>) pool.getResource();
client.sync().ping();
client.sync().getStatefulConnection().close();
}
@Ignore("Redis must have requirepass set to run this test")

32
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
View File

@@ -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<String, String> nativeConnection;
RedisAdvancedClusterCommands<String, String> 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<String, String>) 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<String, String>) 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<String, String>) nativeConnection).pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2);
((RedisHLLCommands<String, String>) 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<String, String>) nativeConnection).pfadd(KEY_1, VALUE_1, VALUE_2);
((RedisHLLCommands<String, String>) 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<String, String>) nativeConnection).pfadd(SAME_SLOT_KEY_1, VALUE_1, VALUE_2);
((RedisHLLCommands<String, String>) 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<String, String>) 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<String, String>) 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());

35
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java
View File

@@ -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<byte[], byte[]> dedicatedConnectionMock;
@Mock RedisConnection<byte[], byte[]> clusterConnection1Mock;
@Mock RedisConnection<byte[], byte[]> clusterConnection2Mock;
@Mock RedisConnection<byte[], byte[]> clusterConnection3Mock;
@Mock RedisClusterAsyncCommands<byte[], byte[]> dedicatedConnectionMock;
@Mock RedisClusterCommands<byte[], byte[]> clusterConnection1Mock;
@Mock RedisClusterCommands<byte[], byte[]> clusterConnection2Mock;
@Mock RedisClusterCommands<byte[], byte[]> clusterConnection3Mock;
LettuceClusterConnection connection;
@@ -112,7 +111,7 @@ public class LettuceClusterConnectionUnitTests {
connection = new LettuceClusterConnection(clusterMock, executor) {
@Override
protected RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
protected RedisClusterAsyncCommands<byte[], byte[]> 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.<byte[]> emptyList());
when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]> emptyList());
when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]> emptyList());
when(clusterConnection1Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]>emptyList());
when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]>emptyList());
when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]>emptyList());
byte[] pattern = LettuceConverters.toBytes("*");
@@ -229,7 +228,7 @@ public class LettuceClusterConnectionUnitTests {
@Test
public void keysShouldOnlyBeRunOnDedicatedNodeWhenPinned() {
when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]> emptyList());
when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]>emptyList());
byte[] pattern = LettuceConverters.toBytes("*");

18
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java
View File

@@ -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<byte[], byte[]>) conn2.getNativeConnection()).ping();
((RedisAsyncCommands<byte[], byte[]>) 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<byte[], byte[]> nativeConn = (RedisAsyncConnection<byte[], byte[]>) connection
RedisAsyncCommands<byte[], byte[]> nativeConn = (RedisAsyncCommands<byte[], byte[]>) connection
.getNativeConnection();
factory.resetConnection();
assertNotSame(nativeConn, factory.getConnection().getNativeConnection());
nativeConn.close();
nativeConn.getStatefulConnection().close();
}
@SuppressWarnings("unchecked")
@Test
public void testInitConnection() {
RedisAsyncConnection<byte[], byte[]> nativeConn = (RedisAsyncConnection<byte[], byte[]>) connection
RedisAsyncCommands<byte[], byte[]> nativeConn = (RedisAsyncCommands<byte[], byte[]>) connection
.getNativeConnection();
factory.initConnection();
RedisConnection newConnection = factory.getConnection();
@@ -190,7 +190,7 @@ public class LettuceConnectionFactoryTests {
@SuppressWarnings("unchecked")
@Test
public void testResetAndInitConnection() {
RedisAsyncConnection<byte[], byte[]> nativeConn = (RedisAsyncConnection<byte[], byte[]>) connection
RedisAsyncCommands<byte[], byte[]> nativeConn = (RedisAsyncCommands<byte[], byte[]>) connection
.getNativeConnection();
factory.resetConnection();
factory.initConnection();

8
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java
View File

@@ -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());
}
}

81
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java
View File

@@ -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();
}
}

113
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java
View File

@@ -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<RedisAsyncConnectionImpl> {
public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase<RedisAsyncCommands> {
protected LettuceConnection connection;
private RedisClient clientMock;
protected StatefulRedisConnection<byte[], byte[]> statefulConnectionMock;
protected RedisAsyncCommands<byte[], byte[]> 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();
}
}
}

8
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConvertersUnitTests.java
View File

@@ -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.<RedisClientInfo> emptyList()));
assertThat(LettuceConverters.toListOfRedisClientInformation(""), equalTo(Collections.<RedisClientInfo>emptyList()));
}
/**
@@ -62,7 +62,7 @@ public class LettuceConvertersUnitTests {
@Test
public void convertingNullToListOfRedisClientInfoShouldReturnEmptyList() {
assertThat(LettuceConverters.toListOfRedisClientInformation(null),
equalTo(Collections.<RedisClientInfo> emptyList()));
equalTo(Collections.<RedisClientInfo>emptyList()));
}
/**
@@ -100,7 +100,7 @@ public class LettuceConvertersUnitTests {
partition.setConnected(true);
partition.setFlags(new HashSet<NodeFlag>(Arrays.asList(NodeFlag.MASTER, NodeFlag.MYSELF)));
partition.setUri(RedisURI.create("redis://" + CLUSTER_HOST + ":" + MASTER_NODE_1_PORT));
partition.setSlots(Arrays.<Integer> asList(1, 2, 3, 4, 5));
partition.setSlots(Arrays.<Integer>asList(1, 2, 3, 4, 5));
partitions.addPartition(partition);

31
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java
View File

@@ -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<String, String> connectionMock;
private @Mock StatefulRedisSentinelConnection<String, String> connectionMock;
private @Mock RedisSentinelCommands<String, String> sentinelCommandsMock;
private @Mock RedisFuture<List<Map<String, String>>> 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.<Map<String, String>>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.<Map<String, String>>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.<Map<String, String>>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));
}
}

20
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java
View File

@@ -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();
}
}

130
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java
View File

@@ -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<byte[], byte[]> pubsub;
StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
private MessageListener listener;
private RedisPubSubCommands<byte[], byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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<byte[]> 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]);
}
}

4
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientResources.java
View File

@@ -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();
}