Ox WebSockets

build.sbt

@@ -186,6 +186,7 @@ lazy val allAggregates = projectsWithOptionalNative ++
   fs2Ce2.projectRefs ++
   fs2.projectRefs ++
   monix.projectRefs ++
+  ox.projectRefs ++
   scalaz.projectRefs ++
   zio1.projectRefs ++
   zio.projectRefs ++
@@ -231,6 +232,7 @@ lazy val allAggregates = projectsWithOptionalNative ++
   slf4jBackend.projectRefs ++
   examplesCe2.projectRefs ++
   examples.projectRefs ++
+  examples3.projectRefs ++
   docs.projectRefs ++
   testServer.projectRefs
 
@@ -439,6 +441,17 @@ lazy val monix = (projectMatrix in file("effects/monix"))
     settings = commonJsSettings ++ commonJsBackendSettings ++ browserChromeTestSettings ++ testServerSettings
   )
 
+lazy val ox = (projectMatrix in file("effects/ox"))
+  .settings(commonJvmSettings)
+  .settings(
+    name := "ox",
+    libraryDependencies ++= Seq(
+      "com.softwaremill.ox" %% "core" % "0.1.1"
+    )
+  )
+  .jvmPlatform(scalaVersions = scala3)
+  .dependsOn(core)
+
 lazy val zio1 = (projectMatrix in file("effects/zio1"))
   .settings(
     name := "zio1",
@@ -1038,6 +1051,21 @@ lazy val examples = (projectMatrix in file("examples"))
     slf4jBackend
   )
 
