Add test for ReadableStream interop

Author: armanbilge

build.sbt

@@ -159,7 +159,8 @@ def configureTest(project: Project): Project =
       libraryDependencies ++= Seq(
         "org.http4s" %%% "http4s-client-testkit" % http4sVersion,
         "org.scalameta" %%% "munit" % munitVersion % Test,
-        "org.typelevel" %%% "munit-cats-effect" % munitCEVersion % Test
+        "org.typelevel" %%% "munit-cats-effect" % munitCEVersion % Test,
+        "org.typelevel" %%% "scalacheck-effect-munit" % "2.0.0-M2" % Test
       ),
       Compile / unmanagedSourceDirectories +=
         (LocalRootProject / baseDirectory).value / "tests" / "src" / "main" / "scala",

dom/src/main/scala/org/http4s/dom/package.scala

@@ -48,14 +48,12 @@ package object dom {
 
   implicit def blobEncoder[F[_]](implicit F: Async[F]): EntityEncoder[F, Blob] =
     EntityEncoder.entityBodyEncoder.contramap { blob =>
-      readReadableStream[F](F.delay(blob.stream()), cancelAfterUse = true)
+      readReadableStream[F](F.delay(blob.stream()))
     }
 
   implicit def readableStreamEncoder[F[_]: Async]
       : EntityEncoder[F, ReadableStream[Uint8Array]] =
-    EntityEncoder.entityBodyEncoder.contramap { rs =>
-      readReadableStream(rs.pure, cancelAfterUse = true)
-    }
+    EntityEncoder.entityBodyEncoder.contramap { rs => readReadableStream(rs.pure) }
 
   private[dom] def fromDomResponse[F[_]](response: DomResponse)(
       implicit F: Async[F]): F[Response[F]] =
@@ -64,7 +62,7 @@ package object dom {
         status = status,
         headers = fromDomHeaders(response.headers),
         body = Stream.fromOption(Option(response.body)).flatMap { rs =>
-          readReadableStream[F](rs.pure, cancelAfterUse = true)
+          readReadableStream[F](rs.pure)
         }
       )
     }
@@ -84,9 +82,8 @@ package object dom {
       headers.map { header => header(0) -> header(1) }.toList
     )
 
-  private def readReadableStream[F[_]](
-      readableStream: F[ReadableStream[Uint8Array]],
-      cancelAfterUse: Boolean
+  private[dom] def readReadableStream[F[_]](
+      readableStream: F[ReadableStream[Uint8Array]]
   )(implicit F: Async[F]): Stream[F, Byte] = {
     def read(readableStream: ReadableStream[Uint8Array]) =
       Stream
@@ -102,10 +99,7 @@ package object dom {
           }
         }
 
-    if (cancelAfterUse)
-      Stream.bracketCase(readableStream)(cancelReadableStream(_, _)).flatMap(read(_))
-    else
-      Stream.eval(readableStream).flatMap(read(_))
+    Stream.bracketCase(readableStream)(cancelReadableStream(_, _)).flatMap(read(_))
   }
 
   private[dom] def cancelReadableStream[F[_], A](

tests/src/test/scala/org/http4s/ReadableStreamSuite.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2021 http4s.org
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.http4s.dom
+
+import cats.effect.IO
+import fs2.Chunk
+import fs2.Stream
+import munit.CatsEffectSuite
+import munit.ScalaCheckEffectSuite
+import org.scalacheck.effect.PropF.forAllF
+
+class ReadableStreamSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
+
+  test("to/read ReadableStream") {
+    forAllF { (chunks: Vector[Vector[Byte]]) =>
+      Stream
+        .emits(chunks)
+        .map(Chunk.seq(_))
+        .unchunks
+        .through(in => Stream.resource(toReadableStream[IO](in)))
+        .flatMap(readable => readReadableStream(IO(readable)))
+        .compile
+        .toVector
+        .assertEquals(chunks.flatten)
+    }
+  }
+
+}