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,53 @@
+package sttp.client4.ox.ws
+
+import ox.*
+import ox.channels.Channel
+import ox.channels.ChannelClosed
+import ox.channels.Sink
+import ox.channels.Source
+import sttp.client4.ws.SyncWebSocket
+import sttp.ws.WebSocketFrame
+
+import scala.util.control.NonFatal
+
+  extension (ws: SyncWebSocket)(using Ox) def asSourceAndSink: (Source[WebSocketFrame], Sink[WebSocketFrame]) = 
+    val srcChannel = Channel.bufferedDefault[WebSocketFrame]
+    fork {
+      repeatWhile {
+        try ws.receive() match 
+          case WebSocketFrame.Close(status, _) if status > 1001 => 
+              srcChannel.error(new Exception(s"WebSocket closed with status $status"))
+              false
+          case _: WebSocketFrame.Close => 
+              srcChannel.done()
+              false
+          case frame =>
+              srcChannel.send(frame)
+              true
+        catch case NonFatal(err) => 
+          srcChannel.error(err)          
+          false
+        }
+    }
+    val sinkChannel = Channel.bufferedDefault[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())
+    }
+    (srcChannel, sinkChannel)

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

@@ -0,0 +1,31 @@
+package sttp.client4.examples
+
+import _root_.ox.*
+import _root_.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) = ws.asSourceAndSink
+      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()