Http4sWebSocketServer.scala

import cats.effect._
import fs2._
import org.http4s._
import org.http4s.blaze.server.BlazeServerBuilder
import org.http4s.dsl.Http4sDsl
import org.http4s.implicits._
import org.http4s.server.websocket.WebSocketBuilder2
import org.http4s.websocket.WebSocketFrame

import scala.concurrent.duration._

object Http4sWebSocketServer extends IOApp.Simple {

  private val connectorPoolSize = Math.max(2, Runtime.getRuntime.availableProcessors() / 8)
  private val responseStream = Stream.repeatEval(IO.realTime.map(ts => WebSocketFrame.Text(s"${ts.toMillis}")).delayBy(500.millis))

  override protected def computeWorkerThreadCount: Int =
    Math.max(2, super.computeWorkerThreadCount / 2)

  def service(wsb: WebSocketBuilder2[IO]): HttpApp[IO] = {
    val dsl = new Http4sDsl[IO] {}
    import dsl._

    val receive: Pipe[IO, WebSocketFrame, Unit] = (_: Stream[IO, WebSocketFrame]) => Stream.eval(IO.unit)

    HttpRoutes
      .of[IO] {
        case GET -> Root / "ts" => wsb.build(responseStream, receive)
      }
      .orNotFound
  }

  private val blaze = BlazeServerBuilder[IO]
    .bindHttp(8888, "0.0.0.0")
    .withMaxConnections(65536)
    .withConnectorPoolSize(connectorPoolSize)
    .withHttpWebSocketApp(wsb => service(wsb))
    .resource
    .useForever

  def run: IO[Unit] = blaze
}