Merge pull request #1913 from softwaremill/sync-ws-api

Introduce a synchronous web socket interface

core/src/main/scala/sttp/client4/SttpApi.scala

@@ -255,72 +255,4 @@ trait SttpApi extends SttpExtensions with UriInterpolator {
       StreamBody(s)(b),
       contentType = Some(MediaType.ApplicationOctetStream)
     )
-
-  // websocket response specifications
-
-  def asWebSocket[F[_], T](f: WebSocket[F] => F[T]): WebSocketResponseAs[F, Either[String, T]] =
-    asWebSocketEither(asStringAlways, asWebSocketAlways(f))
-
-  def asWebSocketWithMetadata[F[_], T](
-      f: (WebSocket[F], ResponseMetadata) => F[T]
-  ): WebSocketResponseAs[F, Either[String, T]] =
-    asWebSocketEither(asStringAlways, asWebSocketAlwaysWithMetadata(f))
-
-  def asWebSocketAlways[F[_], T](f: WebSocket[F] => F[T]): WebSocketResponseAs[F, T] =
-    asWebSocketAlwaysWithMetadata((w, _) => f(w))
-
-  def asWebSocketAlwaysWithMetadata[F[_], T](f: (WebSocket[F], ResponseMetadata) => F[T]): WebSocketResponseAs[F, T] =
-    WebSocketResponseAs(ResponseAsWebSocket(f))
-
-  def asWebSocketUnsafe[F[_]]: WebSocketResponseAs[F, Either[String, WebSocket[F]]] =
-    asWebSocketEither(asStringAlways, asWebSocketAlwaysUnsafe)
-
-  def asWebSocketAlwaysUnsafe[F[_]]: WebSocketResponseAs[F, WebSocket[F]] =
-    WebSocketResponseAs(ResponseAsWebSocketUnsafe())
-
-  def fromMetadata[F[_], T](
-      default: ResponseAs[T],
-      conditions: ConditionalResponseAs[WebSocketResponseAs[F, T]]*
-  ): WebSocketResponseAs[F, T] =
-    WebSocketResponseAs(ResponseAsFromMetadata(conditions.map(_.map(_.delegate)).toList, default.delegate))
-
-  /** Uses the `onSuccess` response specification for 101 responses (switching protocols) on JVM/Native, 200 responses
-    * on JS. Otherwise, use the `onError` specification.
-    */
-  def asWebSocketEither[F[_], A, B](
-      onError: ResponseAs[A],
-      onSuccess: WebSocketResponseAs[F, B]
-  ): WebSocketResponseAs[F, Either[A, B]] =
-    SttpExtensions.asWebSocketEitherPlatform(onError, onSuccess)
-
-  // websocket stream response specifications
-
-  def asWebSocketStream[S](
-      s: Streams[S]
-  )(p: s.Pipe[WebSocketFrame.Data[_], WebSocketFrame]): WebSocketStreamResponseAs[Either[String, Unit], S] =
-    asWebSocketEither(asStringAlways, asWebSocketStreamAlways(s)(p))
-
-  def asWebSocketStreamAlways[S](s: Streams[S])(
-      p: s.Pipe[WebSocketFrame.Data[_], WebSocketFrame]
-  ): WebSocketStreamResponseAs[Unit, S] = WebSocketStreamResponseAs[Unit, S](ResponseAsWebSocketStream(s, p))
-
-  def fromMetadata[T, S](
-      default: ResponseAs[T],
-      conditions: ConditionalResponseAs[WebSocketStreamResponseAs[T, S]]*
-  ): WebSocketStreamResponseAs[T, S] =
-    WebSocketStreamResponseAs[T, S](
-      ResponseAsFromMetadata(conditions.map(_.map(_.delegate)).toList, default.delegate)
-    )
-
-  /** Uses the `onSuccess` response specification for 101 responses (switching protocols), and the `onError`
-    * specification otherwise.
-    */
-  def asWebSocketEither[A, B, S](
-      onError: ResponseAs[A],
-      onSuccess: WebSocketStreamResponseAs[B, S]
-  ): WebSocketStreamResponseAs[Either[A, B], S] =
-    fromMetadata(
-      onError.map(Left(_)),
-      ConditionalResponseAs(_.code == StatusCode.SwitchingProtocols, onSuccess.map(Right(_)))
-    ).showAs(s"either(${onError.show}, ${onSuccess.show})")
 }

