split the streaming API into two separated ones

Author: sunheehnus

Diff for: README.md

@@ -251,15 +251,33 @@ sc.toRedisZSET(zsetRDD, zsetName)
 The above example demonstrates storing data in Redis in a Sorted Set. The `zsetRDD` in the example should contain pairs of members and their scores, whereas `zsetName` is the name for that key.
 
 ### Streaming
-Spark-Redis support streaming data from Redis instance/cluster, currently streaming data are fetched from Redis' List by the `blpop` command. Users are required to provide an array which stores all the List names they are interested in. The [storageLevel](http://spark.apache.org/docs/latest/streaming-programming-guide.html#data-serialization) is `MEMORY_AND_DISK_SER_2` by default, you can change it on your demand. The stream is a `(listName, value)` stream by default, but if you don't care about which list feeds the value, set the `withStreamName` as `false` and you will get the only `value` stream.
+Spark-Redis support streaming data from Redis instance/cluster, currently streaming data are fetched from Redis' List by the `blpop` command. Users are required to provide an array which stores all the List names they are interested in. The [storageLevel](http://spark.apache.org/docs/latest/streaming-programming-guide.html#data-serialization) is `MEMORY_AND_DISK_SER_2` by default, you can change it on your demand.
+`createRedisStream` will create a `(listName, value)` stream, but if you don't care about which list feeds the value, you can use `createRedisStreamWithoutListname` to get the only `value` stream.
+
+Use the following to get a `(listName, value)` stream from `foo` and `bar` list
+
+```
+import org.apache.spark.streaming.{Seconds, StreamingContext}
+import org.apache.spark.storage.StorageLevel
+import com.redislabs.provider.redis._
+val ssc = new StreamingContext(sc, Seconds(1))
+val redisStream = ssc.createRedisStream(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2)
+redisStream.print
+ssc.awaitTermination()
+```
+
+
+Use the following to get a `value` stream from `foo` and `bar` list
+
 ```
 import org.apache.spark.streaming.{Seconds, StreamingContext}
 import org.apache.spark.storage.StorageLevel
 import com.redislabs.provider.redis._
 val ssc = new StreamingContext(sc, Seconds(1))
-val redisStream = ssc.createRedisStream(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2, withStreamName = true)
+val redisStream = ssc.createRedisStreamWithoutListname(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2)
+redisStream.print
+ssc.awaitTermination()
 ```
-The above example will get a (listName, value) stream from `foo` list and `bar` list
 
 
 ### Connecting to Multiple Redis Clusters/Instances

Diff for: src/main/scala/com/redislabs/provider/redis/redisFunctions.scala

@@ -394,19 +394,24 @@ class RedisStreamingContext(@transient val ssc: StreamingContext) extends Serial
   /**
     * @param keys an Array[String] which consists all the Lists we want to listen to
     * @param storageLevel the receiver' storage tragedy of received data, default as MEMORY_AND_DISK_2
-    * @param withStreamName if it is true the streaming will be (listName, value),
-    *                       else the streaming will be only (value)
-    *                       default as true
+    * @return a stream of (listname, value)
     */
   def createRedisStream(keys: Array[String],
-                        storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_2,
-                        withStreamName: Boolean = true)
+                        storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_2)
                        (implicit redisConfig: RedisConfig = new RedisConfig(new
                            RedisEndpoint(ssc.sparkContext.getConf))) = {
-    withStreamName match {
-      case true => new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[(String, String)])
-      case false => new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[String])
-    }
+      new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[(String, String)])
+  }
+  /**
+    * @param keys an Array[String] which consists all the Lists we want to listen to
+    * @param storageLevel the receiver' storage tragedy of received data, default as MEMORY_AND_DISK_2
+    * @return a stream of (value)
+    */
+  def createRedisStreamWithoutListname(keys: Array[String],
+                        storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_2)
+                       (implicit redisConfig: RedisConfig = new RedisConfig(new
+                           RedisEndpoint(ssc.sparkContext.getConf))) = {
+      new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[String])
   }
 }
 

Diff for: src/main/scala/com/redislabs/provider/redis/streaming/RedisInputDStream.scala

@@ -1,14 +1,16 @@
 package com.redislabs.provider.redis.streaming
 
+import com.redislabs.provider.redis.RedisConfig
 import org.apache.curator.utils.ThreadUtils
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.StreamingContext
 import org.apache.spark.streaming.receiver.Receiver
 import org.apache.spark.streaming.dstream.ReceiverInputDStream
+
 import redis.clients.jedis._
+
 import scala.reflect.{ClassTag, classTag}
 import scala.util.control.NonFatal
-import com.redislabs.provider.redis.RedisConfig
 
 class RedisInputDStream[T: ClassTag](_ssc: StreamingContext,
                                      keys: Array[String],