+lazy val examples3 = (projectMatrix in file("examples3"))
+  .settings(commonJvmSettings)
+  .settings(
+    name := "examples3",
+    publish / skip := true,
+    libraryDependencies ++= Seq(
+      logback
+    )
+  )
+  .jvmPlatform(scalaVersions = scala3)
+  .dependsOn(
+    core,
+    ox
+  )
+
 //TODO this should be invoked by compilation process, see #https://github.com/scalameta/mdoc/issues/355
 val compileDocs: TaskKey[Unit] = taskKey[Unit]("Compiles docs module throwing away its output")
 compileDocs := {

effects/ox/src/main/scala/sttp/client4/ox/ws/OxWebSockets.scala

@@ -0,0 +1,92 @@
+package sttp.client4.ox.ws
+
+import ox.*
+import ox.channels.ChannelClosed
+import ox.channels.Sink
+import ox.channels.Source
+import ox.channels.StageCapacity
+import sttp.client4.ws.SyncWebSocket
+import sttp.ws.WebSocketFrame
+
+import scala.util.control.NonFatal
+
+  def asSource(ws: SyncWebSocket, concatenateFragmented: Boolean = true, pongOnPing: Boolean = true)(using Ox, StageCapacity): Source[WebSocketFrame] =
+    val srcChannel = StageCapacity.newChannel[WebSocketFrame]
+    fork {
+      repeatWhile {
+        try
+          ws.receive() match
+            case frame: WebSocketFrame.Data[_] =>
+              srcChannel.send(frame)
+              true
+            case WebSocketFrame.Close(status, msg) if status > 1001 =>
+              srcChannel.error(new WebSocketClosedWithError(status, msg))
+              false
+            case _: WebSocketFrame.Close =>
+              srcChannel.done()
+              false
+            case ping: WebSocketFrame.Ping =>
+              if pongOnPing then ws.send(WebSocketFrame.Pong(ping.payload))            
+              true
+            case _: WebSocketFrame.Pong =>
+              // ignore pongs
+              true
+        catch
+          case NonFatal(err) =>
+            srcChannel.error(err)
+            false
+      }
+    }.discard
+    optionallyConcatenateFrames(srcChannel, concatenateFragmented)
+
+  def asSink(ws: SyncWebSocket)(using Ox, StageCapacity): Sink[WebSocketFrame] =
+    val sinkChannel = StageCapacity.newChannel[WebSocketFrame]
+    fork {
+      try
+        repeatWhile {
+          sinkChannel.receiveOrClosed() match
+            case closeFrame: WebSocketFrame.Close =>
+              ws.send(closeFrame) // TODO should we just let 'send' throw exceptions?
+              false
+            case frame: WebSocketFrame =>
+              ws.send(frame)
+              true
+            case ChannelClosed.Done =>
+              ws.send(WebSocketFrame.close)
+              false
+            case ChannelClosed.Error(err) =>
+              // There's no proper "client error" status. Statuses 4000+ are available for custom cases
+              ws.send(WebSocketFrame.Close(4000, "Client error")) // TODO should we bother the server with client error?
+              false
+        }
+      finally uninterruptible(ws.close())
+    }.discard
+    sinkChannel
+
+  def asSourceAndSink(ws: SyncWebSocket, concatenateFragmented: Boolean = true, pongOnPing: Boolean = true)(using Ox, StageCapacity): (Source[WebSocketFrame], Sink[WebSocketFrame]) =
+    (asSource(ws, concatenateFragmented, pongOnPing), asSink(ws))
+
+final case class WebSocketClosedWithError(statusCode: Int, msg: String) extends Exception(s"WebSocket closed with status $statusCode: $msg")
+
+private def optionallyConcatenateFrames(s: Source[WebSocketFrame], doConcatenate: Boolean)(using
+    Ox
+): Source[WebSocketFrame] =
+  if doConcatenate then
+    type Accumulator = Option[Either[Array[Byte], String]]
+    s.mapStateful(() => None: Accumulator) {
+      case (None, f: WebSocketFrame.Ping)                       => (None, Some(f))
+      case (None, f: WebSocketFrame.Pong)                       => (None, Some(f))
+      case (None, f: WebSocketFrame.Close)                      => (None, Some(f))
+      case (None, f: WebSocketFrame.Data[_]) if f.finalFragment => (None, Some(f))
+      case (None, f: WebSocketFrame.Text)                       => (Some(Right(f.payload)), None)
+      case (None, f: WebSocketFrame.Binary)                     => (Some(Left(f.payload)), None)
+      case (Some(Left(acc)), f: WebSocketFrame.Binary) if f.finalFragment =>
+        (None, Some(f.copy(payload = acc ++ f.payload)))
+      case (Some(Left(acc)), f: WebSocketFrame.Binary) if !f.finalFragment => (Some(Left(acc ++ f.payload)), None)
+      case (Some(Right(acc)), f: WebSocketFrame.Text) if f.finalFragment =>
+        (None, Some(f.copy(payload = acc + f.payload)))
+      case (Some(Right(acc)), f: WebSocketFrame.Text) if !f.finalFragment => (Some(Right(acc + f.payload)), None)
+      case (acc, f) =>
+        throw new IllegalStateException(s"Cannot accumulate web socket frames. Accumulator: $acc, frame: $f.")
+    }.collectAsView { case Some(f: WebSocketFrame) => f }
+  else s

examples3/src/main/scala/sttp/client4/examples/WebSocketOx.scala

@@ -0,0 +1,32 @@
+package sttp.client4.examples
+
+import ox.*
+import ox.channels.Source
+import sttp.client4.*
+import sttp.client4.ox.ws.*
+import sttp.client4.ws.SyncWebSocket
+import sttp.client4.ws.sync.*
+import sttp.ws.WebSocketFrame
+
+@main def wsOxExample =
+  def useWebSocket(ws: SyncWebSocket): Unit =
+    supervised {
+      val inputs = Source.fromValues(1, 2, 3).map(i => WebSocketFrame.text(s"Frame no $i"))
+      val (wsSource, wsSink) = asSourceAndSink(ws)
+      fork {
+        inputs.pipeTo(wsSink)
+      }
+      wsSource.foreach { frame =>
+        println(s"RECEIVED: $frame")
+      }
+    }
+
+  val backend = DefaultSyncBackend()
+  try
+    basicRequest
+      .get(uri"wss://ws.postman-echo.com/raw")
+      .response(asWebSocket(useWebSocket))
+      .send(backend)
+      .discard
+  finally
+    backend.close()