Batch operations in pipeline (#283)

* Refactoring - getting rid of compiler warnings

* Implements batch operations in pipeline mode, refactoring, removes compiler warnings

Author: debasishg

src/main/scala/com/redis/BaseOperations.scala

@@ -47,14 +47,14 @@ trait BaseOperations extends BaseApi {
     send("KEYS", List(pattern))(asList)
 
   override def time[A](implicit format: Format, parse: Parse[A]): Option[List[Option[A]]] =
-    send("TIME")(asList)
+    send("TIME", false)(asList)
 
   @deprecated("use randomkey", "2.8")
   def randkey[A](implicit parse: Parse[A]): Option[A] =
-    send("RANDOMKEY")(asBulk)
+    send("RANDOMKEY", false)(asBulk)
 
   override def randomkey[A](implicit parse: Parse[A]): Option[A] =
-    send("RANDOMKEY")(asBulk)
+    send("RANDOMKEY", false)(asBulk)
 
   override def rename(oldkey: Any, newkey: Any)(implicit format: Format): Boolean =
     send("RENAME", List(oldkey, newkey))(asBoolean)
@@ -63,7 +63,7 @@ trait BaseOperations extends BaseApi {
     send("RENAMENX", List(oldkey, newkey))(asBoolean)
 
   override def dbsize: Option[Long] =
-    send("DBSIZE")(asLong)
+    send("DBSIZE", false)(asLong)
 
   override def exists(key: Any)(implicit format: Format): Boolean =
     send("EXISTS", List(key))(asBoolean)
@@ -104,16 +104,16 @@ trait BaseOperations extends BaseApi {
     })
 
   override def flushdb: Boolean =
-    send("FLUSHDB")(asBoolean)
+    send("FLUSHDB", false)(asBoolean)
 
   override def flushall: Boolean =
-    send("FLUSHALL")(asBoolean)
+    send("FLUSHALL", false)(asBoolean)
 
   override def move(key: Any, db: Int)(implicit format: Format): Boolean =
     send("MOVE", List(key, db))(asBoolean)
 
   override def quit: Boolean =
-    send("QUIT")(disconnect)
+    send("QUIT", false)(disconnect)
 
   override def auth(secret: Any)(implicit format: Format): Boolean =
     send("AUTH", List(secret))(asBoolean)
@@ -125,13 +125,13 @@ trait BaseOperations extends BaseApi {
     send("SCAN", cursor :: ((x: List[Any]) => if (pattern == "*") x else "match" :: pattern :: x) (if (count == 10) Nil else List("count", count)))(asPair)
 
   override def ping: Option[String] =
-    send("PING")(asString)
+    send("PING", false)(asString)
 
   override def watch(key: Any, keys: Any*)(implicit format: Format): Boolean =
     send("WATCH", key :: keys.toList)(asBoolean)
 
   override def unwatch(): Boolean =
-    send("UNWATCH")(asBoolean)
+    send("UNWATCH", false)(asBoolean)
 
   override def getConfig(key: Any = "*")(implicit format: Format): Option[Map[String, Option[String]]] =
     send("CONFIG", List("GET", key))(asList).map { ls =>

src/main/scala/com/redis/IO.scala

@@ -115,7 +115,7 @@ trait IO extends Log {
         disconnect
         throw ex
     }
-    build.result
+    build.result()
   }
 
   def readCounted(count: Int): Array[Byte] = {

src/main/scala/com/redis/NodeOperations.scala

@@ -6,25 +6,25 @@ trait NodeOperations extends NodeApi {
   self: Redis =>
 
   override def save: Boolean =
-    send("SAVE")(asBoolean)
+    send("SAVE", false)(asBoolean)
 
   override def bgsave: Boolean =
-    send("BGSAVE")(asBoolean)
+    send("BGSAVE", false)(asBoolean)
 
   override def lastsave: Option[Long] =
-    send("LASTSAVE")(asLong)
+    send("LASTSAVE", false)(asLong)
 
   override def shutdown: Boolean =
-    send("SHUTDOWN")(asBoolean)
+    send("SHUTDOWN", false)(asBoolean)
 
   override def bgrewriteaof: Boolean =
-    send("BGREWRITEAOF")(asBoolean)
+    send("BGREWRITEAOF", false)(asBoolean)
 
   override def info: Option[String] =
-    send("INFO")(asBulk)
+    send("INFO", false)(asBulk)
 
   override def monitor: Boolean =
-    send("MONITOR")(asBoolean)
+    send("MONITOR", false)(asBoolean)
 
   override def slaveof(options: Any): Boolean = options match {
     case (h: String, p: Int) =>

src/main/scala/com/redis/PubSub.scala

@@ -77,7 +77,7 @@ trait PubSub extends PubOperations { self: Redis =>
   }
 
   def pUnsubscribe(): Unit = {
-    send("PUNSUBSCRIBE")(())
+    send("PUNSUBSCRIBE", false)(())
   }
 
   def pUnsubscribe(channel: String, channels: String*): Unit = {
@@ -98,7 +98,7 @@ trait PubSub extends PubOperations { self: Redis =>
   }
 
   def unsubscribe(): Unit = {
-    val r = send("UNSUBSCRIBE")(())
+    val r = send("UNSUBSCRIBE", false)(())
     pubSub = false
     r
   }

src/main/scala/com/redis/RedisClient.scala

@@ -15,6 +15,10 @@ object RedisClient {
   case object MIN extends Aggregate
   case object MAX extends Aggregate
 
+  sealed trait Mode
+  case object SINGLE extends Mode
+  case object BATCH extends Mode
+
   private def extractDatabaseNumber(connectionUri: java.net.URI): Int = {
     Option(connectionUri.getPath).map(path =>
       if (path.isEmpty) 0
@@ -24,11 +28,22 @@ object RedisClient {
   }
 }
 
-trait Redis extends IO with Protocol {
+import RedisClient._
+abstract class Redis(batch: Mode) extends IO with Protocol {
+  var handlers: Vector[(String, () => Any)] = Vector.empty
+  var commandBuffer: StringBuffer           = new StringBuffer
+  val crlf = "\r\n"
+
 
   def send[A](command: String, args: Seq[Any])(result: => A)(implicit format: Format): A = try {
-    write(Commands.multiBulk(command.getBytes("UTF-8") +: (args map (format.apply))))
-    result
+    if (batch == BATCH) {
+      handlers :+= ((command, () => result))
+      commandBuffer.append((List(command) ++ args.toList).mkString(" ") ++ crlf)
+      null.asInstanceOf[A] // hack
+    } else {
+      write(Commands.multiBulk(command.getBytes("UTF-8") +: (args map (format.apply))))
+      result
+    }
   } catch {
     case e: RedisConnectionException =>
       if (disconnect) send(command, args)(result)
@@ -38,15 +53,26 @@ trait Redis extends IO with Protocol {
       else throw e
   }
 
-  def send[A](command: String)(result: => A): A = try {
-    write(Commands.multiBulk(List(command.getBytes("UTF-8"))))
-    result
+  def send[A](command: String, submissionMode: Boolean = false)(result: => A): A = try {
+    if (batch == BATCH) {
+      if (!submissionMode) {
+        handlers :+= ((command, () => result))
+        commandBuffer.append(command ++ crlf)
+        null.asInstanceOf[A]
+      } else {
+        write(command.getBytes("UTF-8"))
+        result
+      }
+    } else {
+      write(Commands.multiBulk(List(command.getBytes("UTF-8"))))
+      result
+    }
   } catch {
     case e: RedisConnectionException =>
-      if (disconnect) send(command)(result)
+      if (disconnect) send(command, submissionMode)(result)
       else throw e
     case e: SocketException =>
-      if (disconnect) send(command)(result)
+      if (disconnect) send(command, submissionMode)(result)
       else throw e
   }
 
@@ -57,7 +83,7 @@ trait Redis extends IO with Protocol {
 
 }
 
-trait RedisCommand extends Redis
+abstract class RedisCommand(batch: Mode) extends Redis(batch)
   with BaseOperations
   with GeoOperations
   with NodeOperations
@@ -74,7 +100,7 @@ trait RedisCommand extends Redis
   val database: Int = 0
   val secret: Option[Any] = None
 
-  override def onConnect: Unit = {
+  override def onConnect(): Unit = {
     secret.foreach {s =>
       auth(s)
     }
@@ -89,13 +115,12 @@ trait RedisCommand extends Redis
   private def authenticate(): Unit = {
     secret.foreach(auth _)
   }
-
 }
 
-
 class RedisClient(override val host: String, override val port: Int,
-    override val database: Int = 0, override val secret: Option[Any] = None, override val timeout : Int = 0, override val sslContext: Option[SSLContext] = None)
-  extends RedisCommand with PubSub {
+    override val database: Int = 0, override val secret: Option[Any] = None, override val timeout : Int = 0, 
+    override val sslContext: Option[SSLContext] = None, val batch: Mode = RedisClient.SINGLE)
+  extends RedisCommand(batch) with PubSub {
 
   def this() = this("localhost", 6379)
   def this(connectionUri: java.net.URI) = this(
@@ -110,18 +135,19 @@ class RedisClient(override val host: String, override val port: Int,
   )
   override def toString: String = host + ":" + String.valueOf(port) + "/" + database
 
+  // with MULTI/EXEC
   def pipeline(f: PipelineClient => Any): Option[List[Any]] = {
-    send("MULTI")(asString) // flush reply stream
+    send("MULTI", false)(asString) // flush reply stream
     try {
       val pipelineClient = new PipelineClient(this)
       try {
         f(pipelineClient)
       } catch {
         case e: Exception =>
-          send("DISCARD")(asString)
+          send("DISCARD", false)(asString)
           throw e
       }
-      send("EXEC")(asExec(pipelineClient.handlers))
+      send("EXEC", false)(asExec(pipelineClient.responseHandlers))
     } catch {
       case e: RedisMultiExecException =>
         None
@@ -135,7 +161,7 @@ class RedisClient(override val host: String, override val port: Int,
   /**
    * Redis pipelining API without the transaction semantics. The implementation has a non-blocking
    * semantics and returns a <tt>List</tt> of <tt>Promise</tt>. The caller may use <tt>Future.firstCompletedOf</tt> to get the
-   * first completed task before all tasks have been completed.
+   * first completed task before all tasks have been completed. However the commands are submitted one by one and NOT in batch.
    * If an exception is raised in executing any of the commands, then the corresponding <tt>Promise</tt> holds
    * the exception. Here's a sample usage:
    * <pre>
@@ -179,20 +205,32 @@ class RedisClient(override val host: String, override val port: Int,
     ps
   }
 
-  class PipelineClient(parent: RedisClient) extends RedisCommand with PubOperations {
+  // batched pipelines : all commands submitted in batch
+  def batchedPipeline(commands: List[() => Any]): Option[List[Any]] = {
+    assert(batch == BATCH)
+    commands.foreach { command =>
+      command()
+    }
+    val r = send(commandBuffer.toString, true)(Some(handlers.map(_._2).map(_()).toList))
+    handlers = Vector.empty
+    commandBuffer.setLength(0)
+    r
+  }
+
+  class PipelineClient(parent: RedisClient) extends RedisCommand(parent.batch) with PubOperations {
     import com.redis.serialization.Parse
 
-    var handlers: Vector[() => Any] = Vector.empty
+    var responseHandlers: Vector[() => Any] = Vector.empty
 
     override def send[A](command: String, args: Seq[Any])(result: => A)(implicit format: Format): A = {
       write(Commands.multiBulk(command.getBytes("UTF-8") +: (args map (format.apply))))
-      handlers :+= (() => result)
+      responseHandlers :+= (() => result)
       receive(singleLineReply).map(Parse.parseDefault)
       null.asInstanceOf[A] // ugh... gotta find a better way
     }
-    override def send[A](command: String)(result: => A): A = {
+    override def send[A](command: String, submissionMode: Boolean = false)(result: => A): A = {
       write(Commands.multiBulk(List(command.getBytes("UTF-8"))))
-      handlers :+= (() => result)
+      responseHandlers :+= (() => result)
       receive(singleLineReply).map(Parse.parseDefault)
       null.asInstanceOf[A]
     }
@@ -207,7 +245,7 @@ class RedisClient(override val host: String, override val port: Int,
     override def connected = parent.connected
     override def connect = parent.connect
     override def disconnect = parent.disconnect
-    override def clearFd = parent.clearFd
+    override def clearFd() = parent.clearFd()
     override def write(data: Array[Byte]) = parent.write(data)
     override def readLine = parent.readLine
     override def readCounted(count: Int) = parent.readCounted(count)

src/main/scala/com/redis/RedisProtocol.scala

@@ -32,7 +32,7 @@ private [redis] object Commands {
       b ++= arg
       b ++= LS
     }
-    b.result
+    b.result()
   }
 }
 
@@ -111,7 +111,7 @@ private [redis] trait Reply {
       Parsers.parseInt(str) match {
         case -1 => None
         case n if n == handlers.size => 
-          Some(handlers.map(_.apply).toList)
+          Some(handlers.map(_.apply()).toList)
         case n => throw new Exception("Protocol error: Expected "+handlers.size+" results, but got "+n)
       }
   }
@@ -286,6 +286,7 @@ private [redis] trait R extends Reply {
   }
 
   def asAny = receive(integerReply orElse singleLineReply orElse bulkReply orElse multiBulkReply)
+  def asAnyMany(count: Int) = (0 to count - 1).map(_ => asAny).toList
 }
 
 trait Protocol extends R

src/main/scala/com/redis/cluster/HashRing.scala

@@ -76,7 +76,7 @@ case class HashRing[T](nodes: List[T], replicas: Int) {
   def isEmpty: Boolean = ring.isEmpty
 
   private def nodeHashFor(node: T, replica: Int): Long = {
-    calculateChecksum((node + ":" + replica).getBytes("UTF-8").toIndexedSeq)
+    calculateChecksum((s"$node:$replica").getBytes("UTF-8").toIndexedSeq)
   }
 
 }

src/main/scala/com/redis/ds/Deque.scala

@@ -36,9 +36,14 @@ import com.redis.ListOperations
 import com.redis.serialization.Parse.Implicits._
 import com.redis.serialization._
 
-abstract class RedisDeque[A](val blocking: Boolean = false, val timeoutInSecs: Int = 0)(implicit private val format: Format, private val parse: Parse[A])
+trait RedisDeque[A] 
   extends Deque[A] { self: ListOperations =>
 
+  val blocking: Boolean = false
+  val timeoutInSecs: Int = 0
+  implicit val f: Format
+  implicit val pr: Parse[A]
+
   val key: String
 
   def addFirst(a: A): Option[Long] = lpush(key, a) 
@@ -73,18 +78,27 @@ abstract class RedisDeque[A](val blocking: Boolean = false, val timeoutInSecs: I
   }
 }
 
-import com.redis.RedisCommand
+import com.redis._
+
+abstract class MyRedisDeque[A](bloking: Boolean, timoutInSecs: Int)(implicit format: Format, parse: Parse[A]) 
+  extends RedisCommand(RedisClient.SINGLE) with RedisDeque[A] {
+    override val blocking = bloking
+    override val timeoutInSecs: Int = timoutInSecs
+  }
 
 class RedisDequeClient(val h: String, val p: Int, val d: Int = 0, val s: Option[Any] = None, val t : Int =0) {
-  def getDeque[A](k: String, blocking: Boolean = false, timeoutInSecs: Int = 0)(implicit format: Format, parse: Parse[A]) =
-    new RedisDeque(blocking, timeoutInSecs)(format, parse) with RedisCommand {
+  def getDeque[A](k: String, blocking: Boolean = false, timeoutInSecs: Int = 0)(implicit format: Format, parse: Parse[A]) = {
+
+    new MyRedisDeque[A](blocking, timeoutInSecs)(format, parse) {
+      implicit val f = format
+      implicit val pr = parse
       val host = h
       val port = p
       val timeout = t
       val key = k
       override val database = d
       override val secret = s
-
       override def close(): Unit = disconnect
     }
+  }
 }

src/test/scala/com/redis/Bench.scala

@@ -13,7 +13,7 @@ object Bench {
     import Values._
     clients.withClient {
       client => {
-        (1 to count) foreach (i => client.rpush(key, values.next))
+        (1 to count) foreach (i => client.rpush(key, values.next()))
         assert(client.llen(key) == Some(count))
         (1 to count) foreach (i => client.lpop(key))
         assert(client.llen(key) == Some(0))