Chunk - Introduce a Constant Chunk subclass.

core/shared/src/main/scala/fs2/Chunk.scala

@@ -550,6 +550,31 @@ object Chunk
     override def map[O2](f: O => O2): Chunk[O2] = singleton(f(value))
   }
 
+  def constant[A](value: A, size: Int): Chunk[A] =
+    if (size <= 0) empty
+    else if (size == 1) singleton(value)
+    else new Constant(value, size)
+
+  final class Constant[A](value: A, override val size: Int) extends Chunk[A] {
+
+    def apply(i: Int): A =
+      if (0 <= i && i < size) value else throw new IndexOutOfBoundsException()
+
+    def copyToArray[O2 >: A](xs: Array[O2], start: Int): Unit = {
+
+      @tailrec
+      def go(ix: Int): Unit =
+        if (ix < size) {
+          xs(start + ix) = value
+          go(ix + 1)
+        }
+      go(0)
+    }
+
+    protected def splitAtChunk_(n: Int): (Chunk[A], Chunk[A]) =
+      constant(value, n) -> constant(value, size - n)
+  }
+
   /** Creates a chunk backed by a vector. */
   def vector[O](v: Vector[O]): Chunk[O] = indexedSeq(v)
 

core/shared/src/main/scala/fs2/Stream.scala

@@ -3104,7 +3104,7 @@ object Stream extends StreamLowPriority {
     * }}}
     */
   def constant[F[x] >: Pure[x], O](o: O, chunkSize: Int = 256): Stream[F, O] =
-    chunk(Chunk.seq(List.fill(chunkSize)(o))).repeat
+    chunk(Chunk.constant(o, chunkSize)).repeat
 
   /** A continuous stream of the elapsed time, computed using `System.nanoTime`.
     * Note that the actual granularity of these elapsed times depends on the OS, for instance