Web Sockets for netty-loom

[kciesielski]

Closes #3666

This PR adds support for Web Sockects in the tapir-netty-loom backend, using Ox for efficient and safe concurrency.
From the user's perspective, creating a WS endpoint requires defining a pipe function of type Ox ?=> Source[REQ] => Source[RESP], where Source is ox.channels.Source.
For example:

import ox.*
import ox.channels.*
import sttp.tapir.*
import sttp.tapir.server.netty.loom.Id
import sttp.tapir.server.netty.loom.OxStreams
import sttp.tapir.server.netty.loom.OxStreams.Pipe // alias for Ox ?=> Source[A] => Source[B]
import sttp.tapir.server.netty.loom.NettySyncServer
import sttp.ws.WebSocketFrame

import scala.concurrent.duration.*

object WebSocketsNettySyncServer:
  // Web socket endpoint
  val wsEndpoint =
    endpoint.get
      .in("ws") 
      .out(
        webSocketBody[String, CodecFormat.TextPlain, String, CodecFormat.TextPlain](OxStreams)
          .concatenateFragmentedFrames(false) // All these options are supported by tapir-netty
          .ignorePong(true)
          .autoPongOnPing(true)
          .decodeCloseRequests(false)
          .decodeCloseResponses(false)
          .autoPing(Some((10.seconds, WebSocketFrame.Ping("ping-content".getBytes))))
      )

  // Your processor transforming a stream of requests into a stream of responses
  val wsPipe: Pipe[String, String] = requestStream => requestStream.map(_.toUpperCase)
  // Alternatively, requests and responses can be treated separately, for example to emit frames to the client from another source: 
  val wsPipe2: Pipe[String, String] = { in =>
    fork {
      in.drain() // read and ignore requests
    }
    // emit periodic responses
    Source.tick(1.second).map(_ => System.currentTimeMillis()).map(_.toString)
  }

  // The WebSocket endpoint, builds the pipeline in serverLogicSuccess
  val wsServerEndpoint = wsEndpoint.serverLogicSuccess[Id](_ => wsPipe)

  // A regular /GET endpoint
  val helloWorldEndpoint: PublicEndpoint[String, Unit, String, Any] =
    endpoint.get.in("hello").in(query[String]("name")).out(stringBody)

  val helloWorldServerEndpoint = helloWorldEndpoint
    .serverLogicSuccess(name => s"Hello, $name!")

  def main(args: Array[String]): Unit = 
    NettySyncServer()
      .host("0.0.0.0")
      .port(8080)
      .addEndpoints(List(wsServerEndpoint, helloWorldServerEndpoint))
      .startAndWait()

The implementation has been tested with our performance tests and it's latency is really good comparing to other fast WS-supporting backends. The tests were run with all bells and whistles enabled (frame concatenation, very frequent auto ping - 40ms, auto pong on ping, etc).

[kciesielski]

@adamw Can I get a pre-review? The implementation passes all our WS tests, has great latency in perf tests (a few ms on p99.999 on my machine, comparing to 10-20 for other backends). There's no documentation and little comments, probably some missing error handling? Would be great to get some comments 🙇

[kciesielski]

@adamw PR no longer in draft mode, re-requesting a review :)

[adamw]

Nicely done :)

Btw. I presume the commented out tests pass as well?

[kciesielski]

I checked one of them separately already (failing pipe), and the other tests should be rather related to shared netty features (ping/pong stuff). However, I will merge them today (#3676) because I finally found the one causing flakiness, so this will give a final confirmation.\

Update: test merged, and pass