fix 'fs.io.readInputStreamGeneric' overallocation of underlying buffers

Author: seigert

io/jvm/src/main/scala/fs2/io/ioplatform.scala

@@ -43,10 +43,10 @@ private[fs2] trait ioplatform extends iojvmnative {
       F.pure(System.in),
       F.delay(new Array[Byte](bufSize)),
       false
-    ) { (is, buf) =>
+    ) { (is, buf, off) =>
       F.async[Int] { cb =>
         F.delay {
-          val task: Runnable = () => cb(Right(is.read(buf)))
+          val task: Runnable = () => cb(Right(is.read(buf, off, buf.length - off)))
           stdinExecutor.submit(task)
         }.map { fut =>
           Some(F.delay {

io/shared/src/main/scala/fs2/io/io.scala

@@ -45,7 +45,7 @@ package object io extends ioplatform {
       fis,
       F.delay(new Array[Byte](chunkSize)),
       closeAfterUse
-    )((is, buf) => F.blocking(is.read(buf)))
+    )((is, buf, off) => F.blocking(is.read(buf, off, buf.length - off)))
 
   private[io] def readInputStreamCancelable[F[_]](
       fis: F[InputStream],
@@ -57,7 +57,7 @@ package object io extends ioplatform {
       fis,
       F.delay(new Array[Byte](chunkSize)),
       closeAfterUse
-    )((is, buf) => F.blocking(is.read(buf)).cancelable(cancel))
+    )((is, buf, off) => F.blocking(is.read(buf, off, buf.length - off)).cancelable(cancel))
 
   /** Reads all bytes from the specified `InputStream` with a buffer size of `chunkSize`.
     * Set `closeAfterUse` to false if the `InputStream` should not be closed after use.
@@ -76,28 +76,31 @@ package object io extends ioplatform {
       fis,
       F.pure(new Array[Byte](chunkSize)),
       closeAfterUse
-    )((is, buf) => F.blocking(is.read(buf)))
+    )((is, buf, off) => F.blocking(is.read(buf, off, buf.length - off)))
 
-  private def readBytesFromInputStream[F[_]](is: InputStream, buf: Array[Byte])(
-      read: (InputStream, Array[Byte]) => F[Int]
+  private def readBytesFromInputStream[F[_]](is: InputStream, buf: Array[Byte], offset: Int)(
+      read: (InputStream, Array[Byte], Int) => F[Int]
   )(implicit
       F: Sync[F]
   ): F[Option[Chunk[Byte]]] =
-    read(is, buf).map { numBytes =>
-      if (numBytes < 0) None
-      else if (numBytes == 0) Some(Chunk.empty)
-      else if (numBytes < buf.size) Some(Chunk.array(buf, 0, numBytes))
-      else Some(Chunk.array(buf))
+    read(is, buf, offset).flatMap { numBytes =>
+      if (numBytes < 0) {
+        if (offset == 0) F.pure(None)
+        else F.pure(Some(Chunk.array(buf, 0, offset)))
+      } else {
+        if (offset + numBytes == buf.size) F.pure(Some(Chunk.array(buf)))
+        else readBytesFromInputStream(is, buf, offset + numBytes)(read)
+      }
     }
 
   private[fs2] def readInputStreamGeneric[F[_]](
       fis: F[InputStream],
       buf: F[Array[Byte]],
       closeAfterUse: Boolean
-  )(read: (InputStream, Array[Byte]) => F[Int])(implicit F: Sync[F]): Stream[F, Byte] = {
+  )(read: (InputStream, Array[Byte], Int) => F[Int])(implicit F: Sync[F]): Stream[F, Byte] = {
     def useIs(is: InputStream) =
       Stream
-        .eval(buf.flatMap(b => readBytesFromInputStream(is, b)(read)))
+        .eval(buf.flatMap(b => readBytesFromInputStream(is, b, 0)(read)))
         .repeat
         .unNoneTerminate
         .flatMap(c => Stream.chunk(c))