Add FetchFs2Backend (#2726)

Supports websocket and sse streaming. Heavily based of the
FetchZioBackend.

Before submitting pull request:
- [X] Check if the project compiles by running `sbt compile`
- [X] Verify docs compilation by running `sbt compileDocs`
- [X] Check if tests pass by running `sbt test`
- [X] Format code by running `sbt scalafmt`

Author: alfabetacain

build.sbt

@@ -459,7 +459,10 @@ lazy val fs2 = (projectMatrix in file("effects/fs2"))
       )
     )
   )
-  .jsPlatform(scalaVersions = scala2And3, settings = commonJsSettings)
+  .jsPlatform(
+    scalaVersions = scala2And3,
+    settings = commonJsSettings ++ commonJsBackendSettings ++ browserChromeTestSettings ++ testServerSettings
+  )
 
 lazy val monix = (projectMatrix in file("effects/monix"))
   .settings(

docs/backends/fs2.md

@@ -126,6 +126,11 @@ This backend is built on top of [Armeria](https://armeria.dev/docs/client-http).
 Armeria's [ClientFactory](https://armeria.dev/docs/client-factory) manages connections and protocol-specific properties.
 Please visit [the official documentation](https://armeria.dev/docs/client-factory) to learn how to configure it.
 
+## Using JavaScript
+
+The Fs2 backend is also available for the JS platform, see [the `FetchBackend` documentation](javascript/fetch.md).
+The `FetchFs2Backend` companion object contains methods to create the backend directly.
+
 ## Streaming
 
 The fs2 backends support streaming for any instance of the `cats.effect.Effect` typeclass, such as `cats.effect.IO`. If `IO` is used then the type of supported streams is `fs2.Stream[IO, Byte]`. The streams capability is represented as `sttp.client4.fs2.Fs2Streams`.

docs/backends/javascript/fetch.md

@@ -70,6 +70,21 @@ And create the backend instance:
 val backend = FetchCatsBackend[IO]()
 ```
 
+## fs2-based
+
+Any effect implementing the cats-effect `Async` typeclass can be used. To use, add the following dependency to 
+your project:
+
+```
+"com.softwaremill.sttp.client4" %%% "fs2" % "@VERSION@"
+```
+
+And create the backend instance:
+
+```scala
+val backend = FetchFs2Backend[IO]()
+```
+
 ## Node.js
 
 ### CommonJS module
@@ -161,4 +176,4 @@ The backend supports both regular and streaming [websockets](../../other/websock
 
 ## Server-sent events
 
-Received data streams can be parsed to a stream of server-sent events (SSE), when using the [Monix](../monix.md) and [ZIO](../zio.md) variants - the respective documentation pages contain appropriate SSE examples.
+Received data streams can be parsed to a stream of server-sent events (SSE), when using the [Monix](../monix.md), [ZIO](../zio.md), or [Fs2](../fs2.md) variants - the respective documentation pages contain appropriate SSE examples.

effects/fs2/src/main/scalajs/sttp.client4.impl.fs2/FetchFs2Backend.scala

@@ -0,0 +1,123 @@
+package sttp.client4.impl.fs2
+
+import scala.scalajs.js.{Function1, Thenable, |}
+import sttp.client4.testing.WebSocketStreamBackendStub
+import fs2.Stream
+import cats.syntax.all._
+import cats.effect.syntax.all._
+import sttp.client4.fetch.FetchOptions
+import sttp.client4.fetch.AbstractFetchBackend
+import sttp.capabilities.fs2.Fs2Streams
+import cats.effect.kernel.Async
+import sttp.client4.impl.cats.CatsMonadAsyncError
+import sttp.ws.WebSocketFrame.Data
+import sttp.ws.WebSocket
+import sttp.ws.WebSocketFrame
+import scala.scalajs.js
+import sttp.client4.internal.ConvertFromFuture
+import scala.concurrent.Future
+import sttp.client4.WebSocketStreamBackend
+import scala.scalajs.js.typedarray._
+import fs2.Chunk
+import org.scalajs.dom.Request
+import org.scalajs.dom.Response
+import org.scalajs.dom.BodyInit
+
+class FetchFs2Backend[F[_]: Async] private (fetchOptions: FetchOptions, customizeRequest: Request => Request)
+    extends AbstractFetchBackend[F, Fs2Streams[F]](fetchOptions, customizeRequest, new CatsMonadAsyncError)
+    with WebSocketStreamBackend[F, Fs2Streams[F]] {
+
+  override val streams: Fs2Streams[F] = Fs2Streams[F]
+
+  override protected def addCancelTimeoutHook[T](result: F[T], cancel: () => Unit, cleanup: () => Unit): F[T] = {
+    result
+      .onCancel(Async[F].delay(cancel()).voidError)
+      .guarantee(
+        Async[F].delay(cleanup()).voidError
+      )
+  }
+
+  override protected def compileWebSocketPipe(
+      ws: WebSocket[F],
+      pipe: streams.Pipe[Data[_], WebSocketFrame]
+  ): F[Unit] = {
+    Fs2WebSockets.handleThroughPipe[F](ws)(pipe)
+  }
+
+  override protected def handleResponseAsStream(response: Response): F[(streams.BinaryStream, () => F[Unit])] = {
+    Async[F].delay {
+      lazy val reader = response.body.getReader()
+      def read() = Async[F].fromFuture(Async[F].delay(reader.read().toFuture))
+
+      val cancel = Async[F].async[Unit] { callback =>
+        Async[F]
+          .delay {
+            // copied from https://github.com/zio/zio/blob/fca6870f42b1d2b06ebf0e6c13b975bccea72f13/core/js/src/main/scala/zio/ZIOPlatformSpecific.scala#L58
+            val onFulfilled: Function1[Unit, Unit | Thenable[Unit]] = new scala.Function1[Unit, Unit | Thenable[Unit]] {
+              def apply(a: Unit): Unit | Thenable[Unit] = callback(Right(a))
+            }
+            val onRejected: Function1[Any, Unit | Thenable[Unit]] = new scala.Function1[Any, Unit | Thenable[Unit]] {
+              def apply(e: Any): Unit | Thenable[Unit] =
+                callback(Left(e match {
+                  case t: Throwable => t
+                  case _            => js.JavaScriptException(e)
+                }))
+            }
+            reader.cancel("Response body reader cancelled").`then`[Unit](onFulfilled, js.defined(onRejected))
+          }
+          .as(None)
+      }
+
+      val stream = Stream
+        .unfoldChunkEval(()) { case () =>
+          read()
+            .map { chunk =>
+              if (chunk.done) {
+                Option.empty
+              } else {
+                val bytes = new Int8Array(chunk.value.buffer).toArray
+                Option((Chunk.array(bytes), ()))
+              }
+            }
+        }
+      (stream.onComplete(Stream.exec(cancel.voidError)), () => cancel.voidError)
+
+    }
+  }
+
+  override protected def handleStreamBody(s: streams.BinaryStream): F[js.UndefOr[BodyInit]] = {
+    s.chunks
+      .fold(Chunk.empty[Byte])(_ ++ _)
+      .compile
+      .last
+      .map {
+        case None      => js.undefined
+        case Some(res) => res.toArray.toTypedArray
+      }
+  }
+
+  override def convertFromFuture: ConvertFromFuture[F] = {
+    new ConvertFromFuture[F] {
+      override def apply[T](f: Future[T]): F[T] = Async[F].fromFuture(monad.unit(f))
+    }
+  }
+}
+
+object FetchFs2Backend {
+
+  def apply[F[_]: Async](
+      fetchOptions: FetchOptions = FetchOptions.Default,
+      customizeRequest: Request => Request = identity
+  ): WebSocketStreamBackend[F, Fs2Streams[F]] = {
+    new FetchFs2Backend[F](fetchOptions, customizeRequest)
+  }
+
+  /** Create a stub backend for testing, which uses the given [[F]] response wrapper
+    *
+    * See [[WebSocketStreamBackendStub]] for details on how to configure stub responses.
+    */
+  def stub[F[_]: Async]: WebSocketStreamBackendStub[F, Fs2Streams[F]] = WebSocketStreamBackendStub(
+    new CatsMonadAsyncError[F]
+  )
+
+}

effects/fs2/src/test/scalajs/sttp/client4/impl/fs2/FetchFs2HttpTest.scala

@@ -0,0 +1,20 @@
+package sttp.client4.impl.fs2
+
+import cats.effect.IO
+import sttp.capabilities.fs2.Fs2Streams
+import sttp.client4.StreamBackend
+import sttp.client4.testing.{AbstractFetchHttpTest, ConvertToFuture}
+import cats.effect.unsafe.implicits.global
+
+class FetchFs2HttpTest extends AbstractFetchHttpTest[IO, Fs2Streams[IO]] {
+  override implicit val convertToFuture: ConvertToFuture[IO] = sttp.client4.impl.cats.convertCatsIOToFuture()
+
+  override val backend: StreamBackend[IO, Fs2Streams[IO]] = FetchFs2Backend()
+
+  override protected def supportsCustomMultipartContentType = false
+
+  override protected def supportsCustomMultipartEncoding = false
+
+  override def timeoutToNone[T](t: IO[T], timeoutMillis: Int): IO[Option[T]] = super.timeoutToNone(t, timeoutMillis)
+
+}

effects/fs2/src/test/scalajs/sttp/client4/impl/fs2/FetchFs2StreamingTest.scala

@@ -0,0 +1,15 @@
+package sttp.client4.impl.fs2
+
+import cats.effect.IO
+import sttp.capabilities.fs2.Fs2Streams
+import sttp.client4.StreamBackend
+import sttp.client4.testing.streaming.StreamingTest
+
+class FetchFs2StreamingTest extends StreamingTest[IO, Fs2Streams[IO]] with Fs2StreamingTest {
+  override val streams: Fs2Streams[IO] = Fs2Streams[IO]
+
+  override val backend: StreamBackend[IO, Fs2Streams[IO]] = FetchFs2Backend()
+
+  override protected def supportsStreamingMultipartParts: Boolean = false
+
+}

effects/fs2/src/test/scalajs/sttp/client4/impl/fs2/FetchFs2WebSocketTest.scala

@@ -0,0 +1,40 @@
+package sttp.client4.impl.fs2
+
+import sttp.client4.impl.cats.CatsTestBase
+import cats.effect.IO
+import fs2.Stream
+import sttp.capabilities.fs2.Fs2Streams
+import sttp.client4.WebSocketStreamBackend
+import sttp.client4.testing.websocket.{WebSocketStreamingTest, WebSocketTest}
+import sttp.ws.WebSocketFrame
+
+import scala.concurrent.ExecutionContext
+import scala.scalajs.concurrent.JSExecutionContext.Implicits.queue
+
+class FetchFs2WebSocketTest
+    extends WebSocketTest[IO]
+    with WebSocketStreamingTest[IO, Fs2Streams[IO]]
+    with CatsTestBase {
+  implicit override def executionContext: ExecutionContext = queue
+  override def throwsWhenNotAWebSocket: Boolean = true
+  override def supportsReadingWebSocketResponseHeaders: Boolean = false
+  override def supportsReadingSubprotocolWebSocketResponseHeader: Boolean = false
+
+  override val backend: WebSocketStreamBackend[IO, Fs2Streams[IO]] = FetchFs2Backend()
+
+  override val streams: Fs2Streams[IO] = Fs2Streams[IO]
+
+  override def prepend(item: WebSocketFrame.Text)(
+      to: streams.Pipe[WebSocketFrame.Data[_], WebSocketFrame]
+  ): streams.Pipe[WebSocketFrame.Data[_], WebSocketFrame] =
+    to.andThen(rest => Stream(item) ++ rest)
+
+  override def fromTextPipe(function: String => WebSocketFrame): streams.Pipe[WebSocketFrame.Data[_], WebSocketFrame] =
+    Fs2WebSockets.fromTextPipe(function)
+
+  override def functionToPipe(
+      f: WebSocketFrame.Data[_] => Option[WebSocketFrame]
+  ): streams.Pipe[WebSocketFrame.Data[_], WebSocketFrame] = _.map(f).collect { case Some(v) =>
+    v
+  }
+}