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

Make SRP return converted tx results

Browse Source 
DATAREDIS-208
This commit is contained in:
Jennifer Hickey
2013-07-23 14:37:03 -07:00
parent 08d502083c
commit aa8d10b150
6 changed files with 44 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java
View File

@@ -78,7 +78,7 @@ public class SrpConnection implements RedisConnection {
private PipelineTracker callback;
private PipelineTracker txTracker;
private volatile SrpSubscription subscription;
private boolean convertPipelineResults=true;
private boolean convertPipelineAndTxResults=true;
@SuppressWarnings("rawtypes")
@@ -2180,13 +2180,13 @@ public class SrpConnection implements RedisConnection {
/**
* Specifies if pipelined results should be converted to the expected data
* type. If false, results of {@link #closePipeline()} will be of the
* type. If false, results of {@link #closePipeline()} and {@link #exec()} will be of the
* type returned by the Lettuce driver
*
* @param convertPipelineResults Whether or not to convert pipeline results
*/
public void setConvertPipelineResults(boolean convertPipelineResults) {
this.convertPipelineResults = convertPipelineResults;
public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) {
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
private void checkSubscription() {
@@ -2207,14 +2207,14 @@ public class SrpConnection implements RedisConnection {
private void initPipeline() {
if (pipeline == null) {
callback = new PipelineTracker(convertPipelineResults);
callback = new PipelineTracker(convertPipelineAndTxResults);
pipeline = client.pipeline();
}
}
private void initTxTracker() {
if(txTracker == null) {
txTracker = new PipelineTracker(false);
txTracker = new PipelineTracker(convertPipelineAndTxResults);
}
initPipeline();
}

24
src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java
View File

@@ -35,7 +35,7 @@ public class SrpConnectionFactory implements InitializingBean, DisposableBean, R
private String hostName = "localhost";
private int port = 6379;
private BlockingQueue<SrpConnection> trackedConnections = new ArrayBlockingQueue<SrpConnection>(50);
private boolean convertPipelineResults = true;
private boolean convertPipelineAndTxResults = true;
/**
@@ -73,7 +73,7 @@ public class SrpConnectionFactory implements InitializingBean, DisposableBean, R
public RedisConnection getConnection() {
SrpConnection connection = new SrpConnection(hostName, port, trackedConnections);
connection.setConvertPipelineResults(convertPipelineResults);
connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults);
return connection;
}
@@ -119,23 +119,23 @@ public class SrpConnectionFactory implements InitializingBean, DisposableBean, R
/**
* Specifies if pipelined results should be converted to the expected data
* type. If false, results of {@link #closePipeline()} will be of the
* type returned by the Jedis driver
* type. If false, results of {@link SrpConnection#closePipeline()} and {@link SrpConnection#exec()}
* will be of the type returned by the SRP driver
*
* @return Whether or not to convert pipeline results
* @return Whether or not to convert pipeline and tx results
*/
public boolean getConvertPipelineResults() {
return convertPipelineResults;
public boolean getConvertPipelineAndTxResults() {
return convertPipelineAndTxResults;
}
/**
* Specifies if pipelined results should be converted to the expected data
* type. If false, results of {@link #closePipeline()} will be of the
* type returned by the Jedis driver
* type. If false, results of {@link SrpConnection#closePipeline()} and {@link SrpConnection#exec()}
* will be of the type returned by the SRP driver
*
* @param convertPipelineResults Whether or not to convert pipeline results
* @param convertPipelineResults Whether or not to convert pipeline and tx results
*/
public void setConvertPipelineResults(boolean convertPipelineResults) {
this.convertPipelineResults = convertPipelineResults;
public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) {
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
}

9
src/test/java/org/springframework/data/redis/connection/ConvertingAbstractConnectionTransactionIntegrationTests.java
View File

@@ -7,6 +7,7 @@ import java.util.List;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.test.annotation.IfProfileValue;
public class ConvertingAbstractConnectionTransactionIntegrationTests extends
AbstractConnectionIntegrationTests {
@@ -96,6 +97,14 @@ public class ConvertingAbstractConnectionTransactionIntegrationTests extends
connection.watch("foo".getBytes());
}
@Test(expected = UnsupportedOperationException.class)
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testScriptKill() {
// Impossible to call script kill in a tx because you can't issue the
// exec command while Redis is running a script
connection.scriptKill();
}
protected void initConnection() {
connection.multi();
}

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

@@ -61,13 +61,6 @@ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegration
super.testZUnionStoreAggWeights();
}
@Test
@IfProfileValue(name = "redisVersion", values = {"2.4", "2.6"})
public void testGetRangeSetRange() {
// SRP throws an Exception if you try to execute on version lower than 2.4
super.testGetRangeSetRange();
}
@Test
public void testExecuteNoArgs() {
// SRP returns this as String while other drivers return as byte[]

20
src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineTxIntegrationTests.java
View File

@@ -34,18 +34,16 @@ import org.springframework.test.annotation.IfProfileValue;
*/
public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransactionIntegrationTests {
@Test
public void testUsePipelineAfterTxExec() {
connection.set("foo", "bar");
assertNull(connection.exec());
assertNull(connection.get("foo"));
List<Object> results = connection.closePipeline();
assertEquals(Arrays.asList(new Object[] {Collections.singletonList("OK"), "bar"}), results);
assertEquals(Arrays.asList(new Object[] {Collections.emptyList(), "bar"}), results);
assertEquals("bar", connection.get("foo"));
}
@SuppressWarnings("unchecked")
@Test
public void testExec2TransactionsInPipeline() {
connection.set("foo", "bar");
@@ -57,9 +55,8 @@ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransa
assertNull(connection.exec());
List<Object> results = connection.closePipeline();
assertEquals(2, results.size());
// First element of each list is "OK"
assertEquals("bar", new String((byte[])((List<Object>)results.get(0)).get(1)));
assertEquals("baz", new String((byte[])((List<Object>)results.get(1)).get(1)));
assertEquals(Arrays.asList(new Object[] {"bar"}), results.get(0));
assertEquals(Arrays.asList(new Object[] {"baz"}), results.get(1));
}
@Test(expected=RedisPipelineException.class)
@@ -99,17 +96,6 @@ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransa
@SuppressWarnings("unchecked")
protected List<Object> getResults() {
assertNull(connection.exec());
List<Object> pipelined = connection.closePipeline();
// We expect only the results of exec to be in the closed pipeline
assertEquals(1,pipelined.size());
List<Object> txResults = (List<Object>)pipelined.get(0);
// Return exec results and this test should behave exactly like its superclass
return convertResults(txResults);
}
@SuppressWarnings("unchecked")
protected List<Object> getResultsNoConversion() {
assertNull(connection.exec());
List<Object> pipelined = connection.closePipeline();
// We expect only the results of exec to be in the closed pipeline

388
src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java
View File

@@ -15,33 +15,16 @@
*/
package org.springframework.data.redis.connection.srp;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests;
import org.springframework.data.redis.connection.DefaultStringTuple;
import org.springframework.data.redis.connection.ConvertingAbstractConnectionTransactionIntegrationTests;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.connection.StringRedisConnection.StringTuple;
import org.springframework.data.redis.serializer.SerializationUtils;
import org.springframework.test.annotation.IfProfileValue;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import redis.reply.IntegerReply;
import redis.reply.Reply;
import redis.reply.StatusReply;
/**
@@ -52,374 +35,31 @@ import redis.reply.StatusReply;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("SrpConnectionIntegrationTests-context.xml")
public class SrpConnectionTransactionIntegrationTests extends AbstractConnectionTransactionIntegrationTests {
protected boolean convertResultToSet=false;
protected boolean convertResultToStringTuples=false;
protected boolean convertBooleanToLong=false;
protected boolean convertResultsToMap=false;
public class SrpConnectionTransactionIntegrationTests extends ConvertingAbstractConnectionTransactionIntegrationTests {
@Test
public void testDel() {
connection.set("testing","123");
actual.add(connection.del("testing"));
actual.add(connection.exists("testing"));
verifyResults(Arrays.asList(new Object[] { true, false }));
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalReturnArrayOKs() {
// SRP returns the Strings from individual StatusReplys in a MultiBulkReply, while other clients return as byte[]
actual.add(connection.eval("return { redis.call('set','abc','ghk'), redis.call('set','abc','lfdf')}",
ReturnType.MULTI, 0));
verifyResults(Arrays.asList(new Object[] {Arrays.asList(new Object[] {"OK", "OK"} )}));
}
@Test
public void testSAdd() {
convertResultToSet = true;
super.testSAdd();
public void testExecuteNoArgs() {
// SRP returns this as String while other drivers return as byte[]
actual.add(connection.execute("PING"));
verifyResults(Arrays.asList(new Object[] { "PONG" }));
}
@Test
public void testSDiff() {
convertResultToSet = true;
super.testSDiff();
}
@Test
public void testSInter() {
convertResultToSet = true;
super.testSInter();
}
@Test
public void testSRem() {
convertResultToSet = true;
super.testSRem();
}
@Test
public void testSUnion() {
convertResultToSet = true;
super.testSUnion();
}
@Test
public void testSUnionStore() {
convertResultToSet = true;
super.testSUnionStore();
}
@Test
public void testSDiffStore() {
convertResultToSet = true;
super.testSDiffStore();
}
@Test
public void testSInterStore() {
convertResultToSet = true;
super.testSInterStore();
}
@Test
public void testZRemRangeByScore() {
convertResultToSet = true;
super.testZRemRangeByScore();
}
@Test
public void testZAddAndZRange() {
convertResultToSet = true;
super.testZAddAndZRange();
}
@Test
public void testZIncrBy() {
convertResultToSet = true;
actual.add(connection.zAdd("myset", 2, "Bob"));
actual.add(connection.zAdd("myset", 1, "James"));
actual.add(connection.zAdd("myset", 4, "Joe"));
actual.add(connection.zIncrBy("myset", 2, "Joe"));
actual.add(connection.zRangeByScore("myset", 6, 6));
verifyResults(
Arrays.asList(new Object[] { true, true, true, "6",
new LinkedHashSet<String>(Collections.singletonList("Joe")) }));
}
@Test
public void testZInterStore() {
convertResultToSet = true;
super.testZInterStore();
}
@Test(expected=UnsupportedOperationException.class)
@Test(expected = UnsupportedOperationException.class)
public void testZInterStoreAggWeights() {
super.testZInterStoreAggWeights();
}
@Test(expected=UnsupportedOperationException.class)
@Test(expected = UnsupportedOperationException.class)
public void testZUnionStoreAggWeights() {
super.testZUnionStoreAggWeights();
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testHIncrByDouble() {
actual.add(connection.hSet("test", "key", "2.9"));
actual.add(connection.hIncrBy("test", "key", 3.5));
actual.add(connection.hGet("test", "key"));
// SRP returns byte[] from hIncrBy
verifyResults(Arrays.asList(new Object[] { true, "6.4", "6.4" }));
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testIncrByDouble() {
connection.set("tdb", "4.5");
actual.add(connection.incrBy("tdb", 7.2));
actual.add(connection.get("tdb"));
// SRP returns byte[] fro incrBy
verifyResults(Arrays.asList(new Object[] { "11.7", "11.7" }));
}
@Test
public void testBitSet() throws Exception {
convertLongToBoolean = false;
String key = "bitset-test";
connection.setBit(key, 0, false);
connection.setBit(key, 1, true);
actual.add(connection.getBit(key, 0));
actual.add(connection.getBit(key, 1));
// Lettuce setBit returns Long instead of void
verifyResults(Arrays.asList(new Object[] { 0l, 0l, 0l, 1l }));
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testBitCount() {
convertBooleanToLong = true;
super.testBitCount();
}
@Test
public void testSortStoreNullParams() {
convertBooleanToLong = true;
super.testSortNullParams();
}
@Test
public void testZRangeWithScores() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRangeWithScores();
}
@Test
public void testZRangeByScore() {
convertResultToSet = true;
super.testZRangeByScore();
}
@Test
public void testZRangeByScoreOffsetCount() {
convertResultToSet = true;
super.testZRangeByScoreOffsetCount();
}
@Test
public void testZRangeByScoreWithScores() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRangeByScoreWithScores();
}
@Test
public void testZRangeByScoreWithScoresOffsetCount() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRangeByScoreWithScoresOffsetCount();
}
@Test
public void testZRevRange() {
convertResultToSet = true;
super.testZRevRange();
}
@Test
public void testZRevRangeWithScores() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRevRangeWithScores();
}
@Test
public void testZRevRangeByScoreWithScores() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRevRangeByScoreWithScores();
}
@Test
public void testZRevRangeByScoreOffsetCount() {
convertResultToSet = true;
super.testZRevRangeByScoreOffsetCount();
}
@Test
public void testZRevRangeByScore() {
convertResultToSet = true;
super.testZRevRangeByScore();
}
@Test
public void testZRevRangeByScoreWithScoresOffsetCount() {
convertResultToStringTuples = true;
convertResultToSet = true;
super.testZRevRangeByScoreWithScoresOffsetCount();
}
@Test
public void testZRem() {
convertResultToSet = true;
super.testZRem();
}
@Test
public void testZRemRangeByRank() {
convertResultToSet = true;
super.testZRemRangeByRank();
}
@Test
public void testZScore() {
actual.add(connection.zAdd("myset", 2, "Bob"));
actual.add(connection.zAdd("myset", 1, "James"));
actual.add(connection.zAdd("myset", 3, "Joe"));
actual.add(connection.zScore("myset", "Joe"));
verifyResults(Arrays.asList(new Object[] { true, true, true, "3" }));
}
@Test
public void testZUnionStore() {
convertResultToSet = true;
super.testZUnionStore();
}
@Test
public void testHSetGet() throws Exception {
convertResultsToMap = true;
super.testHSetGet();
}
@Test
public void testHKeys() {
convertResultToSet = true;
super.testHKeys();
}
@SuppressWarnings("unchecked")
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalShaArrayStrings() {
getResults();
String sha1 = connection.scriptLoad("return {KEYS[1],ARGV[1]}");
initConnection();
actual.add(connection.evalSha(sha1, ReturnType.MULTI, 1, "key1", "arg1"));
List<Object> results = getResults();
List<String> scriptResults = (List<String>) results.get(0);
assertEquals(Arrays.asList(new Object[] { "key1", "arg1" }),
scriptResults);
}
@SuppressWarnings("unchecked")
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalReturnArrayStrings() {
actual.add(connection.eval("return {KEYS[1],ARGV[1]}", ReturnType.MULTI, 1, "foo", "bar"));
List<String> result = (List<String>) getResults().get(0);
assertEquals(Arrays.asList(new Object[] { "foo", "bar" }), result);
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalReturnFalse() {
actual.add(connection.eval("return false", ReturnType.BOOLEAN, 0));
verifyResults(Arrays.asList(new Object[] { null }));
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalReturnArrayNumbers() {
convertLongToBoolean = false;
super.testEvalReturnArrayNumbers();
}
@Test
@IfProfileValue(name = "redisVersion", value = "2.6")
public void testEvalReturnArrayTrues() {
convertLongToBoolean = false;
super.testEvalReturnArrayTrues();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
protected Object convertResult(Object result) {
Object convertedResult = super.convertResult(result);
if (convertedResult instanceof Set && !(((Set) convertedResult).isEmpty())) {
Object firstResult = ((Set) convertedResult).iterator().next();
if(firstResult instanceof Tuple) {
Set<StringTuple> tuples = new LinkedHashSet<StringTuple>();
for (Tuple value : ((Set<Tuple>) convertedResult)) {
DefaultStringTuple tuple = new DefaultStringTuple(value,stringSerializer.deserialize(value.getValue()));
tuples.add(tuple);
}
return tuples;
}
}else if (convertedResult instanceof Reply[]) {
if (convertResultToStringTuples) {
Set<Tuple> tuples = SrpConverters.toTupleSet((Reply[])convertedResult);
Collection<StringTuple> stringTuples;
if(convertResultToSet) {
stringTuples = new LinkedHashSet<StringTuple>();
}else {
stringTuples = new ArrayList<StringTuple>();
}
for (Tuple tuple : tuples) {
stringTuples.add(new DefaultStringTuple(tuple, new String(tuple.getValue())));
}
return stringTuples;
} else if (convertResultToSet) {
return SerializationUtils.deserialize(SrpConverters.toBytesSet((Reply[])convertedResult),
stringSerializer);
} else if(((Reply[]) convertedResult).length > 0 && ((Reply[])convertedResult)[0] instanceof IntegerReply) {
if(convertLongToBoolean) {
return SrpConverters.toBooleanList((Reply[])convertedResult);
}
List<Long> results = new ArrayList<Long>();
for(Reply reply: (Reply[])convertedResult) {
results.add(((IntegerReply)reply).data());
}
return results;
} else if(((Reply[]) convertedResult).length > 0 && ((Reply[])convertedResult)[0] instanceof StatusReply) {
List<String> statuses = new ArrayList<String>();
for(Reply reply: (Reply[])convertedResult) {
statuses.add(((StatusReply)reply).data());
}
return statuses;
} else if(convertResultsToMap) {
return (SerializationUtils.deserialize(SrpConverters.toBytesMap((Reply[])convertedResult), stringSerializer));
} else {
return SerializationUtils.deserialize(
SrpConverters.toBytesList((Reply[]) convertedResult), stringSerializer);
}
} else if(convertBooleanToLong && result instanceof Boolean) {
if((Boolean)result) {
return 1l;
}
return 0l;
}
return convertedResult;
}
@Override
protected DataAccessException convertException(Exception ex) {
return SrpConverters.toDataAccessException(ex);
}
}