core/src/main/scala/sttp/client4/SttpWebSocketAsyncApi.scala

@@ -0,0 +1,41 @@
+package sttp.client4
+
+import sttp.model.ResponseMetadata
+import sttp.ws.WebSocket
+
+trait SttpWebSocketAsyncApi {
+  def asWebSocket[F[_], T](f: WebSocket[F] => F[T]): WebSocketResponseAs[F, Either[String, T]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlways(f))
+
+  def asWebSocketWithMetadata[F[_], T](
+      f: (WebSocket[F], ResponseMetadata) => F[T]
+  ): WebSocketResponseAs[F, Either[String, T]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlwaysWithMetadata(f))
+
+  def asWebSocketAlways[F[_], T](f: WebSocket[F] => F[T]): WebSocketResponseAs[F, T] =
+    asWebSocketAlwaysWithMetadata((w, _) => f(w))
+
+  def asWebSocketAlwaysWithMetadata[F[_], T](f: (WebSocket[F], ResponseMetadata) => F[T]): WebSocketResponseAs[F, T] =
+    WebSocketResponseAs(ResponseAsWebSocket(f))
+
+  def asWebSocketUnsafe[F[_]]: WebSocketResponseAs[F, Either[String, WebSocket[F]]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlwaysUnsafe)
+
+  def asWebSocketAlwaysUnsafe[F[_]]: WebSocketResponseAs[F, WebSocket[F]] =
+    WebSocketResponseAs(ResponseAsWebSocketUnsafe())
+
+  def fromMetadata[F[_], T](
+      default: ResponseAs[T],
+      conditions: ConditionalResponseAs[WebSocketResponseAs[F, T]]*
+  ): WebSocketResponseAs[F, T] =
+    WebSocketResponseAs(ResponseAsFromMetadata(conditions.map(_.map(_.delegate)).toList, default.delegate))
+
+  /** Uses the `onSuccess` response specification for 101 responses (switching protocols) on JVM/Native, 200 responses
+    * on JS. Otherwise, use the `onError` specification.
+    */
+  def asWebSocketEither[F[_], A, B](
+      onError: ResponseAs[A],
+      onSuccess: WebSocketResponseAs[F, B]
+  ): WebSocketResponseAs[F, Either[A, B]] =
+    SttpExtensions.asWebSocketEitherPlatform(onError, onSuccess)
+}

core/src/main/scala/sttp/client4/SttpWebSocketStreamApi.scala

@@ -0,0 +1,37 @@
+package sttp.client4
+
+import sttp.capabilities.Streams
+import sttp.model.StatusCode
+import sttp.ws.WebSocketFrame
+
+trait SttpWebSocketStreamApi {
+  def asWebSocketStream[S](
+      s: Streams[S]
+  )(p: s.Pipe[WebSocketFrame.Data[_], WebSocketFrame]): WebSocketStreamResponseAs[Either[String, Unit], S] =
+    asWebSocketEither(asStringAlways, asWebSocketStreamAlways(s)(p))
+
+  def asWebSocketStreamAlways[S](s: Streams[S])(
+      p: s.Pipe[WebSocketFrame.Data[_], WebSocketFrame]
+  ): WebSocketStreamResponseAs[Unit, S] = WebSocketStreamResponseAs[Unit, S](ResponseAsWebSocketStream(s, p))
+
+  def fromMetadata[T, S](
+      default: ResponseAs[T],
+      conditions: ConditionalResponseAs[WebSocketStreamResponseAs[T, S]]*
+  ): WebSocketStreamResponseAs[T, S] =
+    WebSocketStreamResponseAs[T, S](
+      ResponseAsFromMetadata(conditions.map(_.map(_.delegate)).toList, default.delegate)
+    )
+
+  /** Uses the `onSuccess` response specification for 101 responses (switching protocols), and the `onError`
+    * specification otherwise.
+    */
+  def asWebSocketEither[A, B, S](
+      onError: ResponseAs[A],
+      onSuccess: WebSocketStreamResponseAs[B, S]
+  ): WebSocketStreamResponseAs[Either[A, B], S] =
+    fromMetadata(
+      onError.map(Left(_)),
+      ConditionalResponseAs(_.code == StatusCode.SwitchingProtocols, onSuccess.map(Right(_)))
+    ).showAs(s"either(${onError.show}, ${onSuccess.show})")
+
+}

