count() now uses a transform instead of ConstantInputDStream

streaming/src/main/scala/spark/streaming/DStream.scala

@@ -441,10 +441,7 @@ abstract class DStream[T: ClassManifest] (
    * Return a new DStream in which each RDD has a single element generated by counting each RDD
    * of this DStream.
    */
-  def count(): DStream[Long] = {
-   val zero = new ConstantInputDStream(context, context.sparkContext.makeRDD(Seq((null, 0L)), 1))
-   this.map(_ => (null, 1L)).union(zero).reduceByKey(_ + _).map(_._2)
-  }
+  def count(): DStream[Long] = this.map(_ => (null, 1L)).transform(_.union(context.sparkContext.makeRDD(Seq((null, 0L)), 1))).reduceByKey(_ + _).map(_._2)
 
   /**
    * Return a new DStream in which each RDD contains the counts of each distinct value in