Added SCAN command

src/main/scala/com/redis/api/KeyOperations.scala

@@ -19,6 +19,9 @@ trait KeyOperations { this: RedisOps =>
   def keys(pattern: String = "*")(implicit timeout: Timeout) =
     clientRef.ask(Keys(pattern)).mapTo[Keys#Ret]
 
+  def scan(cursor:Long = 0, pattern: String = "", count: Long = 0)(implicit timeout: Timeout) =
+    clientRef.ask(Scan(cursor, pattern, count)).mapTo[Scan#Ret]
+
   // RANDOMKEY
   // return a randomly selected key from the currently selected DB.
   def randomkey(implicit timeout: Timeout) =

src/main/scala/com/redis/protocol/KeyCommands.scala

@@ -10,6 +10,14 @@ object KeyCommands {
     def params = pattern +: ANil
   }
 
+  case class Scan(cursor:Long = 0, pattern:String = "", count:Long = 0) extends RedisCommand[(Long, List[String])]("SCAN") {
+    require(cursor >= 0)
+    def params = Seq( Seq(cursor.toString),
+      if(pattern != "") Seq("MATCH", pattern) else Nil,
+      if(count > 0) Seq("COUNT", count.toString) else Nil
+    ).flatten.toArgs
+  }
+
   case object RandomKey extends RedisCommand[Option[String]]("RANDOMKEY") {
     def params = ANil
   }

src/main/scala/com/redis/serialization/PartialDeserializer.scala

@@ -110,6 +110,14 @@ private[serialization] trait CommandSpecificPD { this: LowPriorityPD =>
       }.toMap
     }
 
+  // special deserializer for SCAN
+  implicit def scanResultPD(implicit reader: Reader[String]):PartialDeserializer[(Long, List[String])] = {
+    new PrefixDeserializer[(Long, List[String])](Multi, (x: RawReply) => {
+      parseBulk(x)
+      (parseLong(x), listPD[String].apply(x))
+    })
+  }
+
   // special deserializer for EVAL(SHA)
   def ensureListPD[A](implicit reader: Reader[A]): PartialDeserializer[List[A]] =
     multiBulkPD[A, List].orElse(parsedOptionPD[A].andThen(_.toList))

src/test/scala/com/redis/api/KeyOperationsSpec.scala

@@ -24,12 +24,55 @@ class KeyOperationsSpec extends RedisSpecBase {
     it("should fetch keys with spaces") {
       val prepare = Seq(client.set("anshin 1", "debasish"), client.set("anshin 2", "maulindu"))
       val prepareRes = Future.sequence(prepare).futureValue
-
       val res = client.keys("anshin*")
       res.futureValue should have length (2)
     }
   }
 
+  describe("scan") {
+
+    val keys = Seq(
+      "key:1", "key:2", "key:3", "key:4", "key:5",
+      "key:11", "key:22", "key:33", "key:44", "key:55"
+    )
+
+    def prepare = keys.map(k => client.set(k, k))
+
+    it("should collect all keys in keyspace") {
+      Future.sequence(prepare).futureValue
+      prepare.size should equal(iterateScan().size)
+    }
+
+    it("should filter base on pattern") {
+      Future.sequence(prepare).futureValue
+      val res = iterateScan(pattern = "*5*")
+      res.size should equal(2)
+    }
+
+    it("should contain string encoded keys in result") {
+      Future.sequence(prepare).futureValue
+      val res = client.scan().futureValue
+      res._2.foreach(k => keys.contains(k) should be(true))
+    }
+
+    it("should throw on negative cursor") {
+      an[IllegalArgumentException] shouldBe thrownBy(client.scan(cursor = -1))
+    }
+
+    def iterateScan(count:Long = 0, pattern:String = "") = {
+      var cursor = -1l
+      var keys = Seq[String]()
+      while(cursor != 0) {
+        val res = client.scan(cursor = if(cursor < 0) 0 else cursor, pattern = pattern, count = 2)
+        val w = res.futureValue
+        keys = keys ++ w._2
+        cursor = w._1
+      }
+      keys
+    }
+
+  }
+
   describe("randomkey") {
     it("should give") {
       val prepare = Seq(client.set("anshin-1", "debasish"), client.set("anshin-2", "maulindu"))