Initial implementation of websocket support for ZIO http.

[yabosedira]

This PR is roughly based on this one. It aims to introduce web socket support for zio http.

TODO:

• Update documentation.
• Code cleanup
• Add tests for ping/pong/close frames
• Add tests for fragmented frames Will do this in a follow up PR

[adamw]

Yes, this is definitely the right direction! Though so far, I think this will support only a single incoming message producing multiple outgoing messages?

[yabosedira]

Yes, this is definitely the right direction! Though so far, I think this will support only a single incoming message producing multiple outgoing messages?

Thank Adam for the quick feedback. Yes, that was correct, I have updated the code a bit to handle continuation frames and few other cases.

I have one question, this test is failing locally, and to be honest, it is not entirely clear to me what is the expected behavior, it would be great if you help me understand.

Thanks again, really appreciate the support.

[yabosedira]

Diving a bit in the websocket implementation, I think there is a potential limitation (bug?). AFAIU, it is always the expectation to receive at least on WebSocketFrame from client before invoking the pipe. i.e if we want to setup one way webSockets communication from server to client, we expect at least one message from the client before reacting. Am I missing something ?

This use case explains the issue a bit better.

    testServer(
      endpoint.out(webSocketBody[String, CodecFormat.TextPlain, Option[String], CodecFormat.TextPlain](streams)),
      "Only server send events"
    )((_: Unit) => pureResult(functionToPipe[String, Option[String]] { _ => Some("test") }.asRight[Unit])) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            m1 <- ws.eitherClose(ws.receiveText())
            m2 <- ws.eitherClose(ws.receiveText())
            m3 <- ws.eitherClose(ws.receiveText())
            _ <- ws.close()
          } yield List(m1, m2, m3)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(
          _.body.map(_.map(_.left.map(_.statusCode))) shouldBe Right(
            List(Right("test"), Right("test"), Right("test"), Left(WebSocketFrame.close.statusCode))
          )
        )
    },

[adamw]

As for the test you mentioned, the scenario there is that a client connects, but the server always responds with an empty stream. "End of stream" means in case of web sockets "close the connection", so the only expected communication is a close frame, sent from the server to the client.

[adamw]

For the second question, I think the problem is here:

functionToPipe[String, Option[String]] { _ => Some("test") }

The server-side behavior is to create a pipe, which maps each incoming message to Some("test"). So you'll get a server, which only responds to incoming messages.

To produce some "welcome" messages, I think you should do sth along the lines of:

(in: ZStream[Any, Nothing, String]) => ZStream.fromIterable(List("hello", "world")) ++ in

So for a given stream of incoming messages, you return a stream which emits two values (hello world), and then echoes any of the incoming ones.

[adamw]

Looks great, thank you!