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

Upgrade to SRP 0.5

Browse Source 
DATAREDIS-132, DATAREDIS-130, DATAREDIS-152

- Upgrade to fix broken config_get in SRP 0.2

- Fix syntax errors in sort methods

- Fix syntax errors in zRange/zRevRange methods 
with offset and count
This commit is contained in:
Jennifer Hickey
2013-04-09 16:39:17 -07:00
committed by pivotal
parent aaa28495be
commit 7d485d6463
8 changed files with 248 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
gradle.properties
View File

@@ -16,7 +16,7 @@ mockitoVersion = 1.8.5
jedisVersion = 2.1.0
jredisVersion = 03122010
rjcVersion = 0.6.4
srpVersion = 0.2
srpVersion = 0.5
lettuceVersion = 2.2.0
# --------------------

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

@@ -56,6 +56,8 @@ import com.google.common.util.concurrent.ListenableFuture;
*/
public class SrpConnection implements RedisConnection {
private static final Object[] EMPTY_PARAMS_ARRAY = new Object[0];
private final RedisClient client;
private final BlockingQueue<SrpConnection> queue;
@@ -184,14 +186,14 @@ public class SrpConnection implements RedisConnection {
public List<byte[]> sort(byte[] key, SortParameters params) {
byte[] sort = SrpUtils.sort(params);
Object[] sort = SrpUtils.sortParams(params);
try {
if (isPipelined()) {
pipeline(pipeline.sort(key, sort, null, (Object[]) null));
pipeline(pipeline.sort(key, sort));
return null;
}
return SrpUtils.toBytesList((Reply[]) client.sort(key, sort, null, (Object[]) null).data());
return SrpUtils.toBytesList((Reply[]) client.sort(key, sort).data());
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -199,14 +201,14 @@ public class SrpConnection implements RedisConnection {
public Long sort(byte[] key, SortParameters params, byte[] sortKey) {
byte[] sort = SrpUtils.sort(params, sortKey);
Object[] sort = SrpUtils.sortParams(params, sortKey);
try {
if (isPipelined()) {
pipeline(pipeline.sort(key, sort, null, (Object[]) null));
pipeline(pipeline.sort(key, sort));
return null;
}
return ((Long) client.sort(key, sort, null, (Object[]) null).data());
return ((Long) client.sort(key, sort).data());
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -307,10 +309,10 @@ public class SrpConnection implements RedisConnection {
public Properties info() {
try {
if (isPipelined()) {
pipeline(pipeline.info());
pipeline(pipeline.info(null));
return null;
}
return SrpUtils.info(client.info());
return SrpUtils.info(client.info(null));
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1245,10 +1247,10 @@ public class SrpConnection implements RedisConnection {
public byte[] sRandMember(byte[] key) {
try {
if (isPipelined()) {
pipeline(pipeline.srandmember(key));
pipeline(pipeline.srandmember(key, null));
return null;
}
return client.srandmember(key).data();
return (byte[])client.srandmember(key, null).data();
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1356,22 +1358,12 @@ public class SrpConnection implements RedisConnection {
public Long zInterStore(byte[] destKey, byte[]... sets) {
Object[] args = new Object[2 + sets.length];
args[0] = destKey;
args[1] = sets.length;
int i = 2;
for (byte[] set : sets) {
args[i++] = set;
}
try {
if (isPipelined()) {
pipeline(pipeline.zinterstore(args));
pipeline(pipeline.zinterstore(destKey, sets.length, sets));
return null;
}
return client.zinterstore(args).data();
return client.zinterstore(destKey, sets.length, sets).data();
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1406,10 +1398,10 @@ public class SrpConnection implements RedisConnection {
public Set<byte[]> zRangeByScore(byte[] key, double min, double max) {
try {
if (isPipelined()) {
pipeline(pipeline.zrangebyscore(key, min, max, null, null));
pipeline(pipeline.zrangebyscore(key, min, max, null, EMPTY_PARAMS_ARRAY));
return null;
}
return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, null).data());
return SrpUtils.toSet(client.zrangebyscore(key, min, max, null, EMPTY_PARAMS_ARRAY).data());
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1419,10 +1411,10 @@ public class SrpConnection implements RedisConnection {
public Set<Tuple> zRangeByScoreWithScores(byte[] key, double min, double max) {
try {
if (isPipelined()) {
pipeline(pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null));
pipeline(pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, EMPTY_PARAMS_ARRAY));
return null;
}
return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, null));
return SrpUtils.convertTuple(client.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, EMPTY_PARAMS_ARRAY));
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1444,7 +1436,7 @@ public class SrpConnection implements RedisConnection {
public Set<byte[]> zRangeByScore(byte[] key, double min, double max, long offset, long count) {
try {
byte[] limit = SrpUtils.limit(offset, count);
Object[] limit = SrpUtils.limitParams(offset, count);
if (isPipelined()) {
pipeline(pipeline.zrangebyscore(key, min, max, null, limit));
return null;
@@ -1458,7 +1450,7 @@ public class SrpConnection implements RedisConnection {
public Set<Tuple> zRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) {
try {
byte[] limit = SrpUtils.limit(offset, count);
Object[] limit = SrpUtils.limitParams(offset, count);
if (isPipelined()) {
pipeline(pipeline.zrangebyscore(key, min, max, SrpUtils.WITHSCORES, limit));
return null;
@@ -1472,7 +1464,7 @@ public class SrpConnection implements RedisConnection {
public Set<byte[]> zRevRangeByScore(byte[] key, double min, double max, long offset, long count) {
try {
byte[] limit = SrpUtils.limit(offset, count);
Object[] limit = SrpUtils.limitParams(offset, count);
if (isPipelined()) {
pipeline(pipeline.zrevrangebyscore(key, max, min, null, limit));
return null;
@@ -1487,10 +1479,10 @@ public class SrpConnection implements RedisConnection {
public Set<byte[]> zRevRangeByScore(byte[] key, double min, double max) {
try {
if (isPipelined()) {
pipeline(pipeline.zrevrangebyscore(key, max, min, null, null));
pipeline(pipeline.zrevrangebyscore(key, max, min, null, EMPTY_PARAMS_ARRAY));
return null;
}
return SrpUtils.toSet(client.zrevrangebyscore(key, max, min, null, null).data());
return SrpUtils.toSet(client.zrevrangebyscore(key, max, min, null, EMPTY_PARAMS_ARRAY).data());
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}
@@ -1499,7 +1491,7 @@ public class SrpConnection implements RedisConnection {
public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, double min, double max, long offset, long count) {
try {
byte[] limit = SrpUtils.limit(offset, count);
Object[] limit = SrpUtils.limitParams(offset, count);
if (isPipelined()) {
pipeline(pipeline.zrevrangebyscore(key, max, min, SrpUtils.WITHSCORES, limit));
return null;
@@ -1514,10 +1506,10 @@ public class SrpConnection implements RedisConnection {
public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, double min, double max) {
try {
if (isPipelined()) {
pipeline(pipeline.zrevrangebyscore(key, max, min, SrpUtils.WITHSCORES, null));
pipeline(pipeline.zrevrangebyscore(key, max, min, SrpUtils.WITHSCORES, EMPTY_PARAMS_ARRAY));
return null;
}
return SrpUtils.convertTuple(client.zrevrangebyscore(key, max, min, SrpUtils.WITHSCORES, null));
return SrpUtils.convertTuple(client.zrevrangebyscore(key, max, min, SrpUtils.WITHSCORES, EMPTY_PARAMS_ARRAY));
} catch (Exception ex) {
throw convertSrpAccessException(ex);
}

96
src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java
View File

@@ -45,6 +45,7 @@ import com.google.common.base.Charsets;
* Helper class featuring methods for SRedis connection handling, providing support for exception translation.
*
* @author Costin Leau
* @author Jennifer Hickey
*/
abstract class SrpUtils {
@@ -53,11 +54,11 @@ abstract class SrpUtils {
private static final byte[] BEFORE = "BEFORE".getBytes(Charsets.UTF_8);
private static final byte[] AFTER = "AFTER".getBytes(Charsets.UTF_8);
static final byte[] WITHSCORES = "WITHSCORES".getBytes(Charsets.UTF_8);
private static final byte[] SPACE = "".getBytes(Charsets.UTF_8);
private static final byte[] BY = "BY ".getBytes(Charsets.UTF_8);
private static final byte[] GET = "GET ".getBytes(Charsets.UTF_8);
private static final byte[] ALPHA = "ALPHA ".getBytes(Charsets.UTF_8);
private static final byte[] STORE = "STORE ".getBytes(Charsets.UTF_8);
private static final byte[] SPACE = " ".getBytes(Charsets.UTF_8);
private static final byte[] BY = "BY".getBytes(Charsets.UTF_8);
private static final byte[] GET = "GET".getBytes(Charsets.UTF_8);
private static final byte[] ALPHA = "ALPHA".getBytes(Charsets.UTF_8);
private static final byte[] STORE = "STORE".getBytes(Charsets.UTF_8);
static DataAccessException convertSRedisAccessException(RuntimeException ex) {
@@ -178,6 +179,12 @@ abstract class SrpUtils {
return ("LIMIT " + offset + " " + count).getBytes(Charsets.UTF_8);
}
static Object[] limitParams(long offset, long count) {
return new Object[] { "LIMIT".getBytes(Charsets.UTF_8),
String.valueOf(offset).getBytes(Charsets.UTF_8),
String.valueOf(count).getBytes(Charsets.UTF_8)};
}
static byte[] sort(SortParameters params) {
return sort(params, null);
}
@@ -185,45 +192,18 @@ abstract class SrpUtils {
static byte[] sort(SortParameters params, byte[] sortKey) {
List<byte[]> arrays = new ArrayList<byte[]>();
if (params.getByPattern() != null) {
arrays.add(BY);
arrays.add(params.getByPattern());
Object[] sortParams = sortParams(params, sortKey);
for(Object param: sortParams) {
arrays.add((byte[])param);
arrays.add(SPACE);
}
if (params.getLimit() != null) {
arrays.add(limit(params.getLimit().getStart(), params.getLimit().getCount()));
arrays.add(SPACE);
}
if (params.getGetPattern() != null) {
byte[][] pattern = params.getGetPattern();
for (byte[] bs : pattern) {
arrays.add(GET);
arrays.add(bs);
arrays.add(SPACE);
}
}
if (params.getOrder() != null) {
arrays.add(params.getOrder().name().getBytes(Charsets.UTF_8));
arrays.add(SPACE);
}
if (params.isAlphabetic()) {
arrays.add(ALPHA);
}
if (sortKey != null) {
arrays.add(STORE);
arrays.add(sortKey);
}
arrays.remove(arrays.size()-1);
// concatenate array
int size = 0;
for (byte[] bs : arrays) {
size += bs.length;
for (Object bs : arrays) {
size += ((byte[])bs).length;
}
byte[] result = new byte[size];
@@ -235,4 +215,44 @@ abstract class SrpUtils {
return result;
}
static Object[] sortParams(SortParameters params) {
return sortParams(params, null);
}
static Object[] sortParams(SortParameters params, byte[] sortKey) {
List<byte[]> arrays = new ArrayList<byte[]>();
if (params.getByPattern() != null) {
arrays.add(BY);
arrays.add(params.getByPattern());
}
if (params.getLimit() != null) {
arrays.add(limit(params.getLimit().getStart(), params.getLimit().getCount()));
}
if (params.getGetPattern() != null) {
byte[][] pattern = params.getGetPattern();
for (byte[] bs : pattern) {
arrays.add(GET);
arrays.add(bs);
}
}
if (params.getOrder() != null) {
arrays.add(params.getOrder().name().getBytes(Charsets.UTF_8));
}
if (params.isAlphabetic()) {
arrays.add(ALPHA);
}
if (sortKey != null) {
arrays.add(STORE);
arrays.add(sortKey);
}
return arrays.toArray();
}
}

3
src/test/java/org/springframework/data/redis/connection/AbstractConnectionPipelineIntegrationTests.java
View File

@@ -288,8 +288,7 @@ abstract public class AbstractConnectionPipelineIntegrationTests extends
connection.get("testitnow");
connection.exec();
List<Object> convertedResults = convertResults();
assertEquals(Arrays.asList(new Object[] { Arrays.asList(new String[] { "OK",
"somethingelse" }) }), convertedResults);
assertEquals(Arrays.asList(new Object[] {"somethingelse"}), convertedResults);
}
@Test

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

@@ -61,34 +61,6 @@ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegration
public void testWatch() {
}
@Ignore("DATAREDIS-130, sort not working")
public void testSort() {
}
@Ignore("DATAREDIS-130, sort not working")
public void testSortStore() {
}
@Ignore("DATAREDIS-132 config get broken in SRP 0.2")
public void testGetConfig() {
}
@Ignore("DATAREDIS-152 Syntax error on zRangeByScore when using offset and count")
public void testZRangeByScoreOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRangeByScoreWithScores when using offset and count")
public void testZRangeByScoreWithScoresOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRevRangeByScore when using offset and count")
public void testZRevRangeByScoreOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRevRangeByScoreWithScores when using offset and count")
public void testZRevRangeByScoreWithScoresOffsetCount() {
}
@Ignore("DATAREDIS-156 SRP bRPopLPush ClassCastException")
public void testBRPopLPushTimeout() {
}

40
src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java
View File

@@ -54,50 +54,14 @@ import redis.reply.Reply;
public class SrpConnectionPipelineIntegrationTests extends
AbstractConnectionPipelineIntegrationTests {
@Ignore("DATAREDIS-123, exec does not return command results")
public void testMultiExec() throws Exception {
}
@Ignore("DATAREDIS-123, exec does not return command results")
@Ignore("DATAREDIS-169 SRP discard does not clear txReplies, results in inconsistent results on next tx exec")
public void testMultiDiscard() {
}
@Ignore("DATAREDIS-123, exec does not return command results")
@Ignore("DATAREDIS-168 SRP exec throws TransactionFailedException if watched value modified")
public void testWatch() {
}
@Ignore("DATAREDIS-123, exec does not return command results")
public void testUnwatch() {
}
@Ignore("DATAREDIS-130, sort not working")
public void testSort() {
}
@Ignore("DATAREDIS-130, sort not working")
public void testSortStore() {
}
@Ignore("DATAREDIS-132 config get broken in SRP 0.2")
public void testGetConfig() {
}
@Ignore("DATAREDIS-152 Syntax error on zRangeByScore when using offset and count")
public void testZRangeByScoreOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRangeByScoreWithScores when using offset and count")
public void testZRangeByScoreWithScoresOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRevRangeByScore when using offset and count")
public void testZRevRangeByScoreOffsetCount() {
}
@Ignore("DATAREDIS-152 Syntax error on zRevRangeByScoreWithScores when using offset and count")
public void testZRevRangeByScoreWithScoresOffsetCount() {
}
@Test(expected = UnsupportedOperationException.class)
public void testZInterStoreAggWeights() {
super.testZInterStoreAggWeights();

158
src/test/java/org/springframework/data/redis/connection/srp/SrpUtilsTests.java Normal file
View File

@@ -0,0 +1,158 @@
/*
* 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.srp;
import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.data.redis.connection.DefaultSortParameters;
import org.springframework.data.redis.connection.SortParameters;
/**
* Unit test of {@link SrpUtils}
*
* @author Jennifer Hickey
*
*/
public class SrpUtilsTests {
@Test
public void testSortParamsWithAllParams() {
SortParameters sortParams = new DefaultSortParameters().alpha().asc()
.by("weight_*".getBytes()).get("object_*".getBytes())
.limit(0, 5);
Object[] sort = SrpUtils.sortParams(sortParams, "foo".getBytes());
assertArrayEquals(new String[] { "BY", "weight_*", "LIMIT 0 5", "GET",
"object_*", "ASC", "ALPHA", "STORE", "foo" },
convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyBy() {
SortParameters sortParams = new DefaultSortParameters().numeric().by(
"weight_*".getBytes());
Object[] sort = SrpUtils.sortParams(sortParams);
assertArrayEquals(new String[] { "BY", "weight_*" },
convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyLimit() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.limit(0, 5);
Object[] sort = SrpUtils.sortParams(sortParams);
assertArrayEquals(new String[] { "LIMIT 0 5" }, convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyGetPatterns() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.get("foo".getBytes()).get("bar".getBytes());
Object[] sort = SrpUtils.sortParams(sortParams);
assertArrayEquals(new String[] { "GET", "foo", "GET", "bar" },
convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyOrder() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.desc();
Object[] sort = SrpUtils.sortParams(sortParams);
assertArrayEquals(new String[] { "DESC" }, convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyAlpha() {
SortParameters sortParams = new DefaultSortParameters().alpha();
Object[] sort = SrpUtils.sortParams(sortParams);
assertArrayEquals(new String[] { "ALPHA" }, convertByteArrays(sort));
}
@Test
public void testSortParamsOnlyStore() {
SortParameters sortParams = new DefaultSortParameters().numeric();
Object[] sort = SrpUtils.sortParams(sortParams, "storelist".getBytes());
assertArrayEquals(new String[] { "STORE", "storelist" },
convertByteArrays(sort));
}
@Test
public void testSortWithAllParams() {
SortParameters sortParams = new DefaultSortParameters().alpha().asc()
.by("weight_*".getBytes()).get("object_*".getBytes())
.limit(0, 5);
byte[] sort = SrpUtils.sort(sortParams, "foo".getBytes());
assertEquals("BY weight_* LIMIT 0 5 GET object_* ASC ALPHA STORE foo",
new String(sort));
}
@Test
public void testSortOnlyBy() {
SortParameters sortParams = new DefaultSortParameters().numeric().by(
"weight_*".getBytes());
byte[] sort = SrpUtils.sort(sortParams);
assertEquals("BY weight_*", new String(sort));
}
@Test
public void testSortOnlyLimit() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.limit(0, 5);
byte[] sort = SrpUtils.sort(sortParams);
assertEquals("LIMIT 0 5", new String(sort));
}
@Test
public void testSortOnlyGetPatterns() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.get("foo".getBytes()).get("bar".getBytes());
byte[] sort = SrpUtils.sort(sortParams);
assertEquals("GET foo GET bar", new String(sort));
}
@Test
public void testSortOnlyOrder() {
SortParameters sortParams = new DefaultSortParameters().numeric()
.desc();
byte[] sort = SrpUtils.sort(sortParams);
assertEquals("DESC", new String(sort));
}
@Test
public void testSortOnlyAlpha() {
SortParameters sortParams = new DefaultSortParameters().alpha();
byte[] sort = SrpUtils.sort(sortParams);
assertEquals("ALPHA", new String(sort));
}
@Test
public void testSortOnlyStore() {
SortParameters sortParams = new DefaultSortParameters().numeric();
byte[] sort = SrpUtils.sort(sortParams, "storelist".getBytes());
assertEquals("STORE storelist", new String(sort));
}
private String[] convertByteArrays(Object[] bytes) {
List<String> convertedParams = new ArrayList<String>();
for (Object b : bytes) {
convertedParams.add(new String((byte[]) b));
}
return convertedParams.toArray(new String[0]);
}
}

3
src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java
View File

@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -200,7 +201,7 @@ public class PubSubResubscribeTests {
* a thread executing its PatternSubscriptionTask
* @throws Exception
*/
@Test
@Ignore("DATAREDIS-166 Intermittent corrupted input/output streams subscribing to both patterns and channels in RMLC")
public void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exception {
container.removeMessageListener(adapter);
container.stop();