Merge pull request #201 from dskrvk/deprecate_pubsub0204

Deprecate Publisher and Subscriber; code cleanup; update dependencies. Fixes #200

Author: debasishg

README.md

@@ -30,7 +30,7 @@ Add to `Build.scala` or `build.sbt`
 
 ```scala
 libraryDependencies ++= Seq(
-    "net.debasishg" %% "redisclient" % "3.4"
+    "net.debasishg" %% "redisclient" % "3.5"
 )
 ```
 
@@ -230,6 +230,10 @@ def scatterGatherFirstWithList(opsPerClient: Int)(implicit clients: RedisClientP
 }
 ```
 
+## Using Pub/Sub
+
+See an example implementation using Akka at https://github.com/debasishg/akka-redis-pubsub.
+
 ## License
 
 This software is licensed under the Apache 2 license, quoted below.

project/ScalaRedisProject.scala

@@ -8,9 +8,9 @@ object ScalaRedisProject extends Build
 
   lazy val commonSettings: Seq[Setting[_]] = Seq(
     organization := "net.debasishg",
-    version := "3.4",
-    scalaVersion := "2.11.8",
-    crossScalaVersions := Seq("2.12.1", "2.11.8", "2.10.6"),
+    version := "3.5",
+    scalaVersion := "2.11.12",
+    crossScalaVersions := Seq("2.12.4", "2.11.12", "2.10.7"),
 
     scalacOptions in Compile ++= Seq( "-unchecked", "-feature", "-language:postfixOps", "-deprecation" ),
 
@@ -21,11 +21,11 @@ object ScalaRedisProject extends Build
     name := "RedisClient",
     libraryDependencies := Seq(
       "commons-pool"      %  "commons-pool"            % "1.6",
-      "org.slf4j"         %  "slf4j-api"               % "1.7.2",
-      "org.slf4j"         %  "slf4j-log4j12"           % "1.7.2"      % "provided",
-      "log4j"             %  "log4j"                   % "1.2.16"     % "provided",
-      "junit"             %  "junit"                   % "4.8.1"      % "test",
-      "org.scalatest"     %%  "scalatest"              % "3.0.0" % "test"),
+      "org.slf4j"         %  "slf4j-api"               % "1.7.25",
+      "org.slf4j"         %  "slf4j-log4j12"           % "1.7.25"     % "provided",
+      "log4j"             %  "log4j"                   % "1.2.17"     % "provided",
+      "junit"             %  "junit"                   % "4.12"       % "test",
+      "org.scalatest"     %%  "scalatest"              % "3.0.4"      % "test"),
 
     libraryDependencies += {
       if(scalaVersion.value.startsWith("2.12"))

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.12
+sbt.version=0.13.17

src/main/scala/com/redis/GeoOperations.scala

@@ -86,7 +86,7 @@ trait GeoOperations { self: Redis =>
         , count.fold[List[Any]](Nil)(b => List("COUNT", b))
         , store.fold[List[Any]](Nil)(b => List("STORE", b))
         , storeDist.fold[List[Any]](Nil)(b => List("STOREDIST", b))
-      ).flatMap(x=>x)
+      ).flatten
     send("GEORADIUS", List(key, longitude, latitude, radius, unit) ++ radArgs)(receive(geoRadiusMemberReply))
   }
 
@@ -123,7 +123,7 @@ trait GeoOperations { self: Redis =>
       , count.fold[List[Any]](Nil)(b => List("COUNT", b))
       , store.fold[List[Any]](Nil)(b => List("STORE", b))
       , storeDist.fold[List[Any]](Nil)(b => List("STOREDIST", b))
-    ).flatMap(x=>x)
+    ).flatten
     send("GEORADIUSBYMEMBER", List(key, member, radius, unit) ++ radArgs)(receive(geoRadiusMemberReply))
   }
 

src/main/scala/com/redis/HyperLogLogOperations.scala

@@ -15,6 +15,6 @@ trait HyperLogLogOperations { self: Redis =>
 
   // PFMERGE (>= 2.8.9)
   // Merge existing keys
-  def pfmerge(destination: Any, sources: Any*) =
+  def pfmerge(destination: Any, sources: Any*): Boolean =
     send("PFMERGE", List(destination) ::: sources.toList)(asBoolean)
 }

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

@@ -15,8 +15,8 @@ trait IO extends Log {
   var in: InputStream = _
   var db: Int = _
 
-  def connected = {
-    socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected() && !socket.isInputShutdown() && !socket.isOutputShutdown();
+  def connected: Boolean = {
+    socket != null && socket.isBound && !socket.isClosed && socket.isConnected && !socket.isInputShutdown && !socket.isOutputShutdown
   }
 
   // Connects the socket, and sets the input and output streams.
@@ -34,42 +34,42 @@ trait IO extends Log {
       true
     } catch {
       case x: Throwable =>
-        clearFd
+        clearFd()
         throw new RuntimeException(x)
     }
   }
 
   // Disconnects the socket.
   def disconnect: Boolean = {
     try {
-      socket.close
-      out.close
-      in.close
-      clearFd
+      socket.close()
+      out.close()
+      in.close()
+      clearFd()
       true
     } catch {
       case x: Throwable =>
         false
     }
   }
 
-  def clearFd = {
+  def clearFd(): Unit = {
     socket = null
     out = null
     in = null
   }
 
   // Wrapper for the socket write operation.
-  def write_to_socket(data: Array[Byte])(op: OutputStream => Unit) = op(out)
+  def write_to_socket(data: Array[Byte])(op: OutputStream => Unit): Unit = op(out)
 
   // Writes data to a socket using the specified block.
-  def write(data: Array[Byte]) = {
+  def write(data: Array[Byte]): Unit = {
     ifDebug("C: " + parseStringSafe(data))
-    if (!connected) connect;
+    if (!connected) connect
     write_to_socket(data){ os =>
       try {
         os.write(data)
-        os.flush
+        os.flush()
       } catch {
         case x: Throwable => throw new RedisConnectionException("connection is closed. write error")
       }

src/main/scala/com/redis/Log.scala

@@ -5,28 +5,28 @@ import org.slf4j.LoggerFactory
 trait Log {
  private val log = LoggerFactory.getLogger(getClass)
 
- def ifTrace(message: => String) = if (log.isTraceEnabled) trace(message)
- def trace(message:String, values:AnyRef*) =
+ def ifTrace(message: => String): Unit = if (log.isTraceEnabled) trace(message)
+ def trace(message:String, values:AnyRef*): Unit =
      log.trace(message, values)
- def trace(message:String, error:Throwable) = log.trace(message, error)
+ def trace(message:String, error:Throwable): Unit = log.trace(message, error)
 
- def ifDebug(message: => String) = if (log.isDebugEnabled) debug(message)
- def debug(message:String, values:AnyRef*) =
+ def ifDebug(message: => String): Unit = if (log.isDebugEnabled) debug(message)
+ def debug(message:String, values:AnyRef*): Unit =
      log.debug(message, values)
- def debug(message:String, error:Throwable) = log.debug(message, error)
+ def debug(message:String, error:Throwable): Unit = log.debug(message, error)
 
- def ifInfo(message: => String) = if (log.isInfoEnabled) info(message)
- def info(message:String, values:AnyRef*) =
+ def ifInfo(message: => String): Unit = if (log.isInfoEnabled) info(message)
+ def info(message:String, values:AnyRef*): Unit =
      log.info(message, values)
- def info(message:String, error:Throwable) = log.info(message, error)
+ def info(message:String, error:Throwable): Unit = log.info(message, error)
 
- def ifWarn(message: => String) = if (log.isWarnEnabled) warn(message)
- def warn(message:String, values:AnyRef*) =
+ def ifWarn(message: => String): Unit = if (log.isWarnEnabled) warn(message)
+ def warn(message:String, values:AnyRef*): Unit =
      log.warn(message, values)
- def warn(message:String, error:Throwable) = log.warn(message, error)
+ def warn(message:String, error:Throwable): Unit = log.warn(message, error)
 
- def ifError(message: => String) = if (log.isErrorEnabled) error(message)
- def error(message:String, values:AnyRef*) =
+ def ifError(message: => String): Unit = if (log.isErrorEnabled) error(message)
+ def error(message:String, values:AnyRef*): Unit =
      log.error(message, values)
- def error(message:String, error:Throwable) = log.error(message, error)
-}
+ def error(message:String, error:Throwable): Unit = log.error(message, error)
+}

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

@@ -41,14 +41,13 @@ trait NodeOperations { self: Redis =>
   // SLAVEOF
   // The SLAVEOF command can change the replication settings of a slave on the fly.
   def slaveof(options: Any): Boolean = options match {
-    case (h: String, p: Int) => {
+    case (h: String, p: Int) =>
       send("SLAVEOF", List(h, p))(asBoolean)
-    }
-    case _ => setAsMaster
+    case _ => setAsMaster()
   }
 
   @deprecated("use slaveof", "1.2.0") def slaveOf(options: Any): Boolean = slaveof(options)
 
-  private def setAsMaster: Boolean =
+  private def setAsMaster(): Boolean =
     send("SLAVEOF", List("NO", "ONE"))(asBoolean)
 }

src/main/scala/com/redis/Operations.scala

@@ -26,8 +26,8 @@ trait Operations { self: Redis =>
       , (if (desc) List("DESC") else Nil)
       , (if (alpha) List("ALPHA") else Nil)
       , by.map(b => List("BY", b)).getOrElse(Nil)
-      , get.map(g => List("GET", g)).flatMap(x=>x)
-      ).flatMap(x=>x)
+      , get.flatMap(g => List("GET", g))
+      ).flatten
   }
 
   // SORT with STORE
@@ -128,12 +128,11 @@ trait Operations { self: Redis =>
   // SELECT (index)
   // selects the DB to connect, defaults to 0 (zero).
   def select(index: Int): Boolean =
-    send("SELECT", List(index))(asBoolean match {
-      case true => {
-        db = index
-        true
-      }
-      case _ => false
+    send("SELECT", List(index))(if (asBoolean) {
+      db = index
+      true
+    } else {
+      false
     })
 
 

src/main/scala/com/redis/Pool.scala

@@ -8,7 +8,7 @@ private [redis] class RedisClientFactory(val host: String, val port: Int, val da
   extends PoolableObjectFactory[RedisClient] {
 
   // when we make an object it's already connected
-  def makeObject = {
+  def makeObject: RedisClient = {
     new RedisClient(host, port, database, secret, timeout)
   }
 
@@ -20,7 +20,7 @@ private [redis] class RedisClientFactory(val host: String, val port: Int, val da
 
   // noop: we want to have it connected
   def passivateObject(rc: RedisClient): Unit = {}
-  def validateObject(rc: RedisClient) = rc.connected == true
+  def validateObject(rc: RedisClient): Boolean = rc.connected
 
   // noop: it should be connected already
   def activateObject(rc: RedisClient): Unit = {}
@@ -37,7 +37,7 @@ object RedisClientPool {
 class RedisClientPool(val host: String, val port: Int, val maxIdle: Int = 8, val database: Int = 0, val secret: Option[Any] = None, val timeout : Int = 0, 
     val maxConnections: Int = RedisClientPool.UNLIMITED_CONNECTIONS, val whenExhaustedBehavior: Byte = RedisClientPool.WHEN_EXHAUSTED_BLOCK, val poolWaitTimeout: Long = 3000) {
   val pool = new GenericObjectPool(new RedisClientFactory(host, port, database, secret, timeout), maxConnections, whenExhaustedBehavior, poolWaitTimeout, maxIdle, false, true)
-  override def toString = host + ":" + String.valueOf(port)
+  override def toString: String = host + ":" + String.valueOf(port)
 
   def withClient[T](body: RedisClient => T) = {
     val client = pool.borrowObject
@@ -49,14 +49,14 @@ class RedisClientPool(val host: String, val port: Int, val maxIdle: Int = 8, val
   }
 
   // close pool & free resources
-  def close = pool.close
+  def close: Unit = pool.close()
 }
 
 /**
  *
- * @param poolname must be unique
+ * @param node
  */
 class IdentifiableRedisClientPool(val node: ClusterNode)
   extends RedisClientPool (node.host, node.port, node.maxIdle, node.database, node.secret,node.timeout){
-  override def toString = node.nodename
+  override def toString: String = node.nodename
 }