Redis tests now passes with new STM + misc minor changes to Cluster
This commit is contained in:
Jonas Bonér
parent
73a0648292
commit
36c0266a5d
11 changed files with 131 additions and 96 deletions
|
|
@ -72,11 +72,11 @@ private [akka] object RedisStorageBackend extends
|
|||
* base64(T1):base64("debasish.programming_language") -> "scala"
|
||||
* </i>
|
||||
*/
|
||||
def insertMapStorageEntryFor(name: String, key: Array[Byte], value: Array[Byte]) {
|
||||
def insertMapStorageEntryFor(name: String, key: Array[Byte], value: Array[Byte]): Unit = withErrorHandling {
|
||||
insertMapStorageEntriesFor(name, List((key, value)))
|
||||
}
|
||||
|
||||
def insertMapStorageEntriesFor(name: String, entries: List[Tuple2[Array[Byte], Array[Byte]]]) {
|
||||
def insertMapStorageEntriesFor(name: String, entries: List[Tuple2[Array[Byte], Array[Byte]]]): Unit = withErrorHandling {
|
||||
mset(entries.map(e =>
|
||||
(makeRedisKey(name, e._1), new String(e._2))))
|
||||
}
|
||||
|
|
@ -89,22 +89,22 @@ private [akka] object RedisStorageBackend extends
|
|||
* <li>: is chosen since it cannot appear in base64 encoding charset</li>
|
||||
* <li>both parts of the key need to be based64 encoded since there can be spaces within each of them</li>
|
||||
*/
|
||||
private [this] def makeRedisKey(name: String, key: Array[Byte]): String = {
|
||||
private [this] def makeRedisKey(name: String, key: Array[Byte]): String = withErrorHandling {
|
||||
"%s:%s".format(new String(encode(name.getBytes)), new String(encode(key)))
|
||||
}
|
||||
|
||||
private [this] def makeKeyFromRedisKey(redisKey: String) = {
|
||||
private [this] def makeKeyFromRedisKey(redisKey: String) = withErrorHandling {
|
||||
val nk = redisKey.split(':').map{e: String => decode(e.getBytes)}
|
||||
(nk(0), nk(1))
|
||||
}
|
||||
|
||||
private [this] def mset(entries: List[(String, String)]) {
|
||||
private [this] def mset(entries: List[(String, String)]): Unit = withErrorHandling {
|
||||
entries.foreach {e: (String, String) =>
|
||||
db.set(e._1, e._2)
|
||||
}
|
||||
}
|
||||
|
||||
def removeMapStorageFor(name: String): Unit = {
|
||||
def removeMapStorageFor(name: String): Unit = withErrorHandling {
|
||||
db.keys("%s:*".format(encode(name.getBytes))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -113,18 +113,19 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def removeMapStorageFor(name: String, key: Array[Byte]): Unit = {
|
||||
def removeMapStorageFor(name: String, key: Array[Byte]): Unit = withErrorHandling {
|
||||
db.delete(makeRedisKey(name, key))
|
||||
}
|
||||
|
||||
def getMapStorageEntryFor(name: String, key: Array[Byte]): Option[Array[Byte]] =
|
||||
def getMapStorageEntryFor(name: String, key: Array[Byte]): Option[Array[Byte]] = withErrorHandling {
|
||||
db.get(makeRedisKey(name, key)) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(new String(key) + " not present")
|
||||
case Some(s) => Some(s.getBytes)
|
||||
}
|
||||
}
|
||||
|
||||
def getMapStorageSizeFor(name: String): Int = {
|
||||
def getMapStorageSizeFor(name: String): Int = withErrorHandling {
|
||||
db.keys("%s:*".format(new String(encode(name.getBytes)))) match {
|
||||
case None => 0
|
||||
case Some(keys) =>
|
||||
|
|
@ -132,7 +133,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def getMapStorageFor(name: String): List[(Array[Byte], Array[Byte])] = {
|
||||
def getMapStorageFor(name: String): List[(Array[Byte], Array[Byte])] = withErrorHandling {
|
||||
db.keys("%s:*".format(new String(encode(name.getBytes)))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -143,7 +144,7 @@ private [akka] object RedisStorageBackend extends
|
|||
|
||||
def getMapStorageRangeFor(name: String, start: Option[Array[Byte]],
|
||||
finish: Option[Array[Byte]],
|
||||
count: Int): List[(Array[Byte], Array[Byte])] = {
|
||||
count: Int): List[(Array[Byte], Array[Byte])] = withErrorHandling {
|
||||
|
||||
import scala.collection.immutable.TreeMap
|
||||
val wholeSorted =
|
||||
|
|
@ -188,19 +189,19 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def insertVectorStorageEntryFor(name: String, element: Array[Byte]) {
|
||||
def insertVectorStorageEntryFor(name: String, element: Array[Byte]): Unit = withErrorHandling {
|
||||
db.lpush(new String(encode(name.getBytes)), new String(element))
|
||||
}
|
||||
|
||||
def insertVectorStorageEntriesFor(name: String, elements: List[Array[Byte]]) {
|
||||
def insertVectorStorageEntriesFor(name: String, elements: List[Array[Byte]]): Unit = withErrorHandling {
|
||||
elements.foreach(insertVectorStorageEntryFor(name, _))
|
||||
}
|
||||
|
||||
def updateVectorStorageEntryFor(name: String, index: Int, elem: Array[Byte]) {
|
||||
def updateVectorStorageEntryFor(name: String, index: Int, elem: Array[Byte]): Unit = withErrorHandling {
|
||||
db.lset(new String(encode(name.getBytes)), index, new String(elem))
|
||||
}
|
||||
|
||||
def getVectorStorageEntryFor(name: String, index: Int): Array[Byte] = {
|
||||
def getVectorStorageEntryFor(name: String, index: Int): Array[Byte] = withErrorHandling {
|
||||
db.lindex(new String(encode(name.getBytes)), index) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " does not have element at " + index)
|
||||
|
|
@ -208,7 +209,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def getVectorStorageRangeFor(name: String, start: Option[Int], finish: Option[Int], count: Int): List[Array[Byte]] = {
|
||||
def getVectorStorageRangeFor(name: String, start: Option[Int], finish: Option[Int], count: Int): List[Array[Byte]] = withErrorHandling {
|
||||
/**
|
||||
* <tt>count</tt> is the max number of results to return. Start with
|
||||
* <tt>start</tt> or 0 (if <tt>start</tt> is not defined) and go until
|
||||
|
|
@ -237,11 +238,11 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def insertRefStorageFor(name: String, element: Array[Byte]) {
|
||||
def insertRefStorageFor(name: String, element: Array[Byte]): Unit = withErrorHandling {
|
||||
db.set(new String(encode(name.getBytes)), new String(element))
|
||||
}
|
||||
|
||||
def getRefStorageFor(name: String): Option[Array[Byte]] = {
|
||||
def getRefStorageFor(name: String): Option[Array[Byte]] = withErrorHandling {
|
||||
db.get(new String(encode(name.getBytes))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -250,12 +251,13 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
|
||||
// add to the end of the queue
|
||||
def enqueue(name: String, item: Array[Byte]): Boolean = {
|
||||
def enqueue(name: String, item: Array[Byte]): Boolean = withErrorHandling {
|
||||
db.rpush(new String(encode(name.getBytes)), new String(item))
|
||||
}
|
||||
|
||||
|
||||
// pop from the front of the queue
|
||||
def dequeue(name: String): Option[Array[Byte]] = {
|
||||
def dequeue(name: String): Option[Array[Byte]] = withErrorHandling {
|
||||
db.lpop(new String(encode(name.getBytes))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -265,7 +267,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
|
||||
// get the size of the queue
|
||||
def size(name: String): Int = {
|
||||
def size(name: String): Int = withErrorHandling {
|
||||
db.llen(new String(encode(name.getBytes))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -275,26 +277,28 @@ private [akka] object RedisStorageBackend extends
|
|||
|
||||
// return an array of items currently stored in the queue
|
||||
// start is the item to begin, count is how many items to return
|
||||
def peek(name: String, start: Int, count: Int): List[Array[Byte]] = count match {
|
||||
case 1 =>
|
||||
db.lindex(new String(encode(name.getBytes)), start) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException("No element at " + start)
|
||||
case Some(s) =>
|
||||
List(s.getBytes)
|
||||
}
|
||||
case n =>
|
||||
db.lrange(new String(encode(name.getBytes)), start, start + count - 1) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(
|
||||
"No element found between " + start + " and " + (start + count - 1))
|
||||
case Some(es) =>
|
||||
es.map(_.get.getBytes)
|
||||
}
|
||||
def peek(name: String, start: Int, count: Int): List[Array[Byte]] = withErrorHandling {
|
||||
count match {
|
||||
case 1 =>
|
||||
db.lindex(new String(encode(name.getBytes)), start) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException("No element at " + start)
|
||||
case Some(s) =>
|
||||
List(s.getBytes)
|
||||
}
|
||||
case n =>
|
||||
db.lrange(new String(encode(name.getBytes)), start, start + count - 1) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(
|
||||
"No element found between " + start + " and " + (start + count - 1))
|
||||
case Some(es) =>
|
||||
es.map(_.get.getBytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// completely delete the queue
|
||||
def remove(name: String): Boolean = {
|
||||
def remove(name: String): Boolean = withErrorHandling {
|
||||
db.delete(new String(encode(name.getBytes))) match {
|
||||
case Some(1) => true
|
||||
case _ => false
|
||||
|
|
@ -302,7 +306,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
|
||||
// add item to sorted set identified by name
|
||||
def zadd(name: String, zscore: String, item: Array[Byte]): Boolean = {
|
||||
def zadd(name: String, zscore: String, item: Array[Byte]): Boolean = withErrorHandling {
|
||||
db.zadd(new String(encode(name.getBytes)), zscore, new String(item)) match {
|
||||
case Some(1) => true
|
||||
case _ => false
|
||||
|
|
@ -310,7 +314,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
|
||||
// remove item from sorted set identified by name
|
||||
def zrem(name: String, item: Array[Byte]): Boolean = {
|
||||
def zrem(name: String, item: Array[Byte]): Boolean = withErrorHandling {
|
||||
db.zrem(new String(encode(name.getBytes)), new String(item)) match {
|
||||
case Some(1) => true
|
||||
case _ => false
|
||||
|
|
@ -318,7 +322,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
|
||||
// cardinality of the set identified by name
|
||||
def zcard(name: String): Int = {
|
||||
def zcard(name: String): Int = withErrorHandling {
|
||||
db.zcard(new String(encode(name.getBytes))) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -326,7 +330,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def zscore(name: String, item: Array[Byte]): String = {
|
||||
def zscore(name: String, item: Array[Byte]): String = withErrorHandling {
|
||||
db.zscore(new String(encode(name.getBytes)), new String(item)) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(new String(item) + " not present")
|
||||
|
|
@ -334,7 +338,7 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def zrange(name: String, start: Int, end: Int): List[Array[Byte]] = {
|
||||
def zrange(name: String, start: Int, end: Int): List[Array[Byte]] = withErrorHandling {
|
||||
db.zrange(new String(encode(name.getBytes)), start.toString, end.toString, RedisClient.ASC, false) match {
|
||||
case None =>
|
||||
throw new Predef.NoSuchElementException(name + " not present")
|
||||
|
|
@ -343,5 +347,16 @@ private [akka] object RedisStorageBackend extends
|
|||
}
|
||||
}
|
||||
|
||||
def flushDB = db.flushDb
|
||||
def flushDB = withErrorHandling(db.flushDb)
|
||||
|
||||
private def withErrorHandling[T](body: => T): T = {
|
||||
try {
|
||||
body
|
||||
} catch {
|
||||
case e: java.lang.NullPointerException =>
|
||||
throw new StorageException("Could not connect to Redis server")
|
||||
case e =>
|
||||
throw new StorageException("Error in Redis: " + e.getMessage)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue