diff --git a/gradle.properties b/gradle.properties index 93c98ebed..032c1e877 100644 --- a/gradle.properties +++ b/gradle.properties @@ -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 # -------------------- diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index ca4177d25..98370c675 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -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 queue; @@ -184,14 +186,14 @@ public class SrpConnection implements RedisConnection { public List 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 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 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 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 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 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 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 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 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); } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java index a57ccef5e..79821e00a 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpUtils.java @@ -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 arrays = new ArrayList(); - 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 arrays = new ArrayList(); + + 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(); + } } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionPipelineIntegrationTests.java index 70e7929d5..cc08571fd 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionPipelineIntegrationTests.java @@ -288,8 +288,7 @@ abstract public class AbstractConnectionPipelineIntegrationTests extends connection.get("testitnow"); connection.exec(); List convertedResults = convertResults(); - assertEquals(Arrays.asList(new Object[] { Arrays.asList(new String[] { "OK", - "somethingelse" }) }), convertedResults); + assertEquals(Arrays.asList(new Object[] {"somethingelse"}), convertedResults); } @Test diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java index b3d203192..7ae7d84b8 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionIntegrationTests.java @@ -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() { } diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java index 10f0522d0..5a61d302d 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineIntegrationTests.java @@ -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(); diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpUtilsTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpUtilsTests.java new file mode 100644 index 000000000..29a335795 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpUtilsTests.java @@ -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 convertedParams = new ArrayList(); + for (Object b : bytes) { + convertedParams.add(new String((byte[]) b)); + } + return convertedParams.toArray(new String[0]); + } +} diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index 06c5a1b15..724e7bb1d 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -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();