Streaming: do some cleaning

Author: sunheehnus

src/main/scala/com/redislabs/provider/redis/rdd/RedisRDD.scala

@@ -191,7 +191,7 @@ class RedisKeysRDD(sc: SparkContext,
                    val keyPattern: String = "*",
                    val partitionNum: Int = 3,
                    val keys: Array[String] = null)
-  extends RDD[String](sc, Seq.empty) with Logging with Keys {
+  extends RDD[String](sc, Seq.empty) with Keys {
 
   override protected def getPreferredLocations(split: Partition): Seq[String] = {
     Seq(split.asInstanceOf[RedisPartition].redisConfig.initialAddr)

src/main/scala/com/redislabs/provider/redis/redisFunctions.scala

@@ -7,7 +7,6 @@ import org.apache.spark.rdd.RDD
 import com.redislabs.provider.redis.rdd._
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.StreamingContext
-import redis.clients.util.RedisInputStream
 
 /**
   * RedisContext extends sparkContext's functionality with redis functions
@@ -218,13 +217,6 @@ class RedisContext(@transient val sc: SparkContext) extends Serializable {
     }
   }
 
-  def createRedisStream(ssc: StreamingContext, keys: Array[String], storageLevel: StorageLevel)
-                       (implicit redisConfig: RedisConfig = new RedisConfig(new RedisEndpoint(sc.getConf))): Unit = {
-    new RedisInputDStream(ssc, keys, redisConfig, storageLevel)
-  }
-
-
-
   /**
     * @param kvs Pair RDD of K/V
     * @param ttl time to live
@@ -393,7 +385,15 @@ object RedisContext extends Serializable {
   }
 }
 
+class RedisStreamingContext(@transient val ssc: StreamingContext) extends Serializable {
+  def createRedisStream(keys: Array[String], storageLevel: StorageLevel)
+                       (implicit redisConfig: RedisConfig = new RedisConfig(new
+                           RedisEndpoint(ssc.sparkContext.getConf))) = {
+    new RedisInputDStream(ssc, keys, redisConfig, storageLevel)
+  }
+}
 trait RedisFunctions {
   implicit def toRedisContext(sc: SparkContext): RedisContext = new RedisContext(sc)
+  implicit def toRedisStreamingContext(ssc: StreamingContext): RedisStreamingContext = new RedisStreamingContext(ssc)
 }
 

src/main/scala/com/redislabs/provider/redis/streaming/RedisInputDStream.scala

@@ -11,8 +11,6 @@ import redis.clients.jedis._
 
 import scala.util.control.NonFatal
 
-import scala.collection.JavaConversions._
-
 class RedisInputDStream(_ssc: StreamingContext,
                         keys: Array[String],
                         redisConfig: RedisConfig,
@@ -48,8 +46,10 @@ private class RedisReceiver(keys: Array[String],
       val keys: Array[String] = Array(key)
       try {
         while(!isStopped) {
-          conn.blpop(keys: _*).toList.foreach{ x =>
-            store((key, x))
+          val res = conn.blpop(Integer.MAX_VALUE, keys:_*)
+          res match {
+            case null => "Really?"
+            case _ => store((res.get(0), res.get(1)))
           }
         }
       } catch {

src/main/scala/com/redislabs/provider/redis/streaming/RedisUtils.scala

@@ -0,0 +1,8 @@
+package com.redislabs.provider.redis.streaming
+
+/**
+  * Created by sunheehnus on 16/3/29.
+  */
+class RedisUtils {
+
+}