core/src/main/scala/sttp/client4/SttpWebSocketSyncApi.scala

@@ -0,0 +1,44 @@
+package sttp.client4
+
+import sttp.client4.ws.SyncWebSocket
+import sttp.model.ResponseMetadata
+import sttp.ws.WebSocket
+
+trait SttpWebSocketSyncApi {
+  def asWebSocket[T](f: SyncWebSocket => T): WebSocketResponseAs[Identity, Either[String, T]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlways(f))
+
+  def asWebSocketWithMetadata[T](
+      f: (SyncWebSocket, ResponseMetadata) => T
+  ): WebSocketResponseAs[Identity, Either[String, T]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlwaysWithMetadata(f))
+
+  def asWebSocketAlways[T](f: SyncWebSocket => T): WebSocketResponseAs[Identity, T] =
+    asWebSocketAlwaysWithMetadata((w, _) => f(w))
+
+  def asWebSocketAlwaysWithMetadata[T](
+      f: (SyncWebSocket, ResponseMetadata) => T
+  ): WebSocketResponseAs[Identity, T] =
+    WebSocketResponseAs[Identity, T](ResponseAsWebSocket[Identity, T]((ws, m) => f(new SyncWebSocket(ws), m)))
+
+  def asWebSocketUnsafe: WebSocketResponseAs[Identity, Either[String, SyncWebSocket]] =
+    asWebSocketEither(asStringAlways, asWebSocketAlwaysUnsafe)
+
+  def asWebSocketAlwaysUnsafe: WebSocketResponseAs[Identity, SyncWebSocket] =
+    WebSocketResponseAs[Identity, WebSocket[Identity]](ResponseAsWebSocketUnsafe()).map(new SyncWebSocket(_))
+
+  def fromMetadata[T](
+      default: ResponseAs[T],
+      conditions: ConditionalResponseAs[WebSocketResponseAs[Identity, T]]*
+  ): WebSocketResponseAs[Identity, T] =
+    WebSocketResponseAs(ResponseAsFromMetadata(conditions.map(_.map(_.delegate)).toList, default.delegate))
+
+  /** Uses the `onSuccess` response specification for 101 responses (switching protocols) on JVM/Native, 200 responses
+    * on JS. Otherwise, use the `onError` specification.
+    */
+  def asWebSocketEither[A, B](
+      onError: ResponseAs[A],
+      onSuccess: WebSocketResponseAs[Identity, B]
+  ): WebSocketResponseAs[Identity, Either[A, B]] =
+    SttpExtensions.asWebSocketEitherPlatform(onError, onSuccess)
+}

core/src/main/scala/sttp/client4/ws/SyncWebSocket.scala

