add use case for subscription on key events

Author: drm

src/nl/melp/redis/Redis.java

@@ -78,7 +78,7 @@ void write(List<?> list) throws IOException, IllegalArgumentException {
 
 			for (Object o : list) {
 				if (o instanceof byte[]) {
-					write((byte[])o);
+					write((byte[]) o);
 				} else if (o instanceof String) {
 					write(((String) o).getBytes());
 				} else if (o instanceof Long) {
@@ -136,12 +136,12 @@ static class ServerError extends IOException {
 
 		/**
 		 * Parse incoming data from the stream.
-		 *
+		 * <p>
 		 * Based on each of the markers which will identify the type of data being sent, the parsing
 		 * is delegated to the type-specific methods.
 		 *
 		 * @return The parsed object
-		 * @throws IOException Propagated from the stream
+		 * @throws IOException       Propagated from the stream
 		 * @throws ProtocolException In case unexpected bytes are encountered.
 		 */
 		Object parse() throws IOException, ProtocolException {
@@ -267,8 +267,8 @@ public Redis(Socket socket) throws IOException {
 	/**
 	 * Construct the connection with the specified Socket as the server connection with specified buffer sizes.
 	 *
-	 * @param socket Socket to connect to
-	 * @param inputBufferSize buffer size in bytes for the input stream
+	 * @param socket           Socket to connect to
+	 * @param inputBufferSize  buffer size in bytes for the input stream
 	 * @param outputBufferSize buffer size in bytes for the output stream
 	 * @throws IOException If a socket error occurs.
 	 */
@@ -282,7 +282,7 @@ public Redis(Socket socket, int inputBufferSize, int outputBufferSize) throws IO
 	/**
 	 * Construct with the specified streams to respectively read from and write to.
 	 *
-	 * @param inputStream Read from this stream
+	 * @param inputStream  Read from this stream
 	 * @param outputStream Write to this stream
 	 */
 	public Redis(InputStream inputStream, OutputStream outputStream) {
@@ -294,14 +294,24 @@ public Redis(InputStream inputStream, OutputStream outputStream) {
 	 * Execute a Redis command and return it's result.
 	 *
 	 * @param args Command and arguments to pass into redis.
-	 * @param <T> The expected result type
+	 * @param <T>  The expected result type
 	 * @return Result of redis.
-	 *
 	 * @throws IOException All protocol and io errors are IO exceptions.
 	 */
 	public <T> T call(Object... args) throws IOException {
 		writer.write(Arrays.asList((Object[]) args));
 		writer.flush();
+		return read();
+	}
+
+	/**
+	 * Does a blocking read to wait for redis to send data.
+	 *
+	 * @param <T> The expected result type.
+	 * @return Result of redis
+	 * @throws IOException Propagated
+	 */
+	public <T> T read() throws IOException {
 		return (T) reader.parse();
 	}
 
@@ -363,13 +373,17 @@ public interface FailableConsumer<T, E extends Throwable> {
 	 * Utility method to execute some command with redis and close the connection directly after.
 	 *
 	 * @param callback The callback to perform with redis.
-	 * @param addr Connection IP address
-	 * @param port Connection port
+	 * @param addr     Connection IP address
+	 * @param port     Connection port
 	 * @throws IOException Propagated
 	 */
 	public static void run(FailableConsumer<Redis, IOException> callback, String addr, int port) throws IOException {
 		try (Socket s = new Socket(addr, port)) {
-			callback.accept(new Redis(s));
+			run(callback, s);
 		}
 	}
+
+	public static void run(FailableConsumer<Redis, IOException> callback, Socket s) throws IOException {
+		callback.accept(new Redis(s));
+	}
 }

test/nl/melp/redis/RedisTest.java

@@ -3,12 +3,10 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketTimeoutException;
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoUnit;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -28,6 +26,7 @@ public static void main(String[] args) throws IOException, InterruptedException
 		bufferSizePerformanceTest(16, 16);
 		socketManagementPerformanceTest();
 		binaryTest();
+		subscribeTest();
 	}
 
 	private static void assertEqual(String a, String b) {
@@ -195,6 +194,18 @@ private static void integrationTest() throws InterruptedException {
 		r.shutdown();
 		r.awaitTermination(2, TimeUnit.SECONDS);
 		assertTrue(b.get());
+
+		exec.accept((redis) -> {
+			redis.call("SET", "foo", "val");
+			redis.call("PEXPIRE", "foo", "100");
+			try {
+				Thread.sleep(101);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+				throw new RuntimeException(e);
+			}
+			assertTrue(null == redis.call("GET", "foo"));
+		});
 	}
 
 	private static void bufferSizePerformanceTest(int numBitsFrom, int numBitsTo) throws IOException, InterruptedException {
@@ -344,6 +355,85 @@ public static void binaryTest() throws IOException {
 	}
 
 
+	public static void subscribeTest() throws IOException, InterruptedException {
+		List<String> events =  Collections.synchronizedList(new LinkedList<>());
+
+		ExecutorService s = Executors.newFixedThreadPool(2);
+		Redis.run(
+			redis -> {
+				redis.call("CONFIG", "SET", "notify-keyspace-events", "AKE");
+			}, "localhost", 6379
+		);
+		AtomicInteger n = new AtomicInteger(0);
+
+		s.submit(() -> {
+			try {
+				System.out.println("Subscription starting");
+				Socket timeoutSocket = new Socket("localhost", 6379);
+				timeoutSocket.setSoTimeout(1500);
+				try {
+					Redis.run(redis -> {
+						redis.call("PSUBSCRIBE", "__keyevent@0__:*", "*");
+						LinkedList<Object> result;
+						try {
+							while ((result = redis.read()) != null) {
+								try {
+									System.out.println("Received event " + String.join(", ", result.stream().map(r -> new String((byte[])r)).toArray(String[]::new)));
+								} catch (ClassCastException e) {
+									continue;
+								}
+								if (
+									result.get(0) instanceof byte[] && Arrays.equals((byte[])result.get(0), "pmessage".getBytes())
+									&& result.get(1) instanceof byte[] && Arrays.equals((byte[])result.get(1), "__keyevent@0__:*".getBytes())
+								) {
+									events.add(new String((byte[])result.get(2)).replace("__keyevent@0__:", "") + ":" + new String((byte[])result.get(3)));
+								}
+							}
+						} catch (SocketTimeoutException ignored) {
+						} catch (Exception e) {
+							e.printStackTrace();
+						}
+					}, timeoutSocket);
+				} catch (SocketTimeoutException ignored) {
+				}
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+			n.incrementAndGet();
+		});
+		Thread.sleep(1000);
+
+		s.submit(() -> {
+			try {
+				Redis.run(redis -> {
+					redis.call("SET", "a", "b");
+					redis.call("PEXPIRE", "a", "100");
+					try {
+						// so the key will expire
+						Thread.sleep(500);
+					} catch (InterruptedException e) {
+						e.printStackTrace();
+					}
+				}, "localhost", 6379);
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+			n.incrementAndGet();
+		});
+
+		while (n.get() < 2) {
+			System.out.println("Waiting for both threads to finish");
+			Thread.sleep(100);
+		}
+		s.shutdown();
+		s.awaitTermination(1000, TimeUnit.MILLISECONDS);
+
+		assertTrue(events.get(0).equals("set:a"));
+		assertTrue(events.get(1).equals("expire:a"));
+		assertTrue(events.get(2).equals("expired:a"));
+	}
+
+
 	private static String msg = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Suspendisse in purus in dui cursus dignissim id at neque. Duis porta ullamcorper aliquam. Suspendisse hendrerit urna id felis aliquet rutrum. Fusce ultricies magna elit, id volutpat risus dictum et. Sed pretium elementum arcu, vitae aliquet ligula. Phasellus viverra vel arcu vel dictum. Fusce ac purus fringilla neque dapibus sollicitudin sit amet et felis. Nulla gravida fringilla ex sit amet faucibus. Etiam sit amet nisl id est dictum porttitor eget nec risus. Vivamus et ultrices arcu, vitae accumsan lectus. Phasellus tempus tortor lectus, vitae consequat enim dictum auctor. Ut elementum sapien eu diam tempus condimentum.\n" +
 		"\n" +
 		"Vestibulum ultricies bibendum arcu ut commodo. Morbi tristique dui quis commodo consectetur. Praesent venenatis augue justo, sed placerat lectus aliquam eget. Duis malesuada lobortis quam id congue. Fusce mollis faucibus arcu. Aliquam consectetur leo eu luctus accumsan. Nulla nec diam non ex eleifend fringilla sit amet et ante. Mauris posuere est ut turpis pellentesque hendrerit.\n" +