From a40a88795a00ee33f1771c37fbaa38d22eebf00a Mon Sep 17 00:00:00 2001 From: Jennifer Hickey Date: Fri, 19 Jul 2013 12:06:34 -0700 Subject: [PATCH] SRP closePipeline should return List of tx results and strip individual op results DATAREDIS-223 - closePipeline now returns Lists of exec results instead of flattening results - Temp revert conversion of exec results for consistency --- .../redis/connection/srp/SrpConnection.java | 155 ++++-- ...ConnectionTransactionIntegrationTests.java | 88 ++- ...ConnectionTransactionIntegrationTests.java | 92 +-- .../srp/SrpConnectionIntegrationTests.java | 42 -- ...SrpConnectionPipelineIntegrationTests.java | 67 --- ...pConnectionPipelineTxIntegrationTests.java | 64 +-- ...ConnectionTransactionIntegrationTests.java | 522 +++++++++++------- 7 files changed, 534 insertions(+), 496 deletions(-) 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 449e356d7..f6978975f 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 @@ -76,6 +76,7 @@ public class SrpConnection implements RedisConnection { private boolean pipelineRequested = false; private Pipeline pipeline; private PipelineTracker callback; + private PipelineTracker txTracker; private volatile SrpSubscription subscription; private boolean convertPipelineResults=true; @@ -84,7 +85,12 @@ public class SrpConnection implements RedisConnection { private class PipelineTracker implements FutureCallback { private final List results = Collections.synchronizedList(new ArrayList()); - private final Queue futureResults = new LinkedList(); + private final Queue futureResults = new LinkedList(); + private boolean convertResults; + + public PipelineTracker(boolean convertResults) { + this.convertResults = convertResults; + } public void onSuccess(Reply result) { results.add(result.data()); @@ -94,35 +100,56 @@ public class SrpConnection implements RedisConnection { results.add(t); } + @SuppressWarnings("unchecked") public List complete() { + int txResults = 0; List> futures = new ArrayList>(); - for(SrpGenericResult future: futureResults) { - futures.add(future.getResultHolder()); + for(FutureResult future: futureResults) { + if(future instanceof SrpTxResult) { + txResults++; + } else { + ListenableFuture f = (ListenableFuture) future.getResultHolder(); + futures.add(f); + } } try { Futures.successfulAsList(futures).get(); } catch (Exception ex) { // ignore } - if(futureResults.size() != results.size()) { + if(futureResults.size() != results.size() + txResults) { throw new RedisPipelineException("Received a different number of results than expected. Expected: " + futureResults.size(), results); } List convertedResults = new ArrayList(); - for(Object result: results) { - SrpGenericResult futureResult = futureResults.remove(); - if(result instanceof Exception || !convertPipelineResults) { - convertedResults.add(result); - }else if(!(futureResult instanceof SrpStatusResult)) { - convertedResults.add(futureResult.convert(result)); + + int i = 0; + for(FutureResult future: futureResults) { + if(future instanceof SrpTxResult) { + PipelineTracker txTracker = ((SrpTxResult)future).getResultHolder(); + if(txTracker != null) { + convertedResults.add(getPipelinedResults(txTracker, true)); + }else { + convertedResults.add(null); + } + }else { + Object result = results.get(i); + if(result instanceof Exception || !convertResults) { + convertedResults.add(result); + }else if(!(future instanceof SrpStatusResult)) { + convertedResults.add(future.convert(result)); + } + i++; } } return convertedResults; } - public void addCommand(SrpGenericResult result) { + public void addCommand(FutureResult result) { futureResults.add(result); - Futures.addCallback(result.getResultHolder(), this); + if(!(result instanceof SrpTxResult)) { + Futures.addCallback(((SrpGenericResult)result).getResultHolder(), this); + } } public void close() { @@ -164,6 +191,18 @@ public class SrpConnection implements RedisConnection { } } + private class SrpTxResult extends FutureResult { + public SrpTxResult(PipelineTracker txTracker) { + super(txTracker); + } + public List get() { + if(resultHolder == null) { + return null; + } + return resultHolder.complete(); + } + } + public SrpConnection(String host, int port, BlockingQueue queue) { try { this.client = new RedisClient(host, port); @@ -224,7 +263,7 @@ public class SrpConnection implements RedisConnection { } public boolean isPipelined() { - return (pipeline != null); + return pipelineRequested || (txTracker != null); } @@ -233,10 +272,6 @@ public class SrpConnection implements RedisConnection { initPipeline(); } - public List closePipeline() { - return closePipeline(true); - } - public List sort(byte[] key, SortParameters params) { Object[] sort = sortParams(params); @@ -481,10 +516,9 @@ public class SrpConnection implements RedisConnection { public void discard() { isMulti = false; try { + // discard tracked futures + txTracker = null; client.discard(); - if (!pipelineRequested) { - closePipeline(false); - } } catch (Exception ex) { throw convertSrpAccessException(ex); } @@ -493,33 +527,27 @@ public class SrpConnection implements RedisConnection { public List exec() { isMulti = false; - Exception execException = null; - List results = null; - boolean resultsSet = true; try { Future exec = client.exec(); // Need to wait on execution or subsequent non-pipelined calls may read exec results - resultsSet = exec.get(); - } catch (Exception ex) { - execException = ex; - } finally { - // ensure pipeline is closed even if error in exec - if (!pipelineRequested) { - try { - results = closePipeline(); - } catch (RedisPipelineException e) { - if(execException == null) { - execException = e; - } + boolean resultsSet = exec.get(); + if(!resultsSet) { + // This is the case where a nil MultiBulk Reply was returned b/c watched variable modified + if(pipelineRequested) { + pipeline(new SrpTxResult(null)); } + return null; } + if(pipelineRequested) { + pipeline(new SrpTxResult(txTracker)); + return null; + } + return closeTransaction(); + } catch (Exception ex) { + throw convertSrpAccessException(ex); + } finally { + txTracker = null; } - // If resultsSet is false, it's b/c of nil Multi-Bulk reply (watch case) and null is returned - if(execException != null && resultsSet) { - // Intentionally convert RedisPipelineException too, it's an impl detail - throw convertSrpAccessException(execException); - } - return results; } @@ -603,7 +631,7 @@ public class SrpConnection implements RedisConnection { return; } isMulti = true; - initPipeline(); + initTxTracker(); try { client.multi(); } catch (Exception ex) { @@ -2168,33 +2196,52 @@ public class SrpConnection implements RedisConnection { } // processing method that adds a listener to the future in order to track down the results and close the pipeline - private void pipeline(SrpGenericResult future) { - callback.addCommand(future); + private void pipeline(FutureResult future) { + if(isQueueing()) { + txTracker.addCommand(future); + }else { + callback.addCommand(future); + } } private void initPipeline() { if (pipeline == null) { - callback = new PipelineTracker(); + callback = new PipelineTracker(convertPipelineResults); pipeline = client.pipeline(); } } - private List closePipeline(boolean getResults) { + private void initTxTracker() { + if(txTracker == null) { + txTracker = new PipelineTracker(false); + } + initPipeline(); + } + + public List closePipeline() { pipelineRequested = false; List results = Collections.emptyList(); if (pipeline != null) { pipeline = null; - if(getResults) { - results = getPipelinedResults(); - } + results = getPipelinedResults(callback, true); callback.close(); callback = null; } return results; } - private List getPipelinedResults() { - List execute = new ArrayList(callback.complete()); + private List closeTransaction() { + List results = Collections.emptyList(); + if (txTracker != null) { + results = getPipelinedResults(txTracker, false); + txTracker.close(); + txTracker = null; + } + return results; + } + + private List getPipelinedResults(PipelineTracker tracker, boolean throwPipelineException) { + List execute = new ArrayList(tracker.complete()); if (execute != null && !execute.isEmpty()) { Exception cause = null; for (int i = 0; i < execute.size(); i++) { @@ -2208,7 +2255,11 @@ public class SrpConnection implements RedisConnection { } } if (cause != null) { - throw new RedisPipelineException(cause, execute); + if(throwPipelineException) { + throw new RedisPipelineException(cause, execute); + }else { + throw convertSrpAccessException(cause); + } } return execute; diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionTransactionIntegrationTests.java index da4f3d23b..ab6f3a2f7 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionTransactionIntegrationTests.java @@ -1,6 +1,7 @@ package org.springframework.data.redis.connection; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import java.util.ArrayList; import java.util.Arrays; @@ -344,9 +345,22 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends } @Test + @IfProfileValue(name = "redisVersion", value = "2.6") public void testDumpAndRestore() { convert = false; - super.testDumpAndRestore(); + connection.set("testing", "12"); + actual.add(connection.dump("testing".getBytes())); + List results = getResults(); + initConnection(); + actual.add(connection.del("testing")); + actual.add((connection.get("testing"))); + connection.restore("testing".getBytes(), 0, (byte[]) results.get(results.size() - 1)); + actual.add(connection.get("testing")); + results = getResults(); + assertEquals(3,results.size()); + assertEquals(1l, results.get(0)); + assertNull(results.get(1)); + assertEquals("12", new String((byte[])results.get(2))); } @Test @@ -383,6 +397,24 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends super.testEvalReturnArrayStrings(); } + @Test + @IfProfileValue(name = "redisVersion", value = "2.6") + public void testEvalReturnSingleOK() { + actual.add(connection.eval("return redis.call('set','abc','ghk')", ReturnType.STATUS, 0)); + assertEquals(Arrays.asList(new Object[] { "OK" }), getResultsNoConversion()); + } + + @SuppressWarnings("unchecked") + @Test + @IfProfileValue(name = "redisVersion", value = "2.6") + public void testEvalReturnArrayOKs() { + actual.add(connection.eval( + "return { redis.call('set','abc','ghk'), redis.call('set','abc','lfdf')}", + ReturnType.MULTI, 0)); + List result = (List) getResults().get(0); + assertEquals(Arrays.asList(new Object[] { "OK", "OK" }), result); + } + @Test @IfProfileValue(name = "redisVersion", value = "2.6") public void testRestoreTtl() { @@ -398,6 +430,57 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends connection.scriptKill(); } + @Test + @IfProfileValue(name = "redisVersion", value = "2.6") + public void testBitCount() { + convertLongToBoolean = false; + String key = "bitset-test"; + connection.setBit(key, 0, false); + connection.setBit(key, 1, true); + connection.setBit(key, 2, true); + actual.add(connection.bitCount(key)); + // Lettuce setBit returns Long instead of void + verifyResults(new ArrayList(Arrays.asList(0l, 0l, 0l, 2l))); + } + + @Test + public void testGetRangeSetRange() { + connection.set("rangekey", "supercalifrag"); + actual.add(connection.getRange("rangekey", 0l, 2l)); + connection.setRange("rangekey", "ck", 2); + actual.add(connection.get("rangekey")); + // Lettuce returns a value for setRange + verifyResults(Arrays.asList(new Object[] { "sup", 13l, "suckrcalifrag" })); + } + + @SuppressWarnings("unchecked") + @Test + public void testZRevRangeByScoreOffsetCount() { + actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); + actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); + actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d, 0, 5)); + assertEquals(Arrays.asList(new String[] { "Bob", "James" }),(List) getResults().get(2)); + } + + @SuppressWarnings("unchecked") + @Test + public void testZRevRangeByScore() { + actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); + actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); + actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d)); + assertEquals(Arrays.asList(new String[] { "Bob", "James" }), (List) getResults().get(2)); + } + + @SuppressWarnings("unchecked") + @Test + public void testZRevRangeByScoreWithScoresOffsetCount() { + actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); + actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); + actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d, 0, 5)); + assertEquals(Arrays.asList(new String[] { "Bob", "James" }), + (List) getResults().get(2)); + } + protected void initConnection() { connection.multi(); } @@ -442,6 +525,9 @@ abstract public class AbstractConnectionTransactionIntegrationTests extends && ((List) result).get(0) instanceof byte[]) { return (SerializationUtils.deserialize((List) result, stringSerializer)); } else if (result instanceof byte[]) { + if(convertStringToProps) { + return Converters.toProperties(stringSerializer.deserialize((byte[]) result)); + } return (stringSerializer.deserialize((byte[]) result)); } else if (result instanceof Map && ((Map) result).keySet().iterator().next() instanceof byte[]) { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java index 3ea25dcb5..6f156764b 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.connection.lettuce; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; import java.util.ArrayList; import java.util.Arrays; @@ -32,9 +31,8 @@ import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.AbstractConnectionTransactionIntegrationTests; import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.DefaultStringTuple; -import org.springframework.data.redis.connection.StringRedisConnection; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; -import org.springframework.data.redis.connection.ReturnType; +import org.springframework.data.redis.connection.StringRedisConnection; import org.springframework.data.redis.connection.StringRedisConnection.StringTuple; import org.springframework.data.redis.connection.convert.SetConverter; import org.springframework.test.annotation.IfProfileValue; @@ -174,19 +172,6 @@ public class LettuceConnectionTransactionIntegrationTests extends verifyResults(Arrays.asList(new Object[] { 0l, 0l, 0l, 1l })); } - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testBitCount() { - convertLongToBoolean = false; - String key = "bitset-test"; - connection.setBit(key, 0, false); - connection.setBit(key, 1, true); - connection.setBit(key, 2, true); - actual.add(connection.bitCount(key)); - // Lettuce setBit returns Long instead of void - verifyResults(new ArrayList(Arrays.asList(0l, 0l, 0l, 2l))); - } - @Test public void testHKeys() { convertListToSet = true; @@ -198,16 +183,6 @@ public class LettuceConnectionTransactionIntegrationTests extends super.testBitOpNotMultipleSources(); } - @Test - public void testGetRangeSetRange() { - connection.set("rangekey", "supercalifrag"); - actual.add(connection.getRange("rangekey", 0l, 2l)); - connection.setRange("rangekey", "ck", 2); - actual.add(connection.get("rangekey")); - // Lettuce returns a value for setRange - verifyResults(Arrays.asList(new Object[] { "sup", 13l, "suckrcalifrag" })); - } - @Test(expected = UnsupportedOperationException.class) @IfProfileValue(name = "redisVersion", value = "2.6") public void testSRandMemberCountNegative() { @@ -220,77 +195,12 @@ public class LettuceConnectionTransactionIntegrationTests extends super.testSortStoreNullParams(); } - @SuppressWarnings("unchecked") - @Test - public void testZRevRangeByScoreOffsetCount() { - actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); - actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); - actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d, 0, 5)); - assertEquals(Arrays.asList(new String[] { "Bob", "James" }),(List) getResults().get(2)); - } - - @SuppressWarnings("unchecked") - @Test - public void testZRevRangeByScore() { - actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); - actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); - actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d)); - assertEquals(Arrays.asList(new String[] { "Bob", "James" }), (List) getResults().get(2)); - } - - @SuppressWarnings("unchecked") - @Test - public void testZRevRangeByScoreWithScoresOffsetCount() { - actual.add(byteConnection.zAdd("myset".getBytes(), 2, "Bob".getBytes())); - actual.add(byteConnection.zAdd("myset".getBytes(), 1, "James".getBytes())); - actual.add(byteConnection.zRevRangeByScore("myset".getBytes(), 0d, 3d, 0, 5)); - assertEquals(Arrays.asList(new String[] { "Bob", "James" }), - (List) getResults().get(2)); - } - @Test public void testZRevRangeByScoreWithScores() { convertToStringTuple = false; super.testZRevRangeByScoreWithScores(); } - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testEvalReturnSingleOK() { - actual.add(connection.eval("return redis.call('set','abc','ghk')", ReturnType.STATUS, 0)); - assertEquals(Arrays.asList(new Object[] { "OK" }), getResultsNoConversion()); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testDumpAndRestore() { - convert = false; - connection.set("testing", "12"); - actual.add(connection.dump("testing".getBytes())); - List results = getResults(); - initConnection(); - actual.add(connection.del("testing")); - actual.add((connection.get("testing"))); - connection.restore("testing".getBytes(), 0, (byte[]) results.get(results.size() - 1)); - actual.add(connection.get("testing")); - results = getResults(); - assertEquals(3,results.size()); - assertEquals(1l, results.get(0)); - assertNull(results.get(1)); - assertEquals("12", new String((byte[])results.get(2))); - } - - @SuppressWarnings("unchecked") - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testEvalReturnArrayOKs() { - actual.add(connection.eval( - "return { redis.call('set','abc','ghk'), redis.call('set','abc','lfdf')}", - ReturnType.MULTI, 0)); - List result = (List) getResults().get(0); - assertEquals(Arrays.asList(new Object[] { "OK", "OK" }), result); - } - @Test public void testMove() { connection.set("foo", "bar"); 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 fa7f4e419..6ac067a6f 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 @@ -16,17 +16,12 @@ package org.springframework.data.redis.connection.srp; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; - import java.util.Arrays; -import java.util.List; import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; -import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.ReturnType; import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; @@ -88,41 +83,4 @@ public class SrpConnectionIntegrationTests extends AbstractConnectionIntegration ReturnType.MULTI, 0)); verifyResults(Arrays.asList(new Object[] {Arrays.asList(new Object[] {"OK", "OK"} )})); } - - @SuppressWarnings("unchecked") - @Test - public void testMultiExec() throws Exception { - connection.multi(); - connection.set("key", "value"); - actual.add(connection.get("key")); - actual.add(connection.exec()); - List results = getResults(); - assertNull(results.get(0)); - List execResults = (List) results.get(1); - // SRP is already filtering out the void types since tx uses pipeline, so result size here is only 1 - assertEquals(1, execResults.size()); - // However, since DefaultStringRedisConnection doesn't yet convert tx results, the data is still in raw bytes - assertEquals("value", new String((byte[]) execResults.get(0))); - assertEquals("value", connection.get("key")); - } - - @SuppressWarnings("unchecked") - @Test - public void testUnwatch() throws Exception { - connection.set("testitnow", "willdo"); - connection.watch("testitnow".getBytes()); - connection.unwatch(); - connection.multi(); - DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection( - connectionFactory.getConnection()); - conn2.set("testitnow", "something"); - connection.set("testitnow", "somethingelse"); - actual.add(connection.get("testitnow")); - actual.add(connection.exec()); - List results = getResults(); - assertNull(results.get(0)); - List execResults = (List) results.get(1); - // SRP is already filtering out the void types since tx uses pipeline, so result size here is only 1 - assertEquals("somethingelse", new String((byte[]) execResults.get(0))); - } } 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 55f208434..d99c7f5e0 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 @@ -15,18 +15,12 @@ */ package org.springframework.data.redis.connection.srp; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.fail; - import java.util.Arrays; -import java.util.List; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.AbstractConnectionPipelineIntegrationTests; -import org.springframework.data.redis.connection.DefaultStringRedisConnection; -import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.data.redis.connection.ReturnType; import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; @@ -43,67 +37,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class SrpConnectionPipelineIntegrationTests extends AbstractConnectionPipelineIntegrationTests { - @Test - public void testMultiDiscard() throws Exception { - DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection( - connectionFactory.getConnection()); - conn2.set("testitnow", "willdo"); - connection.multi(); - connection.set("testitnow2", "notok"); - connection.discard(); - // SRP throws Exception on transaction discard, so we expect pipeline - // exception here - try { - getResults(); - fail("Closing the pipeline on a discarded tx should throw a RedisPipelineException"); - } catch (RedisPipelineException e) { - } - assertEquals("willdo", connection.get("testitnow")); - connection.openPipeline(); - // Ensure we can run a new tx after discarding previous one - testMultiExec(); - } - - @Test - public void testMultiExec() throws Exception { - connection.multi(); - connection.set("key", "value"); - connection.get("key"); - actual.add(connection.exec()); - List results = getResults(); - // For the moment, SRP exec results are flattened into pipeline - assertEquals(1, results.size()); - assertEquals("value", new String((byte[]) results.get(0))); - assertEquals("value", connection.get("key")); - } - - @Test - public void testUnwatch() throws Exception { - connection.set("testitnow", "willdo"); - connection.watch("testitnow".getBytes()); - connection.unwatch(); - connection.multi(); - //Give some time for unwatch to be asynch executed - Thread.sleep(500); - DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection( - connectionFactory.getConnection()); - conn2.set("testitnow", "something"); - connection.set("testitnow", "somethingelse"); - connection.get("testitnow"); - actual.add(connection.exec()); - List results = getResults(); - // For the moment, SRP exec results are flattened into pipeline - assertEquals("somethingelse", new String((byte[]) results.get(0))); - } - - // SRP sets results of all commands in the pipeline to RedisException if - // exec returns a - // null multi-bulk reply - @Test(expected = RedisPipelineException.class) - public void testWatch() throws Exception { - super.testWatch(); - } - @Test @IfProfileValue(name = "redisVersion", value = "2.6") public void testEvalReturnArrayOKs() { diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineTxIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineTxIntegrationTests.java index 8f78c0dbf..7ebc86edc 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineTxIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionPipelineTxIntegrationTests.java @@ -17,14 +17,12 @@ package org.springframework.data.redis.connection.srp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; -import static org.junit.Assert.fail; -import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import org.junit.Test; -import org.springframework.data.redis.RedisSystemException; -import org.springframework.data.redis.RedisVersionUtils; import org.springframework.data.redis.connection.RedisPipelineException; import org.springframework.test.annotation.IfProfileValue; @@ -36,30 +34,6 @@ import org.springframework.test.annotation.IfProfileValue; */ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransactionIntegrationTests { - @Test - public void exceptionExecuteNative() throws Exception { - connection.execute("ZadD", getClass() + "#foo\t0.90\titem"); - try { - getResults(); - fail("Expected an Exception to be thrown executing a command with syntax error"); - }catch(RedisPipelineException e) { - // Redis 2.4, Exception occurs when we get the result of execute on closePipeline - if(RedisVersionUtils.atLeast("2.6.4", byteConnection)) { - fail("RedisPipelineException should not be thrown in Redis 2.6"); - } - }catch(RedisSystemException e) { - try { - connection.closePipeline(); - }catch(Exception ex) { - // Gonna get another error on closing the pipeline as results from execute - } - if(!RedisVersionUtils.atLeast("2.6", byteConnection)) { - // Redis 2.6 returns an ErrorReploy on exec, the Exception occurs when we call exec() - // b/c it waits on the response - fail("RedisSystemException should only be thrown in Redis 2.6"); - } - } - } @Test public void testUsePipelineAfterTxExec() { @@ -67,11 +41,11 @@ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransa assertNull(connection.exec()); assertNull(connection.get("foo")); List results = connection.closePipeline(); - assertEquals(1, results.size()); - assertEquals("bar", new String((byte[]) results.get(0))); + assertEquals(Arrays.asList(new Object[] {Collections.singletonList("OK"), "bar"}), results); assertEquals("bar", connection.get("foo")); } + @SuppressWarnings("unchecked") @Test public void testExec2TransactionsInPipeline() { connection.set("foo", "bar"); @@ -83,8 +57,9 @@ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransa assertNull(connection.exec()); List results = connection.closePipeline(); assertEquals(2, results.size()); - assertEquals("bar", new String((byte[]) results.get(0))); - assertEquals("baz", new String((byte[]) results.get(1))); + // First element of each list is "OK" + assertEquals("bar", new String((byte[])((List)results.get(0)).get(1))); + assertEquals("baz", new String((byte[])((List)results.get(1)).get(1))); } @Test(expected=RedisPipelineException.class) @@ -122,14 +97,25 @@ public class SrpConnectionPipelineTxIntegrationTests extends SrpConnectionTransa connection.multi(); } + @SuppressWarnings("unchecked") protected List getResults() { assertNull(connection.exec()); - List actual = connection.closePipeline(); - List serializedResults = new ArrayList(); - for (Object result : actual) { - Object convertedResult = convertResult(result); - serializedResults.add(convertedResult); - } - return serializedResults; + List pipelined = connection.closePipeline(); + // We expect only the results of exec to be in the closed pipeline + assertEquals(1,pipelined.size()); + List txResults = (List)pipelined.get(0); + // Return exec results and this test should behave exactly like its superclass + return convertResults(txResults); + } + + @SuppressWarnings("unchecked") + protected List getResultsNoConversion() { + assertNull(connection.exec()); + List pipelined = connection.closePipeline(); + // We expect only the results of exec to be in the closed pipeline + assertEquals(1,pipelined.size()); + List txResults = (List)pipelined.get(0); + // Return exec results and this test should behave exactly like its superclass + return txResults; } } diff --git a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java index 6c66e0d90..08c68e32d 100644 --- a/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/srp/SrpConnectionTransactionIntegrationTests.java @@ -16,22 +16,31 @@ package org.springframework.data.redis.connection.srp; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; 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.Map; import java.util.Set; -import org.junit.Ignore; import org.junit.Test; -import org.springframework.data.redis.RedisSystemException; +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.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; @@ -41,254 +50,359 @@ import org.springframework.test.annotation.IfProfileValue; * @author Jennifer Hickey * */ -public class SrpConnectionTransactionIntegrationTests extends SrpConnectionIntegrationTests { +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration("SrpConnectionIntegrationTests-context.xml") +public class SrpConnectionTransactionIntegrationTests extends AbstractConnectionTransactionIntegrationTests { - private boolean convert = true; + protected boolean convertResultToSet=false; - @Ignore - public void testMultiDiscard() { - } + protected boolean convertResultToStringTuples=false; - @Ignore - public void testMultiExec() { - } + protected boolean convertResultToTuples=false; - @Ignore - public void testUnwatch() { - } + protected boolean convertBooleanToLong=false; - @Ignore - public void testWatch() { - } - - @Ignore - @Test - public void testExecWithoutMulti() { - } - - @Ignore - @Test - public void testErrorInTx() { - } - - /* - * Using blocking ops inside a tx does not make a lot of sense as it would - * require blocking the entire server in order to execute the block - * atomically, which in turn does not allow other clients to perform a push - * operation. * - */ - - @Ignore - public void testBLPop() { - } - - @Ignore - public void testBRPop() { - } - - @Ignore - public void testBRPopLPush() { - } - - @Ignore - public void testBLPopTimeout() { - } - - @Ignore - public void testBRPopTimeout() { - } - - @Ignore - public void testBRPopLPushTimeout() { - } - - @Ignore("Pub/Sub not supported with transactions") - public void testPubSubWithNamedChannels() throws Exception { - } - - @Ignore("Pub/Sub not supported with transactions") - public void testPubSubWithPatterns() throws Exception { - } - - @Ignore - public void testNullKey() throws Exception { - } - - @Ignore - public void testNullValue() throws Exception { - } - - @Ignore - public void testHashNullKey() throws Exception { - } - - @Ignore - public void testHashNullValue() throws Exception { - } - - @Test(expected = RedisSystemException.class) - public void exceptionExecuteNative() throws Exception { - super.exceptionExecuteNative(); - } - - // SRP tx.exec() is now returning converted data types b/c it uses pipeline, but - // DefaultStringRedisConnection isn't converting byte[]s from exec() yet. - // We implement the conversion in convertResult(), but convertResult's a bit over-eager in certain scenarios, hence these overrides + protected boolean convertResultsToMap=false; @Test - public void testExecute() { - convert = false; - super.testExecute(); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testScriptLoadEvalSha() { - convert = false; - super.testScriptLoadEvalSha(); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testEvalShaArrayStrings() { - convert = false; - super.testEvalShaArrayStrings(); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testEvalReturnString() { - convert = false; - super.testEvalReturnString(); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testEvalReturnArrayStrings() { - convert = false; - super.testEvalReturnArrayStrings(); - } - - @Test - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testDumpAndRestore() { - convert = false; - connection.set("testing", "12"); - actual.add(connection.dump("testing".getBytes())); - List results = getResults(); - initConnection(); + public void testDel() { + connection.set("testing","123"); actual.add(connection.del("testing")); - actual.add((connection.get("testing"))); - connection.restore("testing".getBytes(), 0, (byte[]) results.get(results.size() - 1)); - actual.add(connection.get("testing")); - results = getResults(); - assertEquals(3,results.size()); - assertEquals(1l, results.get(0)); - assertNull(results.get(1)); - assertEquals("12", new String((byte[])results.get(2))); + actual.add(connection.exists("testing")); + verifyResults(Arrays.asList(new Object[] { true, false })); } - @Test(expected = RedisSystemException.class) - @IfProfileValue(name = "redisVersion", value = "2.6") - public void testRestoreExistingKey() { - connection.set("testing", "12"); - actual.add(connection.dump("testing".getBytes())); - List results = getResults(); - initConnection(); - connection.restore("testing".getBytes(), 0, ((String)results.get(0)).getBytes()); - getResults(); + @Test + public void testSAdd() { + convertResultToSet = true; + super.testSAdd(); + } + + @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(Collections.singletonList("Joe")) })); + } + + @Test + public void testZInterStore() { + convertResultToSet = true; + super.testZInterStore(); + } + + @Test(expected=UnsupportedOperationException.class) + public void testZInterStoreAggWeights() { + super.testZInterStoreAggWeights(); + } + + @Test(expected=UnsupportedOperationException.class) + public void testZUnionStoreAggWeights() { + super.testZUnionStoreAggWeights(); + } + + @Test + 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 testRestoreTtl() { - convert = false; - super.testRestoreTtl(); + 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 testZRevRangeByScoreOffsetCount() { - convert = false; - super.testZRevRangeByScoreOffsetCount(); + 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 - public void testZRevRangeByScore() { - convert = false; - super.testZRevRangeByScore(); + public void testBitCount() { + convertBooleanToLong = true; + super.testBitCount(); } @Test - public void testZRevRangeByScoreWithScoresOffsetCount() { - convert = false; - super.testZRevRangeByScoreWithScoresOffsetCount(); + 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() { - convert = false; + convertResultToTuples = true; super.testZRevRangeByScoreWithScores(); } - @Test(expected = UnsupportedOperationException.class) + @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 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(); + 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 results = getResults(); + List scriptResults = (List) results.get(0); + assertEquals(Arrays.asList(new Object[] { "key1", "arg1" }), + scriptResults); } - protected void initConnection() { - connection.multi(); + @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 result = (List) getResults().get(0); + assertEquals(Arrays.asList(new Object[] { "foo", "bar" }), result); } - protected List getResults() { - List actual = connection.exec(); - List serializedResults = new ArrayList(); - for (Object result : actual) { - Object convertedResult = convertResult(result); - serializedResults.add(convertedResult); - } - return serializedResults; + @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) { - if(! convert) { - return result; - } - if (result instanceof List && !(((List) result).isEmpty()) - && ((List) result).get(0) instanceof byte[]) { - return (SerializationUtils.deserialize((List) result, stringSerializer)); - } else if (result instanceof byte[]) { - return (stringSerializer.deserialize((byte[]) result)); - } else if (result instanceof Map - && ((Map) result).keySet().iterator().next() instanceof byte[]) { - return (SerializationUtils.deserialize((Map) result, stringSerializer)); - } else if (result instanceof Set && !(((Set) result).isEmpty())) { - Object firstResult = ((Set) result).iterator().next(); - if(firstResult instanceof byte[]) { - return (SerializationUtils.deserialize((Set) result, stringSerializer)); - } + Object convertedResult = super.convertResult(result); + if (convertedResult instanceof Set && !(((Set) convertedResult).isEmpty())) { + Object firstResult = ((Set) convertedResult).iterator().next(); if(firstResult instanceof Tuple) { Set tuples = new LinkedHashSet(); - for (Tuple value : ((Set) result)) { + for (Tuple value : ((Set) convertedResult)) { DefaultStringTuple tuple = new DefaultStringTuple(value,stringSerializer.deserialize(value.getValue())); tuples.add(tuple); } return tuples; } + }else if (convertedResult instanceof Reply[]) { + if (convertResultToStringTuples) { + Set tuples = SrpConverters.toTupleSet((Reply[])convertedResult); + Collection stringTuples; + if(convertResultToSet) { + stringTuples = new LinkedHashSet(); + }else { + stringTuples = new ArrayList(); + } + for (Tuple tuple : tuples) { + stringTuples.add(new DefaultStringTuple(tuple, new String(tuple.getValue()))); + } + return stringTuples; + } else if(convertResultToTuples) { + return SrpConverters.toTupleSet((Reply[])convertedResult); + } + 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 results = new ArrayList(); + for(Reply reply: (Reply[])convertedResult) { + results.add(((IntegerReply)reply).data()); + } + return results; + } else if(((Reply[]) convertedResult).length > 0 && ((Reply[])convertedResult)[0] instanceof StatusReply) { + List statuses = new ArrayList(); + 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 result; + return convertedResult; } - protected void verifyResults(List expected) { - List expectedTx = new ArrayList(); - for (int i = 0; i < actual.size(); i++) { - expectedTx.add(null); - } - assertEquals(expectedTx, actual); - List results = getResults(); - assertEquals(expected, results); + @Override + protected DataAccessException convertException(Exception ex) { + return SrpConverters.toDataAccessException(ex); } }