Bugfix: WebSocketConfig.withForwardCloseFrames does not forward Close…

… frames #2375  (#2395)

* fix bug #2375 - WebSocketConfig.withForwardCloseFrames does not forward Close frames

fix typos, clarify descriptor

add WebSocketConfig unit test for forwardCloseFrames

* * scalafmtAll

* * scalafmtAll

* * typo fix

* squash!  * typo fix

 * scalafmtAll

* squash! squash!  * typo fix

 * remove unused layer

* squash! squash! squash!  * typo fix

 * run sFix

zio-http/src/main/scala/zio/http/WebSocketConfig.scala

@@ -33,13 +33,13 @@ final case class WebSocketConfig(
 ) { self =>
 
   /**
-   * Close frame to send, when close frame was not send manually.
+   * Close frame to send, when close frame was not sent manually.
    */
   def closeFrame(code: Int, reason: String): WebSocketConfig =
     self.copy(sendCloseFrame = WebSocketConfig.CloseStatus.Custom(code, reason))
 
   /**
-   * Close frame to send, when close frame was not send manually.
+   * Close frame to send, when close frame was not sent manually.
    */
   def closeStatus(status: WebSocketConfig.CloseStatus): WebSocketConfig = self.copy(sendCloseFrame = status)
 
@@ -53,9 +53,10 @@ final case class WebSocketConfig(
     self.copy(forceCloseTimeoutMillis = duration.toMillis)
 
   /**
-   * Close frames should be forwarded
+   * Close frames should be forwarded instead of handled solely by Netty,
+   * invisibly to the Websocket Client
    */
-  def forwardCloseFrames(forward: Boolean): WebSocketConfig = self.copy(handleCloseFrames = forward)
+  def forwardCloseFrames(forward: Boolean): WebSocketConfig = self.copy(handleCloseFrames = !forward)
 
   /**
    * Pong frames should be forwarded

zio-http/src/test/scala/zio/http/WebSocketConfig.scala

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2021 - 2023 Sporta Technologies PVT LTD & the ZIO HTTP contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zio.http
+
+import zio._
+import zio.test.TestAspect._
+import zio.test.{TestClock, assertCompletes, assertTrue, assertZIO, testClock}
+
+import zio.http.ChannelEvent.{Read, Unregistered, UserEvent, UserEventTriggered}
+import zio.http.internal.{DynamicServer, HttpRunnableSpec, serverTestLayer}
+import zio.http.netty.NettyConfig
+import zio.http.{Client, DnsResolver}
+// import zio.http.netty.NettyConfig
+
+object WebSocketConfigSpec extends HttpRunnableSpec {
+
+  val closeFrame = Read(WebSocketFrame.Close(1000, Some("goodbye")))
+
+  private val webSocketConfigSpec = suite("WebSocketConfigSpec")(
+    test("Close frames are received when WebSocketConfig.forwardCloseFrames is true") {
+      for {
+        msg <- MessageCollector.make[WebSocketChannelEvent]
+        url <- DynamicServer.wsURL
+        id  <- DynamicServer.deploy {
+          Handler.webSocket { channel =>
+            channel.receiveAll {
+              case UserEventTriggered(UserEvent.HandshakeComplete) =>
+                channel.send(closeFrame)
+              case _                                               => ZIO.unit
+            }
+          }.toHttpAppWS
+        }
+
+        res <- ZIO.scoped {
+          Handler.webSocket { channel =>
+            channel.receiveAll {
+              case event @ Read(WebSocketFrame.Close(_, _)) =>
+                msg.add(event, true)
+              case _                                        => ZIO.unit
+            }
+          }.connect(url, Headers(DynamicServer.APP_ID, id)) *> {
+            for {
+              events <- msg.await
+              expected = List(closeFrame)
+            } yield assertTrue(events == expected)
+          }
+        }
+      } yield res
+    },
+  )
+
+  def clientWithCloseFrames =
+    ZLayer.succeed(
+      ZClient.Config.default.webSocketConfig(
+        WebSocketConfig.default
+          .forwardCloseFrames(true),
+      ),
+    ) ++
+      ZLayer.succeed(NettyConfig.default) ++
+      DnsResolver.default >>>
+      Client.live
+
+  override def spec = suite("Server") {
+    ZIO.scoped {
+      serve.as(List(webSocketConfigSpec))
+    }
+  }
+    .provideShared(
+      DynamicServer.live,
+      serverTestLayer,
+      clientWithCloseFrames,
+    ) @@
+    timeout(30 seconds) @@
+    diagnose(30.seconds) @@
+    withLiveClock @@
+    sequential
+
+  final class MessageCollector[A](ref: Ref[List[A]], promise: Promise[Nothing, Unit]) {
+    def add(a: A, isDone: Boolean = false): UIO[Unit] = ref.update(_ :+ a) <* promise.succeed(()).when(isDone)
+    def await: UIO[List[A]]                           = promise.await *> ref.get
+    def done: UIO[Boolean]                            = promise.succeed(())
+  }
+
+  object MessageCollector {
+    def make[A]: ZIO[Any, Nothing, MessageCollector[A]] = for {
+      ref <- Ref.make(List.empty[A])
+      prm <- Promise.make[Nothing, Unit]
+    } yield new MessageCollector(ref, prm)
+  }
+}