@@ -0,0 +1,140 @@
+package sttp.client4.ws
+
+import sttp.client4.Identity
+import sttp.model.Headers
+import sttp.ws.{WebSocket, WebSocketClosed, WebSocketFrame}
+
+/** Allows interacting with a web socket. Interactions can happen:
+  *
+  *   - on the frame level, by sending and receiving raw [[WebSocketFrame]] s
+  *   - using the provided `receive*` methods to obtain concatenated data frames, or string/byte payloads, and the
+  *     `send*` method to send string/binary frames.
+  *
+  * The `send*` and `receive*` methods may result in a failed effect, with either one of [[sttp.ws.WebSocketException]]
+  * exceptions, or a backend-specific exception. Specifically, they will fail with [[WebSocketClosed]] if the web socket
+  * is closed.
+  *
+  * See the `either` and `eitherClose` method to lift web socket closed events to the value level.
+  */
+class SyncWebSocket(val delegate: WebSocket[Identity]) {
+
+  /** Receive the next frame from the web socket. This can be a data frame, or a control frame including
+    * [[WebSocketFrame.Close]]. After receiving a close frame, no further interactions with the web socket should
+    * happen.
+    *
+    * However, not all implementations expose the close frame, and web sockets might also get closed without the proper
+    * close frame exchange. In such cases, as well as when invoking `receive`/`send` after receiving a close frame, a
+    * [[WebSocketClosed]] exception will be thrown.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber). Because web socket frames might be fragmented,
+    * calling this method concurrently might result in threads/fibers receiving fragments of the same frame.
+    */
+  def receive(): WebSocketFrame = delegate.receive()
+
+  /** Sends a web socket frame. Can be safely called from multiple threads.
+    *
+    * May result in an exception, in case of a network error, or if the socket is closed.
+    */
+  def send(f: WebSocketFrame, isContinuation: Boolean = false): Unit = delegate.send(f, isContinuation)
+
+  def isOpen(): Boolean = delegate.isOpen()
+
+  /** Receive a single data frame, ignoring others. The frame might be a fragment. Will throw [[WebSocketClosed]] if the
+    * web socket is closed, or if a close frame is received.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber).
+    *
+    * @param pongOnPing
+    *   Should a [[WebSocketFrame.Pong]] be sent when a [[WebSocketFrame.Ping]] is received.
+    */
+  def receiveDataFrame(pongOnPing: Boolean = true): WebSocketFrame.Data[_] = delegate.receiveDataFrame(pongOnPing)
+
+  /** Receive a single text data frame, ignoring others. The frame might be a fragment. To receive whole messages, use
+    * [[receiveText]]. Will throw [[WebSocketClosed]] if the web socket is closed, or if a close frame is received.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber).
+    *
+    * @param pongOnPing
+    *   Should a [[WebSocketFrame.Pong]] be sent when a [[WebSocketFrame.Ping]] is received.
+    */
+  def receiveTextFrame(pongOnPing: Boolean = true): WebSocketFrame.Text = delegate.receiveTextFrame(pongOnPing)
+
+  /** Receive a single binary data frame, ignoring others. The frame might be a fragment. To receive whole messages, use
+    * [[receiveBinary]]. Will throw [[WebSocketClosed]] if the web socket is closed, or if a close frame is received.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber).
+    *
+    * @param pongOnPing
+    *   Should a [[WebSocketFrame.Pong]] be sent when a [[WebSocketFrame.Ping]] is received.
+    */
+  def receiveBinaryFrame(pongOnPing: Boolean = true): WebSocketFrame.Binary = delegate.receiveBinaryFrame(pongOnPing)
+
+  /** Receive a single text message (which might come from multiple, fragmented frames). Ignores non-text frames and
+    * returns combined results. Will throw [[WebSocketClosed]] if the web socket is closed, or if a close frame is
+    * received.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber).
+    *
+    * @param pongOnPing
+    *   Should a [[WebSocketFrame.Pong]] be sent when a [[WebSocketFrame.Ping]] is received.
+    */
+  def receiveText(pongOnPing: Boolean = true): String = delegate.receiveText(pongOnPing)
+
+  /** Receive a single binary message (which might come from multiple, fragmented frames). Ignores non-binary frames and
+    * returns combined results. Will throw [[WebSocketClosed]] if the web socket is closed, or if a close frame is
+    * received.
+    *
+    * *Should be only called sequentially!* (from a single thread/fiber).
+    *
+    * @param pongOnPing
+    *   Should a [[WebSocketFrame.Pong]] be sent when a [[WebSocketFrame.Ping]] is received.
+    */
+  def receiveBinary(pongOnPing: Boolean): Array[Byte] = delegate.receiveBinary(pongOnPing)
+
+  /** Extracts the received close frame (if available) as the left side of an either, or returns the original result on
+    * the right.
+    *
+    * Will throw [[WebSocketClosed]] if the web socket is closed, but no close frame is available.
+    *
+    * @param f
+    *   The effect describing web socket interactions.
+    */
+  def eitherClose[T](f: => T): Either[WebSocketFrame.Close, T] =
+    try Right(f)
+    catch {
+      case WebSocketClosed(Some(close)) => Left(close)
+    }
+
+  /** Returns an effect computing a:
+    *
+    *   - `Left` if the web socket is closed - optionally with the received close frame (if available).
+    *   - `Right` with the original result otherwise.
+    *
+    * Will never throw a [[WebSocketClosed]].
+    *
+    * @param f
+    *   The effect describing web socket interactions.
+    */
+  def either[T](f: => T): Either[Option[WebSocketFrame.Close], T] =
+    try Right(f)
+    catch {
+      case WebSocketClosed(close) => Left(close)
+    }
+
+  /** Sends a web socket frame with the given payload. Can be safely called from multiple threads.
+    *
+    * May result in an exception, in case of a network error, or if the socket is closed.
+    */
+  def sendText(payload: String): Unit = delegate.sendText(payload)
+
+  /** Sends a web socket frame with the given payload. Can be safely called from multiple threads.
+    *
+    * May result in an exception, in case of a network error, or if the socket is closed.
+    */
+  def sendBinary(payload: Array[Byte]): Unit = delegate.sendBinary(payload)
+
+  /** Idempotent when used sequentially. */
+  def close(): Unit = delegate.close()
+
+  def upgradeHeaders: Headers = delegate.upgradeHeaders
+}

