小编sme*_*tca的帖子

这个想法是能够对路由器服务上的 websocket 端点进行单元测试。任何其他类型的端点都可以很容易地使用 a 进行测试Request，但我无法找出一种轻松测试 websocket 响应的方法，因为使用 a 击中端点Request会给您一个 501 错误，因为您没有任何相应的套接字握手。

这是我正在使用的路由器

def router(averages: IO[SomeAverages]): HttpRoutes[F] =
    HttpRoutes.of[F] {
        case GET -> Root =>
            val toClient: Stream[F, WebSocketFrame] = {
                Stream
                  .awakeEvery[F](5.seconds)
                  .map(_ => {
                      val avg = averages.unsafeRunSync
                      WebSocketFrame.Text(avg.toString)
                  })
            }
            val fromClient: Pipe[F, WebSocketFrame, Unit] = _.evalMap {
                case WebSocketFrame.Text(t, _) => F.delay(println(t))
                case f => F.delay(println(s"Unknown type: $f"))
            }
            WebSocketBuilder[F].build(toClient, fromClient)
    }

这个想法是让类似于下面的代码的东西与 websockets 一起工作

val server = Server.initServer[IO]
val router = server.router …