Merge branch 'main' into fix/io-read_input_stream-overallocation

Author: mpilquist

.scalafmt.conf

@@ -1,4 +1,4 @@
-version = "3.7.14"
+version = "3.7.15"
 
 style = default
 

build.sbt

@@ -265,7 +265,7 @@ lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   .settings(
     name := "fs2-core",
     libraryDependencies ++= Seq(
-      "org.scodec" %%% "scodec-bits" % "1.1.37",
+      "org.scodec" %%% "scodec-bits" % "1.1.38",
       "org.typelevel" %%% "cats-core" % "2.10.0",
       "org.typelevel" %%% "cats-effect" % "3.5.2",
       "org.typelevel" %%% "cats-effect-laws" % "3.5.2" % Test,

io/js/src/main/scala/fs2/io/NodeStream.scala

@@ -72,4 +72,5 @@ trait Duplex extends Readable with Writable {
   protected[io] override def destroy(): this.type = js.native
 }
 
+@deprecated("No longer raised", "3.9.3")
 final class StreamDestroyedException private[io] () extends IOException

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

@@ -109,30 +109,30 @@ private[fs2] trait ioplatform {
   /** `Pipe` that converts a stream of bytes to a stream that will emit a single `Readable`,
     * that ends whenever the resulting stream terminates.
     */
-  def toReadable[F[_]](implicit F: Async[F]): Pipe[F, Byte, Readable] =
-    in =>
-      Stream
-        .resource(mkDuplex(in))
-        .flatMap { case (duplex, out) =>
-          Stream
-            .emit(duplex)
-            .merge(out.drain)
-            .concurrently(
-              Stream.eval(
-                F.async_[Unit](cb =>
-                  duplex.end { e =>
-                    cb(e.filterNot(_ == null).toLeft(()).leftMap(js.JavaScriptException))
-                  }
-                )
-              )
-            )
-        }
-        .adaptError { case IOException(ex) => ex }
+  def toReadable[F[_]: Async]: Pipe[F, Byte, Readable] =
+    in => Stream.resource(toReadableResource(in))
 
   /** Like [[toReadable]] but returns a `Resource` rather than a single element stream.
     */
-  def toReadableResource[F[_]: Async](s: Stream[F, Byte]): Resource[F, Readable] =
-    s.through(toReadable).compile.resource.lastOrError
+  def toReadableResource[F[_]](s: Stream[F, Byte])(implicit F: Async[F]): Resource[F, Readable] =
+    mkDuplex(s)
+      .flatMap { case (duplex, out) =>
+        out
+          .concurrently(
+            Stream.eval(
+              F.async_[Unit](cb =>
+                duplex.end { e =>
+                  cb(e.filterNot(_ == null).toLeft(()).leftMap(js.JavaScriptException))
+                }
+              )
+            )
+          )
+          .compile
+          .drain
+          .background
+          .as(duplex)
+      }
+      .adaptError { case IOException(ex) => ex }
 
   /** Writes all bytes to the specified `Writable`.
     */
@@ -206,7 +206,7 @@ private[fs2] trait ioplatform {
       errorDispatcher <- Dispatcher.sequential[F]
       readQueue <- Queue.bounded[F, Option[Chunk[Byte]]](1).toResource
       writeChannel <- Channel.synchronous[F, Chunk[Byte]].toResource
-      error <- F.deferred[Throwable].toResource
+      interrupt <- F.deferred[Either[Throwable, Unit]].toResource
       duplex <- Resource.make {
         F.delay {
           new facade.stream.Duplex(
@@ -236,10 +236,9 @@ private[fs2] trait ioplatform {
 
               var destroy = { (_, err, cb) =>
                 errorDispatcher.unsafeRunAndForget {
-                  error
+                  interrupt
                     .complete(
-                      Option(err)
-                        .fold[Exception](new StreamDestroyedException)(js.JavaScriptException(_))
+                      Option(err).map(js.JavaScriptException(_)).toLeft(())
                     ) *> F.delay(cb(null))
                 }
               }
@@ -254,10 +253,9 @@ private[fs2] trait ioplatform {
       }
       drainIn = in.enqueueNoneTerminatedChunks(readQueue).drain
       out = writeChannel.stream.unchunks
-        .concurrently(Stream.eval(error.get.flatMap(F.raiseError[Unit])))
     } yield (
       duplex,
-      drainIn.merge(out).adaptError { case IOException(ex) => ex }
+      drainIn.merge(out).interruptWhen(interrupt).adaptError { case IOException(ex) => ex }
     )
 
   /** Stream of bytes read asynchronously from standard input. */

io/js/src/test/scala/fs2/io/IoPlatformSuite.scala

@@ -114,4 +114,16 @@ class IoPlatformSuite extends Fs2Suite {
       .timeoutTo(100.millis, IO.unit)
   }
 
+  test("Destroying Node.js stream without error does not raise an exception") {
+    Stream
+      .never[IO]
+      .through {
+        toDuplexAndRead[IO] { duplex =>
+          IO(duplex.destroy())
+        }
+      }
+      .compile
+      .drain
+  }
+
 }

project/build.properties

@@ -1 +1 @@
-sbt.version=1.9.6
+sbt.version=1.9.7