From 4b2ccbecb9998e0b3acace1f4603a84932731dff Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Thu, 13 Mar 2014 15:21:45 +0100 Subject: [PATCH] DATAREDIS-271 - Add support for 'pSetEx'. Support for 'pSetEx' has been added for 'jedis' & 'srp'. For 'lettuce' the command is emulated using 'eval'. There is no support for 'pSetEx' when using 'jredis'. 'DefaultValueOperations.set' used with 'TimeUnit.MILLISECONDS' will fallback to 'setEx' in case 'pSetEx' is not supported by the driver in use. Original pull request: #46. --- .../appendix/appendix-command-reference.xml | 2 +- .../DefaultStringRedisConnection.java | 18 ++++++++ .../redis/connection/RedisStringCommands.java | 11 +++++ .../connection/StringRedisConnection.java | 14 +++++- .../connection/jedis/JedisConnection.java | 30 +++++++++++++ .../connection/jredis/JredisConnection.java | 9 ++++ .../connection/lettuce/LettuceConnection.java | 44 +++++++++++++++++++ .../redis/connection/srp/SrpConnection.java | 23 ++++++++++ .../redis/core/DefaultValueOperations.java | 29 ++++++++++-- .../data/redis/core/ValueOperations.java | 12 ++++- .../AbstractConnectionIntegrationTests.java | 15 +++++++ .../DefaultStringRedisConnectionTests.java | 13 +++++- .../JRedisConnectionIntegrationTests.java | 8 ++++ .../LettuceConnectionIntegrationTests.java | 1 + ...uceConnectionPipelineIntegrationTests.java | 1 + ...eConnectionPipelineTxIntegrationTests.java | 1 + ...ConnectionTransactionIntegrationTests.java | 1 + .../core/DefaultValueOperationsTests.java | 23 ++++++++-- 18 files changed, 244 insertions(+), 11 deletions(-) diff --git a/docs/src/reference/docbook/appendix/appendix-command-reference.xml b/docs/src/reference/docbook/appendix/appendix-command-reference.xml index bef5d3879..a8c5aeb66 100644 --- a/docs/src/reference/docbook/appendix/appendix-command-reference.xml +++ b/docs/src/reference/docbook/appendix/appendix-command-reference.xml @@ -94,7 +94,7 @@ PEXIPREX PEXPIREATX PINGX - PSETEX- + PSETEXX PSUBSCRIBEX PTTLX PUBLISHX diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index fdb338ce3..9d2805c66 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -735,6 +735,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection { delegate.setEx(key, seconds, value); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[]) + */ + @Override + public void pSetEx(byte[] key, long milliseconds, byte[] value) { + delegate.pSetEx(key, milliseconds, value); + } + public Boolean setNX(byte[] key, byte[] value) { Boolean result = delegate.setNX(key, value); if (isFutureConversion()) { @@ -1695,6 +1704,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection { delegate.setEx(serialize(key), seconds, serialize(value)); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pSetEx(java.lang.String, long, java.lang.String) + */ + @Override + public void pSetEx(String key, long seconds, String value) { + pSetEx(serialize(key), seconds, serialize(value)); + } + public Boolean setNX(String key, String value) { Boolean result = delegate.setNX(serialize(key), serialize(value)); if (isFutureConversion()) { diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java index afc951cde..f362be751 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java @@ -87,6 +87,17 @@ public interface RedisStringCommands { */ void setEx(byte[] key, long seconds, byte[] value); + /** + * Set the {@code value} and expiration in {@code milliseconds} for {@code key}. + * + * @see http://redis.io/commands/psetex + * @param key + * @param milliseconds + * @param value + * @since 1.3 + */ + void pSetEx(byte[] key, long milliseconds, byte[] value); + /** * Set multiple keys to multiple values using key-value pairs provided in {@code tuple}. * diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index ed0a62deb..20657682e 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2013 the original author or authors. + * Copyright 2011-2014 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. @@ -29,6 +29,7 @@ import org.springframework.data.redis.serializer.RedisSerializer; * Uses a {@link RedisSerializer} underneath to perform the conversion. * * @author Costin Leau + * @author Christoph Strobl * @see RedisCallback * @see RedisSerializer * @see StringRedisTemplate @@ -93,6 +94,17 @@ public interface StringRedisConnection extends RedisConnection { void setEx(String key, long seconds, String value); + /** + * Set the {@code value} and expiration in {@code milliseconds} for {@code key}. + * + * @see http://redis.io/commands/psetex + * @param key + * @param seconds + * @param value + * @since 1.3 + */ + void pSetEx(String key, long milliseconds, String value); + void mSetString(Map tuple); Boolean mSetNXString(Map tuple); diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index e6ad85f06..67460b754 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -331,6 +331,10 @@ public class JedisConnection implements RedisConnection { return results; } + private void doPipelined(Response response) { + pipeline(new JedisStatusResult(response)); + } + private void pipeline(FutureResult> result) { if (isQueueing()) { transaction(result); @@ -339,6 +343,10 @@ public class JedisConnection implements RedisConnection { } } + private void doQueued(Response response) { + transaction(new JedisStatusResult(response)); + } + private void transaction(FutureResult> result) { txResults.add(result); } @@ -1141,6 +1149,28 @@ public class JedisConnection implements RedisConnection { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[]) + */ + @Override + public void pSetEx(byte[] key, long milliseconds, byte[] value) { + + try { + if (isPipelined()) { + doPipelined(pipeline.psetex(key, (int) milliseconds, value)); + return; + } + if (isQueueing()) { + doQueued(transaction.psetex(key, (int) milliseconds, value)); + return; + } + jedis.psetex(key, (int) milliseconds, value); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + public Boolean setNX(byte[] key, byte[] value) { try { if (isPipelined()) { diff --git a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java index 85c4e9a42..6ecab2ee4 100644 --- a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java @@ -495,6 +495,15 @@ public class JredisConnection implements RedisConnection { throw new UnsupportedOperationException(); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[]) + */ + @Override + public void pSetEx(byte[] key, long milliseconds, byte[] value) { + throw new UnsupportedOperationException(); + } + public Boolean setNX(byte[] key, byte[] value) { try { return jredis.setnx(key, value); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 98044a813..8cdbadc1f 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -56,6 +56,7 @@ import org.springframework.util.ObjectUtils; import com.lambdaworks.redis.RedisAsyncConnection; import com.lambdaworks.redis.RedisClient; import com.lambdaworks.redis.RedisException; +import com.lambdaworks.redis.ScriptOutputType; import com.lambdaworks.redis.SortArgs; import com.lambdaworks.redis.ZStoreArgs; import com.lambdaworks.redis.codec.RedisCodec; @@ -1197,6 +1198,49 @@ public class LettuceConnection implements RedisConnection { } } + /** + * {@code pSetEx} is not directly supported and therefore emulated via {@literal lua script}. + * + * @since 1.3 + * @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[]) + */ + @Override + public void pSetEx(byte[] key, long milliseconds, byte[] value) { + + byte[] script = createRedisScriptForPSetEx(key, milliseconds, value); + byte[][] emptyArgs = new byte[0][0]; + + try { + if (isPipelined()) { + pipeline(new LettuceStatusResult(getAsyncConnection().eval(script, ScriptOutputType.STATUS, emptyArgs, + emptyArgs))); + return; + } + if (isQueueing()) { + transaction(new LettuceTxStatusResult(getConnection().eval(script, ScriptOutputType.STATUS, emptyArgs, + emptyArgs))); + return; + } + this.eval(script, ReturnType.STATUS, 0); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + + private byte[] createRedisScriptForPSetEx(byte[] key, long milliseconds, byte[] value) { + + StringBuilder sb = new StringBuilder("return redis.call('PSETEX'"); + sb.append(",'"); + sb.append(new String(key)); + sb.append("',"); + sb.append(milliseconds); + sb.append(",'"); + sb.append(new String(value)); + sb.append("')"); + + return sb.toString().getBytes(); + } + public Boolean setNX(byte[] key, byte[] value) { try { if (isPipelined()) { 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 2aecab82d..ba7c5a6c3 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 @@ -900,6 +900,24 @@ public class SrpConnection implements RedisConnection { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[]) + */ + @Override + public void pSetEx(byte[] key, long milliseconds, byte[] value) { + + try { + if (isPipelined()) { + doPipelined(pipeline.psetex(key, milliseconds, value)); + return; + } + client.psetex(key, milliseconds, value); + } catch (Exception ex) { + throw convertSrpAccessException(ex); + } + } + public Boolean setNX(byte[] key, byte[] value) { try { if (isPipelined()) { @@ -2113,6 +2131,11 @@ public class SrpConnection implements RedisConnection { } } + @SuppressWarnings("rawtypes") + private void doPipelined(ListenableFuture listenableFuture) { + pipeline(new SrpStatusResult(listenableFuture)); + } + // processing method that adds a listener to the future in order to track down the results and close the pipeline private void pipeline(FutureResult future) { if (isQueueing()) { diff --git a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java index 1e148b426..bdfaf8d2c 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2013 the original author or authors. + * Copyright 2011-2014 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. @@ -30,6 +30,7 @@ import org.springframework.data.redis.connection.RedisConnection; * * @author Costin Leau * @author Jennifer Hickey + * @author Christoph Strobl */ class DefaultValueOperations extends AbstractOperations implements ValueOperations { @@ -173,17 +174,37 @@ class DefaultValueOperations extends AbstractOperations implements V }, true); } - public void set(K key, V value, long timeout, TimeUnit unit) { + public void set(K key, V value, final long timeout, final TimeUnit unit) { final byte[] rawKey = rawKey(key); final byte[] rawValue = rawValue(value); - final long rawTimeout = TimeoutUtils.toSeconds(timeout, unit); execute(new RedisCallback() { public Object doInRedis(RedisConnection connection) throws DataAccessException { - connection.setEx(rawKey, rawTimeout, rawValue); + + potentiallyUsePsetEx(connection); return null; } + + public void potentiallyUsePsetEx(RedisConnection connection) { + + if (!TimeUnit.MILLISECONDS.equals(unit) || !failsafeInvokePsetEx(connection)) { + connection.setEx(rawKey, TimeoutUtils.toSeconds(timeout, unit), rawValue); + } + } + + private boolean failsafeInvokePsetEx(RedisConnection connection) { + + boolean failed = false; + try { + connection.pSetEx(rawKey, timeout, rawValue); + } catch (UnsupportedOperationException e) { + // in case the connection does not support pSetEx return false to allow fallback to other operation. + failed = true; + } + return !failed; + } + }, true); } diff --git a/src/main/java/org/springframework/data/redis/core/ValueOperations.java b/src/main/java/org/springframework/data/redis/core/ValueOperations.java index 2cbe1a5dc..41570ecef 100644 --- a/src/main/java/org/springframework/data/redis/core/ValueOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ValueOperations.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2013 the original author or authors. + * Copyright 2011-2014 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. @@ -24,11 +24,21 @@ import java.util.concurrent.TimeUnit; * Redis operations for simple (or in Redis terminology 'string') values. * * @author Costin Leau + * @author Christoph Strobl */ public interface ValueOperations { void set(K key, V value); + /** + * Set {@code key} to hold the string {@code value} until {@code timeout}. + * + * @param key + * @param value + * @param timeout + * @param units + * @see http://redis.io/commands/set + */ void set(K key, V value, long timeout, TimeUnit unit); Boolean setIfAbsent(K key, V value); diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 4a0943f3b..6fd8135e5 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -66,6 +66,7 @@ import org.springframework.test.annotation.ProfileValueSourceConfiguration; * @author Costin Leau * @author Jennifer Hickey * @author Christoph Strobl + * @author Thomas Darimont */ @ProfileValueSourceConfiguration(RedisTestProfileValueSource.class) public abstract class AbstractConnectionIntegrationTests { @@ -359,6 +360,20 @@ public abstract class AbstractConnectionIntegrationTests { assertTrue(waitFor(new KeyExpired("expy"), 2500l)); } + /** + * @see DATAREDIS-271 + */ + @Test + @IfProfileValue(name = "runLongTests", value = "true") + public void testPsetEx() throws Exception { + + connection.pSetEx("expy", 500L, "yep"); + actual.add(connection.get("expy")); + + verifyResults(Arrays.asList(new Object[] { "yep" })); + assertTrue(waitFor(new KeyExpired("expy"), 2500L)); + } + @Test @IfProfileValue(name = "runLongTests", value = "true") public void testBRPopTimeout() throws Exception { diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java index ea1a3d3e5..816a0594c 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013 the original author or authors. + * Copyright 2013-2014 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. @@ -45,6 +45,7 @@ import org.springframework.data.redis.serializer.StringRedisSerializer; * Unit test of {@link DefaultStringRedisConnection} * * @author Jennifer Hickey + * @auhtor Christoph Strobl */ public class DefaultStringRedisConnectionTests { @@ -907,6 +908,16 @@ public class DefaultStringRedisConnectionTests { verifyResults(Arrays.asList(new Object[] { true })); } + /** + * @see DATAREDIS-271 + */ + @Test + public void testPSetExShouldDelegateCallToNativeConnection() { + + connection.pSetEx(fooBytes, 10L, barBytes); + verify(nativeConnection, times(1)).pSetEx(eq(fooBytes), eq(10L), eq(barBytes)); + } + @Test public void testSInterBytes() { doReturn(bytesSet).when(nativeConnection).sInter(fooBytes, barBytes); diff --git a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java index e7326aa33..88806fe5d 100644 --- a/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java @@ -801,4 +801,12 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat IsCollectionContaining.hasItems("awesome".getBytes(), "cool".getBytes(), "supercalifragilisticexpialidocious".getBytes())); } + + /** + * @see DATAREDIS-271 + */ + @Test(expected = UnsupportedOperationException.class) + public void testPsetEx() throws Exception { + super.testPsetEx(); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index eeaf4d218..5ef58a80c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -287,6 +287,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra @SuppressWarnings("unchecked") @Test public void testExecuteShouldConvertArrayReplyCorrectly() { + connection.set("spring", "awesome"); connection.set("data", "cool"); connection.set("redis", "supercalifragilisticexpialidocious"); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java index 88e03211f..58a6943e3 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java @@ -112,4 +112,5 @@ public class LettuceConnectionPipelineIntegrationTests extends AbstractConnectio factory2.destroy(); } } + } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxIntegrationTests.java index 294a3b522..59aa57d2f 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxIntegrationTests.java @@ -82,4 +82,5 @@ public class LettuceConnectionPipelineTxIntegrationTests extends LettuceConnecti // 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/lettuce/LettuceConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java index 1c48f12aa..56b460f69 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 @@ -75,4 +75,5 @@ public class LettuceConnectionTransactionIntegrationTests extends AbstractConnec public void testSelect() { super.testSelect(); } + } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsTests.java index ce25f621a..d0dc5a530 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013 the original author or authors. + * Copyright 2013-2014 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. @@ -44,6 +44,7 @@ import org.springframework.data.redis.connection.RedisConnection; * Integration test of {@link DefaultValueOperations} * * @author Jennifer Hickey + * @author Christoph Strobl */ @RunWith(Parameterized.class) public class DefaultValueOperationsTests { @@ -173,7 +174,23 @@ public class DefaultValueOperationsTests { @Test public void testSetWithExpiration() { - // 1 ms timeout gets upgraded to 1 sec timeout at the moment + assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true")); + final K key1 = keyFactory.instance(); + V value1 = valueFactory.instance(); + valueOps.set(key1, value1, 1, TimeUnit.SECONDS); + waitFor(new TestCondition() { + public boolean passes() { + return (!redisTemplate.hasKey(key1)); + } + }, 1000); + } + + /** + * @see DATAREDIS-271 + */ + @Test + public void testSetWithExpirationWithTimeUnitMilliseconds() { + assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true")); final K key1 = keyFactory.instance(); V value1 = valueFactory.instance(); @@ -182,7 +199,7 @@ public class DefaultValueOperationsTests { public boolean passes() { return (!redisTemplate.hasKey(key1)); } - }, 1000); + }, 500); } @Test