From fc483b1140e893403e82ed2d163fbc502c64578a Mon Sep 17 00:00:00 2001 From: kciesielski Date: Wed, 20 Mar 2024 08:38:46 +0100 Subject: [PATCH] Initial implementation --- .../examples/WebSocketsNettyCatsServer.scala | 75 ++++++++ .../tapir/perf/netty/cats/NettyCats.scala | 26 ++- .../server/netty/cats/NettyCatsServer.scala | 5 +- .../cats/NettyCatsServerInterpreter.scala | 5 +- .../cats/internal/Fs2StreamCompatible.scala | 27 +-- .../cats/internal/NettyCatsRequestBody.scala | 4 +- .../internal/WebSocketPipeProcessor.scala | 94 ++++++++++ .../netty/cats/NettyCatsServerTest.scala | 6 +- .../cats/NettyCatsTestServerInterpreter.scala | 5 +- .../server/netty/loom/NettyIdServer.scala | 2 + .../sttp/tapir/server/netty/NettyConfig.scala | 13 +- .../server/netty/NettyFutureServer.scala | 2 + .../server/netty/NettyResponseContent.scala | 12 ++ .../netty/internal/NettyBootstrap.scala | 9 +- .../netty/internal/NettyServerHandler.scala | 11 ++ .../internal/NettyToStreamsResponseBody.scala | 32 +++- .../internal/ReactiveWebSocketHandler.scala | 161 ++++++++++++++++++ .../netty/internal/StreamCompatible.scala | 7 +- .../internal/WebSocketAutoPingHandler.scala | 45 +++++ .../WebSocketControlFrameHandler.scala | 33 ++++ .../internal/WebSocketFrameConverters.scala | 32 ++++ .../server/netty/zio/NettyZioServer.scala | 3 +- .../zio/internal/ZioStreamCompatible.scala | 11 +- .../server/tests/ServerWebSocketTests.scala | 6 +- 24 files changed, 582 insertions(+), 44 deletions(-) create mode 100644 examples/src/main/scala/sttp/tapir/examples/WebSocketsNettyCatsServer.scala create mode 100644 server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/WebSocketPipeProcessor.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/ReactiveWebSocketHandler.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketAutoPingHandler.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketControlFrameHandler.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketFrameConverters.scala diff --git a/examples/src/main/scala/sttp/tapir/examples/WebSocketsNettyCatsServer.scala b/examples/src/main/scala/sttp/tapir/examples/WebSocketsNettyCatsServer.scala new file mode 100644 index 0000000000..50729f0c2e --- /dev/null +++ b/examples/src/main/scala/sttp/tapir/examples/WebSocketsNettyCatsServer.scala @@ -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 + } +} diff --git a/perf-tests/src/main/scala/sttp/tapir/perf/netty/cats/NettyCats.scala b/perf-tests/src/main/scala/sttp/tapir/perf/netty/cats/NettyCats.scala index 85d3ac7c2b..10639eb7a9 100644 --- a/perf-tests/src/main/scala/sttp/tapir/perf/netty/cats/NettyCats.scala +++ b/perf-tests/src/main/scala/sttp/tapir/perf/netty/cats/NettyCats.scala @@ -3,16 +3,36 @@ package sttp.tapir.perf.netty.cats import cats.effect.IO import cats.effect.kernel.Resource import cats.effect.std.Dispatcher +import fs2.Stream +import sttp.tapir.{CodecFormat, webSocketBody} import sttp.tapir.perf.Common._ import sttp.tapir.perf.apis._ import sttp.tapir.server.ServerEndpoint import sttp.tapir.server.netty.cats.NettyCatsServer import sttp.tapir.server.netty.cats.NettyCatsServerOptions +import sttp.ws.WebSocketFrame +import sttp.capabilities.fs2.Fs2Streams -object Tapir extends Endpoints +import scala.concurrent.duration._ -object NettyCats { +object Tapir extends Endpoints { + val wsResponseStream = Stream.fixedRate[IO](WebSocketSingleResponseLag, dampen = false) + val wsEndpoint = wsBaseEndpoint + .out( + webSocketBody[Long, CodecFormat.TextPlain, Long, CodecFormat.TextPlain](Fs2Streams[IO]) + .concatenateFragmentedFrames(false) + .autoPongOnPing(false) + .ignorePong(true) + .autoPing(None) + ) +} +object NettyCats { + val wsServerEndpoint = Tapir.wsEndpoint.serverLogicSuccess(_ => + IO.pure { (in: Stream[IO, Long]) => + Tapir.wsResponseStream.evalMap(_ => IO.realTime.map(_.toMillis)).concurrently(in.as(())) + } + ) def runServer(endpoints: List[ServerEndpoint[Any, IO]], withServerLog: Boolean = false): IO[ServerRunner.KillSwitch] = { val declaredPort = Port val declaredHost = "0.0.0.0" @@ -25,7 +45,7 @@ object NettyCats { server .port(declaredPort) .host(declaredHost) - .addEndpoints(endpoints) + .addEndpoints(wsServerEndpoint :: endpoints) .start() )(binding => binding.stop()) } yield ()).allocated.map(_._2) diff --git a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServer.scala b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServer.scala index 1b7a449326..abe2b64be4 100644 --- a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServer.scala +++ b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServer.scala @@ -23,12 +23,14 @@ import java.util.UUID import java.util.concurrent.atomic.AtomicBoolean import scala.concurrent.Future import scala.concurrent.duration._ +import sttp.capabilities.WebSockets +import sttp.tapir.server.netty.internal.ReactiveWebSocketHandler case class NettyCatsServer[F[_]: Async](routes: Vector[Route[F]], options: NettyCatsServerOptions[F], config: NettyConfig) { def addEndpoint(se: ServerEndpoint[Fs2Streams[F], F]): NettyCatsServer[F] = addEndpoints(List(se)) def addEndpoint(se: ServerEndpoint[Fs2Streams[F], F], overrideOptions: NettyCatsServerOptions[F]): NettyCatsServer[F] = addEndpoints(List(se), overrideOptions) - def addEndpoints(ses: List[ServerEndpoint[Fs2Streams[F], F]]): NettyCatsServer[F] = addRoute( + def addEndpoints(ses: List[ServerEndpoint[Fs2Streams[F] with WebSockets, F]]): NettyCatsServer[F] = addRoute( NettyCatsServerInterpreter(options).toRoute(ses) ) def addEndpoints(ses: List[ServerEndpoint[Fs2Streams[F], F]], overrideOptions: NettyCatsServerOptions[F]): NettyCatsServer[F] = addRoute( @@ -75,6 +77,7 @@ case class NettyCatsServer[F[_]: Async](routes: Vector[Route[F]], options: Netty NettyBootstrap( config, new NettyServerHandler(route, unsafeRunAsync, channelGroup, isShuttingDown, config.serverHeader), + new ReactiveWebSocketHandler(route, channelGroup, unsafeRunAsync, config.sslContext.isDefined), eventLoopGroup, socketOverride ) diff --git a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala index 9e1c4722c0..8f65e39a73 100644 --- a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala +++ b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala @@ -14,12 +14,13 @@ import sttp.tapir.server.interpreter.{BodyListener, FilterServerEndpoints, Serve import sttp.tapir.server.netty.internal.{NettyBodyListener, RunAsync, _} import sttp.tapir.server.netty.cats.internal.NettyCatsRequestBody import sttp.tapir.server.netty.{NettyResponse, NettyServerRequest, Route} +import sttp.capabilities.WebSockets trait NettyCatsServerInterpreter[F[_]] { implicit def async: Async[F] def nettyServerOptions: NettyCatsServerOptions[F] - def toRoute(ses: List[ServerEndpoint[Fs2Streams[F], F]]): Route[F] = { + def toRoute(ses: List[ServerEndpoint[Fs2Streams[F] with WebSockets, F]]): Route[F] = { implicit val monad: MonadError[F] = new CatsMonadError[F] val runAsync = new RunAsync[F] { @@ -31,7 +32,7 @@ trait NettyCatsServerInterpreter[F[_]] { val createFile = nettyServerOptions.createFile val deleteFile = nettyServerOptions.deleteFile - val serverInterpreter = new ServerInterpreter[Fs2Streams[F], F, NettyResponse, Fs2Streams[F]]( + val serverInterpreter = new ServerInterpreter[Fs2Streams[F] with WebSockets, F, NettyResponse, Fs2Streams[F]]( FilterServerEndpoints(ses), new NettyCatsRequestBody(createFile, Fs2StreamCompatible[F](nettyServerOptions.dispatcher)), new NettyToStreamsResponseBody(Fs2StreamCompatible[F](nettyServerOptions.dispatcher)), diff --git a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/Fs2StreamCompatible.scala b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/Fs2StreamCompatible.scala index 3849fb4a19..28167f4422 100644 --- a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/Fs2StreamCompatible.scala +++ b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/Fs2StreamCompatible.scala @@ -1,22 +1,19 @@ package sttp.tapir.server.netty.cats.internal +import cats.effect.kernel.{Async, Sync} +import cats.effect.std.Dispatcher +import fs2.interop.reactivestreams.{StreamSubscriber, StreamUnicastPublisher} +import fs2.io.file.{Files, Flags, Path} +import fs2.{Chunk, Pipe} import io.netty.buffer.Unpooled +import io.netty.handler.codec.http.websocketx.WebSocketFrame import io.netty.handler.codec.http.{DefaultHttpContent, HttpContent} -import org.reactivestreams.Publisher -import sttp.tapir.FileRange +import org.reactivestreams.{Processor, Publisher} +import sttp.capabilities.fs2.Fs2Streams import sttp.tapir.server.netty.internal._ +import sttp.tapir.{FileRange, WebSocketBodyOutput} import java.io.InputStream -import cats.effect.std.Dispatcher -import sttp.capabilities.fs2.Fs2Streams -import fs2.io.file.Path -import fs2.io.file.Files -import cats.effect.kernel.Async -import fs2.io.file.Flags -import fs2.interop.reactivestreams.StreamUnicastPublisher -import cats.effect.kernel.Sync -import fs2.Chunk -import fs2.interop.reactivestreams.StreamSubscriber object Fs2StreamCompatible { @@ -68,6 +65,12 @@ object Fs2StreamCompatible { override def emptyStream: streams.BinaryStream = fs2.Stream.empty + override def asWsProcessor[REQ, RESP]( + pipe: Pipe[F, REQ, RESP], + o: WebSocketBodyOutput[Pipe[F, REQ, RESP], REQ, RESP, ?, Fs2Streams[F]] + ): Processor[WebSocketFrame, WebSocketFrame] = + new WebSocketPipeProcessor[F, REQ, RESP](pipe, dispatcher, o) + private def inputStreamToFs2(inputStream: () => InputStream, chunkSize: Int) = fs2.io.readInputStream( Sync[F].blocking(inputStream()), diff --git a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/NettyCatsRequestBody.scala b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/NettyCatsRequestBody.scala index e81f66fd00..f801d29ec2 100644 --- a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/NettyCatsRequestBody.scala +++ b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/NettyCatsRequestBody.scala @@ -12,6 +12,7 @@ import sttp.tapir.TapirFile import sttp.tapir.integ.cats.effect.CatsMonadError import sttp.tapir.model.ServerRequest import sttp.tapir.server.netty.internal.{NettyStreamingRequestBody, StreamCompatible} +import sttp.capabilities.WebSockets private[cats] class NettyCatsRequestBody[F[_]: Async]( val createFile: ServerRequest => F[TapirFile], @@ -24,7 +25,8 @@ private[cats] class NettyCatsRequestBody[F[_]: Async]( streamCompatible.fromPublisher(publisher, maxBytes).compile.to(Chunk).map(_.toArray[Byte]) override def writeToFile(serverRequest: ServerRequest, file: TapirFile, maxBytes: Option[Long]): F[Unit] = - (toStream(serverRequest, maxBytes).asInstanceOf[streamCompatible.streams.BinaryStream]) + (toStream(serverRequest, maxBytes) + .asInstanceOf[streamCompatible.streams.BinaryStream]) .through( Files[F](Files.forAsync[F]).writeAll(Path.fromNioPath(file.toPath)) ) diff --git a/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/WebSocketPipeProcessor.scala b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/WebSocketPipeProcessor.scala new file mode 100644 index 0000000000..518db27e65 --- /dev/null +++ b/server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/WebSocketPipeProcessor.scala @@ -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 +} diff --git a/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsServerTest.scala b/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsServerTest.scala index e7a1280719..e3252a25a1 100644 --- a/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsServerTest.scala +++ b/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsServerTest.scala @@ -40,7 +40,11 @@ class NettyCatsServerTest extends TestSuite with EitherValues { new ServerStreamingTests(createServerTest).tests(Fs2Streams[IO])(drainFs2) ++ new ServerCancellationTests(createServerTest)(m, IO.asyncForIO).tests() ++ new NettyFs2StreamingCancellationTest(createServerTest).tests() ++ - new ServerGracefulShutdownTests(createServerTest, ioSleeper).tests() + new ServerGracefulShutdownTests(createServerTest, ioSleeper).tests() ++ + new ServerWebSocketTests(createServerTest, Fs2Streams[IO]) { + override def functionToPipe[A, B](f: A => B): streams.Pipe[A, B] = in => in.map(f) + override def emptyPipe[A, B]: fs2.Pipe[IO, A, B] = _ => fs2.Stream.empty + }.tests() IO.pure((tests, eventLoopGroup)) } { case (_, eventLoopGroup) => diff --git a/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsTestServerInterpreter.scala b/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsTestServerInterpreter.scala index 61be7e6f4d..9334504a2f 100644 --- a/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsTestServerInterpreter.scala +++ b/server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsTestServerInterpreter.scala @@ -4,6 +4,7 @@ import cats.data.NonEmptyList import cats.effect.std.Dispatcher import cats.effect.{IO, Resource} import io.netty.channel.nio.NioEventLoopGroup +import sttp.capabilities.WebSockets import sttp.tapir.server.ServerEndpoint import sttp.tapir.server.netty.{NettyConfig, Route} import sttp.tapir.server.tests.TestServerInterpreter @@ -12,8 +13,8 @@ import sttp.capabilities.fs2.Fs2Streams import scala.concurrent.duration.FiniteDuration class NettyCatsTestServerInterpreter(eventLoopGroup: NioEventLoopGroup, dispatcher: Dispatcher[IO]) - extends TestServerInterpreter[IO, Fs2Streams[IO], NettyCatsServerOptions[IO], Route[IO]] { - override def route(es: List[ServerEndpoint[Fs2Streams[IO], IO]], interceptors: Interceptors): Route[IO] = { + extends TestServerInterpreter[IO, Fs2Streams[IO] with WebSockets, NettyCatsServerOptions[IO], Route[IO]] { + override def route(es: List[ServerEndpoint[Fs2Streams[IO] with WebSockets, IO]], interceptors: Interceptors): Route[IO] = { val serverOptions: NettyCatsServerOptions[IO] = interceptors( NettyCatsServerOptions.customiseInterceptors[IO](dispatcher) ).options diff --git a/server/netty-server/loom/src/main/scala/sttp/tapir/server/netty/loom/NettyIdServer.scala b/server/netty-server/loom/src/main/scala/sttp/tapir/server/netty/loom/NettyIdServer.scala index 46918fc8fd..2de073f763 100644 --- a/server/netty-server/loom/src/main/scala/sttp/tapir/server/netty/loom/NettyIdServer.scala +++ b/server/netty-server/loom/src/main/scala/sttp/tapir/server/netty/loom/NettyIdServer.scala @@ -25,6 +25,7 @@ import scala.concurrent.Future import scala.concurrent.Promise import scala.concurrent.duration.FiniteDuration import scala.util.control.NonFatal +import sttp.tapir.server.netty.internal.ReactiveWebSocketHandler case class NettyIdServer(routes: Vector[IdRoute], options: NettyIdServerOptions, config: NettyConfig) { private val executor = Executors.newVirtualThreadPerTaskExecutor() @@ -100,6 +101,7 @@ case class NettyIdServer(routes: Vector[IdRoute], options: NettyIdServerOptions, isShuttingDown, config.serverHeader ), + new ReactiveWebSocketHandler(route, channelGroup, unsafeRunF, config.sslContext.isDefined), eventLoopGroup, socketOverride ) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala index 07e43954c5..1d6d66144d 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala @@ -61,7 +61,7 @@ case class NettyConfig( sslContext: Option[SslContext], eventLoopConfig: EventLoopConfig, socketConfig: NettySocketConfig, - initPipeline: NettyConfig => (ChannelPipeline, ChannelHandler) => Unit, + initPipeline: NettyConfig => (ChannelPipeline, ChannelHandler, ChannelHandler) => Unit, gracefulShutdownTimeout: Option[FiniteDuration], serverHeader: Option[String] ) { @@ -96,7 +96,7 @@ case class NettyConfig( def eventLoopConfig(elc: EventLoopConfig): NettyConfig = copy(eventLoopConfig = elc) def eventLoopGroup(elg: EventLoopGroup): NettyConfig = copy(eventLoopConfig = EventLoopConfig.useExisting(elg)) - def initPipeline(f: NettyConfig => (ChannelPipeline, ChannelHandler) => Unit): NettyConfig = copy(initPipeline = f) + def initPipeline(f: NettyConfig => (ChannelPipeline, ChannelHandler, ChannelHandler) => Unit): NettyConfig = copy(initPipeline = f) def withGracefulShutdownTimeout(t: FiniteDuration) = copy(gracefulShutdownTimeout = Some(t)) def noGracefulShutdown = copy(gracefulShutdownTimeout = None) @@ -120,14 +120,15 @@ object NettyConfig { sslContext = None, eventLoopConfig = EventLoopConfig.auto, socketConfig = NettySocketConfig.default, - initPipeline = cfg => defaultInitPipeline(cfg)(_, _), + initPipeline = cfg => defaultInitPipeline(cfg)(_, _, _), serverHeader = Some(s"tapir/${buildinfo.BuildInfo.version}") ) - def defaultInitPipeline(cfg: NettyConfig)(pipeline: ChannelPipeline, handler: ChannelHandler): Unit = { + def defaultInitPipeline(cfg: NettyConfig)(pipeline: ChannelPipeline, handler: ChannelHandler, wsHandler: ChannelHandler): Unit = { cfg.sslContext.foreach(s => pipeline.addLast(s.newHandler(pipeline.channel().alloc()))) - pipeline.addLast(new HttpServerCodec()) - pipeline.addLast(new HttpStreamsServerHandler()) + pipeline.addLast("serverCodecHandler", new HttpServerCodec()) + pipeline.addLast("streamsHandler", new HttpStreamsServerHandler()) + pipeline.addLast("wsHandler", wsHandler) pipeline.addLast(handler) if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler()) () diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyFutureServer.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyFutureServer.scala index f510e8e85b..2c972c3226 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyFutureServer.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyFutureServer.scala @@ -16,6 +16,7 @@ import java.util.UUID import java.util.concurrent.atomic.AtomicBoolean import scala.concurrent.duration.FiniteDuration import scala.concurrent.{ExecutionContext, Future, blocking} +import sttp.tapir.server.netty.internal.ReactiveWebSocketHandler case class NettyFutureServer(routes: Vector[FutureRoute], options: NettyFutureServerOptions, config: NettyConfig)(implicit ec: ExecutionContext @@ -72,6 +73,7 @@ case class NettyFutureServer(routes: Vector[FutureRoute], options: NettyFutureSe NettyBootstrap( config, new NettyServerHandler(route, unsafeRunAsync, channelGroup, isShuttingDown, config.serverHeader), + new ReactiveWebSocketHandler(route, channelGroup, unsafeRunAsync, config.sslContext.isDefined), eventLoopGroup, socketOverride ) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyResponseContent.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyResponseContent.scala index b17335f60b..d3aea1dda6 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyResponseContent.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyResponseContent.scala @@ -5,6 +5,10 @@ import io.netty.channel.ChannelPromise import io.netty.handler.codec.http.HttpContent import io.netty.handler.stream.{ChunkedFile, ChunkedStream} import org.reactivestreams.Publisher +import org.reactivestreams.Processor +import io.netty.handler.codec.http.websocketx.WebSocketFrame +import sttp.ws.{WebSocketFrame => SttpWebSocketFrame} +import scala.concurrent.duration.FiniteDuration sealed trait NettyResponseContent { def channelPromise: ChannelPromise @@ -17,4 +21,12 @@ object NettyResponseContent { final case class ChunkedFileNettyResponseContent(channelPromise: ChannelPromise, chunkedFile: ChunkedFile) extends NettyResponseContent final case class ReactivePublisherNettyResponseContent(channelPromise: ChannelPromise, publisher: Publisher[HttpContent]) extends NettyResponseContent + final case class ReactiveWebSocketProcessorNettyResponseContent( + channelPromise: ChannelPromise, + processor: Processor[WebSocketFrame, WebSocketFrame], + ignorePong: Boolean, + autoPongOnPing: Boolean, + decodeCloseRequests: Boolean, + autoPing: Option[(FiniteDuration, SttpWebSocketFrame.Ping)] + ) extends NettyResponseContent } diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyBootstrap.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyBootstrap.scala index 71d57519a8..68835c8f07 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyBootstrap.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyBootstrap.scala @@ -12,6 +12,7 @@ object NettyBootstrap { def apply[F[_]]( nettyConfig: NettyConfig, handler: => NettyServerHandler[F], + wsHandler: => ReactiveWebSocketHandler[F], eventLoopGroup: EventLoopGroup, overrideSocketAddress: Option[SocketAddress] ): ChannelFuture = { @@ -27,8 +28,12 @@ object NettyBootstrap { nettyConfig.requestTimeout match { case Some(requestTimeout) => - nettyConfigBuilder(ch.pipeline().addLast(new ReadTimeoutHandler(requestTimeout.toSeconds.toInt)), handler) - case None => nettyConfigBuilder(ch.pipeline(), handler) + nettyConfigBuilder( + ch.pipeline().addLast("readTimeoutHandler", new ReadTimeoutHandler(requestTimeout.toSeconds.toInt)), + handler, + wsHandler + ) + case None => nettyConfigBuilder(ch.pipeline(), handler, wsHandler) } connectionCounterOpt.map(counter => { diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala index c3333a7ff4..3da92d8f56 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala @@ -4,6 +4,7 @@ import io.netty.buffer.{ByteBuf, Unpooled} import io.netty.channel._ import io.netty.channel.group.ChannelGroup import io.netty.handler.codec.http._ +import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory import io.netty.handler.stream.{ChunkedFile, ChunkedStream} import org.playframework.netty.http.{DefaultStreamedHttpResponse, StreamedHttpRequest} import org.reactivestreams.Publisher @@ -25,6 +26,8 @@ import scala.collection.mutable.{Queue => MutableQueue} import scala.concurrent.{ExecutionContext, Future} import scala.util.control.NonFatal import scala.util.{Failure, Success} +import sttp.tapir.EndpointInput.AuthType.Http +import sttp.tapir.server.netty.NettyResponseContent.ReactiveWebSocketProcessorNettyResponseContent /** @param unsafeRunAsync * Function which dispatches given effect to run asynchronously, returning its result as a Future, and function of type `() => @@ -211,6 +214,12 @@ class NettyServerHandler[F[_]]( }) ctx.writeAndFlush(res, channelPromise).closeIfNeeded(req) + }, + wsHandler = (channelPromise) => { + logger.error("Unexpected WebSocket processor response received in NettyServerHandler, it should be handled only in the ReactiveWebSocketHandler") + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR) + res.headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) + val _ = ctx.writeAndFlush(res) }, noBodyHandler = () => { val res = new DefaultFullHttpResponse( @@ -234,6 +243,7 @@ class NettyServerHandler[F[_]]( chunkedStreamHandler: (ChannelPromise, ChunkedStream) => Unit, chunkedFileHandler: (ChannelPromise, ChunkedFile) => Unit, reactiveStreamHandler: (ChannelPromise, Publisher[HttpContent]) => Unit, + wsHandler: ChannelPromise => Unit, noBodyHandler: () => Unit ): Unit = { r.body match { @@ -245,6 +255,7 @@ class NettyServerHandler[F[_]]( case r: ChunkedStreamNettyResponseContent => chunkedStreamHandler(r.channelPromise, r.chunkedStream) case r: ChunkedFileNettyResponseContent => chunkedFileHandler(r.channelPromise, r.chunkedFile) case r: ReactivePublisherNettyResponseContent => reactiveStreamHandler(r.channelPromise, r.publisher) + case r: ReactiveWebSocketProcessorNettyResponseContent => wsHandler(r.channelPromise) } } case None => noBodyHandler() diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyToStreamsResponseBody.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyToStreamsResponseBody.scala index 0f335a7b14..ae94a180b3 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyToStreamsResponseBody.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyToStreamsResponseBody.scala @@ -12,12 +12,14 @@ import sttp.tapir.{CodecFormat, RawBodyType, WebSocketBodyOutput} import java.nio.ByteBuffer import java.nio.charset.Charset +import sttp.tapir.server.netty.NettyResponseContent.ReactiveWebSocketProcessorNettyResponseContent -/** Common logic for producing response body in all Netty backends that support streaming. These backends use streaming libraries - * like fs2 or zio-streams to obtain reactive Publishers representing responses like InputStreamBody, InputStreamRangeBody or FileBody. - * Other kinds of raw responses like directly available String, ByteArray or ByteBuffer can be returned without wrapping into a Publisher. +/** Common logic for producing response body in all Netty backends that support streaming. These backends use streaming libraries like fs2 + * or zio-streams to obtain reactive Publishers representing responses like InputStreamBody, InputStreamRangeBody or FileBody. Other kinds + * of raw responses like directly available String, ByteArray or ByteBuffer can be returned without wrapping into a Publisher. */ -private[netty] class NettyToStreamsResponseBody[S <: Streams[S]](streamCompatible: StreamCompatible[S]) extends ToResponseBody[NettyResponse, S] { +private[netty] class NettyToStreamsResponseBody[S <: Streams[S]](streamCompatible: StreamCompatible[S]) + extends ToResponseBody[NettyResponse, S] { override val streams: S = streamCompatible.streams @@ -37,7 +39,10 @@ private[netty] class NettyToStreamsResponseBody[S <: Streams[S]](streamCompatibl case RawBodyType.InputStreamBody => (ctx: ChannelHandlerContext) => - new ReactivePublisherNettyResponseContent(ctx.newPromise(), streamCompatible.publisherFromInputStream(() => v, DefaultChunkSize, length = None)) + new ReactivePublisherNettyResponseContent( + ctx.newPromise(), + streamCompatible.publisherFromInputStream(() => v, DefaultChunkSize, length = None) + ) case RawBodyType.InputStreamRangeBody => (ctx: ChannelHandlerContext) => @@ -47,7 +52,8 @@ private[netty] class NettyToStreamsResponseBody[S <: Streams[S]](streamCompatibl ) case RawBodyType.FileBody => - (ctx: ChannelHandlerContext) => new ReactivePublisherNettyResponseContent(ctx.newPromise(), streamCompatible.publisherFromFile(v, DefaultChunkSize)) + (ctx: ChannelHandlerContext) => + new ReactivePublisherNettyResponseContent(ctx.newPromise(), streamCompatible.publisherFromFile(v, DefaultChunkSize)) case _: RawBodyType.MultipartBody => throw new UnsupportedOperationException } @@ -68,5 +74,17 @@ private[netty] class NettyToStreamsResponseBody[S <: Streams[S]](streamCompatibl override def fromWebSocketPipe[REQ, RESP]( pipe: streams.Pipe[REQ, RESP], o: WebSocketBodyOutput[streams.Pipe[REQ, RESP], REQ, RESP, _, S] - ): NettyResponse = throw new UnsupportedOperationException + ): NettyResponse = (ctx: ChannelHandlerContext) => { + new ReactiveWebSocketProcessorNettyResponseContent( + ctx.newPromise(), + streamCompatible.asWsProcessor( + pipe.asInstanceOf[streamCompatible.streams.Pipe[REQ, RESP]], + o.asInstanceOf[WebSocketBodyOutput[streamCompatible.streams.Pipe[REQ, RESP], REQ, RESP, _, S]] + ), + ignorePong = o.ignorePong, + autoPongOnPing = o.autoPongOnPing, + decodeCloseRequests = o.decodeCloseRequests, + autoPing = o.autoPing + ) + } } diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/ReactiveWebSocketHandler.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/ReactiveWebSocketHandler.scala new file mode 100644 index 0000000000..7fc695ab78 --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/ReactiveWebSocketHandler.scala @@ -0,0 +1,161 @@ +package sttp.tapir.server.netty.internal + +import io.netty.channel.group.ChannelGroup +import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} +import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory +import io.netty.handler.codec.http._ +import io.netty.util.ReferenceCountUtil +import org.playframework.netty.http.DefaultWebSocketHttpResponse +import org.slf4j.LoggerFactory +import sttp.monad.MonadError +import sttp.monad.syntax._ +import sttp.tapir.server.model.ServerResponse +import sttp.tapir.server.netty.NettyResponseContent.ReactiveWebSocketProcessorNettyResponseContent +import sttp.tapir.server.netty.{NettyResponse, NettyServerRequest, Route} + +import scala.concurrent.{ExecutionContext, Future} +import scala.util.control.NonFatal +import scala.util.{Failure, Success} + +/** Handles a WS handshake and initiates the communication by calling Tapir interpreter to get a Pipe, then sends that Pipe to the rest of + * the processing pipeline and removes itself from the pipeline. + */ +class ReactiveWebSocketHandler[F[_]]( + route: Route[F], + channelGroup: ChannelGroup, + unsafeRunAsync: (() => F[ServerResponse[NettyResponse]]) => (Future[ServerResponse[NettyResponse]], () => Future[Unit]), + isSsl: Boolean +)(implicit m: MonadError[F]) + extends ChannelInboundHandlerAdapter { + + // By using the Netty event loop assigned to this channel we get two benefits: + // 1. We can avoid the necessary hopping around of threads since Netty pipelines will + // only pass events up and down from within the event loop to which it is assigned. + // That means calls to ctx.read(), and ctx.write(..), would have to be trampolined otherwise. + // 2. We get serialization of execution: the EventLoop is a serial execution queue so + // we can rest easy knowing that no two events will be executed in parallel. + private[this] var eventLoopContext: ExecutionContext = _ + + private val logger = LoggerFactory.getLogger(getClass.getName) + + def isWsHandshake(req: HttpRequest): Boolean = + "Websocket".equalsIgnoreCase(req.headers().get(HttpHeaderNames.UPGRADE)) && + "Upgrade".equalsIgnoreCase(req.headers().get(HttpHeaderNames.CONNECTION)) + + override def handlerAdded(ctx: ChannelHandlerContext): Unit = + if (ctx.channel.isActive) { + initHandler(ctx) + } + override def channelActive(ctx: ChannelHandlerContext): Unit = { + channelGroup.add(ctx.channel) + initHandler(ctx) + } + + private[this] def initHandler(ctx: ChannelHandlerContext): Unit = { + if (eventLoopContext == null) + eventLoopContext = ExecutionContext.fromExecutor(ctx.channel.eventLoop) + } + + override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable): Unit = { + logger.error("Error while processing the request", cause) + } + + override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { + def writeError500(req: HttpRequest, reason: Throwable): Unit = { + logger.error("Error while processing the request", reason) + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR) + res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0) + val _ = ctx.writeAndFlush(res) + } + msg match { + case req: FullHttpRequest if isWsHandshake(req) => + ctx.pipeline().remove(this) + ctx.pipeline().remove("readTimeoutHandler") + ReferenceCountUtil.release(msg) + val (runningFuture, _) = unsafeRunAsync { () => + route(NettyServerRequest(req.retain())) + .map { + case Some(response) => response + case None => ServerResponse.notFound + } + } + + val _ = runningFuture.transform { + case Success(serverResponse) => + try { + serverResponse.body match { + case Some(function) => { + val content = function(ctx) + content match { + case r: ReactiveWebSocketProcessorNettyResponseContent => { + ctx + .pipeline() + .addAfter( + "serverCodecHandler", + "wsControlFrameHandler", + new NettyControlFrameHandler( + ignorePong = r.ignorePong, + autoPongOnPing = r.autoPongOnPing, + decodeCloseRequests = r.decodeCloseRequests + ) + ) + r.autoPing.foreach { case (interval, pingMsg) => + ctx.pipeline().addFirst("wsAutoPingHandler", new WebSocketAutoPingHandler(interval, pingMsg)) + } + r.channelPromise.setSuccess() + val _ = ctx.writeAndFlush( + // Push a special message down the pipeline, it will be handled by HttpStreamsServerHandler + // and from now on that handler will take control of the flow (our NettyServerHandler will not receive messages) + new DefaultWebSocketHttpResponse( + req.protocolVersion(), + HttpResponseStatus.valueOf(200), + r.processor, // the Processor (Pipe) created by Tapir interpreter will be used by HttpStreamsServerHandler + new WebSocketServerHandshakerFactory(wsUrl(req), null, false) + ) + ) + } + case otherContent => + logger.error(s"Unexpected response content: $otherContent") + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR) + res.headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) + otherContent.channelPromise.setFailure(new IllegalStateException("Unexpected response content")) + val _ = ctx.writeAndFlush(res) + } + } + case None => + logger.error("Missing response body, expected WebSocketProcessorNettyResponseContent") + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR) + res.headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) + val _ = ctx.writeAndFlush(res) + } + Success(()) + } catch { + case NonFatal(ex) => + writeError500(req, ex) + Failure(ex) + } finally { + val _ = req.release() + } + case Failure(NonFatal(ex)) => + try { + writeError500(req, ex) + Failure(ex) + } finally { + val _ = req.release() + } + case Failure(fatalException) => Failure(fatalException) + }(eventLoopContext) + + case other => + // not a WS handshake, from now on process messages as normal HTTP requests in this channel + ctx.pipeline.remove(this) + val _ = ctx.fireChannelRead(other) + } + } + + // Only ancient WS protocol versions will use this in the response header. + private def wsUrl(req: FullHttpRequest): String = { + val scheme = if (isSsl) "wss" else "ws" + s"$scheme://${req.headers().get(HttpHeaderNames.HOST)}${req.uri()}" + } +} diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/StreamCompatible.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/StreamCompatible.scala index 6d5da177bd..b973aef7eb 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/StreamCompatible.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/StreamCompatible.scala @@ -1,9 +1,10 @@ package sttp.tapir.server.netty.internal import io.netty.handler.codec.http.HttpContent -import org.reactivestreams.Publisher +import io.netty.handler.codec.http.websocketx.WebSocketFrame +import org.reactivestreams.{Processor, Publisher} import sttp.capabilities.Streams -import sttp.tapir.FileRange +import sttp.tapir.{FileRange, WebSocketBodyOutput} import java.io.InputStream @@ -27,4 +28,6 @@ private[netty] trait StreamCompatible[S <: Streams[S]] { def publisherFromInputStream(is: () => InputStream, chunkSize: Int, length: Option[Long]): Publisher[HttpContent] = asPublisher(fromInputStream(is, chunkSize, length)) + + def asWsProcessor[REQ, RESP](pipe: streams.Pipe[REQ, RESP], o: WebSocketBodyOutput[streams.Pipe[REQ, RESP], REQ, RESP, _, S]): Processor[WebSocketFrame, WebSocketFrame] } diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketAutoPingHandler.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketAutoPingHandler.scala new file mode 100644 index 0000000000..62addee363 --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketAutoPingHandler.scala @@ -0,0 +1,45 @@ +package sttp.tapir.server.netty.internal + +import io.netty.buffer.Unpooled +import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame +import org.slf4j.LoggerFactory + +import java.util.concurrent.{ScheduledFuture, TimeUnit} +import scala.concurrent.duration.FiniteDuration + +/** If auto ping is enabled for an endpoint, this handler will be plugged into the pipeline. Its responsibility is to manage start and stop + * of the ping scheduler. + * TODO: should we include logic for closing the channel and reporting an error if some kind of ping timeout is exceeded? + * @param pingInterval + * time interval to be used between sending pings to the client. + * @param frame + * desired content of the Ping frame, as specified in the Tapir endpoint output. + */ +class WebSocketAutoPingHandler(pingInterval: FiniteDuration, frame: sttp.ws.WebSocketFrame.Ping) extends ChannelInboundHandlerAdapter { + val nettyFrame = new PingWebSocketFrame(Unpooled.copiedBuffer(frame.payload)) + private var pingTask: ScheduledFuture[_] = _ + + private val logger = LoggerFactory.getLogger(getClass.getName) + + override def handlerAdded(ctx: ChannelHandlerContext): Unit = { + if (ctx.channel.isActive) { + logger.debug(s"STARTING WebSocket Ping scheduler for channel ${ctx.channel}, interval = $pingInterval") + val sendPing: Runnable = new Runnable { + override def run(): Unit = { + logger.trace(s"Sending PING WebSocket frame for channel ${ctx.channel}") + val _ = ctx.writeAndFlush(nettyFrame.retain()) + } + } + pingTask = ctx.channel().eventLoop().scheduleAtFixedRate(sendPing, pingInterval.toMillis, pingInterval.toMillis, TimeUnit.MILLISECONDS) + } + } + + override def channelInactive(ctx: ChannelHandlerContext): Unit = { + super.channelInactive(ctx) + logger.debug(s"STOPPING WebSocket Ping scheduler for channel ${ctx.channel}") + if (pingTask != null) { + val _ = pingTask.cancel(false) + } + } +} diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketControlFrameHandler.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketControlFrameHandler.scala new file mode 100644 index 0000000000..098e76bcfe --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketControlFrameHandler.scala @@ -0,0 +1,33 @@ +package sttp.tapir.server.netty.internal + +import io.netty.channel.ChannelInboundHandlerAdapter +import io.netty.channel.ChannelHandlerContext +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame + +/** + * Handles Ping, Pong, and Close frames for WebSockets. + */ +class NettyControlFrameHandler(ignorePong: Boolean, autoPongOnPing: Boolean, decodeCloseRequests: Boolean) + extends ChannelInboundHandlerAdapter { + + override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { + msg match { + case ping: PingWebSocketFrame if autoPongOnPing => + val _ = ctx.writeAndFlush(new PongWebSocketFrame(ping.content().retain())) + case pong: PongWebSocketFrame if !ignorePong => + val _ = ctx.fireChannelRead(pong) + case close: CloseWebSocketFrame => + if (decodeCloseRequests) { + // Passing the Close frame for further processing + val _ = ctx.fireChannelRead(close) + } else { + // Responding with Close immediately + val _ = ctx.writeAndFlush(close) + } + case other => + val _ = ctx.fireChannelRead(other) + } + } +} diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketFrameConverters.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketFrameConverters.scala new file mode 100644 index 0000000000..2d5bc06615 --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/WebSocketFrameConverters.scala @@ -0,0 +1,32 @@ +package sttp.tapir.server.netty.internal + +import io.netty.handler.codec.http.websocketx._ +import io.netty.handler.codec.http.websocketx.{WebSocketFrame => NettyWebSocketFrame} +import sttp.ws.WebSocketFrame +import io.netty.buffer.Unpooled + +object WebSocketFrameConverters { + + def nettyFrameToFrame(nettyFrame: NettyWebSocketFrame): WebSocketFrame = { + nettyFrame match { + case text: TextWebSocketFrame => WebSocketFrame.Text(text.text, text.isFinalFragment, Some(text.rsv)) + case close: CloseWebSocketFrame => WebSocketFrame.Close(close.statusCode, close.reasonText) + case ping: PingWebSocketFrame => WebSocketFrame.Ping(ping.content().nioBuffer().array()) + case pong: PongWebSocketFrame => WebSocketFrame.Pong(pong.content().nioBuffer().array()) + case _ => WebSocketFrame.Binary(nettyFrame.content().nioBuffer().array(), nettyFrame.isFinalFragment, Some(nettyFrame.rsv)) + } + } + + def frameToNettyFrame(w: WebSocketFrame): NettyWebSocketFrame = w match { + case WebSocketFrame.Text(payload, finalFragment, rsvOpt) => + new TextWebSocketFrame(finalFragment, rsvOpt.getOrElse(0), payload) + case WebSocketFrame.Close(statusCode, reasonText) => + new CloseWebSocketFrame(statusCode, reasonText) + case WebSocketFrame.Ping(payload) => + new PingWebSocketFrame(Unpooled.wrappedBuffer(payload)) + case WebSocketFrame.Pong(payload) => + new PongWebSocketFrame(Unpooled.wrappedBuffer(payload)) + case WebSocketFrame.Binary(payload, finalFragment, rsvOpt) => + new BinaryWebSocketFrame(finalFragment, rsvOpt.getOrElse(0), Unpooled.wrappedBuffer(payload)) + } +} diff --git a/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/NettyZioServer.scala b/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/NettyZioServer.scala index 8eb00c55bc..da8c9082d3 100644 --- a/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/NettyZioServer.scala +++ b/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/NettyZioServer.scala @@ -7,7 +7,7 @@ import io.netty.util.concurrent.DefaultEventExecutor import sttp.capabilities.zio.ZioStreams import sttp.tapir.server.ServerEndpoint import sttp.tapir.server.model.ServerResponse -import sttp.tapir.server.netty.internal.{NettyBootstrap, NettyServerHandler} +import sttp.tapir.server.netty.internal.{NettyBootstrap, NettyServerHandler, ReactiveWebSocketHandler} import sttp.tapir.server.netty.zio.internal.ZioUtil.{nettyChannelFutureToScala, nettyFutureToScala} import sttp.tapir.server.netty.{NettyConfig, NettyResponse, Route} import sttp.tapir.ztapir.{RIOMonadError, ZServerEndpoint} @@ -93,6 +93,7 @@ case class NettyZioServer[R](routes: Vector[RIO[R, Route[RIO[R, *]]]], options: isShuttingDown, config.serverHeader ), + new ReactiveWebSocketHandler[RIO[R, *]](route, channelGroup, unsafeRunAsync(runtime), config.sslContext.isDefined), eventLoopGroup, socketOverride ) diff --git a/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/internal/ZioStreamCompatible.scala b/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/internal/ZioStreamCompatible.scala index 60a5a2b067..14ad954784 100644 --- a/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/internal/ZioStreamCompatible.scala +++ b/server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/internal/ZioStreamCompatible.scala @@ -4,11 +4,12 @@ import _root_.zio._ import _root_.zio.interop.reactivestreams._ import _root_.zio.stream.{Stream, ZStream} import io.netty.buffer.Unpooled +import io.netty.handler.codec.http.websocketx.WebSocketFrame import io.netty.handler.codec.http.{DefaultHttpContent, HttpContent} -import org.reactivestreams.Publisher +import org.reactivestreams.{Processor, Publisher} import sttp.capabilities.zio.ZioStreams -import sttp.tapir.FileRange import sttp.tapir.server.netty.internal._ +import sttp.tapir.{FileRange, WebSocketBodyOutput} import java.io.InputStream @@ -64,6 +65,12 @@ private[zio] object ZioStreamCompatible { override def emptyStream: streams.BinaryStream = ZStream.empty + + override def asWsProcessor[REQ, RESP]( + pipe: Stream[Throwable, REQ] => Stream[Throwable, RESP], + o: WebSocketBodyOutput[Stream[Throwable, REQ] => Stream[Throwable, RESP], REQ, RESP, ?, ZioStreams] + ): Processor[WebSocketFrame, WebSocketFrame] = + throw new UnsupportedOperationException("TODO") } } } diff --git a/server/tests/src/main/scala/sttp/tapir/server/tests/ServerWebSocketTests.scala b/server/tests/src/main/scala/sttp/tapir/server/tests/ServerWebSocketTests.scala index 378fd86ddf..05cfaad8bb 100644 --- a/server/tests/src/main/scala/sttp/tapir/server/tests/ServerWebSocketTests.scala +++ b/server/tests/src/main/scala/sttp/tapir/server/tests/ServerWebSocketTests.scala @@ -43,11 +43,13 @@ abstract class ServerWebSocketTests[F[_], S <: Streams[S], OPTIONS, ROUTE]( _ <- ws.sendText("test2") m1 <- ws.receiveText() m2 <- ws.receiveText() - } yield List(m1, m2) + _ <- ws.close() + m3 <- ws.eitherClose(ws.receiveText()) + } yield List(m1, m2, m3) }) .get(baseUri.scheme("ws")) .send(backend) - .map(_.body shouldBe Right(List("echo: test1", "echo: test2"))) + .map(_.body shouldBe Right(List("echo: test1", "echo: test2", Left(WebSocketFrame.Close(1000, "normal closure"))))) }, { val reqCounter = newRequestCounter[F]