add the receiver stop function

When application is stoping , it will pop last a data from redis. It throws exception because it can't store the last data,and this data will lose. This commition has solved this bug in a way that stop the connect. It has a deficiencies.When stopd the connection ,it will throws ConnectException.

Author: stupidxian

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

@@ -29,28 +29,37 @@ private class RedisReceiver[T: ClassTag](keys: Array[String],
                                          redisConfig: RedisConfig,
                                          streamType: Class[T])
   extends Receiver[T](storageLevel) {
+    
+  var jedisConnect: Jedis = null
 
   def onStart() {
     val executorPool = ThreadUtils.newFixedThreadPool(keys.length, "BlockLists Streaming")
     try {
       /* start a executor for each interested List */
-      keys.foreach{ key =>
-        executorPool.submit(new MessageHandler(redisConfig.connectionForKey(key), key))
+      keys.foreach{ key =>{
+        jedisConnect = redisConfig.connectionForKey(key)
+        executorPool.submit(new MessageHandler(jedisConnect, key))
+      }
       }
     } finally {
       executorPool.shutdown()
     }
   }
 
   def onStop() {
+    /* quit the connect*/
+    if (jedisConnect != null) {
+      jedisConnect.quit()
+      jedisConnect = null
+    }
   }
 
   private class MessageHandler(conn: Jedis, key: String) extends Runnable {
     def run() {
       try {
         while(!isStopped) {
           val response = conn.blpop(2, key)
-          if (response == null) {
+          if (response == null || response.isEmpty) {
 
           } else if (classTag[T] == classTag[String]) {
             store(response.get(1).asInstanceOf[T])