Spark Streaming stop application mechanism

This is to continue the discussion, that started at mailing list regarding idiomatic way to stop spark streaming app inside a receiver. 

Following is the example that shows how can a streaming application can be stopped using stopApp function. The example will stop streaming app once an end of stream is reached on receiver. I am not sure if its a great thing to do. But can be useful incase we want to detect a situation like connect exception and we would not want to continue. Currently streaming continues even if receiver did not start at all ! 



```scala
package spark.examples

import spark.storage.StorageLevel
import spark.streaming.dstream.NetworkReceiver
import java.net.Socket
import java.io.{BufferedReader, InputStreamReader}
import spark.streaming.{Seconds, StreamingContext}
import spark.streaming.StreamingContext._

class SocketTextStreamReceiver(host: String,
                               port: Int
                                ) extends NetworkReceiver[String] {

  protected lazy val blocksGenerator: BlockGenerator =
    new BlockGenerator(StorageLevel.MEMORY_ONLY_SER_2)

  protected def onStart() = {
    blocksGenerator.start()
    val socket = new Socket(host, port)
    val dataInputStream = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"))
    var data: String = ""
    while (data != null) {
      data = dataInputStream.readLine();
      blocksGenerator += data
    }
    println("Reached end of stream!!")
    stopApp() //<-- this can be called to stop the streaming application.
  }

  protected def onStop() {
    blocksGenerator.stop()
  }

}

object DocSample {

  def main(args: Array[String]) {
    val ssc = new StreamingContext("local[2]", "WordCountCustomStreamSource",
      Seconds(1))
    val lines = ssc.networkStream[String](new SocketTextStreamReceiver(
      "localhost", 8445))
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

    wordCounts.print()
    ssc.start()

  }
}
```