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.

Author: christophstrobl

docs/src/reference/docbook/appendix/appendix-command-reference.xml

@@ -94,7 +94,7 @@
 				<row><entry><code>PEXIPRE</code></entry><entry>X</entry></row>
 				<row><entry><code>PEXPIREAT</code></entry><entry>X</entry></row>
 				<row><entry><code>PING</code></entry><entry>X</entry></row>
-				<row><entry><code>PSETEX</code></entry><entry>-</entry></row>
+				<row><entry><code>PSETEX</code></entry><entry>X</entry></row>
 				<row><entry><code>PSUBSCRIBE</code></entry><entry>X</entry></row>
 				<row><entry><code>PTTL</code></entry><entry>X</entry></row>
 				<row><entry><code>PUBLISH</code></entry><entry>X</entry></row>

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -735,6 +735,15 @@ public void setEx(byte[] key, long seconds, byte[] value) {
 		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 void setEx(String key, long seconds, String value) {
 		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()) {

src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java

@@ -87,6 +87,17 @@ public enum BitOperation {
 	 */
 	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}.
 	 * 

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 @@
  * 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 StringTuple extends Tuple {
 
 	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<String, String> tuple);
 
 	Boolean mSetNXString(Map<String, String> tuple);

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -331,6 +331,10 @@ private List<Object> convertPipelineResults() {
 		return results;
 	}
 
+	private void doPipelined(Response<?> response) {
+		pipeline(new JedisStatusResult(response));
+	}
+
 	private void pipeline(FutureResult<Response<?>> result) {
 		if (isQueueing()) {
 			transaction(result);
@@ -339,6 +343,10 @@ private void pipeline(FutureResult<Response<?>> result) {
 		}
 	}
 
+	private void doQueued(Response<?> response) {
+		transaction(new JedisStatusResult(response));
+	}
+
 	private void transaction(FutureResult<Response<?>> result) {
 		txResults.add(result);
 	}
@@ -1141,6 +1149,28 @@ public void setEx(byte[] key, long time, byte[] value) {
 		}
 	}
 
+	/*
+	 * (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()) {

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java

@@ -495,6 +495,15 @@ public void setEx(byte[] key, long seconds, byte[] value) {
 		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);

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -56,6 +56,7 @@
 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 void setEx(byte[] key, long time, byte[] value) {
 		}
 	}
 
+	/**
+	 * {@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()) {

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

@@ -900,6 +900,24 @@ public void setEx(byte[] key, long time, byte[] value) {
 		}
 	}
 
+	/*
+	 * (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 @@ private void checkSubscription() {
 		}
 	}
 
+	@SuppressWarnings("rawtypes")
+	private void doPipelined(ListenableFuture<Reply> 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()) {

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 @@
  * 
  * @author Costin Leau
  * @author Jennifer Hickey
+ * @author Christoph Strobl
  */
 class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements ValueOperations<K, V> {
 
@@ -173,17 +174,37 @@ protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
 		}, 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<Object>() {
 
 			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);
 	}
 

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 @@
  * Redis operations for simple (or in Redis terminology 'string') values.
  * 
  * @author Costin Leau
+ * @author Christoph Strobl
  */
 public interface ValueOperations<K, V> {
 
 	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);

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -66,6 +66,7 @@
  * @author Costin Leau
  * @author Jennifer Hickey
  * @author Christoph Strobl
+ * @author Thomas Darimont
  */
 @ProfileValueSourceConfiguration(RedisTestProfileValueSource.class)
 public abstract class AbstractConnectionIntegrationTests {
@@ -359,6 +360,20 @@ public void testSetEx() throws Exception {
 		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 {

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 @@
  * Unit test of {@link DefaultStringRedisConnection}
  * 
  * @author Jennifer Hickey
+ * @auhtor Christoph Strobl
  */
 public class DefaultStringRedisConnectionTests {
 
@@ -907,6 +908,16 @@ public void testSetNX() {
 		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);