Update ox to 0.5.1 (#2310)

build.sbt

@@ -150,7 +150,7 @@ val zio2Version = "2.1.11"
 val zio1InteropRsVersion = "1.3.12"
 val zio2InteropRsVersion = "2.0.2"
 
-val oxVersion = "0.4.0"
+val oxVersion = "0.5.1"
 val sttpModelVersion = "1.7.11"
 val sttpSharedVersion = "1.3.22"
 

effects/ox/src/main/scala/sttp/client4/impl/ox/sse/OxServerSentEvents.scala

@@ -1,14 +1,13 @@
 package sttp.client4.impl.ox.sse
 
-import ox.*
-import ox.channels.Source
+import ox.flow.Flow
 import sttp.model.sse.ServerSentEvent
 
 import java.io.InputStream
 
 object OxServerSentEvents:
-  def parse(is: InputStream)(using Ox): Source[ServerSentEvent] =
-    Source
+  def parse(is: InputStream): Flow[ServerSentEvent] =
+    Flow
       .fromInputStream(is)
       .linesUtf8
       .mapStatefulConcat(() => List.empty[String])(
@@ -18,5 +17,5 @@ object OxServerSentEvents:
           else None
         }
       )
-      .filterAsView(_.nonEmpty)
+      .filter(_.nonEmpty)
       .map(ServerSentEvent.parse)

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

@@ -6,21 +6,23 @@ import sttp.client4.ws.SyncWebSocket
 import sttp.ws.WebSocketFrame
 
 import scala.util.control.NonFatal
+import ox.flow.Flow
 
-/** Converts a [[SyncWebSocket]] into a pair of `Source` of server responses and a `Sink` for client requests. The
+/** Converts a [[SyncWebSocket]] into a pair of [[Source]] of server responses and a [[Sink]] for client requests. The
   * `Source` starts receiving frames immediately, its internal buffer size can be adjusted with an implicit
-  * [[ox.channels.StageCapacity]]. Make sure that the `Source` is contiunually read. This will guarantee that
-  * server-side Close signal is received and handled. If you don't want to process frames from the server, you can at
+  * [[ox.channels.BufferCapacity]]. Make sure that the `Source` is contiunually read. This will guarantee that
+  * server-side close signal is received and handled. If you don't want to process frames from the server, you can at
   * least handle it with a `fork { source.drain() }`.
   *
   * You don't need to manually call `ws.close()` when using this approach, this will be handled automatically
   * underneath, according to following rules:
-  *   - If the request `Sink` is closed due to an upstream error, a Close frame is sent, and the `Source` with incoming
-  *     responses gets completed as `Done`.
-  *   - If the request `Sink` completes as `Done`, a `Close` frame is sent, and the response `Sink` keeps receiving
+  *   - If the request sink is closed due to an upstream error, a close frame is sent. The response sink keeps receiving
+  *     responses, until the enclosing [[Ox]] scope ends (that is controlled by the caller). When this happens, the fork
+  *     which populates the response channel will be interrupted.
+  *   - If the request sink completes as done, a close frame is sent. As above, the response sink keeps receiving
   *     responses until the server closes communication.
-  *   - If the response `Source` is closed by a Close frome from the server or due to an error, the request Sink is
-  *     closed as `Done`, which will still send all outstanding buffered frames, and then finish.
+  *   - If the response source is closed by a close frame from the server or due to an error, the request sink is closed
+  *     as done. This will attempt to send all outstanding buffered frames, unless the enclosing scope ends beforehand).
   *
   * @param ws
   *   a `SyncWebSocket` where the underlying `Sink` will send requests, and where the `Source` will pull responses from.
@@ -29,38 +31,30 @@ import scala.util.control.NonFatal
   */
 def asSourceAndSink(ws: SyncWebSocket, concatenateFragmented: Boolean = true)(using
     Ox,
-    StageCapacity
+    BufferCapacity
 ): (Source[WebSocketFrame], Sink[WebSocketFrame]) =
