-
Notifications
You must be signed in to change notification settings - Fork 422
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
00cada7
commit fc483b1
Showing
24 changed files
with
582 additions
and
44 deletions.
There are no files selected for viewing
75 changes: 75 additions & 0 deletions
75
examples/src/main/scala/sttp/tapir/examples/WebSocketsNettyCatsServer.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package sttp.tapir.examples | ||
|
|
||
| import cats.effect.{IO, IOApp} | ||
| import sttp.client3._ | ||
| import sttp.model.StatusCode | ||
| import sttp.tapir.server.netty.cats.NettyCatsServer | ||
| import sttp.tapir.* | ||
| import scala.concurrent.duration._ | ||
| import sttp.capabilities.fs2.Fs2Streams | ||
| import sttp.ws.WebSocket | ||
| import sttp.client3.pekkohttp.PekkoHttpBackend | ||
| import scala.concurrent.Future | ||
|
|
||
| object WebSocketsNettyCatsServer extends IOApp.Simple { | ||
| // One endpoint on GET /hello with query parameter `name` | ||
| val helloWorldEndpoint: PublicEndpoint[String, Unit, String, Any] = | ||
| endpoint.get.in("hello").in(query[String]("name")).out(stringBody) | ||
|
|
||
| val wsEndpoint = | ||
| endpoint.get.in("ws").out(webSocketBody[String, CodecFormat.TextPlain, String, CodecFormat.TextPlain](Fs2Streams[IO])) | ||
|
|
||
| val wsServerEndpoint = wsEndpoint.serverLogicSuccess(_ => | ||
| IO.pure(in => in.evalMap(str => IO.println(s"responding with ${str.toUpperCase}") >> IO.pure(str.toUpperCase()))) | ||
| ) | ||
| // Just returning passed name with `Hello, ` prepended | ||
| val helloWorldServerEndpoint = helloWorldEndpoint | ||
| .serverLogic(name => IO.pure[Either[Unit, String]](Right(s"Hello, $name!"))) | ||
|
|
||
| private val declaredPort = 9090 | ||
| private val declaredHost = "localhost" | ||
|
|
||
| // Creating handler for netty bootstrap | ||
| override def run = NettyCatsServer | ||
| .io() | ||
| .use { server => | ||
| for { | ||
| binding <- server | ||
| .port(declaredPort) | ||
| .host(declaredHost) | ||
| .addEndpoints(List(wsServerEndpoint, helloWorldServerEndpoint)) | ||
| .start() | ||
| result <- IO | ||
| .fromFuture(IO.delay { | ||
| val port = binding.port | ||
| val host = binding.hostName | ||
| println(s"Server started at port = ${binding.port}") | ||
| import scala.concurrent.ExecutionContext.Implicits.global | ||
| def useWebSocket(ws: WebSocket[Future]): Future[Unit] = { | ||
| def send(i: Int) = ws.sendText(s"Hello $i!") | ||
| def receive() = ws.receiveText().map(t => println(s"Client RECEIVED: $t")) | ||
| for { | ||
| _ <- send(1) | ||
| _ <- receive() | ||
| _ <- send(2) | ||
| _ <- send(3) | ||
| _ <- receive() | ||
| } yield () | ||
| } | ||
| val backend = PekkoHttpBackend() | ||
|
|
||
| val url = uri"ws://$host:$port/ws" | ||
| val allGood = uri"http://$host:$port/hello?name=Netty" | ||
| basicRequest.response(asStringAlways).get(allGood).send(backend).map(r => println(r.body)) | ||
| .flatMap { _ => | ||
| basicRequest | ||
| .response(asWebSocket(useWebSocket)) | ||
| .get(url) | ||
| .send(backend) | ||
| } | ||
| .andThen { case _ => backend.close() } | ||
| }) | ||
| .guarantee(binding.stop()) | ||
| } yield result | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
94 changes: 94 additions & 0 deletions
94
...er/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/WebSocketPipeProcessor.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package sttp.tapir.server.netty.cats.internal | ||
|
|
||
| import cats.Applicative | ||
| import cats.effect.kernel.Async | ||
| import cats.effect.std.Dispatcher | ||
| import fs2.interop.reactivestreams.{StreamSubscriber, StreamUnicastPublisher} | ||
| import fs2.{Pipe, Stream} | ||
| import io.netty.handler.codec.http.websocketx.{WebSocketFrame => NettyWebSocketFrame} | ||
| import org.reactivestreams.{Processor, Publisher, Subscriber, Subscription} | ||
| import sttp.capabilities.fs2.Fs2Streams | ||
| import sttp.tapir.model.WebSocketFrameDecodeFailure | ||
| import sttp.tapir.server.netty.internal.WebSocketFrameConverters._ | ||
| import sttp.tapir.{DecodeResult, WebSocketBodyOutput} | ||
| import sttp.ws.WebSocketFrame | ||
|
|
||
| import scala.concurrent.ExecutionContext.Implicits | ||
| import scala.concurrent.Promise | ||
| import scala.util.{Failure, Success} | ||
|
|
||
| class WebSocketPipeProcessor[F[_]: Async, REQ, RESP]( | ||
| pipe: Pipe[F, REQ, RESP], | ||
| dispatcher: Dispatcher[F], | ||
| o: WebSocketBodyOutput[Pipe[F, REQ, RESP], REQ, RESP, ?, Fs2Streams[F]] | ||
| ) extends Processor[NettyWebSocketFrame, NettyWebSocketFrame] { | ||
| private var subscriber: StreamSubscriber[F, NettyWebSocketFrame] = _ | ||
| private val publisher: Promise[Publisher[NettyWebSocketFrame]] = Promise[Publisher[NettyWebSocketFrame]]() | ||
| private var subscription: Subscription = _ | ||
|
|
||
| override def onSubscribe(s: Subscription): Unit = { | ||
| subscriber = dispatcher.unsafeRunSync( | ||
| StreamSubscriber[F, NettyWebSocketFrame](bufferSize = 1) | ||
| ) | ||
| subscription = s | ||
| val in: Stream[F, NettyWebSocketFrame] = subscriber.sub.stream(Applicative[F].unit) | ||
| val sttpFrames = in.map { f => | ||
| val sttpFrame = nettyFrameToFrame(f) | ||
| f.release() | ||
| sttpFrame | ||
| } | ||
| val stream: Stream[F, NettyWebSocketFrame] = | ||
| optionallyConcatenateFrames(sttpFrames, o.concatenateFragmentedFrames) | ||
| .map(f => | ||
| o.requests.decode(f) match { | ||
| case x: DecodeResult.Value[REQ] => x.v | ||
| case failure: DecodeResult.Failure => throw new WebSocketFrameDecodeFailure(f, failure) | ||
| } | ||
| ) | ||
| .through(pipe) | ||
| .map(r => frameToNettyFrame(o.responses.encode(r))) | ||
| .append(fs2.Stream(frameToNettyFrame(WebSocketFrame.close))) | ||
|
|
||
| subscriber.sub.onSubscribe(s) | ||
| publisher.success(StreamUnicastPublisher(stream, dispatcher)) | ||
| } | ||
|
|
||
| override def onNext(t: NettyWebSocketFrame): Unit = { | ||
| subscriber.sub.onNext(t) | ||
| } | ||
|
|
||
| override def onError(t: Throwable): Unit = { | ||
| subscriber.sub.onError(t) | ||
| } | ||
|
|
||
| override def onComplete(): Unit = { | ||
| subscriber.sub.onComplete() | ||
| } | ||
|
|
||
| override def subscribe(s: Subscriber[_ >: NettyWebSocketFrame]): Unit = { | ||
| publisher.future.onComplete { | ||
| case Success(p) => | ||
| p.subscribe(s) | ||
| case Failure(ex) => | ||
| subscriber.sub.onError(ex) | ||
| subscription.cancel | ||
| }(Implicits.global) | ||
| } | ||
|
|
||
| private def optionallyConcatenateFrames(s: Stream[F, WebSocketFrame], doConcatenate: Boolean): Stream[F, WebSocketFrame] = | ||
| if (doConcatenate) { | ||
| type Accumulator = Option[Either[Array[Byte], String]] | ||
|
|
||
| s.mapAccumulate(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 (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.") | ||
| }.collect { case (_, Some(f)) => f } | ||
| } else s | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.