DATAREDIS-271 - Add support for 'pSetEx'.

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>

gradle.properties

@@ -4,7 +4,7 @@ jredisVersion=06052013
 jedisVersion=2.4.1
 springVersion=3.2.8.RELEASE
 log4jVersion=1.2.17
-version=1.3.0.BUILD-SNAPSHOT
+version=1.3.0.DATAREDIS-271-SNAPSHOT
 srpVersion=0.7
 jacksonVersion=1.8.8
 fasterXmlJacksonVersion=2.2.0

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 {