-
Notifications
You must be signed in to change notification settings - Fork 411
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Test client socket improvements and tests (#1753)
* Various cleanup. Add HttpSocket alias. * Work on more user-facing/SocketApp oriented Channel * More removals and notes * More API auditing. Warn about long-running socket apps in TestClient. * Start on Server failure contract app * Complete Server failure test for Live Client * Get rid of SocketApp Channel experimentation I need to grok the backpressure that Tushar discussed more before taking another run at this. * Fix broken refs to experiments * Create/use general contract. Temporarily quiet Netty exception output. * Cleanup * Restructure contract function * Conver other test to new contract function * Warn about unexpected events sent to server in TestClient * TestClient test for broken Client app * Cleanup for PR * More cleanup * Restore noisy Netty output * rm HTTPSocket type alias
- Loading branch information
Showing
6 changed files
with
166 additions
and
110 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
199 changes: 112 additions & 87 deletions
199
zio-http-testkit/src/test/scala/zio/http/SocketContractSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,114 +1,139 @@ | ||
| package zio.http | ||
|
|
||
| import zio.Console.printLine | ||
| import zio._ | ||
| import zio.http.ChannelEvent.{ChannelRead, ChannelUnregistered, UserEvent, UserEventTriggered} | ||
| import zio.http.ServerConfig.LeakDetectionLevel | ||
| import zio.http.model.Status | ||
| import zio.http.netty.server.NettyDriver | ||
| import zio.http.socket._ | ||
| import zio.test._ | ||
|
|
||
| object SocketContractSpec extends ZIOSpecDefault { | ||
| val testServerConfig: ZLayer[Any, Nothing, ServerConfig] = | ||
| ZLayer.succeed(ServerConfig.default.port(0).leakDetection(LeakDetectionLevel.PARANOID)) | ||
|
|
||
| val severTestLayer = testServerConfig >+> Server.live | ||
|
|
||
| val messageFilter: Http[Any, Nothing, WebSocketChannelEvent, (Channel[WebSocketFrame], String)] = | ||
| private val messageFilter: Http[Any, Nothing, WebSocketChannelEvent, (Channel[WebSocketFrame], String)] = | ||
| Http.collect[WebSocketChannelEvent] { case ChannelEvent(channel, ChannelRead(WebSocketFrame.Text(message))) => | ||
| (channel, message) | ||
| } | ||
|
|
||
| val messageSocketServer: Http[Any, Throwable, WebSocketChannelEvent, Unit] = messageFilter >>> | ||
| Http.collectZIO[(WebSocketChannel, String)] { | ||
| case (ch, text) if text.contains("Hi Server") => | ||
| ZIO.debug("Server got message: " + text) *> ch.close() | ||
| case (_, text) => // TODO remove? | ||
| ZIO.debug("Unrecognized message sent to server: " + text) | ||
| } | ||
|
|
||
| def channelSocketServer(p: Promise[Throwable, Unit]): Http[Any, Throwable, WebSocketChannelEvent, Unit] = | ||
| Http.collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(ch, UserEventTriggered(UserEvent.HandshakeComplete)) => | ||
| ch.writeAndFlush(WebSocketFrame.text("Hi Client")) | ||
|
|
||
| case ChannelEvent(_, ChannelRead(WebSocketFrame.Close(status, reason))) => | ||
| p.succeed(()) *> | ||
| Console.printLine("Closing channel with status: " + status + " and reason: " + reason) | ||
| case ChannelEvent(_, ChannelUnregistered) => | ||
| p.succeed(()) *> | ||
| Console.printLine("Server Channel unregistered") | ||
| case ChannelEvent(ch, ChannelRead(WebSocketFrame.Text("Hi Server"))) => | ||
| ch.write(WebSocketFrame.text("Hi Client")) | ||
|
|
||
| case ChannelEvent(_, other) => | ||
| Console.printLine("Server Other: " + other) | ||
| } | ||
|
|
||
| val protocol = SocketProtocol.default.withSubProtocol(Some("json")) | ||
|
|
||
| val decoder = SocketDecoder.default.withExtensions(allowed = true) | ||
|
|
||
| def socketAppServer(p: Promise[Throwable, Unit]): SocketApp[Any] = | ||
| (messageSocketServer ++ channelSocketServer(p)).toSocketApp | ||
| .withDecoder(decoder) | ||
| .withProtocol(protocol) | ||
| private val warnOnUnrecognizedEvent = Http.collectZIO[WebSocketChannelEvent] { case other => | ||
| ZIO.fail(new Exception("Unexpected event: " + other)) | ||
| } | ||
|
|
||
| sys.props.put("ZIOHttpLogLevel", "DEBUG") | ||
| def spec = | ||
| suite("SocketOps")( | ||
| contract( | ||
| "Live", | ||
| ZIO.serviceWithZIO[Server](server => | ||
| for { | ||
| p <- Promise.make[Throwable, Unit] | ||
| _ <- server.install(socketAppServer(p).toHttp) | ||
|
|
||
| } yield (server.port, p), | ||
| ), | ||
| ).provide(Client.default, Scope.default, TestServer.layer, NettyDriver.default, ServerConfig.liveOnOpenPort), | ||
| contract( | ||
| "Test", { | ||
| for { | ||
| p <- Promise.make[Throwable, Unit] | ||
| _ <- TestClient.addSocketApp(socketAppServer(p)) | ||
|
|
||
| } yield (0, p) | ||
| contract("Successful Multi-message application") { p => | ||
| def channelSocketServer: Http[Any, Throwable, WebSocketChannelEvent, Unit] = | ||
| Http | ||
| .collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(ch, UserEventTriggered(UserEvent.HandshakeComplete)) => | ||
| ch.writeAndFlush(WebSocketFrame.text("Hi Client")) | ||
| case ChannelEvent(_, ChannelUnregistered) => | ||
| p.succeed(()) *> | ||
| printLine("Server Channel unregistered") | ||
| case ChannelEvent(ch, ChannelRead(WebSocketFrame.Text("Hi Server"))) => | ||
| ch.close() | ||
| case ChannelEvent(_, other) => | ||
| printLine("Server Unexpected: " + other) | ||
| } | ||
| .defaultWith(warnOnUnrecognizedEvent) | ||
|
|
||
| val messageSocketServer: Http[Any, Throwable, WebSocketChannelEvent, Unit] = messageFilter >>> | ||
| Http.collectZIO[(WebSocketChannel, String)] { | ||
| case (ch, text) if text.contains("Hi Server") => | ||
| printLine("Server got message: " + text) *> ch.close() | ||
| } | ||
|
|
||
| messageSocketServer | ||
| .defaultWith(channelSocketServer) | ||
| } { _ => | ||
| val messageSocketClient: Http[Any, Throwable, WebSocketChannelEvent, Unit] = messageFilter >>> | ||
| Http.collectZIO[(WebSocketChannel, String)] { | ||
| case (ch, text) if text.contains("Hi Client") => | ||
| ch.writeAndFlush(WebSocketFrame.text("Hi Server"), await = true) | ||
| } | ||
|
|
||
| val channelSocketClient: Http[Any, Throwable, WebSocketChannelEvent, Unit] = | ||
| Http.collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(_, ChannelUnregistered) => | ||
| printLine("Client Channel unregistered") | ||
|
|
||
| case ChannelEvent(_, other) => | ||
| printLine("Client received Unexpected event: " + other) | ||
| } | ||
|
|
||
| messageSocketClient.defaultWith(channelSocketClient) | ||
| }, | ||
| contract("Application where server app fails")(_ => | ||
| Http.collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(_, UserEventTriggered(UserEvent.HandshakeComplete)) => | ||
| ZIO.fail(new Exception("Broken server")) | ||
| }, | ||
| ) | ||
| .provide(TestClient.layer, Scope.default), | ||
| ) | ||
|
|
||
| def contract[R](name: String, serverSetup: ZIO[R, Nothing, (Int, Promise[Throwable, Unit])]) = | ||
| test(name) { | ||
| val messageSocketClient: Http[Any, Throwable, WebSocketChannelEvent, Unit] = messageFilter >>> | ||
| Http.collectZIO[(WebSocketChannel, String)] { | ||
| case (ch, text) if text.contains("Hi Client") => | ||
| ch.writeAndFlush(WebSocketFrame.text("Hi Server"), await = true).debug("Client got message: " + text) | ||
| ) { p => | ||
| Http.collectZIO[WebSocketChannelEvent] { case ChannelEvent(ch, ChannelUnregistered) => | ||
| printLine("Server failed and killed socket. Should complete promise.") *> | ||
| p.succeed(()).unit | ||
| } | ||
|
|
||
| val channelSocketClient: Http[Any, Throwable, WebSocketChannelEvent, Unit] = | ||
| }, | ||
| contract("Application where client app fails")(p => | ||
| Http.collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(_, ChannelUnregistered) => | ||
| Console.printLine("Client Channel unregistered") | ||
|
|
||
| case ChannelEvent(_, other) => | ||
| Console.printLine("Client received other event: " + other) | ||
| case ChannelEvent(_, UserEventTriggered(UserEvent.HandshakeComplete)) => ZIO.unit | ||
| case ChannelEvent(_, ChannelUnregistered) => | ||
| printLine("Client failed and killed socket. Should complete promise.") *> | ||
| p.succeed(()).unit | ||
| }, | ||
| ) { _ => | ||
| Http.collectZIO[WebSocketChannelEvent] { | ||
| case ChannelEvent(_, UserEventTriggered(UserEvent.HandshakeComplete)) => | ||
| ZIO.fail(new Exception("Broken client")) | ||
| } | ||
| }, | ||
| ) | ||
|
|
||
| val httpSocketClient: Http[Any, Throwable, WebSocketChannelEvent, Unit] = | ||
| messageSocketClient ++ channelSocketClient | ||
|
|
||
| val socketAppClient: SocketApp[Any] = | ||
| httpSocketClient.toSocketApp | ||
| .withDecoder(decoder) | ||
| .withProtocol(protocol) | ||
| private def contract( | ||
| name: String, | ||
| )( | ||
| serverApp: Promise[Throwable, Unit] => Http[Any, Throwable, WebSocketChannelEvent, Unit], | ||
| )(clientApp: Promise[Throwable, Unit] => Http[Any, Throwable, WebSocketChannelEvent, Unit]) = { | ||
| suite(name)( | ||
| test("Live") { | ||
| for { | ||
| portAndPromise <- liveServerSetup(serverApp) | ||
| (port, promise) = portAndPromise | ||
| response <- ZIO.serviceWithZIO[Client]( | ||
| _.socket(s"ws://localhost:$port/", clientApp(promise).toSocketApp), | ||
| ) | ||
| _ <- promise.await.timeout(10.seconds) | ||
| } yield assertTrue(response.status == Status.SwitchingProtocols) | ||
| }.provide(Client.default, Scope.default, TestServer.layer, NettyDriver.default, ServerConfig.liveOnOpenPort), | ||
| test("Test") { | ||
| for { | ||
| portAndPromise <- testServerSetup(serverApp) | ||
| (port, promise) = portAndPromise | ||
| response <- ZIO.serviceWithZIO[Client]( | ||
| _.socket(s"ws://localhost:$port/", clientApp(promise).toSocketApp), | ||
| ) | ||
| _ <- promise.await.timeout(10.seconds) | ||
| } yield assertTrue(response.status == Status.SwitchingProtocols) | ||
| }.provide(TestClient.layer, Scope.default), | ||
| ) | ||
| } | ||
|
|
||
| private def liveServerSetup( | ||
| serverApp: Promise[Throwable, Unit] => Http[Any, Throwable, WebSocketChannelEvent, Unit], | ||
| ) = | ||
| ZIO.serviceWithZIO[Server](server => | ||
| for { | ||
| portAndPromise <- serverSetup | ||
| response <- ZIO.serviceWithZIO[Client](_.socket(s"ws://localhost:${portAndPromise._1}/", socketAppClient)) | ||
| _ <- portAndPromise._2.await | ||
| } yield assertTrue(response.status == Status.SwitchingProtocols) | ||
| } | ||
| p <- Promise.make[Throwable, Unit] | ||
| _ <- server.install(serverApp(p).toSocketApp.toHttp) | ||
| } yield (server.port, p), | ||
| ) | ||
|
|
||
| private def testServerSetup( | ||
| serverApp: Promise[Throwable, Unit] => Http[Any, Throwable, WebSocketChannelEvent, Unit], | ||
| ) = | ||
| for { | ||
| p <- Promise.make[Throwable, Unit] | ||
| _ <- TestClient.installSocketApp(serverApp(p)) | ||
| } yield (0, p) | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters