Add Redis 2.6 millisecond expiration commands

DATAREDIS-178

Add pExpire, pExpireAt, pTTl ops to connections

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

@@ -1166,4 +1166,28 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 	public Object execute(String command, String... args) {
 		return execute(command, serializeMulti(args));
 	}
+
+	public Boolean pExpire(byte[] key, long millis) {
+		return delegate.pExpire(key, millis);
+	}
+
+	public Boolean pExpireAt(byte[] key, long unixTimeInMillis) {
+		return delegate.pExpireAt(key, unixTimeInMillis);
+	}
+
+	public Long pTtl(byte[] key) {
+		return delegate.pTtl(key);
+	}
+
+	public Boolean pExpire(String key, long millis) {
+		return pExpire(serialize(key), millis);
+	}
+
+	public Boolean pExpireAt(String key, long unixTimeInMillis) {
+		return pExpireAt(serialize(key), unixTimeInMillis);
+	}
+
+	public Long pTtl(String key) {
+		return pTtl(serialize(key));
+	}
 }

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

@@ -42,14 +42,20 @@ public interface RedisKeyCommands {
 
 	Boolean expire(byte[] key, long seconds);
 
+	Boolean pExpire(byte[] key, long millis);
+
 	Boolean expireAt(byte[] key, long unixTime);
 
+	Boolean pExpireAt(byte[] key, long unixTimeInMillis);
+
 	Boolean persist(byte[] key);
 
 	Boolean move(byte[] key, int dbIndex);
 
 	Long ttl(byte[] key);
 
+	Long pTtl(byte[] key);
+
 	// sort commands
 	List<byte[]> sort(byte[] key, SortParameters params);
 

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

@@ -60,14 +60,20 @@ public interface StringRedisConnection extends RedisConnection {
 
 	Boolean expire(String key, long seconds);
 
+	Boolean pExpire(String key, long millis);
+
 	Boolean expireAt(String key, long unixTime);
 
+	Boolean pExpireAt(String key, long unixTimeInMillis);
+
 	Boolean persist(String key);
 
 	Boolean move(String key, int dbIndex);
 
 	Long ttl(String key);
 
+	Long pTtl(String key);
+
 	String echo(String message);
 
 	// sort commands

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

@@ -811,6 +811,17 @@ public class JedisConnection implements RedisConnection {
 		}
 	}
 
+	public Boolean pExpire(byte[] key, long millis) {
+		throw new UnsupportedOperationException();
+	}
+
+	public Boolean pExpireAt(byte[] key, long unixTimeInMillis) {
+		throw new UnsupportedOperationException();
+	}
+
+	public Long pTtl(byte[] key) {
+		throw new UnsupportedOperationException();
+	}
 
 	public DataType type(byte[] key) {
 		try {

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

@@ -343,6 +343,17 @@ public class JredisConnection implements RedisConnection {
 		}
 	}
 
+	public Boolean pExpire(byte[] key, long millis) {
+		throw new UnsupportedOperationException();
+	}
+
+	public Boolean pExpireAt(byte[] key, long unixTimeInMillis) {
+		throw new UnsupportedOperationException();
+	}
+
+	public Long pTtl(byte[] key) {
+		throw new UnsupportedOperationException();
+	}
 
 	public Set<byte[]> keys(byte[] pattern) {
 		try {

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

@@ -555,6 +555,41 @@ public class LettuceConnection implements RedisConnection {
 		}
 	}
 
+	public Boolean pExpire(byte[] key, long millis) {
+		try {
+			if (isPipelined()) {
+				pipeline(getAsyncConnection().pexpire(key, millis));
+				return null;
+			}
+			return getConnection().pexpire(key, millis);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
+	public Boolean pExpireAt(byte[] key, long unixTimeInMillis) {
+		try {
+			if (isPipelined()) {
+				pipeline(getAsyncConnection().pexpireat(key, unixTimeInMillis));
+				return null;
+			}
+			return getConnection().pexpireat(key, unixTimeInMillis);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
+	public Long pTtl(byte[] key) {
+		try {
+			if (isPipelined()) {
+				pipeline(getAsyncConnection().pttl(key));
+				return null;
+			}
+			return getConnection().pttl(key);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
 
 	public Set<byte[]> keys(byte[] pattern) {
 		try {

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

@@ -479,6 +479,29 @@ public class SrpConnection implements RedisConnection {
 		}
 	}
 
+	public Boolean pExpire(byte[] key, long millis) {
+		try {
+			if (isPipelined()) {
+				pipeline(pipeline.pexpire(key, millis));
+				return null;
+			}
+			return SrpUtils.asBoolean(client.pexpire(key, millis));
+		} catch (Exception ex) {
+			throw convertSrpAccessException(ex);
+		}
+	}
+
+	public Boolean pExpireAt(byte[] key, long unixTimeInMillis) {
+		try {
+			if (isPipelined()) {
+				pipeline(pipeline.pexpireat(key, unixTimeInMillis));
+				return null;
+			}
+			return SrpUtils.asBoolean(client.pexpireat(key, unixTimeInMillis));
+		} catch (Exception ex) {
+			throw convertSrpAccessException(ex);
+		}
+	}
 
 	public Set<byte[]> keys(byte[] pattern) {
 		try {
@@ -596,6 +619,17 @@ public class SrpConnection implements RedisConnection {
 		}
 	}
 
+	public Long pTtl(byte[] key) {
+		try {
+			if (isPipelined()) {
+				pipeline(pipeline.pttl(key));
+				return null;
+			}
+			return client.pttl(key).data();
+		} catch (Exception ex) {
+			throw convertSrpAccessException(ex);
+		}
+	}
 
 	public DataType type(byte[] key) {
 		try {

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

@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
 
 import redis.client.RedisException;
 import redis.reply.BulkReply;
+import redis.reply.IntegerReply;
 import redis.reply.MultiBulkReply;
 import redis.reply.Reply;
 
@@ -175,6 +176,13 @@ abstract class SrpUtils {
 		return map;
 	}
 
+	static Boolean asBoolean(IntegerReply reply) {
+		if (reply == null) {
+			return null;
+		}
+		return (Long.valueOf(1).equals(reply.data()));
+	}
+
 	static byte[] limit(long offset, long count) {
 		return ("LIMIT " + offset + " " + count).getBytes(Charsets.UTF_8);
 	}

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

@@ -114,6 +114,30 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertTrue(waitFor(new KeyExpired("exp2"), 3000l));
 	}
 
+	@Test
+	public void testPExpire() {
+		connection.set("exp", "true");
+		assertTrue(connection.pExpire("exp", 100));
+		assertTrue(waitFor(new KeyExpired("exp"), 300l));
+	}
+
+	@Test
+	public void testPExpireKeyNotExists() {
+		assertFalse(connection.pExpire("nonexistent", 100));
+	}
+
+	@Test
+	public void testPExpireAt() {
+		connection.set("exp", "true");
+		assertTrue(connection.pExpireAt("exp", System.currentTimeMillis() + 200));
+		assertTrue(waitFor(new KeyExpired("exp"), 600l));
+	}
+
+	@Test
+	public void testPExpireAtKeyNotExists() {
+		assertFalse(connection.pExpireAt("nonexistent", System.currentTimeMillis() + 200));
+	}
+
 	@Test
 	@IfProfileValue(name = "runLongTests", value = "true")
 	public void testPersist() throws Exception {
@@ -517,6 +541,20 @@ public abstract class AbstractConnectionIntegrationTests {
 		verifyResults(Arrays.asList(new Object[] { -1L }), actual);
 	}
 
+	@Test
+	public void testPTtlNoExpire() {
+		connection.set("whatup", "yo");
+		actual.add(connection.pTtl("whatup"));
+		verifyResults(Arrays.asList(new Object[] { -1L }), actual);
+	}
+
+	@Test
+	public void testPTtl() {
+		connection.set("whatup", "yo");
+		connection.pExpire("whatup", 9000l);
+		assertTrue(connection.pTtl("whatup") > -1);
+	}
+
 	@Test
 	public void testType() {
 		connection.set("something", "yo");

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

@@ -115,6 +115,46 @@ abstract public class AbstractConnectionPipelineIntegrationTests extends
 		verifyResults(Arrays.asList(new Object[] { "bar" }), actual);
 	}
 
+	@Test
+	public void testPExpire() {
+		connection.set("exp", "true");
+		actual.add(connection.pExpire("exp", 100));
+		verifyResults(Arrays.asList(new Object[] { 1l }), actual);
+		assertTrue(waitFor(new KeyExpired("exp"), 1000l));
+	}
+
+	@Test
+	public void testPExpireKeyNotExists() {
+		actual.add(connection.pExpire("nonexistent", 100));
+		verifyResults(Arrays.asList(new Object[] { 0l }), actual);
+	}
+
+	@Test
+	public void testPExpireAt() {
+		connection.set("exp2", "true");
+		actual.add(connection.pExpireAt("exp2", System.currentTimeMillis() + 200));
+		verifyResults(Arrays.asList(new Object[] { 1l }), actual);
+		assertTrue(waitFor(new KeyExpired("exp2"), 1000l));
+	}
+
+	@Test
+	public void testPExpireAtKeyNotExists() {
+		actual.add(connection.pExpireAt("nonexistent", System.currentTimeMillis() + 200));
+		verifyResults(Arrays.asList(new Object[] { 0l }), actual);
+	}
+
+	@Test
+	public void testPTtl() {
+		connection.set("whatup", "yo");
+		actual.add(connection.pExpire("whatup", 9000l));
+		verifyResults(Arrays.asList(new Object[] { 1l }), actual);
+		assertTrue(waitFor(new TestCondition() {
+			public boolean passes() {
+				return (connection.pTtl("whatup") > -1);
+			}
+		}, 1000l));
+	}
+
 	@Test
 	@IfProfileValue(name = "runLongTests", value = "true")
 	public void testExpire() throws Exception {

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionIntegrationTests.java

@@ -50,6 +50,36 @@ public class JedisConnectionIntegrationTests extends AbstractConnectionIntegrati
 		connection = null;
 	}
 
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpire() {
+		super.testPExpire();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireKeyNotExists() {
+		super.testPExpireKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAt() {
+		super.testPExpireAt();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAtKeyNotExists() {
+		super.testPExpireAtKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtl() {
+		super.testPTtl();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtlNoExpire() {
+		super.testPTtlNoExpire();
+	}
+
 	@Test
 	public void testIncrDecrByLong() {
 		String key = "test.count";

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionPipelineIntegrationTests.java

@@ -92,6 +92,36 @@ public class JedisConnectionPipelineIntegrationTests extends
 	}
 
 	// Unsupported Ops
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpire() {
+		super.testPExpire();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireKeyNotExists() {
+		super.testPExpireKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAt() {
+		super.testPExpireAt();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAtKeyNotExists() {
+		super.testPExpireAtKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtl() {
+		super.testPTtl();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtlNoExpire() {
+		super.testPTtlNoExpire();
+	}
+
 	@Test(expected = RedisSystemException.class)
 	public void testBitSet() throws Exception {
 		super.testBitSet();

src/test/java/org/springframework/data/redis/connection/jredis/JRedisConnectionIntegrationTests.java

@@ -288,6 +288,36 @@ public class JRedisConnectionIntegrationTests extends AbstractConnectionIntegrat
 		super.testZRevRangeByScoreWithScoresOffsetCount();
 	}
 
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpire() {
+		super.testPExpire();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireKeyNotExists() {
+		super.testPExpireKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAt() {
+		super.testPExpireAt();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPExpireAtKeyNotExists() {
+		super.testPExpireAtKeyNotExists();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtl() {
+		super.testPTtl();
+	}
+
+	@Test(expected=UnsupportedOperationException.class)
+	public void testPTtlNoExpire() {
+		super.testPTtlNoExpire();
+	}
+
 	// Jredis returns null for rPush
 	@Test
 	public void testSort() {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java

@@ -16,7 +16,6 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
@@ -292,6 +291,46 @@ public class LettuceConnectionPipelineIntegrationTests extends
 		assertTrue(waitFor(new KeyExpired("exp"), 2500));
 	}
 
+	@Test
+	public void testPExpire() {
+		connection.set("exp", "true");
+		actual.add(connection.pExpire("exp", 100));
+		verifyResults(Arrays.asList(new Object[] { true }), actual);
+		assertTrue(waitFor(new KeyExpired("exp"), 1000l));
+	}
+
+	@Test
+	public void testPExpireKeyNotExists() {
+		actual.add(connection.pExpire("nonexistent", 100));
+		verifyResults(Arrays.asList(new Object[] { false }), actual);
+	}
+
+	@Test
+	public void testPExpireAt() {
+		connection.set("exp2", "true");
+		actual.add(connection.pExpireAt("exp2", System.currentTimeMillis() + 200));
+		verifyResults(Arrays.asList(new Object[] { true }), actual);
+		assertTrue(waitFor(new KeyExpired("exp2"), 1000l));
+	}
+
+	@Test
+	public void testPExpireAtKeyNotExists() {
+		actual.add(connection.pExpireAt("nonexistent", System.currentTimeMillis() + 200));
+		verifyResults(Arrays.asList(new Object[] { false }), actual);
+	}
+
+	@Test
+	public void testPTtl() {
+		connection.set("whatup", "yo");
+		actual.add(connection.pExpire("whatup", 9000l));
+		verifyResults(Arrays.asList(new Object[] { true }), actual);
+		assertTrue(waitFor(new TestCondition() {
+			public boolean passes() {
+				return (connection.pTtl("whatup") > -1);
+			}
+		}, 1000l));
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	protected Object convertResult(Object result) {
 		Object convertedResult = super.convertResult(result);