fixes issue debasishg#18 debasishg#18 (reconnect after idle time) - implemented retry in RedisClient#send

Author: debasishg

.gitignore

@@ -1,3 +1,4 @@
 .DS_Store
 target
-project/boot
+project/boot
+*.swp

project/ScalaRedisProject.scala

@@ -3,13 +3,15 @@ import Keys._
 
 object ScalaRedisProject extends Build
 {
+  import Resolvers._
   lazy val root = Project("RedisClient", file(".")) settings(coreSettings : _*)
 
   lazy val commonSettings: Seq[Setting[_]] = Seq(
     organization := "net.debasishg",
     version := "2.6",
     scalaVersion := "2.9.2",
-    scalacOptions ++= Seq("-deprecation", "-unchecked")
+    scalacOptions ++= Seq("-deprecation", "-unchecked"),
+    resolvers ++= Seq(twitterRepo)
   )
 
   lazy val coreSettings = commonSettings ++ template ++ Seq(
@@ -21,8 +23,8 @@ object ScalaRedisProject extends Build
       "log4j"          % "log4j"         % "1.2.16" % "provided",
       "junit"          % "junit"         % "4.8.1"  % "test",
       "org.scalatest"  % "scalatest_2.9.1" % "1.6.1" % "test",
-      "com.twitter"    % "util"          % "1.11.4" % "test" intransitive(),
-      "com.twitter"    % "finagle-core"  % "1.9.0" % "test"),
+      "com.twitter"    % "util_2.9.1"    % "1.12.13" % "test" intransitive(),
+      "com.twitter"    % "finagle-core_2.9.1"  % "4.0.2" % "test"),
 
     parallelExecution in Test := false,
     publishTo <<= version { (v: String) => 
@@ -84,3 +86,7 @@ object ScalaRedisProject extends Build
       (output ** filter).get
   }
 }
+
+object Resolvers {
+  val twitterRepo = "release" at "http://maven.twttr.com"
+}

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

@@ -1,7 +1,7 @@
 package com.redis
 
 import java.io._
-import java.net.Socket
+import java.net.{Socket, InetSocketAddress}
 
 import serialization.Parse.parseStringSafe
 
@@ -15,7 +15,7 @@ trait IO extends Log {
   var db: Int = _
 
   def connected = {
-    socket != null
+    socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected() && !socket.isInputShutdown() && !socket.isOutputShutdown();
   }
   def reconnect = {
     disconnect && connect
@@ -25,9 +25,11 @@ trait IO extends Log {
   def connect: Boolean = {
     try {
       socket = new Socket(host, port)
+
       socket.setSoTimeout(0)
       socket.setKeepAlive(true)
       socket.setTcpNoDelay(true)
+
       out = socket.getOutputStream
       in = new BufferedInputStream(socket.getInputStream)
       true
@@ -83,9 +85,7 @@ trait IO extends Log {
     var found: List[Int] = Nil
     var build = new scala.collection.mutable.ArrayBuilder.ofByte
     while (delimiter != Nil) {
-      val next = try {
-        in.read
-      } catch {case e => -1}
+      val next = in.read
       if (next < 0) return null
       if (next == delimiter.head) {
         found ::= delimiter.head

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

@@ -14,14 +14,25 @@ object RedisClient {
 }
 
 trait Redis extends IO with Protocol {
-  def send[A](command: String, args: Seq[Any])(result: => A)(implicit format: Format): A = {
+
+  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
+  } catch {
+    case e: RedisConnectionException =>
+      if (reconnect) send(command, args)(result)
+      else throw e
   }
-  def send[A](command: String)(result: => A): A = {
+
+  def send[A](command: String)(result: => A): A = try {
     write(Commands.multiBulk(List(command.getBytes("UTF-8"))))
     result
+  } catch {
+    case e: RedisConnectionException =>
+      if (reconnect) send(command)(result)
+      else throw e
   }
+
   def cmd(args: Seq[Array[Byte]]) = Commands.multiBulk(args)
 
   protected def flattenPairs(in: Iterable[Product2[Any, Any]]): List[Any] =

src/test/scala/com/redis/StringOperationsSpec.scala

@@ -255,4 +255,17 @@ class StringOperationsSpec extends Spec
       r.getbit("mykey", 100) should equal(Some(0))
     }
   }
+
+/** uncomment to test timeout : need a custom redis.conf
+  describe("timeout") {
+    it("should append value to that of a key") {
+      r.set("mykey", "Hello World")
+      r.strlen("mykey") should equal(Some(11))
+      r.strlen("nonexisting") should equal(Some(0))
+      Thread.sleep(150000)
+      r.set("nonexisting", "Hello World")
+      r.strlen("nonexisting") should equal(Some(11))
+    }
+  }
+**/
 }