some cleanup

Author: drm

README.md

@@ -32,7 +32,7 @@ System.out.println((String)l.get(0)); // prints "A"
 ```
 
 You will have to do some casting yourself in case of List responses. The
-reasoning here is that you know what data to expect so you're responsible for
+reasoning here is that you know what data to expect, so you're responsible for
 applying the correct casts in the correct context.  
 
 Refer to the Redis documentation for the return types of all calls and
@@ -61,7 +61,7 @@ redis.pipeline()
 This will result in a `List<Object>` containing a list for each of the responses.
 
 ## Is the connection thread safe?
-No. You need to make sure that the connection is accessed atomically. However,
+No. You need to make sure to access the connection atomically. However,
 there is a `Redis.run()` method which you can use to do some simple redis
 operations in isolation. It creates a connection and closes it directly after:
 
@@ -73,11 +73,11 @@ nl.melp.redis.Redis.run((redis) -> redis.call("INCR", "mycounter"));
 However you wish. This library is a protocol implementation only. Managing a
 connection pool is not at all Redis-specific, so it doesn't belong here.
 
-Having said that, this mostly depends on your use case. Typically if you have a
+Having said that, this mostly depends on your use case. Typically, if you have a
 webserver with 20 threads, you can have a socket per thread managed somewhere
-(e.g. ThreadLocal) and if you run a threadpool with 20 workers you can have one
+(e.g. ThreadLocal) and if you run a thread pool with 20 workers you can have one
 socket per thread there as well. This keeps things simple and practical to
-reason about and you don't have to worry who needs to create the socket at what
+reason about, and you don't have to worry who needs to create the socket at what
 point in time. You do need some error handling which, in case of managing the
 sockets centrally is a tiny bit easier. On the other hand, if you expect to be
 creating a lot of threads and not a lot of these threads need the connection,

src/nl/melp/redis/protocol/Encoder.java

@@ -11,15 +11,15 @@ public class Encoder {
 	/**
 	 * CRLF is used a lot.
 	 */
-	private static byte[] CRLF = new byte[]{'\r', '\n'};
+	private static final byte[] CRLF = new byte[]{'\r', '\n'};
 
 	/**
 	 * This stream we will write to.
 	 */
 	private final OutputStream out;
 
 	/**
-	 * Construct the encoder with the passed outputstream the encoder will write to.
+	 * Construct the encoder with the passed output stream the encoder will write to.
 	 *
 	 * @param out Will be used to write all encoded data to.
 	 */

src/nl/melp/redis/protocol/Parser.java

@@ -1,7 +1,5 @@
 package nl.melp.redis.protocol;
 
-import nl.melp.redis.Redis;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
@@ -96,7 +94,7 @@ public Object parse() throws IOException, ProtocolException {
 	 * @return The parsed response
 	 * @throws IOException Propagated from underlying stream.
 	 */
-	private byte[] parseBulkString() throws IOException, ProtocolException {
+	private byte[] parseBulkString() throws IOException {
 		final long expectedLength = parseNumber();
 		if (expectedLength == -1) {
 			return null;
@@ -127,14 +125,26 @@ private byte[] parseBulkString() throws IOException, ProtocolException {
 	 * @throws IOException Propagated from underlying stream.
 	 */
 	private byte[] parseSimpleString() throws IOException {
-		return scanCr(1024);
+		return scanCr();
 	}
 
+	/**
+	 * Parse a number (as long)
+	 * @return The number
+	 * @throws IOException Propagated from underlying stream
+	 */
 	private long parseNumber() throws IOException {
-		return Long.valueOf(new String(scanCr(1024)));
+		return Long.parseLong(new String(scanCr()));
 	}
 
-	private byte[] scanCr(int size) throws IOException {
+	/**
+	 * Scan the input stream for the next CR character
+	 *
+	 * @return Byte array.
+	 * @throws IOException Propagated from underlying stream
+	 */
+	private byte[] scanCr() throws IOException {
+		int size = 1024;
 		int idx = 0;
 		int ch;
 		byte[] buffer = new byte[size];

test/nl/melp/redis/RedisTest.java

@@ -26,16 +26,24 @@
 import java.util.function.Supplier;
 
 public class RedisTest {
+	private static final String REDIS_HOST = System.getProperty("redis.host", "localhost");
+	private static final int REDIS_PORT = Integer.parseInt(System.getProperty("redis.port", "6379"));
+	private static final int numThreads = Integer.parseInt(System.getProperty("redis.test.num-threads", "200"));
+	private static final int numMessages = Integer.parseInt(System.getProperty("redis.test.num-messages", "25000"));
+
+	// 16 bits buffer size (2 ^ 16, 1<<16) seems to be the most efficient for any value size.
+	// tweak these numbers to check if this is the case for you.
+	private static final int numBitsFrom = Integer.parseInt(System.getProperty("redis.test.num-bits-from", "16"));
+	private static final int numBitsTo = Integer.parseInt(System.getProperty("redis.test.num-bits-to", "16"));
+
 	public static void main(String[] args) throws IOException, InterruptedException {
 		if (args.length == 0) {
 			testParse();
+			binaryTest();
 			managedTest();
 			integrationTest();
-			// 16 bits buffer size (2 ^ 16, 1<<16) seems to be the most efficient for any value size.
-			// tweak these numbers to check if this is the case for you.
-			bufferSizePerformanceTest(16, 16);
+			bufferSizePerformanceTest();
 			socketManagementPerformanceTest();
-			binaryTest();
 			subscribeTest();
 
 			System.out.println("\nEverything seems to be alright.");
@@ -77,15 +85,15 @@ private static void testParse() throws IOException {
 			"01234\r\n56789"
 		);
 
-		List arr = (List) new Parser(new ByteArrayInputStream("*5\r\n:1\r\n:2\r\n:3\r\n:4\r\n:5\r\n".getBytes())).parse();
+		List<?> arr = (List<?>) new Parser(new ByteArrayInputStream("*5\r\n:1\r\n:2\r\n:3\r\n:4\r\n:5\r\n".getBytes())).parse();
 		assertEqual(arr.size(), 5);
 		assertEqual((Long) arr.get(0), 1);
 		assertEqual((Long) arr.get(1), 2);
 		assertEqual((Long) arr.get(2), 3);
 		assertEqual((Long) arr.get(3), 4);
 		assertEqual((Long) arr.get(4), 5);
 
-		List arr2 = (List) new Parser(new ByteArrayInputStream("*1\r\n$5\r\n12345\r\n".getBytes())).parse();
+		List<?> arr2 = (List<?>) new Parser(new ByteArrayInputStream("*1\r\n$5\r\n12345\r\n".getBytes())).parse();
 		assertEqual(arr2.size(), 1);
 		assertEqual(new String((byte[])arr2.get(0)), "12345");
 		System.out.println("Tests passed successfully: testParse");
@@ -95,7 +103,7 @@ private static void integrationTest() throws InterruptedException {
 		final String keyName = RedisTest.class.getCanonicalName();
 		Consumer<Redis.FailableConsumer<Redis, IOException>> exec = (consumer) -> {
 			try {
-				Redis.run(consumer, "127.0.0.1", 6379);
+				Redis.run(consumer, "127.0.0.1", REDIS_PORT);
 			} catch (IOException e) {
 				e.printStackTrace();
 				throw new RuntimeException(e);
@@ -142,7 +150,7 @@ private static void integrationTest() throws InterruptedException {
 
 		exec.accept(
 			(redis) -> {
-				redis = new Redis(new Socket("127.0.0.1", 6379));
+				redis = new Redis(new Socket("127.0.0.1", REDIS_PORT));
 				List<Object> result = redis.pipeline()
 					.call("INCR", keyName)
 					.call("INCR", keyName)
@@ -174,18 +182,16 @@ private static void integrationTest() throws InterruptedException {
 				assertEqual("QUEUED", new String((byte[])result.get(2)));
 				assertEqual("QUEUED", new String((byte[])result.get(3)));
 				assertEqual("QUEUED", new String((byte[])result.get(4)));
-				assertEqual(1, (Long) ((List<Object>) result.get(5)).get(0));
-				assertEqual(2, (Long) ((List<Object>) result.get(5)).get(1));
-				assertEqual(3, (Long) ((List<Object>) result.get(5)).get(2));
-				assertEqual(4, (Long) ((List<Object>) result.get(5)).get(3));
+				assertEqual(1, (Long) ((List<?>) result.get(5)).get(0));
+				assertEqual(2, (Long) ((List<?>) result.get(5)).get(1));
+				assertEqual(3, (Long) ((List<?>) result.get(5)).get(2));
+				assertEqual(4, (Long) ((List<?>) result.get(5)).get(3));
 				redis.call("DEL", keyName);
 			}
 		);
 
 		exec.accept(
-			(redis) -> {
-				redis.call("INFO");
-			}
+			(redis) -> redis.call("INFO")
 		);
 
 		// This test checks if `call` will return null from a disconnected connection.
@@ -221,10 +227,7 @@ private static void integrationTest() throws InterruptedException {
 		});
 	}
 
-	private static void bufferSizePerformanceTest(int numBitsFrom, int numBitsTo) throws IOException, InterruptedException {
-		final int numThreads = 200;
-		final int numMessages = 25000;
-
+	private static void bufferSizePerformanceTest() throws IOException, InterruptedException {
 		// Test in- and output buffer sizes.
 		for (Map.Entry<MessageProducerTypeName, Supplier<String>> type : messageProducerTypes.entrySet()) {
 			for (int ab = 0; ab <= 1; ab++) {
@@ -246,9 +249,9 @@ private static void bufferSizePerformanceTest(int numBitsFrom, int numBitsTo) th
 						() -> {
 							try {
 								if (isInputBufferTest) {
-									return new Redis(new Socket("127.0.0.1", 6379), bufSize, 1 << 16);
+									return new Redis(new Socket("127.0.0.1", REDIS_PORT), bufSize, 1 << 16);
 								} else {
-									return new Redis(new Socket("127.0.0.1", 6379), 1 << 16, bufSize);
+									return new Redis(new Socket("127.0.0.1", REDIS_PORT), 1 << 16, bufSize);
 								}
 							} catch (IOException e) {
 								e.printStackTrace();
@@ -280,11 +283,11 @@ private static void socketManagementPerformanceTest() throws IOException, Interr
 					try {
 						if (isThreadLocalTest) {
 							if (redis.get() == null) {
-								redis.set(new Redis(new Socket("127.0.0.1", 6379)));
+								redis.set(new Redis(new Socket("127.0.0.1", REDIS_PORT)));
 							}
 							return redis.get();
 						}
-						return new Redis(new Socket("127.0.0.1", 6379));
+						return new Redis(new Socket("127.0.0.1", REDIS_PORT));
 					} catch (IOException e) {
 						e.printStackTrace();
 					}
@@ -325,7 +328,7 @@ private static void performanceTest(String description, int numThreads, int numM
 					try {
 						Redis redis2 = connector.get();
 						for (int n = 0; n < numMessages / numThreads; n++) {
-							List value = redis2.call("BLPOP", queueKeyName, "0");
+							List<?> value = redis2.call("BLPOP", queueKeyName, "0");
 							assertTrue(value.get(1) instanceof byte[]);
 							size.addAndGet(((byte[])value.get(1)).length);
 							count.incrementAndGet();
@@ -361,10 +364,8 @@ public static void binaryTest() throws IOException {
 
 		Redis.run((redis) -> {
 			redis.call("SET", "foo", bytes);
-			System.out.printf("Byte length: %d", redis.<byte[]>call("GET", "foo").length);
-
-//			assertTrue(redis.<String>call("GET", "foo").getBytes().length == 1024);
-		}, "localhost", 6379);
+			assertEqual(1024, redis.<byte[]>call("GET", "foo").length);
+		}, REDIS_HOST, REDIS_PORT);
 	}
 
 	public static void managedTest() throws IOException {
@@ -374,7 +375,7 @@ public static void managedTest() throws IOException {
 				Redis.run((r) -> {
 					String response = new String(r.<byte[]>call("CLIENT", "LIST", "TYPE", "normal"));
 					numClients.set(response.split("\n").length);
-				}, "localhost", 6379);
+				}, REDIS_HOST, REDIS_PORT);
 			} catch (IOException e) {
 				throw new RuntimeException(e);
 			}
@@ -384,7 +385,7 @@ public static void managedTest() throws IOException {
 		};
 
 		int before = countClients.get();
-		try (Redis.Managed r = Redis.connect("localhost", 6379)) {
+		try (Redis.Managed r = Redis.connect(REDIS_HOST, REDIS_PORT)) {
 			if (countClients.get() <= before) {
 				throw new IllegalStateException("Expected number of connected clients to be less than " + before);
 			}
@@ -400,16 +401,14 @@ public static void subscribeTest() throws IOException, InterruptedException {
 
 		ExecutorService s = Executors.newFixedThreadPool(2);
 		Redis.run(
-			redis -> {
-				redis.call("CONFIG", "SET", "notify-keyspace-events", "AKE");
-			}, "localhost", 6379
+			redis -> redis.call("CONFIG", "SET", "notify-keyspace-events", "AKE"), REDIS_HOST, REDIS_PORT
 		);
 		AtomicInteger n = new AtomicInteger(0);
 
 		s.submit(() -> {
 			try {
 				System.out.println("Subscription starting");
-				Socket timeoutSocket = new Socket("localhost", 6379);
+				Socket timeoutSocket = new Socket(REDIS_HOST, REDIS_PORT);
 				timeoutSocket.setSoTimeout(1500);
 				try {
 					Redis.run(redis -> {
@@ -454,7 +453,7 @@ public static void subscribeTest() throws IOException, InterruptedException {
 					} catch (InterruptedException e) {
 						e.printStackTrace();
 					}
-				}, "localhost", 6379);
+				}, REDIS_HOST, REDIS_PORT);
 			} catch (IOException e) {
 				e.printStackTrace();
 			}
@@ -548,13 +547,11 @@ enum MessageProducerTypeName {
 		final String small = msg.substring(0, 256);
 		final String larger = msg;
 		final AtomicInteger rotate = new AtomicInteger(0);
-		final String large = (((Supplier<String>) () -> {
-			StringBuilder tmp = new StringBuilder();
-			for (int i = 0; i < 10; i++) {
-				tmp.append(msg);
-			}
-			return tmp.toString();
-		}).get());
+		StringBuilder tmp = new StringBuilder();
+		for (int i = 0; i < 10; i++) {
+			tmp.append(msg);
+		}
+		final String large = (tmp.toString());
 
 		strings = new String[]{
 			small, larger, large