Merge pull request #3600 from typelevel/topic/backport-3599

Backport #3599

Author: mpilquist

io/jvm/src/main/scala/fs2/io/net/tls/TLSEngine.scala

@@ -215,9 +215,7 @@ private[tls] object TLSEngine {
               else
                 binding.read(engine.getSession.getPacketBufferSize).flatMap {
                   case Some(c) => unwrapBuffer.input(c) >> unwrapHandshake
-                  case None =>
-                    unwrapBuffer.inputRemains
-                      .flatMap(x => if (x > 0) Applicative[F].unit else stopUnwrap)
+                  case None    => stopUnwrap
                 }
             }
           case SSLEngineResult.HandshakeStatus.NEED_UNWRAP_AGAIN =>

io/jvm/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala

@@ -219,5 +219,60 @@ class TLSSocketSuite extends TLSSuite {
         .to(Chunk)
         .assertEquals(msg)
     }
+
+    test("endOfOutput during handshake results in termination") {
+      val msg = Chunk.array(("Hello, world! " * 20000).getBytes)
+
+      def limitWrites(raw: Socket[IO], limit: Int): Socket[IO] = new Socket[IO] {
+        def endOfInput = raw.endOfInput
+        def endOfOutput = raw.endOfOutput
+        def isOpen = raw.isOpen
+        def localAddress = raw.localAddress
+        def read(maxBytes: Int) = raw.read(maxBytes)
+        def readN(numBytes: Int) = raw.readN(numBytes)
+        def reads = raw.reads
+        def remoteAddress = raw.remoteAddress
+        def writes = raw.writes
+
+        private var totalWritten: Int = 0
+        def write(bytes: Chunk[Byte]) =
+          if (totalWritten >= limit) endOfOutput
+          else {
+            val b = bytes.take(limit - totalWritten)
+            raw.write(b) >> IO(totalWritten += b.size) >> IO(totalWritten >= limit)
+              .ifM(endOfOutput, IO.unit)
+          }
+      }
+
+      // Setup an HTTPS echo server & a client that starts a TLS handshake but only sends the first few bytes and then signals no more output
+      // Doing so should not cause the server to peg a CPU
+      val setup = for {
+        tlsContext <- Resource.eval(testTlsContext)
+        addressAndConnections <- Network[IO].serverResource(Some(ip"127.0.0.1"))
+        (serverAddress, server) = addressAndConnections
+        echoServer =
+          server
+            .flatMap(s => Stream.resource(tlsContext.serverBuilder(s).withLogger(logger).build))
+            .map { socket =>
+              socket.reads.chunks.foreach(socket.write)
+            }
+            .parJoinUnbounded
+        client <- Network[IO].client(serverAddress).flatMap { rawClient =>
+          tlsContext.client(limitWrites(rawClient, 10))
+        }
+      } yield echoServer -> client
+
+      Stream
+        .resource(setup)
+        .flatMap { case (echoServer, clientSocket) =>
+          val client =
+            Stream.exec(clientSocket.write(msg)).onFinalize(clientSocket.endOfOutput) ++
+              clientSocket.reads.take(msg.size.toLong)
+
+          client.concurrently(echoServer)
+        }
+        .compile
+        .drain
+    }
   }
 }