Redis db harness

pom.xml

@@ -217,6 +217,13 @@
       <artifactId>jsr305</artifactId>
       <version>2.0.1</version>
     </dependency>
+    <dependency>
+        <groupId>redis.clients</groupId>
+        <artifactId>jedis</artifactId>
+        <version>2.9.0</version>
+        <type>jar</type>
+        <scope>compile</scope>
+    </dependency>
     <!-- Test Deps -->
     <dependency>
       <groupId>junit</groupId>

src/main/java/com/urbanairship/datacube/WriteBuilder.java

@@ -6,13 +6,9 @@
 
 import java.util.Map;
 
-import com.google.common.base.Function;
 import com.google.common.collect.HashMultimap;
-import com.google.common.collect.ImmutableSetMultimap;
-import com.google.common.collect.ListMultimap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
-import com.google.common.collect.Multimaps;
 import com.google.common.collect.SetMultimap;
 
 public class WriteBuilder {

src/main/java/com/urbanairship/datacube/dbharnesses/MapDbHarness.java

@@ -4,19 +4,6 @@
 
 package com.urbanairship.datacube.dbharnesses;
 
-import com.google.common.base.Optional;
-import com.urbanairship.datacube.Address;
-import com.urbanairship.datacube.Batch;
-import com.urbanairship.datacube.BoxedByteArray;
-import com.urbanairship.datacube.DbHarness;
-import com.urbanairship.datacube.Deserializer;
-import com.urbanairship.datacube.IdService;
-import com.urbanairship.datacube.Op;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.lang.NotImplementedException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
@@ -27,6 +14,20 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.lang.NotImplementedException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Optional;
+import com.urbanairship.datacube.Address;
+import com.urbanairship.datacube.Batch;
+import com.urbanairship.datacube.BoxedByteArray;
+import com.urbanairship.datacube.DbHarness;
+import com.urbanairship.datacube.Deserializer;
+import com.urbanairship.datacube.IdService;
+import com.urbanairship.datacube.Op;
+
 /**
  * For testing, this is is a backing store for a cube that lives in memory. It saves us from 
  * calling a DB just to test the cube logic.
@@ -196,7 +197,7 @@ public List<Optional<T>> multiGet(List<Address> addresses) throws IOException {
         throw new NotImplementedException();
     }
 
-    private static class NullFuture implements Future<Object> {
+    public static class NullFuture implements Future<Object> {
         @Override
         public boolean cancel(boolean mayInterruptIfRunning) {
             return false;

src/main/java/com/urbanairship/datacube/dbharnesses/RedisDbHarness.java

@@ -0,0 +1,159 @@
+package com.urbanairship.datacube.dbharnesses;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+import org.apache.commons.lang.NotImplementedException;
+
+import com.google.common.base.Optional;
+import com.urbanairship.datacube.Address;
+import com.urbanairship.datacube.Batch;
+import com.urbanairship.datacube.DbHarness;
+import com.urbanairship.datacube.Deserializer;
+import com.urbanairship.datacube.IdService;
+import com.urbanairship.datacube.Op;
+
+import redis.clients.jedis.Jedis;
+
+
+public class RedisDbHarness<T extends Op> implements DbHarness<T>
+{
+	private static final Future<?> nullFuture = new MapDbHarness.NullFuture();
+
+	private final Jedis jedis;
+	private final int ttlSeconds;
+	private final Deserializer<T> deserializer;
+	private final CommitType commitType;
+	private final IdService idService;
+
+	public RedisDbHarness(Jedis jedis, int ttlSeconds, Deserializer<T> deserializer,
+	                      CommitType commitType, IdService idService)
+	{
+		this.jedis = jedis;
+		this.ttlSeconds = ttlSeconds;
+		this.deserializer = deserializer;
+		this.commitType = commitType;
+		this.idService = idService;
+	}
+
+	public RedisDbHarness(Jedis jedis, Deserializer<T> deserializer,
+	                      CommitType commitType, IdService idService)
+	{
+		this(jedis, -1, deserializer, commitType, idService);
+	}
+
+	@Override
+	public Future<?> runBatchAsync(Batch<T> batch, AfterExecute<T> afterExecute) throws FullQueueException
+	{
+		for(Map.Entry<Address,T> entry: batch.getMap().entrySet()) {
+			Address address = entry.getKey();
+			T opFromBatch = entry.getValue();
+
+			byte[] redisKey;
+			try {
+				redisKey = address.toWriteKey(idService);
+			} catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+
+			if(commitType == CommitType.READ_COMBINE_CAS) {
+				Optional<byte[]> oldBytes;
+				try {
+					oldBytes = getRaw(address);
+				} catch (Exception e) {
+					throw new RuntimeException(e);
+				}
+
+				T newOp;
+				if(oldBytes.isPresent()) {
+					T oldOp = deserializer.fromBytes(oldBytes.get());
+					newOp = (T)opFromBatch.add(oldOp);
+				} else {
+					newOp = opFromBatch;
+				}
+
+				set(redisKey, newOp);
+			}
+			else if(commitType == CommitType.OVERWRITE) {
+				set(redisKey, opFromBatch);
+			}
+			else {
+				throw new AssertionError("Unsupported commit type: " + commitType);
+			}
+		}
+
+		batch.reset();
+		afterExecute.afterExecute(null); // null throwable => success
+		return nullFuture;
+	}
+
+	@Override
+	public Optional<T> get(Address address) throws IOException, InterruptedException
+	{
+		Optional<byte[]> bytes = getRaw(address);
+		if(bytes.isPresent()) {
+			return Optional.of(deserializer.fromBytes(bytes.get()));
+		} else {
+			return Optional.absent();
+		}
+	}
+
+	@Override
+	public void set(Address address, T op) throws IOException, InterruptedException
+	{
+		byte[] redisKey;
+		try {
+			redisKey = address.toWriteKey(idService);
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+
+		set(redisKey, op);
+	}
+
+	private void set(byte[] redisKey, T op)
+	{
+		if (ttlSeconds > 0)
+		{
+			jedis.setex(redisKey, ttlSeconds, op.serialize());
+		}
+		else
+		{
+			jedis.set(redisKey, op.serialize());
+		}
+	}
+
+	@Override
+	public List<Optional<T>> multiGet(List<Address> addresses) throws IOException {
+		throw new NotImplementedException();
+	}
+
+	@Override
+	public void flush() throws InterruptedException {
+		return; // all ops are synchronously applied, nothing to do
+	}
+
+	private Optional<byte[]> getRaw(Address address) throws InterruptedException {
+		byte[] redisKey;
+		try {
+			final Optional<byte[]> maybeKey = address.toReadKey(idService);
+			if (maybeKey.isPresent()) {
+				redisKey = maybeKey.get();
+			} else {
+				return Optional.absent();
+			}
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+
+		byte[] bytes = jedis.get(redisKey);
+
+		if(bytes == null) {
+			return Optional.absent();
+		} else {
+			return Optional.of(bytes);
+		}
+	}
+}