Fix spinloop bug in TLSEngine

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,6 +219,65 @@ class TLSSocketSuite extends TLSSuite {
         .to(Chunk)
         .assertEquals(msg)
     }
+
+    test("endOfOutput during handshake results in termination".only) {
+      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
+        @deprecated("", "")
+        def isOpen = raw.isOpen
+        @deprecated("", "")
+        def localAddress = raw.localAddress
+        def peerAddress = raw.peerAddress
+        def read(maxBytes: Int) = raw.read(maxBytes)
+        def readN(numBytes: Int) = raw.readN(numBytes)
+        def reads = raw.reads
+        @deprecated("", "")
+        def remoteAddress = raw.remoteAddress
+        def writes = raw.writes
+
+        def address = raw.address
+        def getOption[A](key: SocketOption.Key[A]) = raw.getOption(key)
+        def setOption[A](key: SocketOption.Key[A], value: A) = raw.setOption(key, value)
+        def supportedOptions = raw.supportedOptions
+
+        private var totalWritten: Long = 0
+        def write(bytes: Chunk[Byte]) =
+          if (totalWritten >= limit) endOfOutput
+          else {
+            val b = bytes.take(limit)
+            raw.write(b) >> IO(totalWritten += b.size)
+          }
+      }
+
+      val setup = for {
+        tlsContext <- Resource.eval(testTlsContext)
+        serverSocket <- Network[IO].bind(SocketAddress(ip"127.0.0.1", Port.Wildcard))
+        client <- Network[IO].connect(serverSocket.address).flatMap { rawClient =>
+          tlsContext.clientBuilder(rawClient).withLogger(logger).build
+        }
+      } yield serverSocket.accept
+        .flatMap(s => Stream.resource(tlsContext.server(limitWrites(s, 100)))) -> client
+
+      Stream
+        .resource(setup)
+        .flatMap { case (server, clientSocket) =>
+          val echoServer = server.map { socket =>
+            socket.reads.chunks.foreach(socket.write(_))
+          }.parJoinUnbounded
+
+          val client =
+            Stream.exec(clientSocket.write(msg)).onFinalize(clientSocket.endOfOutput) ++
+              clientSocket.reads.take(msg.size.toLong)
+
+          client.concurrently(echoServer)
+        }
+        .compile
+        .drain
+        .intercept[javax.net.ssl.SSLException]
+    }
   }
 
   group("TLSContextBuilder") {