-  val requestsChannel = StageCapacity.newChannel[WebSocketFrame]
-  val responsesChannel = StageCapacity.newChannel[WebSocketFrame]
-  fork {
-    try
+  val requestsChannel = BufferCapacity.newChannel[WebSocketFrame]
+
+  val responsesChannel = Flow
+    .usingEmit[WebSocketFrame] { emit =>
       repeatWhile {
         ws.receive() match
-          case frame: WebSocketFrame.Data[_] =>
-            responsesChannel.sendOrClosed(frame) match
-              case _: ChannelClosed => false
-              case _                => true
-          case WebSocketFrame.Close(status, msg) if status > 1001 =>
-            responsesChannel.errorOrClosed(new WebSocketClosedWithError(status, msg)).discard
-            false
-          case _: WebSocketFrame.Close =>
-            responsesChannel.doneOrClosed().discard
-            false
+          case frame: WebSocketFrame.Data[_]                      => emit(frame); true
+          case WebSocketFrame.Close(status, msg) if status > 1001 => throw new WebSocketClosedWithError(status, msg)
+          case _: WebSocketFrame.Close                            => false
           case ping: WebSocketFrame.Ping =>
             requestsChannel.sendOrClosed(WebSocketFrame.Pong(ping.payload)).discard
-            // Keep receiving even if pong couldn't be send due to closed request channel. We want to process
+            // Keep receiving even if pong couldn't be sent due to closed request channel. We want to process
             // whatever responses there are still coming from the server until it signals the end with a Close frome.
             true
           case _: WebSocketFrame.Pong =>
             // ignore pongs
             true
       }
-    catch
-      case NonFatal(err) =>
-        responsesChannel.errorOrClosed(err).discard
-    finally requestsChannel.doneOrClosed().discard
-  }.discard
+    }
+    .pipe(optionallyConcatenateFrames(_, concatenateFragmented))
+    .onComplete(requestsChannel.doneOrClosed().discard)
+    .runToChannel()
 
   fork {
     try
@@ -78,7 +72,7 @@ def asSourceAndSink(ws: SyncWebSocket, concatenateFragmented: Boolean = true)(us
           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"))
-            responsesChannel.doneOrClosed().discard
+            // Assuming the responsesChannel fork will get interrupted because the enclosing scope will end
             false
       }
     catch
@@ -87,17 +81,17 @@ def asSourceAndSink(ws: SyncWebSocket, concatenateFragmented: Boolean = true)(us
         if (!responsesChannel.isClosedForReceive) requestsChannel.errorOrClosed(err).discard
   }.discard
 
-  (optionallyConcatenateFrames(responsesChannel, concatenateFragmented), requestsChannel)
+  (responsesChannel, requestsChannel)
 
 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
+private def optionallyConcatenateFrames(f: Flow[WebSocketFrame], doConcatenate: Boolean)(using
     Ox
-): Source[WebSocketFrame] =
+): Flow[WebSocketFrame] =
   if doConcatenate then
     type Accumulator = Option[Either[Array[Byte], String]]
-    s.mapStateful(() => None: Accumulator) {
+    f.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))
@@ -115,5 +109,5 @@ private def optionallyConcatenateFrames(s: Source[WebSocketFrame], doConcatenate
           s"Unexpected WebSocket frame received during concatenation. Frame received: ${f.getClass
               .getSimpleName()}, accumulator type: ${acc.map(_.getClass.getSimpleName)}"
         )
-    }.collectAsView { case Some(f: WebSocketFrame) => f }
-  else s
+    }.collect { case Some(f: WebSocketFrame) => f }
+  else f

effects/ox/src/test/scala/sttp/client4/impl/ox/ws/OxSseTest.scala

@@ -3,7 +3,6 @@ package sttp.client4.impl.ox.sse
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
-import ox.*
 import sttp.client4.*
 import sttp.client4.testing.HttpTest.*
 import sttp.model.sse.ServerSentEvent
@@ -14,7 +13,7 @@ class OxServerSentEventsTest extends AnyFlatSpec with Matchers with BeforeAndAft
 
   behavior of "OxServerSentEvents"
 
-  it should "parse SSEs" in supervised {
+  it should "parse SSEs" in {
     val sseData = "text1 in line1\ntext2 in line2"
     val expectedEvent = ServerSentEvent(data = Some(sseData), eventType = Some("test-event"), retry = Some(42000))
     val expectedEvents =
@@ -23,7 +22,7 @@ class OxServerSentEventsTest extends AnyFlatSpec with Matchers with BeforeAndAft
       .post(uri"$endpoint/sse/echo3")
       .body(sseData)
       .response(asInputStreamAlways { is =>
-        OxServerSentEvents.parse(is).take(3).toList shouldBe expectedEvents
+        OxServerSentEvents.parse(is).take(3).runToList() shouldBe expectedEvents
         ()
       })
       .send(backend)

effects/ox/src/test/scala/sttp/client4/impl/ox/ws/OxWebSocketsTest.scala

@@ -7,6 +7,7 @@ import org.scalatest.concurrent.Eventually
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import ox.*
+import ox.flow.Flow
 import ox.channels.ChannelClosed
 import ox.channels.Sink
 import ox.channels.Source
@@ -62,7 +63,7 @@ class OxWebSocketTest extends AnyFlatSpec with BeforeAndAfterAll with Matchers w
         val (wsSource, wsSink) = asSourceAndSink(ws)
         wsSink.send(WebSocketFrame.text("test1"))
         wsSink.error(new Exception("failed source"))
-        eventually(wsSource.isClosedForReceiveDetail shouldBe Some(ChannelClosed.Done))
+        eventually(wsSource.isClosedForReceiveDetail should matchPattern { case Some(ChannelClosed.Error(_)) => })
       })
       .send(backend)
   }
@@ -117,7 +118,7 @@ class OxWebSocketTest extends AnyFlatSpec with BeforeAndAfterAll with Matchers w
       .response(asWebSocket { ws =>
         val (wsSource, wsSink) = asSourceAndSink(ws, concatenateFragmented = false)
         sendText(wsSink, 1)
-        wsSource.take(3).toList shouldBe List(
+        Flow.fromSource(wsSource).take(3).runToList() shouldBe List(
           WebSocketFrame.Text("echo: ", false, None),
           WebSocketFrame.Text("test1", false, None),
           WebSocketFrame.Text("", true, None)