core/src/main/scala/sttp/client4/ws/package.scala

@@ -0,0 +1,7 @@
+package sttp.client4
+
+package object ws {
+  object async extends SttpWebSocketAsyncApi
+  object sync extends SttpWebSocketSyncApi
+  object stream extends SttpWebSocketStreamApi
+}

core/src/main/scalajs/sttp/client4/SttpExtensions.scala

@@ -28,7 +28,7 @@ object SttpExtensions {
       onError: ResponseAs[A],
       onSuccess: WebSocketResponseAs[F, B]
   ): WebSocketResponseAs[F, Either[A, B]] =
-    fromMetadata(
+    ws.async.fromMetadata(
       onError.map(Left(_)),
       ConditionalResponseAs(_.code == StatusCode.Ok, onSuccess.map(Right(_)))
     ).showAs(s"either(${onError.show}, ${onSuccess.show})")

core/src/main/scalajvm/sttp/client4/SttpExtensions.scala

@@ -39,8 +39,10 @@ object SttpExtensions {
       onError: ResponseAs[A],
       onSuccess: WebSocketResponseAs[F, B]
   ): WebSocketResponseAs[F, Either[A, B]] =
-    fromMetadata(
-      onError.map(Left(_)),
-      ConditionalResponseAs(_.code == StatusCode.SwitchingProtocols, onSuccess.map(Right(_)))
-    ).showAs(s"either(${onError.show}, ${onSuccess.show})")
+    ws.async
+      .fromMetadata(
+        onError.map(Left(_)),
+        ConditionalResponseAs(_.code == StatusCode.SwitchingProtocols, onSuccess.map(Right(_)))
+      )
+      .showAs(s"either(${onError.show}, ${onSuccess.show})")
 }

core/src/main/scalanative/sttp/client4/SttpExtensions.scala

@@ -39,7 +39,7 @@ object SttpExtensions {
       onError: ResponseAs[A],
       onSuccess: WebSocketResponseAs[F, B]
   ): WebSocketResponseAs[F, Either[A, B]] =
-    fromMetadata(
+    sttp.client4.ws.async.fromMetadata(
       onError.map(Left(_)),
       ConditionalResponseAs(_.code == StatusCode.SwitchingProtocols, onSuccess.map(Right(_)))
     ).showAs(s"either(${onError.show}, ${onSuccess.show})")

core/src/test/scala/sttp/client4/testing/BackendStubTests.scala

@@ -7,6 +7,7 @@ import sttp.client4.SttpClientException.ReadException
 import sttp.client4._
 import sttp.client4.internal._
 import sttp.client4.monad.IdMonad
+import sttp.client4.ws.async._
 import sttp.model._
 import sttp.monad.{FutureMonad, TryMonad}
 import sttp.ws.WebSocketFrame

core/src/test/scala/sttp/client4/testing/websocket/WebSocketBufferOverflowTest.scala

@@ -6,6 +6,7 @@ import org.scalatest.Suite
 import org.scalatest.flatspec.AsyncFlatSpecLike
 import sttp.client4.testing.HttpTest.wsEndpoint
 import sttp.client4._
+import sttp.client4.ws.async._
 import sttp.monad.MonadError
 import sttp.client4.testing.ConvertToFuture