Fix NPEs and incorrect results in several ops within a Lettuce transaction

DATAREDIS-164

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

@@ -47,6 +47,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.ScoredValue;
 import com.lambdaworks.redis.SortArgs;
 import com.lambdaworks.redis.ZStoreArgs;
 import com.lambdaworks.redis.codec.RedisCodec;
@@ -515,7 +516,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.keys(pattern));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.keys(pattern));
+			final List<byte[]> results = con.keys(pattern);
+			return results != null ? new LinkedHashSet<byte[]>(results) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -633,7 +635,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.type(key));
 				return null;
 			}
-			return DataType.fromCode(con.type(key));
+			final String type = con.type(key);
+			return type != null ? DataType.fromCode(type) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -853,7 +856,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.getbit(key, offset));
 				return null;
 			}
-			return Long.valueOf(1).equals(con.getbit(key, offset));
+			final Long result = con.getbit(key, offset);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1121,7 +1125,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.sadd(key, value));
 				return null;
 			}
-			return con.sadd(key, value) == 1;
+			final Long result = con.sadd(key, value);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1259,7 +1264,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.srem(key, value));
 				return null;
 			}
-			return Long.valueOf(1).equals(con.srem(key, value));
+			final Long result = con.srem(key, value);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1301,7 +1307,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zadd(key, score, value));
 				return null;
 			}
-			return Long.valueOf(1).equals(con.zadd(key, score, value));
+			final Long result = con.zadd(key, score, value);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1376,7 +1383,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zrange(key, start, end));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.zrange(key, start, end));
+			final List<byte[]> results = con.zrange(key, start, end);
+			return results != null ? new LinkedHashSet<byte[]>(results) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1402,7 +1410,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zrangebyscore(key, min, max));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.zrangebyscore(key, min, max));
+			final List<byte[]> results = con.zrangebyscore(key, min, max);
+			return results != null ? new LinkedHashSet<byte[]>(results) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1441,7 +1450,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zrangebyscore(key, min, max, offset, count));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.zrangebyscore(key, min, max, offset, count));
+			final List<byte[]> results = con.zrangebyscore(key, min, max, offset, count);
+			return results != null ? new LinkedHashSet<byte[]>(results) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1531,7 +1541,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zrem(key, value));
 				return null;
 			}
-			return Long.valueOf(1).equals(con.zrem(key, value));
+			final Long result = con.zrem(key, value);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1568,7 +1579,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.zrevrange(key, start, end));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.zrevrange(key, start, end));
+			final List<byte[]> results = con.zrevrange(key, start, end);
+			return results != null ? new LinkedHashSet<byte[]>(results) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1663,7 +1675,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.hdel(key, field));
 				return null;
 			}
-			return Long.valueOf(1).equals(con.hdel(key, field));
+			final Long result = con.hdel(key, field);
+			return result != null ? result == 1 : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
@@ -1726,7 +1739,8 @@ public class LettuceConnection implements RedisConnection {
 				pipeline(asyncConn.hkeys(key));
 				return null;
 			}
-			return new LinkedHashSet<byte[]>(con.hkeys(key));
+			final List<byte[]> result = con.hkeys(key);
+			return result != null ? new LinkedHashSet<byte[]>(result) : null;
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}

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

@@ -62,6 +62,9 @@ abstract class LettuceUtils {
 	}
 
 	static Properties info(String reply) {
+		if(reply == null) {
+			return null;
+		}
 		Properties info = new Properties();
 		StringReader stringReader = new StringReader(reply);
 		try {
@@ -84,6 +87,9 @@ abstract class LettuceUtils {
 	}
 
 	static Set<Tuple> convertTuple(List<ScoredValue<byte[]>> zrange) {
+		if(zrange == null) {
+			return null;
+		}
 		Set<Tuple> tuples = new LinkedHashSet<Tuple>(zrange.size());
 
 		for (int i = 0; i < zrange.size(); i++) {

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

@@ -0,0 +1,43 @@
+package org.springframework.data.redis.connection.lettuce;
+
+import static org.junit.Assert.assertNull;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.data.redis.connection.RedisPipelineException;
+
+/**
+ * Integration test of {@link LettuceConnection} transactions within a pipeline
+ *
+ * @author Jennifer Hickey
+ *
+ */
+public class LettuceConnectionPipelineTxIntegrationTests extends
+		LettuceConnectionTransactionIntegrationTests {
+
+	@Test(expected = RedisPipelineException.class)
+	public void exceptionExecuteNative() throws Exception {
+		connection.execute("ZadD", getClass() + "#foo\t0.90\titem");
+		getResults();
+	}
+
+	protected void initConnection() {
+		connection.openPipeline();
+		connection.multi();
+	}
+
+	protected List<Object> getResults() {
+		assertNull(connection.exec());
+		List<Object> results = new ArrayList<Object>();
+		List<Object> pipelinedResults = connection.closePipeline();
+		for (Object result : pipelinedResults) {
+			if (!"OK".equals(result) && !("QUEUED").equals(result)) {
+				results.add(result);
+			}
+		}
+		return results;
+	}
+
+}

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

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2011-2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection.lettuce;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * Integration test of {@link LettuceConnection} functionality within a
+ * transaction
+ *
+ * @author Jennifer Hickey
+ *
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration("LettuceConnectionIntegrationTests-context.xml")
+public class LettuceConnectionTransactionIntegrationTests extends
+		LettuceConnectionPipelineIntegrationTests {
+
+	@Ignore
+	public void testMultiDiscard() {
+	}
+
+	@Ignore
+	public void testMultiExec() {
+	}
+
+	@Ignore
+	public void testUnwatch() {
+	}
+
+	@Ignore
+	public void testWatch() {
+	}
+
+	/*
+	 * 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. Also, Lettuce always times out in these scenarios b/c it waits
+	 * for an actual response instead of accepting the null returned by op in tx
+	 * *
+	 */
+
+	@Ignore
+	public void testBLPop() {
+	}
+
+	@Ignore
+	public void testBRPop() {
+	}
+
+	@Ignore
+	public void testBRPopLPush() {
+	}
+
+	@Ignore
+	public void testBLPopTimeout() {
+	}
+
+	@Ignore
+	public void testBRPopTimeout() {
+	}
+
+	@Ignore
+	public void testBRPopLPushTimeout() {
+	}
+
+	@Test
+	public void testDbSize() {
+		connection.set("dbparam", "foo");
+		assertNull(connection.dbSize());
+		List<Object> results = getResults();
+		assertEquals(1, results.size());
+		assertTrue((Long) results.get(0) > 0);
+	}
+
+	@Test
+	public void exceptionExecuteNative() throws Exception {
+		actual.add(connection.execute("ZadD", getClass() + "#foo\t0.90\titem"));
+		// Syntax error on queued commands are swallowed and no results are
+		// returned
+		verifyResults(Arrays.asList(new Object[] {}), actual);
+	}
+
+	protected void initConnection() {
+		connection.multi();
+	}
+
+	protected List<Object> getResults() {
+		List<Object> results = new ArrayList<Object>();
+		List<Object> txResults = connection.exec();
+		for (Object result : txResults) {
+			if (!"OK".equals(result) && !("QUEUED").equals(result)) {
+				results.add(result);
+			}
+		}
+		return results;
+	}
+}